Line data    Source code

       1             : #ifndef HEADER_fd_src_disco_stem_fd_stem_h
       2             : #define HEADER_fd_src_disco_stem_fd_stem_h
       3             : 
       4             : #include "../fd_disco_base.h"
       5             : 
       6           0 : #define FD_STEM_SCRATCH_ALIGN (128UL)
       7             : 
       8             : struct fd_stem_context {
       9             :    fd_frag_meta_t ** mcaches;
      10             :    ulong *           seqs;
      11             :    ulong *           depths;
      12             : 
      13             :    ulong *           cr_avail;
      14             :    ulong *           min_cr_avail;
      15             :    ulong             cr_decrement_amount;
      16             :    int *             out_reliable;
      17             :    ulong const *     cons_seq;
      18             :    struct fd_stem_tile_in * in;
      19             : };
      20             : 
      21             : typedef struct fd_stem_context fd_stem_context_t;
      22             : 
      23             : struct __attribute__((aligned(64))) fd_stem_tile_in {
      24             :   fd_frag_meta_t const * mcache;   /* local join to this in's mcache */
      25             :   uint                   depth;    /* == fd_mcache_depth( mcache ), depth of this in's cache (const) */
      26             :   uint                   idx;      /* index of this in in the list of providers, [0, in_cnt) */
      27             :   ulong                  seq;      /* sequence number of next frag expected from the upstream producer,
      28             :                                       updated when frag from this in is published */
      29             :   fd_frag_meta_t const * mline;    /* == mcache + fd_mcache_line_idx( seq, depth ), location to poll next */
      30             :   ulong *                fseq;     /* local join to the fseq used to return flow control credits to the in */
      31             :   uint                   accum[6]; /* local diagnostic accumulators.  These are drained during in housekeeping. */
      32             :                                    /* Assumes FD_FSEQ_DIAG_{PUB_CNT,PUB_SZ,FILT_CNT,FILT_SZ,OVRNP_CNT,OVRNP_FRAG_CNT} are 0:5 */
      33             : };
      34             : 
      35             : typedef struct fd_stem_tile_in fd_stem_tile_in_t;
      36             : 
      37             : static inline ulong
      38             : fd_stem_publish( fd_stem_context_t * stem,
      39             :                  ulong               out_idx,
      40             :                  ulong               sig,
      41             :                  ulong               chunk,
      42             :                  ulong               sz,
      43             :                  ulong               ctl,
      44             :                  ulong               tsorig,
      45          63 :                  ulong               tspub ) {
      46          63 :   fd_frag_meta_t * mcache = stem->mcaches[ out_idx ];
      47          63 :   ulong            depth  = stem->depths [ out_idx ];
      48          63 :   ulong *          seqp   = &stem->seqs  [ out_idx ];
      49          63 :   ulong            seq    = *seqp;
      50          63 : # if FD_HAS_AVX
      51          63 :   fd_mcache_publish_avx( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
      52             : # elif FD_HAS_ARM
      53             :   fd_mcache_publish_arm( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
      54             : # else
      55             :   fd_mcache_publish    ( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
      56             : # endif
      57          63 :   if( FD_LIKELY( stem->out_reliable[ out_idx ] ) ) {
      58          63 :     if( FD_UNLIKELY( stem->cr_avail[ out_idx ]<stem->cr_decrement_amount ) ) { /* Ensure producer BURST is set correctly */
      59           0 :       FD_LOG_ERR(( "BURST underprovisioned out_idx=%lu cr_avail=%lu min_cr_avail=%lu cr_decrement_amount=%lu", out_idx, stem->cr_avail[ out_idx ], *stem->min_cr_avail, stem->cr_decrement_amount ));
      60           0 :     }
      61          63 :     stem->cr_avail[ out_idx ] -= stem->cr_decrement_amount;
      62          63 :     *stem->min_cr_avail        = fd_ulong_min( stem->cr_avail[ out_idx ], *stem->min_cr_avail );
      63          63 :   }
      64          63 :   *seqp = fd_seq_inc( seq, 1UL );
      65          63 :   return seq;
      66          63 : }
      67             : 
      68             : static inline ulong
      69             : fd_stem_advance( fd_stem_context_t * stem,
      70           0 :                  ulong               out_idx ) {
      71           0 :   ulong * seqp = &stem->seqs[ out_idx ];
      72           0 :   ulong   seq  = *seqp;
      73           0 :   if( FD_LIKELY( stem->out_reliable[ out_idx ] ) ) {
      74           0 :     stem->cr_avail[ out_idx ] -= stem->cr_decrement_amount;
      75           0 :     *stem->min_cr_avail        = fd_ulong_min( stem->cr_avail[ out_idx ], *stem->min_cr_avail );
      76           0 :   }
      77           0 :   *seqp = fd_seq_inc( seq, 1UL );
      78           0 :   return seq;
      79           0 : }
      80             : 
      81             : #endif /* HEADER_fd_src_disco_stem_fd_stem_h */