Line data Source code
1 : #ifndef HEADER_fd_src_disco_events_fd_circq_h 2 : #define HEADER_fd_src_disco_events_fd_circq_h 3 : 4 : /* The circular buffer is a structure, which stores a queue of messages, 5 : supporting two operations: push_back and pop_front. Unlike a regular 6 : queue, the circular buffer is fixed size and push_back must always 7 : succeed. 8 : 9 : To ensure push_back always succeeds, the circular buffer will evict 10 : old messages if necessary to make room for the new one. 11 : 12 : One more complication is that the circular buffer must store 13 : metadata about the messages in the data buffer itself, as it does not 14 : have a separate metadata region. The structure of the buffer then 15 : looks as follows: 16 : 17 : +-------+-----+------+-----+-------+-----+------+-----+-------+-----+------+ 18 : + meta0 | pad | msg0 | pad | meta1 | pad | msg1 | pad | meta2 | pad | msg2 | 19 : +-------+-----+------+-----+-------+-----+------+-----+-------+-----+------+ 20 : ^ | ^ | ^ | 21 : | +-----next---------next--+ +------------------------+ | 22 : | | 23 : head tail 24 : 25 : Here, the meta elements are fd_circq_message_t, which each point to 26 : the next message in the queue, and head, tail are the head and tail 27 : of the queue respectively. */ 28 : 29 : #include "../../util/fd_util_base.h" 30 : 31 0 : #define FD_CIRCQ_ALIGN (4096UL) 32 : 33 : struct __attribute__((aligned(FD_CIRCQ_ALIGN))) fd_circq_private { 34 : /* Current count of elements in the queue. */ 35 : ulong cnt; 36 : ulong bytes_used; 37 : 38 : /* These are offsets relative to the end of this struct of the 39 : metadata for the first, and last message in the queue, 40 : respectively. */ 41 : ulong head; 42 : ulong tail; 43 : 44 : ulong size; 45 : 46 : ulong cursor; /* Current offset in buffer for iteration, or ULONG_MAX if at end */ 47 : ulong cursor_seq; /* Monotonic counter - cursor value for current position */ 48 : ulong cursor_push_seq; /* Monotonic counter - incremented on each push */ 49 : 50 : struct { 51 : ulong drop_cnt; 52 : } metrics; 53 : 54 : /* padding out to 4k here ... */ 55 : }; 56 : 57 : typedef struct fd_circq_private fd_circq_t; 58 : 59 : FD_PROTOTYPES_BEGIN 60 : 61 : FD_FN_CONST ulong 62 : fd_circq_align( void ); 63 : 64 : FD_FN_CONST ulong 65 : fd_circq_footprint( ulong sz ); 66 : 67 : void * 68 : fd_circq_new( void * shmem, 69 : ulong sz ); 70 : 71 : fd_circq_t * 72 : fd_circq_join( void * shbuf ); 73 : 74 : void * 75 : fd_circq_leave( fd_circq_t * buf ); 76 : 77 : void * 78 : fd_circq_delete( void * shbuf ); 79 : 80 : /* fd_circq_push_back appends a message of size sz into the circular 81 : buffer, evicting any old messages if they would be overwritten when 82 : the buffer wraps around. Returns the address of the memory contents 83 : in the buffer on success, or NULL on failure. The only two reasons 84 : for failure are if the requested sz (along with the message metadata) 85 : exceeds the size of the entire buffer and can't fit, or if the 86 : requested alignment is not a power of 2, or is larger than 4096. */ 87 : 88 : uchar * 89 : fd_circq_push_back( fd_circq_t * circq, 90 : ulong align, 91 : ulong footprint ); 92 : 93 : void 94 : fd_circq_resize_back( fd_circq_t * circq, 95 : ulong new_footprint ); 96 : 97 : /* fd_circq_cursor_advance moves an internal cursor forward to the next 98 : message in the circular buffer, returning the message at the previous 99 : cursor position, or NULL if there are no more messages. 100 : 101 : Moving the cursor does not remove the message from the circular 102 : buffer, which only happens when fd_circq_pop_until is called. */ 103 : 104 : uchar const * 105 : fd_circq_cursor_advance( fd_circq_t * circq, 106 : ulong * msg_sz ); 107 : 108 : /* fd_circq_cursor_peek returns 1 and sets *msg_sz to the size of the 109 : message the next fd_circq_cursor_advance would return, or returns 0 110 : if there is none. Does not move the cursor. */ 111 : 112 : int 113 : fd_circq_cursor_peek( fd_circq_t const * circq, 114 : ulong * msg_sz ); 115 : 116 : /* fd_circq_pop_until removes messages from the front of the circular 117 : buffer up to and including the message with the given cursor value. 118 : Returns 0 on success, or -1 if the given cursor value is invalid 119 : (i.e., larger than highest cursor value returned by cursor_advance). */ 120 : 121 : int 122 : fd_circq_pop_until( fd_circq_t * circq, 123 : ulong cursor ); 124 : 125 : /* fd_circq_reset_cursor resets the internal cursor to the front of the 126 : circular buffer. This is useful if you want to re-process all 127 : messages in the buffer from the start. */ 128 : 129 : void 130 : fd_circq_reset_cursor( fd_circq_t * circq ); 131 : 132 : /* fd_circq_bytes_used returns the bytes held by queued messages, 133 : including their metadata and alignment padding. The unused tail a 134 : wrapping push skips over is not counted. */ 135 : 136 : ulong 137 : fd_circq_bytes_used( fd_circq_t const * circq ); 138 : 139 : /* fd_circq_unsent_cnt returns the number of messages in the queue that 140 : the cursor has not yet advanced past, i.e. messages still waiting to 141 : be sent. The remaining messages (cnt - unsent) have been sent and are 142 : awaiting acknowledgement before they are popped. */ 143 : 144 : ulong 145 : fd_circq_unsent_cnt( fd_circq_t const * circq ); 146 : 147 : FD_PROTOTYPES_END 148 : 149 : #endif /* HEADER_fd_src_disco_events_fd_circq_h */
