Line data    Source code

       1             : #include "fd_store.h"
       2             : 
       3             : void *
       4           0 : fd_store_new( void * mem, ulong lo_wmark_slot ) {
       5           0 :   if( FD_UNLIKELY( !mem ) ) {
       6           0 :     FD_LOG_WARNING( ( "NULL mem" ) );
       7           0 :     return NULL;
       8           0 :   }
       9             : 
      10           0 :   if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)mem, fd_store_align() ) ) ) {
      11           0 :     FD_LOG_WARNING( ( "misaligned mem" ) );
      12           0 :     return NULL;
      13           0 :   }
      14             : 
      15           0 :   fd_memset( mem, 0, fd_store_footprint() );
      16             : 
      17           0 :   fd_store_t * store = (fd_store_t *)mem;
      18           0 :   store->first_turbine_slot = FD_SLOT_NULL;
      19           0 :   store->curr_turbine_slot = FD_SLOT_NULL;
      20           0 :   store->root = FD_SLOT_NULL;
      21           0 :   fd_repair_backoff_map_new( store->repair_backoff_map );
      22           0 :   store->pending_slots = fd_pending_slots_new( (uchar *)mem + fd_store_footprint(), lo_wmark_slot );
      23           0 :   if( FD_UNLIKELY( !store->pending_slots ) ) {
      24           0 :     return NULL;
      25           0 :   }
      26             : 
      27           0 :   return mem;
      28           0 : }
      29             : 
      30             : fd_store_t *
      31           0 : fd_store_join( void * store ) {
      32           0 :   if( FD_UNLIKELY( !store ) ) {
      33           0 :     FD_LOG_WARNING( ( "NULL store" ) );
      34           0 :     return NULL;
      35           0 :   }
      36             : 
      37           0 :   if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)store, fd_store_align() ) ) ) {
      38           0 :     FD_LOG_WARNING( ( "misaligned replay" ) );
      39           0 :     return NULL;
      40           0 :   }
      41             : 
      42           0 :   fd_store_t * store_ = (fd_store_t *)store;
      43           0 :   fd_repair_backoff_map_join( store_->repair_backoff_map );
      44           0 :   store_->pending_slots = fd_pending_slots_join( store_->pending_slots );
      45           0 :   if( FD_UNLIKELY( !store_->pending_slots ) ) {
      46           0 :     return NULL;
      47           0 :   }
      48             : 
      49           0 :   return store_;
      50           0 : }
      51             : 
      52             : void *
      53           0 : fd_store_leave( fd_store_t const * store ) {
      54           0 :   if( FD_UNLIKELY( !store ) ) {
      55           0 :     FD_LOG_WARNING( ( "NULL store" ) );
      56           0 :     return NULL;
      57           0 :   }
      58             : 
      59           0 :   if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)store, fd_store_align() ) ) ) {
      60           0 :     FD_LOG_WARNING( ( "misaligned store" ) );
      61           0 :     return NULL;
      62           0 :   }
      63             : 
      64           0 :   return (void *)store;
      65           0 : }
      66             : 
      67             : void *
      68           0 : fd_store_delete( void * store ) {
      69           0 :   if( FD_UNLIKELY( !store ) ) {
      70           0 :     FD_LOG_WARNING( ( "NULL store" ) );
      71           0 :     return NULL;
      72           0 :   }
      73             : 
      74           0 :   if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)store, fd_store_align() ) ) ) {
      75           0 :     FD_LOG_WARNING( ( "misaligned store" ) );
      76           0 :     return NULL;
      77           0 :   }
      78             : 
      79           0 :   return store;
      80           0 : }
      81             : 
      82             : void
      83           0 : fd_store_expected_shred_version( fd_store_t * store, ulong expected_shred_version ) {
      84           0 :   store->expected_shred_version = expected_shred_version;
      85           0 : }
      86             : 
      87             : int
      88             : fd_store_slot_prepare( fd_store_t *   store,
      89             :                        ulong          slot,
      90             :                        ulong *        repair_slot_out,
      91             :                        uchar const ** block_out,
      92           0 :                        ulong *        block_sz_out ) {
      93           0 :   fd_blockstore_start_read( store->blockstore );
      94             : 
      95           0 :   ulong re_adds[2];
      96           0 :   uint re_adds_cnt           = 0U;
      97           0 :   long re_add_delays[2];
      98             : 
      99           0 :   *repair_slot_out = 0;
     100           0 :   int rc = FD_STORE_SLOT_PREPARE_CONTINUE;
     101             : 
     102           0 :   fd_block_t * block = fd_blockstore_block_query( store->blockstore, slot );
     103           0 :   fd_block_map_t * block_map_entry = fd_blockstore_block_map_query( store->blockstore, slot );
     104             : 
     105             : 
     106             :   /* We already executed this block */
     107           0 :   if( FD_UNLIKELY( block && fd_uchar_extract_bit( block_map_entry->flags, FD_BLOCK_FLAG_REPLAYING ) ) ) {
     108           0 :     rc = FD_STORE_SLOT_PREPARE_ALREADY_EXECUTED;
     109           0 :     goto end;
     110           0 :   }
     111             : 
     112           0 :   if( FD_UNLIKELY( block && fd_uchar_extract_bit( block_map_entry->flags, FD_BLOCK_FLAG_PROCESSED ) ) ) {
     113           0 :     rc = FD_STORE_SLOT_PREPARE_ALREADY_EXECUTED;
     114           0 :     goto end;
     115           0 :   }
     116             : 
     117           0 :   if( FD_UNLIKELY( !block_map_entry ) ) {
     118             :     /* I know nothing about this block yet */
     119           0 :     rc = FD_STORE_SLOT_PREPARE_NEED_REPAIR;
     120           0 :     *repair_slot_out = slot;
     121           0 :     re_add_delays[re_adds_cnt] = FD_REPAIR_BACKOFF_TIME;
     122           0 :     re_adds[re_adds_cnt++] = slot;
     123           0 :     goto end;
     124           0 :   }
     125             : 
     126           0 :   ulong            parent_slot  = block_map_entry->parent_slot;
     127           0 :   fd_block_map_t * parent_block_map_entry = fd_blockstore_block_map_query( store->blockstore, parent_slot );
     128             : 
     129             :   /* If the parent slot meta is missing, this block is an orphan and the ancestry needs to be
     130             :    * repaired before we can replay it. */
     131           0 :   if( FD_UNLIKELY( !parent_block_map_entry ) ) {
     132           0 :     rc = FD_STORE_SLOT_PREPARE_NEED_ORPHAN;
     133           0 :     *repair_slot_out = slot;
     134           0 :     re_add_delays[re_adds_cnt] = FD_REPAIR_BACKOFF_TIME;
     135           0 :     re_adds[re_adds_cnt++] = slot;
     136             : 
     137           0 :     re_add_delays[re_adds_cnt] = FD_REPAIR_BACKOFF_TIME;
     138           0 :     re_adds[re_adds_cnt++] = parent_slot;
     139           0 :     goto end;
     140           0 :   }
     141             : 
     142           0 :   fd_block_t * parent_block = fd_blockstore_block_query( store->blockstore, parent_slot );
     143             : 
     144             :   /* We have a parent slot meta, and therefore have at least one shred of the parent block, so we
     145             :      have the ancestry and need to repair that block directly (as opposed to calling repair orphan).
     146             :   */
     147           0 :   if( FD_UNLIKELY( !parent_block ) ) {
     148           0 :     rc = FD_STORE_SLOT_PREPARE_NEED_REPAIR;
     149           0 :     *repair_slot_out = parent_slot;
     150           0 :     re_add_delays[re_adds_cnt] = FD_REPAIR_BACKOFF_TIME;
     151           0 :     re_adds[re_adds_cnt++] = parent_slot;
     152           0 :     re_add_delays[re_adds_cnt] = FD_REPAIR_BACKOFF_TIME;
     153           0 :     re_adds[re_adds_cnt++] = slot;
     154             : 
     155           0 :     goto end;
     156           0 :   }
     157             : 
     158             :   /* See if the parent is executed yet */
     159           0 :   if( FD_UNLIKELY( !fd_uchar_extract_bit( parent_block_map_entry->flags, FD_BLOCK_FLAG_PROCESSED ) ) ) {
     160           0 :     rc = FD_STORE_SLOT_PREPARE_NEED_PARENT_EXEC;
     161             :     // FD_LOG_WARNING(("NEED PARENT EXEC %lu %lu", slot, parent_slot));
     162           0 :     if( FD_UNLIKELY( !fd_uchar_extract_bit( parent_block_map_entry->flags, FD_BLOCK_FLAG_REPLAYING ) ) ) {
     163             :       /* ... but it is not prepared */
     164           0 :       re_add_delays[re_adds_cnt] = (long)5e6;
     165           0 :       re_adds[re_adds_cnt++] = slot;
     166           0 :     }
     167           0 :     re_add_delays[re_adds_cnt] = (long)5e6;
     168           0 :     re_adds[re_adds_cnt++] = parent_slot;
     169           0 :     goto end;
     170           0 :   }
     171             : 
     172             :   /* The parent is executed, but the block is still incomplete. Ask for more shreds. */
     173           0 :   if( FD_UNLIKELY( !block ) ) {
     174           0 :     rc = FD_STORE_SLOT_PREPARE_NEED_REPAIR;
     175           0 :     *repair_slot_out = slot;
     176           0 :     re_add_delays[re_adds_cnt] = FD_REPAIR_BACKOFF_TIME;
     177           0 :     re_adds[re_adds_cnt++] = slot;
     178           0 :     goto end;
     179           0 :   }
     180             : 
     181             :   /* Prepare the replay_slot struct. */
     182           0 :   *block_out    = fd_blockstore_block_data_laddr( store->blockstore, block );
     183           0 :   *block_sz_out = block->data_sz;
     184             : 
     185             :   /* Mark the block as prepared, and thus unsafe to remove. */
     186           0 :   block_map_entry->flags = fd_uchar_set_bit( block_map_entry->flags, FD_BLOCK_FLAG_REPLAYING );
     187             : 
     188           0 : end:
     189             :   /* Block data ptr remains valid outside of the rw lock for the lifetime of the block alloc. */
     190           0 :   fd_blockstore_end_read( store->blockstore );
     191             : 
     192           0 :   for (uint i = 0; i < re_adds_cnt; ++i)
     193           0 :     fd_store_add_pending( store, re_adds[i], re_add_delays[i], 0, 0 );
     194             : 
     195           0 :   return rc;
     196           0 : }
     197             : 
     198             : int
     199             : fd_store_shred_insert( fd_store_t * store,
     200           0 :                        fd_shred_t const * shred ) {
     201             : 
     202           0 :   if( FD_UNLIKELY( shred->version != store->expected_shred_version ) ) {
     203           0 :     FD_LOG_WARNING(( "received shred version %lu instead of %lu", (ulong)shred->version, store->expected_shred_version ));
     204           0 :     return FD_BLOCKSTORE_OK;
     205           0 :   }
     206             : 
     207           0 :   fd_blockstore_t * blockstore = store->blockstore;
     208             : 
     209           0 :   if (shred->slot < blockstore->smr) {
     210           0 :     return FD_BLOCKSTORE_OK;
     211           0 :   }
     212           0 :   uchar shred_type = fd_shred_type( shred->variant );
     213             :   // FD_LOG_INFO(("is chained: %u", fd_shred_is_chained(shred_type) ));
     214           0 :   if( shred_type != FD_SHRED_TYPE_LEGACY_DATA
     215           0 :       && shred_type != FD_SHRED_TYPE_MERKLE_DATA
     216           0 :       && shred_type != FD_SHRED_TYPE_MERKLE_DATA_CHAINED
     217           0 :       && shred_type != FD_SHRED_TYPE_MERKLE_DATA_CHAINED_RESIGNED ) {
     218           0 :     return FD_BLOCKSTORE_OK;
     219           0 :   }
     220             : 
     221           0 :   if( store->root!=FD_SLOT_NULL && shred->slot<store->root ) {
     222           0 :     FD_LOG_WARNING(( "shred slot is behind root, dropping shred - root: %lu, shred_slot: %lu", store->root, shred->slot ));
     223           0 :     return FD_BLOCKSTORE_OK;
     224           0 :   }
     225             : 
     226           0 :   fd_blockstore_start_write( blockstore );
     227           0 :   if( fd_blockstore_block_query( blockstore, shred->slot ) != NULL ) {
     228           0 :     fd_blockstore_end_write( blockstore );
     229           0 :     return FD_BLOCKSTORE_OK;
     230           0 :   }
     231           0 :   int rc = fd_buf_shred_insert( blockstore, shred );
     232           0 :   fd_blockstore_end_write( blockstore );
     233             : 
     234             :   /* FIXME */
     235           0 :   if( FD_UNLIKELY( rc < FD_BLOCKSTORE_OK ) ) {
     236           0 :     FD_LOG_ERR( ( "failed to insert shred. reason: %d", rc ) );
     237           0 :   } else if ( rc == FD_BLOCKSTORE_OK_SLOT_COMPLETE ) {
     238           0 :     fd_store_add_pending( store, shred->slot, (long)5e6, 0, 1 );
     239           0 :   } else {
     240           0 :     fd_store_add_pending( store, shred->slot, FD_REPAIR_BACKOFF_TIME, 0, 0 );
     241           0 :     fd_repair_backoff_t * backoff = fd_repair_backoff_map_query( store->repair_backoff_map, shred->slot, NULL );
     242           0 :     if( FD_LIKELY( backoff==NULL ) ) {
     243             :       /* new backoff entry */
     244           0 :       backoff = fd_repair_backoff_map_insert( store->repair_backoff_map, shred->slot );
     245           0 :       backoff->last_backoff_duration = FD_REPAIR_BACKOFF_TIME;
     246           0 :       backoff->last_repair_time = store->now;
     247           0 :     } else if( ( backoff->last_repair_time+backoff->last_backoff_duration )
     248           0 :         >( store->now + FD_REPAIR_BACKOFF_TIME ) ) {
     249           0 :       backoff->last_backoff_duration = FD_REPAIR_BACKOFF_TIME;
     250           0 :       backoff->last_repair_time = store->now;
     251           0 :     }
     252           0 :   }
     253           0 :   return rc;
     254           0 : }
     255             : 
     256             : void
     257             : fd_store_shred_update_with_shred_from_turbine( fd_store_t * store,
     258           0 :                                                fd_shred_t const * shred ) {
     259           0 :   if( FD_UNLIKELY( store->first_turbine_slot == FD_SLOT_NULL ) ) {
     260           0 :     FD_LOG_NOTICE(("first turbine slot: %lu", shred->slot));
     261             :     // ulong slot = shred->slot;
     262             :     // while ( slot > store->snapshot_slot ) {
     263             :     //   fd_store_add_pending( store, slot, 0 );
     264             :     //   slot -= 10;
     265             :     // }
     266           0 :     store->first_turbine_slot = shred->slot;
     267           0 :     store->curr_turbine_slot = shred->slot;
     268           0 :   }
     269             : 
     270           0 :   store->curr_turbine_slot = fd_ulong_max(shred->slot, store->curr_turbine_slot);
     271           0 : }
     272             : 
     273             : void
     274             : fd_store_add_pending( fd_store_t * store,
     275             :                       ulong slot,
     276             :                       long delay,
     277             :                       int should_backoff,
     278           0 :                       int reset_backoff ) {
     279           0 :                         (void)should_backoff;
     280           0 :                         (void)reset_backoff;
     281             :   // fd_repair_backoff_t * backoff = fd_repair_backoff_map_query( store->repair_backoff_map, slot, NULL );
     282             :   // long existing_when = fd_pending_slots_get( store->pending_slots, slot );
     283             :   // if( existing_when!=0L && existing_when!=LONG_MAX ) {
     284             :   //   if( !should_backoff && delay > ( existing_when-store->now ) ) {
     285             :   //     return;
     286             :   //   }
     287             :   // }
     288             :   // // if( existing_when!=0L && existing_when!=LONG_MAX ) {
     289             :   // //   if( !should_backoff && delay < ( existing_when-store->now ) ) {
     290             :   // //     FD_LOG_WARNING(( "hey! %lu %ld %ld ", slot, delay, ( existing_when-store->now )));
     291             :   // //   } else {
     292             :   // //     FD_LOG_WARNING(( "eep %lu %lu %lu %d %lu", slot, delay/1000000, (existing_when - store->now)/1000000, should_backoff ));
     293             :   //     //  return;
     294             :   // //   }
     295             :   // // }
     296             :   // if( backoff==NULL ) {
     297             :   //   backoff = fd_repair_backoff_map_insert( store->repair_backoff_map, slot );
     298             :   //   backoff->slot = slot;
     299             :   //   backoff->last_backoff = delay;
     300             :   // } else if( reset_backoff ) {
     301             :   //   backoff->last_backoff = delay;
     302             :   // } else if( should_backoff ) {
     303             :   //   ulong backoff->last_backoff + (backoff->last_backoff>>3);
     304             :   //   backoff->last_backoff =
     305             :   //   delay = backoff->last_backoff;
     306             :   // } else {
     307             :   //   delay = backoff->last_backoff;
     308             :   // }
     309             :   // if( should_backoff ) FD_LOG_INFO(("PENDING %lu %d %lu %ld", slot, should_backoff, delay/1000000, (existing_when-store->now)/1000000L));
     310           0 :   if( store->root!=FD_SLOT_NULL && slot<store->root) {
     311           0 :     FD_LOG_WARNING(( "slot is older than root, skipping adding slot to pending queue - root: %lu, slot: %lu",
     312           0 :         store->root, slot ));
     313           0 :     return;
     314           0 :   }
     315           0 :   fd_pending_slots_add( store->pending_slots, slot, store->now + (long)delay );
     316           0 : }
     317             : 
     318             : void
     319             : fd_store_set_root( fd_store_t * store,
     320           0 :                    ulong        root ) {
     321           0 :   store->root = root;
     322           0 :   fd_pending_slots_set_lo_wmark( store->pending_slots, root );
     323             : 
     324             :   /* remove old roots */
     325           0 :   for( ulong i = 0; i<fd_repair_backoff_map_slot_cnt(); i++ ) {
     326           0 :     if( store->repair_backoff_map[ i ].slot <= root ) {
     327           0 :       fd_repair_backoff_map_remove( store->repair_backoff_map, &store->repair_backoff_map[ i ] );
     328           0 :     }
     329           0 :   }
     330           0 : }
     331             : 
     332             : ulong
     333             : fd_store_slot_repair( fd_store_t * store,
     334             :                       ulong slot,
     335             :                       fd_repair_request_t * out_repair_reqs,
     336           0 :                       ulong out_repair_reqs_sz ) {
     337           0 :   if( out_repair_reqs_sz==0UL ) {
     338           0 :     return 0UL;
     339           0 :   }
     340             : 
     341           0 :   fd_repair_backoff_t * backoff = fd_repair_backoff_map_query( store->repair_backoff_map, slot, NULL );
     342           0 :   if( FD_LIKELY( backoff!=NULL ) ) {
     343           0 :     if( store->now<( backoff->last_repair_time+backoff->last_backoff_duration ) ) {
     344           0 :       return 0UL;
     345           0 :     }
     346           0 :   } else {
     347             :     /* new backoff entry */
     348           0 :     backoff = fd_repair_backoff_map_insert( store->repair_backoff_map, slot );
     349           0 :     backoff->last_backoff_duration = FD_REPAIR_BACKOFF_TIME;
     350           0 :   }
     351           0 :   backoff->last_repair_time = store->now;
     352             : 
     353           0 :   ulong repair_req_cnt = 0;
     354           0 :   fd_block_map_t * block_map_entry = fd_blockstore_block_map_query( store->blockstore, slot );
     355             : 
     356           0 :   if( FD_LIKELY( !block_map_entry ) ) {
     357             :     /* We haven't received any shreds for this slot yet */
     358             : 
     359           0 :     fd_repair_request_t * repair_req = &out_repair_reqs[repair_req_cnt++];
     360           0 :     repair_req->shred_index = 0;
     361           0 :     repair_req->slot = slot;
     362           0 :     repair_req->type = FD_REPAIR_REQ_TYPE_NEED_HIGHEST_WINDOW_INDEX;
     363           0 :   } else {
     364             :     /* We've received at least one shred, so fill in what's missing */
     365             : 
     366           0 :     uint complete_idx = block_map_entry->complete_idx;
     367             : 
     368             :     /* We don't know the last index yet */
     369           0 :     if( FD_UNLIKELY( complete_idx == UINT_MAX ) ) {
     370           0 :       complete_idx = block_map_entry->received_idx - 1;
     371           0 :       fd_repair_request_t * repair_req = &out_repair_reqs[repair_req_cnt++];
     372           0 :       repair_req->shred_index = complete_idx;
     373           0 :       repair_req->slot = slot;
     374           0 :       repair_req->type = FD_REPAIR_REQ_TYPE_NEED_HIGHEST_WINDOW_INDEX;
     375           0 :     }
     376             : 
     377           0 :     if( repair_req_cnt==out_repair_reqs_sz ) {
     378           0 :       backoff->last_backoff_duration += backoff->last_backoff_duration>>2;
     379           0 :       FD_LOG_INFO( ( "[repair] MAX need %lu [%u, %u], sent %lu requests (backoff: %ld ms)", slot, block_map_entry->consumed_idx + 1, complete_idx, repair_req_cnt, backoff->last_backoff_duration/(long)1e6 ) );
     380           0 :       return repair_req_cnt;
     381           0 :     }
     382             : 
     383             :     /* First make sure we are ready to execute this block soon. Look for an ancestor that was executed. */
     384           0 :     ulong anc_slot = slot;
     385           0 :     int good = 0;
     386           0 :     for( uint i = 0; i < 6; ++i ) {
     387           0 :       anc_slot  = fd_blockstore_parent_slot_query( store->blockstore, anc_slot );
     388           0 :       fd_block_t * anc_block = fd_blockstore_block_query( store->blockstore, anc_slot );
     389           0 :       fd_block_map_t * anc_block_map_entry = fd_blockstore_block_map_query( store->blockstore, anc_slot );
     390           0 :       if( anc_block && fd_uchar_extract_bit( anc_block_map_entry->flags, FD_BLOCK_FLAG_PROCESSED ) ) {
     391           0 :         good = 1;
     392           0 :         out_repair_reqs_sz /= (i>>1)+1U; /* Slow roll blocks that are further out */
     393           0 :         break;
     394           0 :       }
     395           0 :     }
     396             : 
     397           0 :     if( !good ) {
     398           0 :       return repair_req_cnt;
     399           0 :     }
     400             : 
     401             :     /* Fill in what's missing */
     402           0 :     for( uint i = block_map_entry->consumed_idx + 1; i <= complete_idx; i++ ) {
     403           0 :       if( fd_buf_shred_query( store->blockstore, slot, i ) != NULL ) continue;
     404             : 
     405           0 :       fd_repair_request_t * repair_req = &out_repair_reqs[repair_req_cnt++];
     406           0 :       repair_req->shred_index = i;
     407           0 :       repair_req->slot = slot;
     408           0 :       repair_req->type = FD_REPAIR_REQ_TYPE_NEED_WINDOW_INDEX;
     409             : 
     410           0 :       if( repair_req_cnt == out_repair_reqs_sz ) {
     411           0 :         backoff->last_backoff_duration += backoff->last_backoff_duration>>2;
     412           0 :         FD_LOG_INFO( ( "[repair] MAX need %lu [%u, %u], sent %lu requests (backoff: %ld ms)", slot, block_map_entry->consumed_idx + 1, complete_idx, repair_req_cnt, backoff->last_backoff_duration/(long)1e6 ) );
     413           0 :         return repair_req_cnt;
     414           0 :       }
     415           0 :     }
     416           0 :     if( repair_req_cnt ) {
     417           0 :       backoff->last_backoff_duration += backoff->last_backoff_duration>>2;
     418           0 :       FD_LOG_INFO( ( "[repair] need %lu [%u, %u], sent %lu requests (backoff: %ld ms)", slot, block_map_entry->consumed_idx + 1, complete_idx, repair_req_cnt, backoff->last_backoff_duration/(long)1e6 ) );
     419           0 :     }
     420           0 :   }
     421           0 :   return repair_req_cnt;
     422           0 : }