Line data    Source code

       1           0 :   case FD_VINYL_REQ_TYPE_FETCH: {
       2           0 :     fd_vinyl_key_t const * req_key = MAP_REQ_GADDR( req->key_gaddr, fd_vinyl_key_t, batch_cnt );
       3             : 
       4           0 :     if( FD_UNLIKELY( (!!batch_cnt) & (!req_key) ) ) break;
       5             : 
       6           0 :     for( ulong batch_idx=0UL; batch_idx<batch_cnt; batch_idx++ ) {
       7             : 
       8             :       /* Query vinyl meta for key */
       9             : 
      10           0 :       fd_vinyl_key_t const * key = req_key + batch_idx;
      11             : 
      12           0 :       ulong memo = fd_vinyl_key_memo( meta_seed, key );
      13             : 
      14           0 :       ulong _ele_idx; /* avoid pointer escape */
      15           0 :       int   err = fd_vinyl_meta_query_fast( ele0, ele_max, key, memo, &_ele_idx );
      16           0 :       ulong ele_idx = _ele_idx; /* In [0,ele_max) */
      17             : 
      18           0 :       if( FD_UNLIKELY( err                                 ) ||
      19           0 :           FD_UNLIKELY( ele0[ ele_idx ].phdr.ctl==ULONG_MAX ) ) continue; /* Nothing to fetch */
      20             : 
      21             :       /* At this point, pair key exists at seq_present.  If pair key is
      22             :          already cached, we set the priority to MRU (to reflect the
      23             :          handling when the key is not cached). */
      24             : 
      25           0 :       ulong line_idx = ele0[ ele_idx ].line_idx;
      26             : 
      27           0 :       FD_CRIT( (line_idx<line_cnt) | (line_idx==ULONG_MAX), "corruption detected" );
      28             : 
      29           0 :       if( FD_LIKELY( line_idx<line_cnt ) ) {
      30             : 
      31           0 :         FD_CRIT( line[ line_idx ].ele_idx==ele_idx, "corruption detected" );
      32             : 
      33           0 :         fd_vinyl_line_evict_prio( &vinyl->line_idx_lru, line, line_cnt, line_idx, FD_VINYL_LINE_EVICT_PRIO_MRU );
      34             : 
      35           0 :         continue;
      36           0 :       }
      37             : 
      38             :       /* At this point, pair key existsat seq_present but is not cached.
      39             :          Evict the least recently used evictable line to make room to
      40             :          cache this pair.  Connect this line to meta element ele_idx,
      41             :          set the line's reference count to zero, bump the line's version
      42             :          and set the eviction priority to MRU.  We don't modify any
      43             :          shared fields in meta element ele_idx so we can do the
      44             :          modification fast.
      45             : 
      46             :          We do this upfront to free data cache for the alloc if the LRU
      47             :          line is in use and to handle the same pair appearing multiple
      48             :          times in an acquire.
      49             : 
      50             :          The mechanics for fetch requests with redundant keys are
      51             :          similar to acquire-for-read requests.  In this case, trailing
      52             :          redundant fetches will see the pair as cached (due to the first
      53             :          redundant fetch ... this one), set the eviction priority to MRU
      54             :          (again) and then continue. */
      55             : 
      56           0 :       ulong pair_ctl =        ele0[ ele_idx ].phdr.ctl;
      57           0 :       ulong val_sz   = (ulong)ele0[ ele_idx ].phdr.info.val_sz;
      58             : 
      59           0 :       FD_CRIT( fd_vinyl_bstream_ctl_type( pair_ctl )==FD_VINYL_BSTREAM_CTL_TYPE_PAIR, "corruption detected" );
      60           0 :       FD_CRIT( val_sz<=FD_VINYL_VAL_MAX,                                              "corruption detected" );
      61             : 
      62           0 :       line_idx = fd_vinyl_line_evict_lru( &vinyl->line_idx_lru, line, line_cnt, ele0, ele_max, data );
      63             : 
      64           0 :       ulong line_ctl = line[ line_idx ].ctl;
      65             : 
      66           0 :       ulong ver = fd_vinyl_line_ctl_ver( line_ctl );
      67             : 
      68           0 :       line[ line_idx ].ele_idx = ele_idx; ele0[ ele_idx ].line_idx = line_idx;
      69           0 :       line[ line_idx ].ctl     = fd_vinyl_line_ctl( ver+1UL, 0L );
      70             : 
      71           0 :       fd_vinyl_line_evict_prio( &vinyl->line_idx_lru, line, line_cnt, line_idx, FD_VINYL_LINE_EVICT_PRIO_MRU );
      72             : 
      73             :       /* Allocate an appropriately sized object to hold this pair,
      74             :          connect it to this line and start reading the encoded pair data
      75             :          into obj. */
      76             : 
      77           0 :       ulong szc = fd_vinyl_data_szc( val_sz );
      78             : 
      79           0 :       fd_vinyl_data_obj_t * obj = fd_vinyl_data_alloc( data, szc );
      80           0 :       if( FD_UNLIKELY( !obj ) ) FD_LOG_CRIT(( "increase data cache size" ));
      81             : 
      82           0 :       line[ line_idx ].obj = obj; obj->line_idx = line_idx;
      83             : 
      84             :       /* Start reading encoded pair data and defer the validation and
      85             :          decoding to later (and then in whatever order the I/O layer
      86             :          sees fit). */
      87             : 
      88           0 :       obj->rd_active = (short)1;
      89             : 
      90           0 :       int   style   = fd_vinyl_bstream_ctl_style( pair_ctl );
      91           0 :       ulong val_esz = fd_vinyl_bstream_ctl_sz   ( pair_ctl );
      92             : 
      93           0 :       FD_CRIT( val_esz<=FD_VINYL_VAL_MAX,                                   "corruption detected" );
      94           0 :       FD_CRIT( (style!=FD_VINYL_BSTREAM_CTL_STYLE_RAW) | (val_sz==val_esz), "corruption detected" );
      95             : 
      96           0 :       fd_vinyl_data_obj_t * cobj;
      97             : 
      98           0 :       if( FD_LIKELY( style==FD_VINYL_BSTREAM_CTL_STYLE_RAW ) ) cobj = obj;
      99           0 :       else {
     100           0 :         cobj = fd_vinyl_data_alloc( data, fd_vinyl_data_szc( val_esz ) );
     101           0 :         if( FD_UNLIKELY( !cobj ) ) FD_LOG_CRIT(( "increase data cache size" ));
     102           0 :       }
     103             : 
     104           0 :       cobj->rd->ctx = (ulong)obj;
     105           0 :       cobj->rd->seq = ele0[ ele_idx ].seq;
     106           0 :       cobj->rd->dst = fd_vinyl_data_obj_phdr( cobj );
     107           0 :       cobj->rd->sz  = fd_vinyl_bstream_pair_sz( val_esz );
     108             : 
     109           0 :       cobj->rd_err = (schar *)cobj->unused;
     110             : 
     111           0 :       fd_vinyl_io_read( io, cobj->rd );
     112           0 :       read_cnt++;
     113           0 :     }
     114             : 
     115           0 :     break;
     116           0 :   }