Line data Source code
1 : #ifndef HEADER_fd_src_waltz_mib_fd_dbl_buf_h 2 : #define HEADER_fd_src_waltz_mib_fd_dbl_buf_h 3 : 4 : /* fd_dbl_buf.h provides a concurrent lock-free double buffer. A double 5 : buffer contains two buffers that take turns holding a message for 6 : consumers and receiving a new message by a producer. 7 : 8 : Supports a single producer thread and an arbitrary number of consumer 9 : threads. Optimized for rare updates and frequent polling (e.g. config). 10 : Use an fd_tango mcache/dcache pair if you need frequent updates. 11 : 12 : Currently assumes a memory model that preserves store order across 13 : threads (e.g. x86-TSO). Does not use atomics or hardware fences. */ 14 : 15 : #include "../../util/bits/fd_bits.h" 16 : #include "../../util/log/fd_log.h" 17 : #if FD_HAS_SSE 18 : #include <emmintrin.h> 19 : #endif 20 : 21 : /* FIXME COULD ALLOW FOR IN-PLACE READS WITH PODs BY ADDING A MSG ALIGN ARGUMENT */ 22 : 23 : /* fd_dbl_buf_t is the header of a dbl_buf object. May not be locally 24 : declared. */ 25 : 26 : union __attribute__((aligned(16UL))) fd_dbl_buf { 27 : 28 : struct { 29 : ulong magic; /* ==FD_DBL_BUF_MAGIC */ 30 : ulong mtu; 31 : ulong buf0; /* offset to first buffer from beginning of struct */ 32 : ulong buf1; /* — " — second — " — */ 33 : ulong seq; /* latest msg seq no */ 34 : ulong sz; /* latest msg size */ 35 : ulong pad[2]; 36 : /* objects follow here */ 37 : }; 38 : 39 : # if FD_HAS_SSE 40 : struct { 41 : __m128i magic_mtu; 42 : __m128i buf0_buf1; 43 : __m128i seq_sz; 44 : __m128i pad2; 45 : }; 46 : # endif 47 : 48 : }; 49 : 50 : typedef union fd_dbl_buf fd_dbl_buf_t; 51 : 52 0 : #define FD_DBL_BUF_MAGIC (0xa6c6f85d431c03ceUL) /* random */ 53 : 54 0 : #define FD_DBL_BUF_ALIGN (16UL) 55 : #define FD_DBL_BUF_FOOTPRINT(mtu) \ 56 0 : FD_LAYOUT_FINI( FD_LAYOUT_APPEND( FD_LAYOUT_APPEND( FD_LAYOUT_INIT, \ 57 0 : FD_DBL_BUF_ALIGN, sizeof(fd_dbl_buf_t) ), \ 58 0 : FD_DBL_BUF_ALIGN, FD_ULONG_ALIGN_UP( mtu, FD_DBL_BUF_ALIGN )<<1UL ), \ 59 0 : FD_DBL_BUF_ALIGN ) 60 : 61 : FD_PROTOTYPES_BEGIN 62 : 63 : /* fd_dbl_buf_{align,footprint} describe the memory region of a double 64 : buffer. mtu is the largest possible message size. */ 65 : 66 : ulong 67 : fd_dbl_buf_align( void ); 68 : 69 : ulong 70 : fd_dbl_buf_footprint( ulong mtu ); 71 : 72 : /* fd_dbl_buf_new formats a memory region for use as a double buffer. 73 : shmem points to the memory region matching fd_dbl_buf_{align,footprint}. 74 : Initially, the active object of the double buffer will have sequence 75 : number seq0 and zero byte size. */ 76 : 77 : void * 78 : fd_dbl_buf_new( void * shmem, 79 : ulong mtu, 80 : ulong seq0 ); 81 : 82 : fd_dbl_buf_t * 83 : fd_dbl_buf_join( void * shbuf ); 84 : 85 : void * 86 : fd_dbl_buf_leave( fd_dbl_buf_t * buf ); 87 : 88 : /* fd_dbl_buf_delete unformats the memory region backing a dbl_buf and 89 : releases ownership back to the caller. Returns shbuf. */ 90 : 91 : void * 92 : fd_dbl_buf_delete( void * shbuf ); 93 : 94 : /* fd_dbl_buf_obj_mtu returns the max message size a dbl_buf can store. */ 95 : 96 : static inline ulong 97 0 : fd_dbl_buf_obj_mtu( fd_dbl_buf_t * buf ) { 98 0 : return buf->mtu; 99 0 : } 100 : 101 : /* fd_dbl_buf_seq_query peeks the current sequence number. */ 102 : 103 : static inline ulong 104 0 : fd_dbl_buf_seq_query( fd_dbl_buf_t * buf ) { 105 0 : FD_COMPILER_MFENCE(); 106 0 : ulong seq = FD_VOLATILE_CONST( buf->seq ); 107 0 : FD_COMPILER_MFENCE(); 108 0 : return seq; 109 0 : } 110 : 111 : /* fd_dbl_buf_slot returns a pointer to the buffer for the given sequence 112 : number. */ 113 : 114 : FD_FN_PURE static inline void * 115 : fd_dbl_buf_slot( fd_dbl_buf_t * buf, 116 0 : ulong seq ) { 117 0 : return (seq&1) ? ((char *)buf)+buf->buf1 : ((char *)buf)+buf->buf0; 118 0 : } 119 : 120 : /* fd_dbl_buf_insert appends a message to the double buffer. 121 : 122 : Note: It is NOT safe to call this function from multiple threads. */ 123 : 124 : void 125 : fd_dbl_buf_insert( fd_dbl_buf_t * buf, 126 : void const * msg, 127 : ulong sz ); 128 : 129 : /* fd_dbl_buf_try_read does a speculative read the most recent message 130 : (from the caller's POV). The read may be overrun by a writer. out 131 : points to a buffer of fd_dbl_buf_obj_mtu(buf) bytes. opt_seqp points to 132 : a ulong or NULL. 133 : 134 : On success: 135 : - returns the size of the message read 136 : - a copy of the message is stored at out 137 : - *opt_seqp is set to the msg sequence number (if non-NULL) 138 : 139 : On failure (due to overrun): 140 : - returns ULONG_MAX 141 : - out buffer is clobbered 142 : - *opt_seq is clobbered (if non-NULL) */ 143 : 144 : static inline ulong 145 : fd_dbl_buf_try_read( fd_dbl_buf_t * buf, 146 : void * out, 147 : ulong out_sz, 148 0 : ulong * opt_seqp ) { 149 0 : ulong seq = fd_dbl_buf_seq_query( buf ); 150 0 : void * src = fd_dbl_buf_slot( buf, seq ); 151 0 : ulong sz = FD_VOLATILE_CONST( buf->sz ); 152 0 : if( out_sz<sz ) FD_LOG_ERR(( "fd_dbl_buf_try_read failed: output buffer too small: out_sz: %lu, sz: %lu", out_sz, sz )); 153 0 : fd_memcpy( out, src, sz ); 154 0 : if( FD_UNLIKELY( seq!=fd_dbl_buf_seq_query( buf ) ) ) return ULONG_MAX; 155 0 : fd_ulong_store_if( !!opt_seqp, opt_seqp, seq ); 156 0 : return sz; 157 0 : } 158 : 159 : /* fd_dbl_buf_read does a blocking */ 160 : 161 : ulong 162 : fd_dbl_buf_read( fd_dbl_buf_t * buf, 163 : ulong buf_sz, 164 : void * obj, 165 : ulong * opt_seqp ); 166 : 167 : FD_PROTOTYPES_END 168 : 169 : #endif /* HEADER_fd_src_waltz_mib_fd_dbl_buf_h */
