Line data    Source code

       1             : #include "fd_circq.h"
       2             : 
       3             : struct __attribute__((aligned(8UL))) fd_circq_message_private {
       4             :   ulong align;
       5             :   ulong footprint;
       6             :   
       7             :   /* Offset withn the circular buffer data region of where the next
       8             :      message starts, if there is one.  This is not always the same as
       9             :      aligning up this message + footprint, because the next message may
      10             :      have wrapped around to the start of the buffer. */
      11             :   ulong next;
      12             : };
      13             : 
      14             : typedef struct fd_circq_message_private fd_circq_message_t;
      15             : 
      16             : FD_FN_CONST ulong
      17           0 : fd_circq_align( void ) {
      18           0 :   return FD_CIRCQ_ALIGN;
      19           0 : }
      20             : 
      21             : FD_FN_CONST ulong
      22           0 : fd_circq_footprint( ulong sz ) {
      23           0 :   return sizeof( fd_circq_t ) + sz;
      24           0 : }
      25             : 
      26             : void *
      27             : fd_circq_new( void * shmem,
      28          12 :                 ulong  sz ) {
      29          12 :   fd_circq_t * circq = (fd_circq_t *)shmem;
      30          12 :   circq->cnt  = 0UL;
      31          12 :   circq->head = 0UL;
      32          12 :   circq->tail = 0UL;
      33          12 :   circq->size = sz;
      34          12 :   return shmem;
      35          12 : }
      36             : 
      37             : fd_circq_t *
      38          12 : fd_circq_join( void * shbuf ) {
      39          12 :   return (fd_circq_t *)shbuf;
      40          12 : }
      41             : 
      42             : void *
      43           0 : fd_circq_leave( fd_circq_t * buf ) {
      44           0 :   return (void *)buf;
      45           0 : }
      46             : 
      47             : void *
      48           0 : fd_circq_delete( void * shbuf ) {
      49           0 :   return shbuf;
      50           0 : }
      51             : 
      52             : static inline void FD_FN_UNUSED
      53           0 : verify( fd_circq_t * circq ) {
      54           0 :   FD_TEST( circq->head<circq->size );
      55           0 :   FD_TEST( circq->tail<circq->size );
      56           0 :   FD_TEST( circq->tail!=circq->head || circq->cnt<=1 );
      57           0 :   if( !circq->cnt ) {
      58           0 :     FD_TEST( circq->head==0UL );
      59           0 :     FD_TEST( circq->tail==0UL );
      60           0 :   } else if( circq->cnt==1UL ) {
      61           0 :     FD_TEST( circq->head==circq->tail );
      62           0 :   }
      63           0 : 
      64           0 :   uchar * buf = (uchar *)(circq+1);
      65           0 : 
      66           0 :   ulong current = circq->head;
      67           0 :   int wrapped = 0;
      68           0 :   for( ulong i=0UL; i<circq->cnt; i++ ) {
      69           0 :     fd_circq_message_t * message = (fd_circq_message_t *)(buf+current);
      70           0 :     ulong start = current;
      71           0 :     ulong end = fd_ulong_align_up( start+sizeof( fd_circq_message_t ), message->align ) + message->footprint;
      72           0 :     if( wrapped ) FD_TEST( end<=circq->head );
      73           0 :     FD_TEST( start<end );
      74           0 :     FD_TEST( end<=circq->size );
      75           0 :     current = message->next;
      76           0 :     if( current<start ) wrapped = 1;
      77           0 :   }
      78           0 : }
      79             : 
      80             : static void
      81             : evict( fd_circq_t * circq,
      82             :        ulong        from,
      83   330267006 :        ulong        to ) {
      84   330267006 :   uchar * buf = (uchar *)(circq+1);
      85             : 
      86   531593601 :   for(;;) {
      87   531593601 :     if( FD_UNLIKELY( !circq->cnt ) ) return;
      88             : 
      89   439887978 :     fd_circq_message_t * head = (fd_circq_message_t *)(buf+circq->head);
      90             : 
      91   439887978 :     ulong start = circq->head;
      92   439887978 :     ulong end = fd_ulong_align_up( start + sizeof( fd_circq_message_t ), head->align ) + head->footprint;
      93             : 
      94   439887978 :     if( FD_UNLIKELY( (start<to && end>from) ) ) {
      95   201326595 :       circq->cnt--;
      96   201326595 :       circq->metrics.drop_cnt++;
      97   201326595 :       if( FD_LIKELY( !circq->cnt ) ) circq->head = circq->tail = 0UL;
      98   109621608 :       else                           circq->head = head->next;
      99   238561383 :     } else {
     100   238561383 :       break;
     101   238561383 :     }
     102   439887978 :   }
     103   330267006 : }
     104             : 
     105             : uchar *
     106             : fd_circq_push_back( fd_circq_t * circq,
     107             :                     ulong        align,
     108   201351195 :                     ulong        footprint ) {
     109   201351195 :   if( FD_UNLIKELY( !fd_ulong_is_pow2( align ) ) ) {
     110       23766 :     FD_LOG_WARNING(( "align must be a power of 2" ));
     111       23766 :     return NULL;
     112       23766 :   }
     113   201327429 :   if( FD_UNLIKELY( align>FD_CIRCQ_ALIGN ) ) {
     114           0 :     FD_LOG_WARNING(( "align must be at most %lu", FD_CIRCQ_ALIGN ));
     115           0 :     return NULL;
     116           0 :   }
     117             : 
     118   201327429 :   ulong required = fd_ulong_align_up( sizeof( fd_circq_message_t ), align ) + footprint;
     119   201327429 :   if( FD_UNLIKELY( required>circq->size ) ) {
     120         180 :     FD_LOG_WARNING(( "tried to push message which was too large %lu>%lu", required, circq->size ));
     121         180 :     return NULL;
     122         180 :   }
     123             : 
     124   201327249 :   uchar * buf = (uchar *)(circq+1);
     125             : 
     126   201327249 :   ulong current = 0UL;
     127   201327249 :   fd_circq_message_t * message = NULL;
     128   201327249 :   if( FD_LIKELY( circq->cnt ) ) {
     129   201326613 :     message = (fd_circq_message_t *)(buf+circq->tail);
     130   201326613 :     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 ) );
     131   201326613 :   }
     132             : 
     133   201327249 :   if( FD_UNLIKELY( current+required>circq->size ) ) {
     134   128939757 :     evict( circq, current, circq->size );
     135   128939757 :     evict( circq, 0UL, required );
     136             : 
     137   128939757 :     circq->tail = 0UL;
     138   128939757 :     if( FD_LIKELY( circq->cnt && message ) ) message->next = 0UL;
     139   128939757 :   } else {
     140    72387492 :     evict( circq, current, current+required );
     141             : 
     142    72387492 :     circq->tail = current;
     143    72387492 :     if( FD_LIKELY( circq->cnt && message ) ) message->next = current;
     144    72387492 :   }
     145             : 
     146   201327249 :   circq->cnt++;
     147   201327249 :   fd_circq_message_t * next_message = (fd_circq_message_t *)(buf+circq->tail);
     148   201327249 :   next_message->align = align;
     149   201327249 :   next_message->footprint = footprint;
     150   201327249 :   return (uchar *)(next_message+1);
     151   201327429 : }
     152             : 
     153             : uchar const *
     154       12411 : fd_circq_pop_front( fd_circq_t * circq ) {
     155       12411 :   if( FD_UNLIKELY( !circq->cnt ) ) return NULL;
     156             : 
     157         645 :   circq->cnt--;
     158         645 :   fd_circq_message_t * message = (fd_circq_message_t *)((uchar *)(circq+1)+circq->head);
     159         645 :   if( FD_UNLIKELY( !circq->cnt ) ) circq->head = circq->tail = 0UL;
     160          15 :   else                             circq->head = message->next;
     161         645 :   FD_TEST( circq->head<circq->size );
     162         645 :   return (uchar *)(message+1);
     163         645 : }