Line data    Source code

       1             : #ifndef HEADER_fd_src_vinyl_io_fd_vinyl_io_ur_wq_ring_h
       2             : #define HEADER_fd_src_vinyl_io_fd_vinyl_io_ur_wq_ring_h
       3             : 
       4             : /* wq_ring.h provides a reorder buffer for write completions.  This is
       5             :    required because io_uring might post completions out-of-order. */
       6             : 
       7             : #include "../../bstream/fd_vinyl_bstream.h"
       8             : 
       9             : struct wq_desc {
      10             :   ulong seq;
      11             :   uint  sz;
      12             :   uint  done : 1;
      13             : };
      14             : 
      15             : typedef struct wq_desc wq_desc_t;
      16             : 
      17             : struct wq_ring {
      18             :   ulong wq0;
      19             :   ulong wq1;
      20             :   ulong seq;
      21             :   ulong max;
      22             :   __extension__ wq_desc_t ring[0];
      23             : };
      24             : 
      25             : typedef struct wq_ring wq_ring_t;
      26             : 
      27             : FD_PROTOTYPES_BEGIN
      28             : 
      29             : static inline wq_ring_t *
      30             : wq_ring_init( wq_ring_t * ring,
      31             :               ulong       seq,
      32           0 :               ulong       max ) {
      33           0 :   if( FD_UNLIKELY( !fd_ulong_is_pow2( max ) ) ) {
      34           0 :     FD_LOG_CRIT(( "max (%lu) is not a power of 2", max ));
      35           0 :   }
      36           0 :   *ring = (wq_ring_t) {
      37           0 :     .wq0 = 0UL,
      38           0 :     .wq1 = 0UL,
      39           0 :     .seq = seq,
      40           0 :     .max = max
      41           0 :   };
      42           0 :   return ring;
      43           0 : }
      44             : 
      45             : static inline long
      46           0 : wq_ring_free_cnt( wq_ring_t const * ring ) {
      47           0 :   return (long)( ring->max - (ring->wq1 - ring->wq0) );
      48           0 : }
      49             : 
      50             : static inline int
      51           0 : wq_ring_is_full( wq_ring_t const * ring ) {
      52           0 :   return (ring->wq1 - ring->wq0) >= ring->max;
      53           0 : }
      54             : 
      55             : FD_FN_PURE static inline wq_desc_t *
      56             : wq_ring_desc( wq_ring_t * ring,
      57           0 :               ulong       wq ) {
      58           0 :   ulong mask = ring->max-1UL;
      59           0 :   return &ring->ring[ wq&mask ];
      60           0 : }
      61             : 
      62             : static inline ulong
      63             : wq_ring_enqueue( wq_ring_t * ring,
      64           0 :                  ulong       seq ) {
      65           0 :   FD_CRIT( !wq_ring_is_full( ring ), "wq_ring overflow" );
      66           0 :   ulong wq   = ring->wq1;
      67           0 :   ulong mask = ring->max-1UL;
      68           0 :   ring->ring[ wq&mask ] = (wq_desc_t){
      69           0 :     .seq  = seq,
      70           0 :     .sz   = 0U,
      71           0 :     .done = 0
      72           0 :   };
      73           0 :   ring->wq1 = wq+1UL;
      74           0 :   return wq;
      75           0 : }
      76             : 
      77             : FD_FN_UNUSED static ulong
      78             : wq_ring_complete( wq_ring_t * ring,
      79           0 :                   ulong       wq ) {
      80           0 :   FD_CRIT( fd_vinyl_seq_gt( ring->wq1, ring->wq0 ), "stray wq_ring completion" );
      81           0 :   ulong mask = ring->max-1UL;
      82           0 :   FD_CRIT( !ring->ring[ wq&mask ].done, "wq_ring entry already completed" );
      83           0 :   ring->ring[ wq&mask ].done = 1;
      84           0 :   ulong seq = ring->seq;
      85           0 :   while( fd_vinyl_seq_gt( ring->wq1, ring->wq0 ) ) {
      86           0 :     wq_desc_t const * desc = &ring->ring[ ring->wq0&mask ];
      87           0 :     if( !desc->done ) break;
      88           0 :     seq = desc->seq;
      89           0 :     ring->wq0++;
      90           0 :   }
      91           0 :   ring->seq = seq;
      92           0 :   return seq;
      93           0 : }
      94             : 
      95             : FD_PROTOTYPES_END
      96             : 
      97             : #endif /* HEADER_fd_src_vinyl_io_fd_vinyl_io_ur_wq_ring_h */