prqueue.h

Go to the documentation of this file.

 
 
#pragma once
 
#include <cx/cx.h>
#include <cx/thread/aspin.h>
#include <cx/thread/atomic.h>
#include <cx/thread/mutex.h>
 
#if DEBUG_LEVEL >= 2 && _64BIT
#define PRQ_PERF_STATS
#endif
CX_C_BEGIN
 
typedef enum PrqGrowthEnum {
    PRQ_Grow_None = 1,   
    PRQ_Grow_25,         
    PRQ_Grow_50,         
    PRQ_Grow_100,        
    PRQ_Grow_150,        
    PRQ_Grow_200,        
} PrqGrowth;
 
typedef struct PrqSegment PrqSegment;
 
#ifdef PRQ_PERF_STATS
typedef struct PrqPerfStats {
    atomic(uint64) grow;
    atomic(uint64) grow_collision;
    atomic(uint64) shrink;
    atomic(uint64) shrink_collision;
    atomic(uint64) head_contention;
    atomic(uint64) reserved_contention;
    atomic(uint64) push;
    atomic(uint64) push_optimal;
    atomic(uint64) push_fast;
    atomic(uint64) push_slow;
    atomic(uint64) push_appeared_full;
    atomic(uint64) push_actually_full;
    atomic(uint64) push_collision;
    atomic(uint64) push_retry;
    atomic(uint64) push_full_retry;
    atomic(uint64) push_noreserve_retiring;
    atomic(uint64) pop;
    atomic(uint64) pop_optimal;
    atomic(uint64) pop_fast;
    atomic(uint64) pop_slow;
    atomic(uint64) pop_nonobvious_empty;
    atomic(uint64) pop_assist;
    atomic(uint64) pop_assist_fail;
    atomic(uint64) pop_segtraverse;
    atomic(uint64) pop_collision;
    atomic(uint64) gc_run;
    atomic(uint64) seg_retired;
    atomic(uint64) seg_dealloc;
    atomic(uint64) seg_dealloc_failinuse;
} PrqPerfStats;
#endif
 
typedef struct PrQueue {
    // Initial size of the queue as well as minimum size.
    uint32 minsz;
 
    // Ideal size of the queue.
    uint32 targetsz;
 
    // Maximum size of the queue.
    uint32 maxsz;
 
    // How much to grow the queue at a time.
    PrqGrowth growth;
 
    // How much to shrink the queue at a time.
    PrqGrowth shrink;
 
    // Concurrence factor. Defaults to the number of logcal CPUs in the system. This value is
    // used to determine how many queue entries there must be before threads start assisting
    // each other to complete operations.
    uint32 concurrence;
 
    // Buffer segment that is currently the target for queue pushes and pops. The head of a linked
    // list of buffer segments when the queue is being grown.
    atomic(ptr) current;
 
    // Linked list of buffer segments that have been retired. These are segments that have been
    // fully emptied and no longer have any valid queue entries, nor can they have any entries
    // pushed into them, but are being held to deallocate later until it can be guaranteed that
    // no threads attempting a pop operation still have a pointer to the segment.
    PrqSegment* retired;
 
    // Access counter. This is used internally to close a small gap between when a thread retrieves
    // the current segment pointers and when it increments the use counter, since it cannot do that
    // atomically while another thread is retiring the segment.
    atomic(int32) access;
 
    // Lower 32 bits of timestamp of the last time a segment was added to grow or shrink the queue,
    // because this needs to be atomic and 64-bit atomics don't exist on all platforms.
    atomic(uint32) chgtime;
 
    // Minimum number of milliseconds the queue must wait to shrink after growing or shrinking
    // (default 500ms).
    uint32 shrinkms;
 
    // Running average to track the total queue size across GC cycles for possible shrinking.
    uint32 avgcount;
    uint32 avgcount_num;
 
    // Only 1 thread may run the garbage collection operation at a time. It's recommended to try to
    // run GC optimistcally when a thread has nothing else to do. For example, a consumer thread
    // that is about to sleep.
    Mutex gcmtx;
 
#ifdef PRQ_PERF_STATS
    // Performance stats for debugging
    PrqPerfStats stats;
#endif
} PrQueue;
 
typedef struct PrqSegment {
    // Next segment in the chain. When the queue grows, it allocates a larger segment which is
    // temporarily chained to from the original segment in order to handle the transition while
    // many other threads may be still using the original.
    atomic(ptr) nextseg;
 
    // Next retired segment in retired chain.
    // NOTE: We cannot reuse nextseg for this. nextseg needs to continue to point to the actual
    // segment that replaced this one, so that any threads which grabbed a pointer to this segment
    // before it was retired can still follow it.
    PrqSegment* nextretired;
 
    // Atomic counter of how many threads are actively using this segment. This, along with the
    // access counter, act as a non-blocking optimistic lock similar to a reader-writer lock but
    // less intrusive. They block garbage collection from deallocating this segment if there is
    // a chance that a thread may still be reading from it (or about to read from it).
    atomic(int32) inuse;
 
    // Total number of queue slots in this buffer.
    uint32 size;
 
    // Number of queue slots in this buffer that are used. This number may be slightly higher
    // than the actual number of slots that has been written to. This is the authoritative
    // source for how much of the queue is used.
    atomic(uint32) count;
 
    // Head of the queue; points at the slot that is first in line to be read. This is cached
    // information for performance optimization only and is not authoritative.
    atomic(uint32) head;
 
    // Number of write reservations on this segment. Only used when the buffer is expandable, to
    // prevent GC from retiring the segment while there are pending write operations. The high
    // bit is used to signal that the segment is transitioning to the retired status and further
    // writes may not be started.
    atomic(uint32) reserved;
 
    // The actual ringbuffer
    atomic(ptr) buffer[];
} PrqSegment;
 
void prqInitFixed(_Out_ PrQueue* prq, uint32 sz);
 
void prqInitDynamic(_Out_ PrQueue* prq, uint32 minsz, uint32 targetsz, uint32 maxsz,
                    PrqGrowth growth, PrqGrowth shrink);
 
_Success_(return) bool prqDestroy(_Pre_valid_ _Post_invalid_ PrQueue* prq);
 
_Success_(return) bool prqPush(_Inout_ PrQueue* prq, _Pre_notnull_ _Post_invalid_ void* ptr);
 
_Must_inspect_result_ _Ret_maybenull_ void* prqPop(_Inout_ PrQueue* prq);
 
bool prqCollect(_Inout_ PrQueue* prq);
 
uint32 prqCount(_In_ PrQueue* prq);
 
void* prqPeek(_In_ PrQueue* prq, uint32 n);
 
CX_C_END
 
// end of thread_prqueue group