Line data    Source code

       1           0 :   case FD_VINYL_REQ_TYPE_ERASE: {
       2             : 
       3           0 :     fd_vinyl_key_t const * req_key = MAP_REQ_GADDR( req->key_gaddr, fd_vinyl_key_t, batch_cnt );
       4           0 :     schar *                req_err = MAP_REQ_GADDR( req->err_gaddr, schar,          batch_cnt );
       5             : 
       6           0 :     if( FD_UNLIKELY( (!!batch_cnt) & ((!req_key) | (!req_err)) ) ) {
       7           0 :       comp_err = FD_VINYL_ERR_INVAL;
       8           0 :       break;
       9           0 :     }
      10             : 
      11           0 :     for( ulong batch_idx=0UL; batch_idx<batch_cnt; batch_idx++ ) {
      12             : 
      13             :       /* Query vinyl meta for key.  If pair key does not exist, fail
      14             :          with KEY. */
      15             : 
      16           0 :       fd_vinyl_key_t const * key = req_key + batch_idx;
      17             : 
      18           0 :       ulong memo = fd_vinyl_key_memo( meta_seed, key );
      19             : 
      20           0 :       ulong _ele_idx; /* avoid pointer escape */
      21           0 :       int   err = fd_vinyl_meta_query_fast( ele0, ele_max, key, memo, &_ele_idx );
      22           0 :       ulong ele_idx = _ele_idx; /* In [0,ele_max) */
      23             : 
      24           0 :       if( FD_UNLIKELY( err ) ) {
      25           0 :         FD_COMPILER_MFENCE();
      26           0 :         req_err[ batch_idx ] = (schar)FD_VINYL_ERR_KEY;
      27           0 :         FD_COMPILER_MFENCE();
      28           0 :         fail_cnt++;
      29           0 :         continue;
      30           0 :       }
      31             : 
      32             :       /* At this point, pair key exists. */
      33             : 
      34           0 :       ulong line_idx = ele0[ ele_idx ].line_idx;
      35             : 
      36           0 :       if( FD_LIKELY( line_idx<line_cnt ) ) {
      37             : 
      38             :         /* At this point, pair key is cached.  If it is currently
      39             :            acquired for read or modify, fail with AGAIN.  Otherwise,
      40             :            evict it. */
      41             : 
      42           0 :         FD_CRIT( line[ line_idx ].ele_idx==ele_idx, "corruption detected" );
      43             : 
      44           0 :         fd_vinyl_data_obj_t * obj = line[ line_idx ].obj;
      45             : 
      46           0 :         FD_ALERT( fd_vinyl_data_is_valid_obj( obj, vol, vol_cnt ), "corruption detected" );
      47           0 :         FD_CRIT ( obj->line_idx==line_idx,                         "corruption detected" );
      48           0 :         FD_CRIT ( !obj->rd_active,                                 "corruption detected" );
      49             : 
      50           0 :         ulong ctl = line[ line_idx ].ctl;
      51             : 
      52           0 :         ulong ver = fd_vinyl_line_ctl_ver( ctl );
      53           0 :         long  ref = fd_vinyl_line_ctl_ref( ctl );
      54             : 
      55           0 :         if( FD_UNLIKELY( ref ) ) {
      56           0 :           FD_COMPILER_MFENCE();
      57           0 :           req_err[ batch_idx ] = (schar)FD_VINYL_ERR_AGAIN;
      58           0 :           FD_COMPILER_MFENCE();
      59           0 :           fail_cnt++;
      60           0 :           continue;
      61           0 :         }
      62             : 
      63           0 :         line[ line_idx ].obj     = NULL;
      64           0 :         line[ line_idx ].ele_idx = ULONG_MAX; //ele0[ ele_idx ].line_idx = ULONG_MAX; /* Technically not necessary given below */
      65           0 :         line[ line_idx ].ctl     = fd_vinyl_line_ctl( ver+1UL, 0L ); /* bump version */
      66             : 
      67           0 :         fd_vinyl_line_evict_prio( &vinyl->line_idx_lru, line, line_cnt, line_idx, FD_VINYL_LINE_EVICT_PRIO_LRU );
      68             : 
      69           0 :         fd_vinyl_data_free( data, obj );
      70             : 
      71           0 :       } else {
      72             : 
      73           0 :         FD_CRIT( line_idx==ULONG_MAX, "corruption detected" );
      74             : 
      75           0 :       }
      76             : 
      77             :       /* At this point, pair key exists and is not cached.  Append a
      78             :          dead block and remove it from the meta.  This generates two
      79             :          pieces of bstream garbage: the old pair and the dead block
      80             :          itself (the dead block is only needed for recovery and then
      81             :          only while the old pair is in the bstream's past). */
      82             : 
      83             :       /* FIXME: COMPACT SEQUENTIAL DEADS IN THE BSTREAM TO BE MORE
      84             :          SPACE EFFICIENT? */
      85             : 
      86           0 :       ulong val_esz = fd_vinyl_bstream_ctl_sz( ele0[ ele_idx ].phdr.ctl );
      87             : 
      88           0 :       accum_garbage_cnt += 2UL;
      89           0 :       accum_garbage_sz  += fd_vinyl_bstream_pair_sz( val_esz ) + FD_VINYL_BSTREAM_BLOCK_SZ;
      90             : 
      91           0 :       fd_vinyl_io_append_dead( io, &ele0[ ele_idx ].phdr, NULL, 0UL );
      92           0 :       append_cnt++;
      93           0 :       accum_dead_cnt++;
      94             : 
      95           0 :       fd_vinyl_meta_remove_fast( ele0, ele_max, lock, lock_shift, line, line_cnt, ele_idx );
      96             : 
      97           0 :       ulong pair_cnt = vinyl->pair_cnt;
      98           0 :       FD_CRIT( pair_cnt, "corruption detected" );
      99           0 :       vinyl->pair_cnt = pair_cnt - 1UL;
     100             : 
     101           0 :       FD_COMPILER_MFENCE();
     102           0 :       req_err[ batch_idx ] = (schar)FD_VINYL_SUCCESS;
     103           0 :       FD_COMPILER_MFENCE();
     104           0 :     }
     105             : 
     106           0 :     comp_err = FD_VINYL_SUCCESS;
     107           0 :     break;
     108           0 :   }