Line data    Source code

       1             : #include "fd_circq.h"
       2             : 
       3             : #include "../../util/log/fd_log.h"
       4             : 
       5             : struct __attribute__((aligned(8UL))) fd_circq_message_private {
       6             :   ulong align;
       7             :   ulong footprint;
       8             : 
       9             :   /* Offset withn the circular buffer data region of where the next
      10             :      message starts, if there is one.  This is not always the same as
      11             :      aligning up this message + footprint, because the next message may
      12             :      have wrapped around to the start of the buffer. */
      13             :   ulong next;
      14             : };
      15             : 
      16             : typedef struct fd_circq_message_private fd_circq_message_t;
      17             : 
      18             : FD_FN_CONST ulong
      19           0 : fd_circq_align( void ) {
      20           0 :   return FD_CIRCQ_ALIGN;
      21           0 : }
      22             : 
      23             : FD_FN_CONST ulong
      24           0 : fd_circq_footprint( ulong sz ) {
      25           0 :   return sizeof( fd_circq_t ) + sz;
      26           0 : }
      27             : 
      28             : void *
      29             : fd_circq_new( void * shmem,
      30           0 :               ulong  sz ) {
      31           0 :   fd_circq_t * circq = (fd_circq_t *)shmem;
      32           0 :   circq->cnt  = 0UL;
      33           0 :   circq->bytes_used = 0UL;
      34           0 :   circq->head = 0UL;
      35           0 :   circq->tail = 0UL;
      36           0 :   circq->size = sz;
      37           0 :   circq->cursor = ULONG_MAX;
      38           0 :   circq->cursor_seq = 0UL;
      39           0 :   circq->cursor_push_seq = 0UL;
      40             : 
      41           0 :   memset( &circq->metrics, 0, sizeof( circq->metrics ) );
      42             : 
      43           0 :   return shmem;
      44           0 : }
      45             : 
      46             : fd_circq_t *
      47           0 : fd_circq_join( void * shbuf ) {
      48           0 :   return (fd_circq_t *)shbuf;
      49           0 : }
      50             : 
      51             : void *
      52           0 : fd_circq_leave( fd_circq_t * buf ) {
      53           0 :   return (void *)buf;
      54           0 : }
      55             : 
      56             : void *
      57           0 : fd_circq_delete( void * shbuf ) {
      58           0 :   return shbuf;
      59           0 : }
      60             : 
      61             : static inline void FD_FN_UNUSED
      62           0 : verify( fd_circq_t * circq ) {
      63           0 :   FD_TEST( circq->head<circq->size );
      64           0 :   FD_TEST( circq->tail<circq->size );
      65           0 :   FD_TEST( circq->tail!=circq->head || circq->cnt<=1 );
      66           0 :   if( !circq->cnt ) {
      67           0 :     FD_TEST( circq->head==0UL );
      68           0 :     FD_TEST( circq->tail==0UL );
      69           0 :   } else if( circq->cnt==1UL ) {
      70           0 :     FD_TEST( circq->head==circq->tail );
      71           0 :   }
      72           0 : 
      73           0 :   uchar * buf = (uchar *)(circq+1);
      74           0 : 
      75           0 :   ulong current = circq->head;
      76           0 :   int wrapped = 0;
      77           0 :   for( ulong i=0UL; i<circq->cnt; i++ ) {
      78           0 :     fd_circq_message_t * message = (fd_circq_message_t *)(buf+current);
      79           0 :     ulong start = current;
      80           0 :     ulong end = fd_ulong_align_up( start+sizeof( fd_circq_message_t ), message->align ) + message->footprint;
      81           0 :     if( wrapped ) FD_TEST( end<=circq->head );
      82           0 :     FD_TEST( start<end );
      83           0 :     FD_TEST( end<=circq->size );
      84           0 :     current = message->next;
      85           0 :     if( current<start ) wrapped = 1;
      86           0 :   }
      87           0 : }
      88             : 
      89             : /* Recover from eviction logic removing elements at the cursor */
      90             : 
      91             : static inline void
      92           0 : overrun_recover( fd_circq_t * circq ) {
      93           0 :   if( FD_UNLIKELY( circq->cursor==ULONG_MAX ) ) return;
      94             : 
      95           0 :   ulong oldest_seq = circq->cursor_push_seq - circq->cnt;
      96           0 :   if( FD_UNLIKELY( circq->cursor_seq<=oldest_seq ) ) circq->cursor = ULONG_MAX;
      97           0 : }
      98             : 
      99             : static void
     100             : evict( fd_circq_t * circq,
     101             :        ulong        from,
     102           0 :        ulong        to ) {
     103           0 :   uchar * buf = (uchar *)(circq+1);
     104             : 
     105           0 :   for(;;) {
     106           0 :     if( FD_UNLIKELY( !circq->cnt ) ) return;
     107             : 
     108           0 :     fd_circq_message_t * head = (fd_circq_message_t *)(buf+circq->head);
     109             : 
     110           0 :     ulong start = circq->head;
     111           0 :     ulong end = fd_ulong_align_up( start + sizeof( fd_circq_message_t ), head->align ) + head->footprint;
     112             : 
     113           0 :     if( FD_UNLIKELY( (start<to && end>from) ) ) {
     114           0 :       circq->bytes_used -= end-start;
     115           0 :       circq->cnt--;
     116           0 :       circq->metrics.drop_cnt++;
     117           0 :       if( FD_LIKELY( !circq->cnt ) ) circq->head = circq->tail = 0UL;
     118           0 :       else                           circq->head = head->next;
     119           0 :       overrun_recover( circq );
     120           0 :     } else {
     121           0 :       break;
     122           0 :     }
     123           0 :   }
     124           0 : }
     125             : 
     126             : uchar *
     127             : fd_circq_push_back( fd_circq_t * circq,
     128             :                     ulong        align,
     129           0 :                     ulong        footprint ) {
     130           0 :   if( FD_UNLIKELY( !fd_ulong_is_pow2( align ) ) ) {
     131           0 :     FD_LOG_WARNING(( "align must be a power of 2" ));
     132           0 :     return NULL;
     133           0 :   }
     134           0 :   if( FD_UNLIKELY( align>FD_CIRCQ_ALIGN ) ) {
     135           0 :     FD_LOG_WARNING(( "align must be at most %lu", FD_CIRCQ_ALIGN ));
     136           0 :     return NULL;
     137           0 :   }
     138             : 
     139           0 :   ulong required = fd_ulong_align_up( sizeof( fd_circq_message_t ), align ) + footprint;
     140           0 :   if( FD_UNLIKELY( required>circq->size ) ) {
     141           0 :     FD_LOG_WARNING(( "tried to push message which was too large %lu>%lu", required, circq->size ));
     142           0 :     return NULL;
     143           0 :   }
     144             : 
     145           0 :   uchar * buf = (uchar *)(circq+1);
     146             : 
     147           0 :   ulong current = 0UL;
     148           0 :   fd_circq_message_t * message = NULL;
     149           0 :   if( FD_LIKELY( circq->cnt ) ) {
     150           0 :     message = (fd_circq_message_t *)(buf+circq->tail);
     151           0 :     current = fd_ulong_align_up( fd_ulong_align_up( circq->tail+sizeof( fd_circq_message_t ), message->align )+message->footprint, alignof( fd_circq_message_t ) );
     152           0 :   }
     153             : 
     154           0 :   if( FD_UNLIKELY( current+required>circq->size ) ) {
     155           0 :     evict( circq, current, circq->size );
     156           0 :     evict( circq, 0UL, required );
     157             : 
     158           0 :     circq->tail = 0UL;
     159           0 :     if( FD_LIKELY( circq->cnt && message ) ) message->next = 0UL;
     160           0 :   } else {
     161           0 :     evict( circq, current, current+required );
     162             : 
     163           0 :     circq->tail = current;
     164           0 :     if( FD_LIKELY( circq->cnt && message ) ) message->next = current;
     165           0 :   }
     166             : 
     167           0 :   circq->cnt++;
     168           0 :   circq->bytes_used += fd_ulong_align_up( circq->tail+sizeof( fd_circq_message_t ), align ) + footprint - circq->tail; /* the extent pop and evict subtract */
     169           0 :   fd_circq_message_t * next_message = (fd_circq_message_t *)(buf+circq->tail);
     170           0 :   next_message->align = align;
     171           0 :   next_message->footprint = footprint;
     172           0 :   next_message->next = ULONG_MAX;
     173           0 :   circq->cursor_push_seq++;
     174           0 :   return (uchar *)(next_message+1);
     175           0 : }
     176             : 
     177             : void
     178             : fd_circq_resize_back( fd_circq_t * circq,
     179           0 :                       ulong        new_footprint ) {
     180           0 :   FD_TEST( circq->cnt );
     181             : 
     182           0 :   uchar * buf = (uchar *)(circq+1);
     183           0 :   fd_circq_message_t * message = (fd_circq_message_t *)(buf+circq->tail);
     184           0 :   FD_TEST( new_footprint<=message->footprint );
     185             : 
     186           0 :   circq->bytes_used -= message->footprint-new_footprint;
     187           0 :   message->footprint = new_footprint;
     188           0 : }
     189             : 
     190             : uchar const *
     191             : fd_circq_cursor_advance( fd_circq_t * circq,
     192           0 :                          ulong *      msg_sz ) {
     193             :   /* First call or after reset - start from head */
     194           0 :   if( FD_UNLIKELY( circq->cursor==ULONG_MAX ) ) {
     195           0 :     if( FD_UNLIKELY( !circq->cnt ) ) return NULL;
     196           0 :     circq->cursor = circq->head;
     197           0 :     circq->cursor_seq = circq->cursor_push_seq - circq->cnt;
     198           0 :   } else {
     199             :     /* Already iterating - move to next */
     200           0 :     if( FD_UNLIKELY( circq->cursor_seq >= circq->cursor_push_seq ) ) return NULL;
     201             : 
     202           0 :     uchar * buf = (uchar *)(circq+1);
     203           0 :     fd_circq_message_t * message = (fd_circq_message_t *)(buf+circq->cursor);
     204           0 :     circq->cursor = message->next;
     205           0 :   }
     206             : 
     207           0 :   uchar * buf = (uchar *)(circq+1);
     208           0 :   fd_circq_message_t * current_msg = (fd_circq_message_t *)(buf+circq->cursor);
     209           0 :   circq->cursor_seq++;
     210           0 :   if( FD_LIKELY( msg_sz ) ) *msg_sz = current_msg->footprint;
     211           0 :   return (uchar *)(current_msg+1);
     212           0 : }
     213             : 
     214             : int
     215             : fd_circq_cursor_peek( fd_circq_t const * circq,
     216           0 :                       ulong *            msg_sz ) {
     217           0 :   uchar const * buf = (uchar const *)(circq+1);
     218           0 :   ulong off;
     219           0 :   if( FD_UNLIKELY( circq->cursor==ULONG_MAX ) ) {
     220           0 :     if( FD_UNLIKELY( !circq->cnt ) ) return 0;
     221           0 :     off = circq->head;
     222           0 :   } else {
     223           0 :     if( FD_UNLIKELY( circq->cursor_seq>=circq->cursor_push_seq ) ) return 0;
     224           0 :     off = ((fd_circq_message_t const *)(buf+circq->cursor))->next;
     225           0 :   }
     226           0 :   *msg_sz = ((fd_circq_message_t const *)(buf+off))->footprint;
     227           0 :   return 1;
     228           0 : }
     229             : 
     230             : int
     231             : fd_circq_pop_until( fd_circq_t * circq,
     232           0 :                     ulong        cursor ) {
     233           0 :   if( FD_UNLIKELY( cursor>=circq->cursor_seq ) ) return -1;
     234             : 
     235           0 :   ulong oldest_seq = circq->cursor_push_seq-circq->cnt;
     236           0 :   if( FD_UNLIKELY( cursor<oldest_seq ) ) return 0;
     237             : 
     238           0 :   ulong to_pop = fd_ulong_min( cursor-oldest_seq+1UL, circq->cnt );
     239             : 
     240           0 :   uchar * buf = (uchar *)(circq+1);
     241           0 :   for( ulong i=0UL; i<to_pop; i++ ) {
     242           0 :     fd_circq_message_t * message = (fd_circq_message_t *)(buf+circq->head);
     243           0 :     circq->bytes_used -= fd_ulong_align_up( circq->head+sizeof( fd_circq_message_t ), message->align ) + message->footprint - circq->head;
     244           0 :     circq->cnt--;
     245             : 
     246           0 :     if( FD_UNLIKELY( !circq->cnt ) ) {
     247           0 :       circq->head = circq->tail = 0UL;
     248           0 :     } else {
     249           0 :       circq->head = message->next;
     250           0 :       FD_TEST( circq->head<circq->size );
     251           0 :     }
     252           0 :   }
     253             : 
     254           0 :   if( FD_UNLIKELY( !circq->cnt ) ) circq->cursor = ULONG_MAX;
     255           0 :   overrun_recover( circq );
     256           0 :   return 0;
     257           0 : }
     258             : 
     259             : void
     260           0 : fd_circq_reset_cursor( fd_circq_t * circq ) {
     261           0 :   circq->cursor = ULONG_MAX;
     262           0 : }
     263             : 
     264             : ulong
     265           0 : fd_circq_bytes_used( fd_circq_t const * circq ) {
     266           0 :   return circq->bytes_used;
     267           0 : }
     268             : 
     269             : ulong
     270           0 : fd_circq_unsent_cnt( fd_circq_t const * circq ) {
     271           0 :   if( FD_UNLIKELY( circq->cursor==ULONG_MAX ) ) return circq->cnt;
     272           0 :   return fd_ulong_min( circq->cursor_push_seq - circq->cursor_seq, circq->cnt );
     273           0 : }