Revert "repair: queue-based repair replacement for tree-traversal"

This reverts commit fbff822.

Author: lidatong

src/discof/forest/fd_forest.c

@@ -51,7 +51,7 @@ fd_inflights_join( void * shmem ) {
 }
 
 void
-fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t const * pubkey, ulong slot, ulong shred_idx ) {
+fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t const * pubkey ) {
   if( FD_UNLIKELY( !fd_inflight_pool_free( table->pool ) ) ) {
     fd_inflight_t * evict = fd_inflight_dlist_ele_pop_head( table->dlist, table->pool );
     fd_inflight_map_ele_remove( table->map, &evict->nonce, NULL, table->pool );
@@ -62,10 +62,8 @@ fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t co
   inflight_req->nonce        = nonce;
   inflight_req->timestamp_ns = fd_log_wallclock();
   inflight_req->pubkey       = *pubkey;
-  inflight_req->slot         = slot;
-  inflight_req->shred_idx    = shred_idx;
 
-  fd_inflight_map_ele_insert     ( table->map,   inflight_req, table->pool );
+  fd_inflight_map_ele_insert( table->map, inflight_req, table->pool );
   fd_inflight_dlist_ele_push_tail( table->dlist, inflight_req, table->pool );
 }
 
@@ -86,16 +84,6 @@ fd_inflights_request_remove( fd_inflights_t * table, ulong nonce, fd_pubkey_t *
   return 0;
 }
 
-void
-fd_inflights_request_pop( fd_inflights_t * table, ulong * nonce_out, ulong * slot_out, ulong * shred_idx_out ) {
-  fd_inflight_t * inflight_req = fd_inflight_dlist_ele_pop_head( table->dlist, table->pool );
-  fd_inflight_map_ele_remove( table->map, &inflight_req->nonce, NULL, table->pool );
-  *nonce_out     = inflight_req->nonce;
-  *slot_out      = inflight_req->slot;
-  *shred_idx_out = inflight_req->shred_idx;
-  fd_inflight_pool_ele_release( table->pool, inflight_req );
-}
-
 fd_inflight_t *
 fd_inflights_request_query( fd_inflights_t * table, ulong nonce ) {
   return fd_inflight_map_ele_query( table->map, &nonce, NULL, table->pool );

src/discof/forest/fd_forest.h

@@ -21,9 +21,6 @@ struct __attribute__((aligned(128UL))) fd_inflight {
   long          timestamp_ns;  /* timestamp when request was created (nanoseconds) */
   fd_pubkey_t   pubkey;        /* public key of the peer */
 
-  ulong         slot;          /* slot of the request */
-  ulong         shred_idx;     /* shred index of the request */
-
   /* Reserved for DLL eviction */
   ulong          prevll;      /* pool index of previous element in DLL */
   ulong          nextll;      /* pool index of next element in DLL */
@@ -77,30 +74,11 @@ fd_inflights_t *
 fd_inflights_join( void * shmem );
 
 void
-fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t const * pubkey, ulong slot, ulong shred_idx );
+fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t const * pubkey );
 
 long
 fd_inflights_request_remove( fd_inflights_t * table, ulong nonce, fd_pubkey_t * peer_out );
 
-/* Important! Caller must guarantee that the request list is not empty.
-   This function cannot fail and will always try to populate the output
-   parameters. Typical use should only call this after
-   fd_inflights_should_drain returns true. */
-
-void
-fd_inflights_request_pop( fd_inflights_t * table, ulong * nonce_out, ulong * slot_out, ulong * shred_idx_out );
-
-static inline int
-fd_inflights_should_drain( fd_inflights_t * table, long now ) {
-  /* peek at head */
-  if( FD_UNLIKELY( fd_inflight_dlist_is_empty( table->dlist, table->pool ) ) ) return 0;
-
-  fd_inflight_t * inflight_req = fd_inflight_dlist_ele_peek_head( table->dlist, table->pool );
-  if( FD_UNLIKELY( inflight_req->timestamp_ns + 100e6L < now ) ) return 1;
-  return 0;
-}
-
-
 fd_inflight_t *
 fd_inflights_request_query ( fd_inflights_t * table, ulong nonce );
 

src/discof/forest/test_forest.c

@@ -41,6 +41,7 @@ fd_policy_new( void * shmem, ulong dedup_max, ulong peer_max, ulong seed ) {
   policy->peers.pool    = fd_peer_pool_new        ( peers_pool, peer_max        );
   policy->peers.fast    = fd_peer_dlist_new       ( peers_fast                  );
   policy->peers.slow    = fd_peer_dlist_new       ( peers_slow                  );
+  policy->iterf.ele_idx = ULONG_MAX;
   policy->turbine_slot0 = ULONG_MAX;
   policy->tsreset       = 0;
   policy->nonce         = 1;
@@ -201,13 +202,18 @@ fd_policy_next( fd_policy_t * policy, fd_forest_t * forest, fd_repair_t * repair
     }
   }
 
-  /**********************/
-  /* ADVANCE ITERATOR   */
-  /**********************/
+  /* Every so often we'll need to reset the frontier iterator to the
+     head of frontier, because we could end up traversing down a very
+     long tree if we are far behind. */
 
-  fd_forest_iter_next( forest );
-  if( FD_UNLIKELY( fd_forest_iter_done( forest ) ) ) {
-    // This happens when we have already requested all the shreds we know about.
+  if( FD_UNLIKELY( now_ms - policy->tsreset > 100UL /* ms */ ||
+                   policy->iterf.frontier_ver != fd_fseq_query( fd_forest_ver_const( forest ) ) ) ) {
+    fd_policy_reset( policy, forest );
+  }
+
+  fd_forest_blk_t * ele = fd_forest_pool_ele( pool, policy->iterf.ele_idx );
+  if( FD_UNLIKELY( !ele ) ) {
+    // This happens when we are fully caught up i.e. we have all the shreds of every slot we know about.
     return NULL;
   }
 
@@ -225,34 +231,53 @@ fd_policy_next( fd_policy_t * policy, fd_forest_t * forest, fd_repair_t * repair
      next valid requestable element. */
 
   int req_made = 0;
+  while( !req_made ) {
+    ele = fd_forest_pool_ele( pool, policy->iterf.ele_idx );
+
+    if( FD_UNLIKELY( !passes_throttle_threshold( policy, ele ) ) ) {
+      /* We are not ready to repair this slot yet.  But it's possible we
+         have another fork that we need to repair... so we just
+         should skip to the next SLOT in the consumed iterator.  The
+         likelihood that this ele is the head of turbine is high, which
+         means that the shred_idx of the iterf is likely to be UINT_MAX,
+         which means calling fd_forest_iter_next will advance the iterf
+         to the next slot. */
+      policy->iterf.shred_idx = UINT_MAX; // heinous... i'm sorry
+      policy->iterf = fd_forest_iter_next( policy->iterf, forest );
+      if( FD_UNLIKELY( fd_forest_iter_done( policy->iterf, forest ) ) ) {
+         policy->iterf = fd_forest_iter_init( forest );
+         break;
+      }
+      continue;
+    }
 
-  fd_forest_blk_t * ele = fd_forest_pool_ele( pool, forest->iter.ele_idx );
-  if( FD_UNLIKELY( !passes_throttle_threshold( policy, ele ) ) ) {
-    /* We are not ready to repair this slot yet.  But it's possible we
-       have another fork that we need to repair... so we just
-       should skip to the next SLOT in the consumed iterator.  The
-       likelihood that this ele is the head of turbine is high, which
-       means that the shred_idx of the iterf is likely to be UINT_MAX,
-       which means calling fd_forest_iter_next will advance the iterf
-       to the next slot. */
-    //forest->iter.shred_idx = UINT_MAX; // heinous... i'm sorry
-    //fd_forest_iter_next( forest );
-    //if( FD_UNLIKELY( fd_forest_iter_done( forest->iter, forest ) ) ) break;
-    //continue;
-  }
+    if( FD_UNLIKELY( policy->iterf.shred_idx == UINT_MAX ) ) {
+      ulong key = fd_policy_dedup_key( FD_REPAIR_KIND_HIGHEST_SHRED, ele->slot, 0 );
+      if( FD_UNLIKELY( ele->slot < highest_known_slot && !dedup_next( policy, key, now ) ) ) {
+        // We'll never know the the highest shred for the current turbine slot, so there's no point in requesting it.
+        out = fd_repair_highest_shred( repair, fd_policy_peer_select( policy ), now_ms, policy->nonce, ele->slot, 0 );
+        policy->nonce++;
+        req_made = 1;
+      }
+    } else {
+      ulong key = fd_policy_dedup_key( FD_REPAIR_KIND_SHRED, ele->slot, policy->iterf.shred_idx );
+      if( FD_UNLIKELY( !dedup_next( policy, key, now ) ) ) {
+        out = fd_repair_shred( repair, fd_policy_peer_select( policy ), now_ms, policy->nonce, ele->slot, policy->iterf.shred_idx );
+        policy->nonce++;
+        if( FD_UNLIKELY( ele->first_req_ts == 0 ) ) ele->first_req_ts = fd_tickcount();
+        req_made = 1;
+      }
+    }
 
-  if( FD_UNLIKELY( forest->iter.shred_idx == UINT_MAX ) ) {
-    if( FD_UNLIKELY( ele->slot < highest_known_slot ) ) {
-      // We'll never know the the highest shred for the current turbine slot, so there's no point in requesting it.
-      out = fd_repair_highest_shred( repair, fd_policy_peer_select( policy ), now_ms, policy->nonce, ele->slot, 0 );
-      policy->nonce++;
-      req_made = 1;
+    /* Even if we have a request ready, we need to advance the iterator.
+       Otherwise on the next call of policy_next, we'll try to re-request the
+       same shred and it will get deduped. */
+
+    policy->iterf = fd_forest_iter_next( policy->iterf, forest );
+    if( FD_UNLIKELY( fd_forest_iter_done( policy->iterf, forest ) ) ) {
+      policy->iterf = fd_forest_iter_init( forest );
+      break;
     }
-  } else {
-    out = fd_repair_shred( repair, fd_policy_peer_select( policy ), now_ms, policy->nonce, ele->slot, forest->iter.shred_idx );
-    policy->nonce++;
-    if( FD_UNLIKELY( ele->first_req_ts == 0 ) ) ele->first_req_ts = fd_tickcount();
-    req_made = 1;
   }
 
   if( FD_UNLIKELY( !req_made ) ) return NULL;
@@ -340,6 +365,12 @@ fd_policy_peer_response_update( fd_policy_t * policy, fd_pubkey_t const * to, lo
   }
 }
 
+void
+fd_policy_reset( fd_policy_t * policy, fd_forest_t * forest ) {
+  policy->iterf   = fd_forest_iter_init( forest );
+  policy->tsreset = ts_ms( fd_log_wallclock() );
+}
+
 void
 fd_policy_set_turbine_slot0( fd_policy_t * policy, ulong slot ) {
   policy->turbine_slot0 = slot;

src/discof/repair/fd_inflight.c

@@ -180,6 +180,7 @@ struct fd_policy {
   long              tsmax; /* maximum time for an iteration before resetting the DFS to root */
   long              tsref; /* reference timestamp for resetting DFS */
 
+  fd_forest_iter_t  iterf; /* forest iterator */
   ulong             tsreset; /* ms timestamp of last reset of iterf */
 
   ulong turbine_slot0;
@@ -286,4 +287,7 @@ fd_policy_peer_response_update( fd_policy_t * policy, fd_pubkey_t const * to, lo
 void
 fd_policy_set_turbine_slot0( fd_policy_t * policy, ulong slot );
 
+void
+fd_policy_reset( fd_policy_t * policy, fd_forest_t * forest );
+
 #endif /* HEADER_fd_src_discof_repair_fd_policy_h */

src/discof/repair/fd_inflight.h

@@ -310,6 +310,7 @@ struct ctx {
 
   /* Slot-level metrics */
   fd_repair_metrics_t * slot_metrics;
+
   ulong turbine_slot0;  // catchup considered complete after this slot
 };
 typedef struct ctx ctx_t;
@@ -619,8 +620,8 @@ after_sign( ctx_t             * ctx,
   }
 
   int is_regular_request = pending->msg.kind != FD_REPAIR_KIND_PONG && pending->msg.shred.nonce > 0;
-  if( FD_LIKELY( is_regular_request && pending->msg.kind == FD_REPAIR_KIND_SHRED ) ) {
-    fd_inflights_request_insert( ctx->inflight, pending->msg.shred.nonce, &pending->msg.shred.to, pending->msg.shred.slot, pending->msg.shred.shred_idx );
+  if( FD_LIKELY( is_regular_request ) ) {
+    fd_inflights_request_insert( ctx->inflight, pending->msg.shred.nonce, &pending->msg.shred.to );
     fd_policy_peer_request_update( ctx->policy, &pending->msg.shred.to );
   }
   send_packet( ctx, stem, 1, active->ip4, active->port, src_ip4, pending->buf, pending->buflen, fd_frag_meta_ts_comp( fd_tickcount() ) );
@@ -649,6 +650,12 @@ after_shred( ctx_t      * ctx,
     int ref_tick      = shred->data.flags & FD_SHRED_DATA_REF_TICK_MASK;
     fd_forest_blk_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off );
     fd_forest_data_shred_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off, shred->idx, shred->fec_set_idx, slot_complete, ref_tick, src );
+
+    /* Check if there are FECs to force complete. Algorithm: window
+       through the idxs in interval [i, j). If j = next fec_set_idx
+       then we know we can force complete the FEC set interval [i, j)
+       (assuming it wasn't already completed based on `cmpl`). */
+
   } else {
     fd_forest_code_shred_insert( ctx->forest, shred->slot, shred->idx );
   }
@@ -744,6 +751,7 @@ after_frag( ctx_t * ctx,
   uint in_kind = ctx->in_kind[ in_idx ];
   if( FD_UNLIKELY( in_kind==IN_KIND_GENESIS ) ) {
     fd_forest_init( ctx->forest, 0 );
+    fd_policy_reset( ctx->policy, ctx->forest );
     return;
   }
 
@@ -759,7 +767,10 @@ after_frag( ctx_t * ctx,
 
   if( FD_UNLIKELY( in_kind==IN_KIND_TOWER ) ) {
     fd_tower_slot_done_t const * msg = (fd_tower_slot_done_t const *)fd_type_pun_const( ctx->buffer );
-    if( FD_LIKELY( msg->new_root ) ) fd_forest_publish( ctx->forest, msg->root_slot );
+    if( FD_LIKELY( msg->new_root ) ) {
+      fd_forest_publish( ctx->forest, msg->root_slot );
+      fd_policy_reset  ( ctx->policy, ctx->forest );
+    }
     return;
   }
 
@@ -849,7 +860,19 @@ after_frag( ctx_t * ctx,
         }
       }
     }
-    /* update metrics */
+
+    ulong max_repaired_slot = 0;
+    fd_forest_conslist_t const * conslist = fd_forest_conslist_const( ctx->forest );
+    fd_forest_cns_t const *      conspool = fd_forest_conspool_const( ctx->forest );
+    fd_forest_blk_t const *      pool     = fd_forest_pool_const( ctx->forest );
+    for( fd_forest_conslist_iter_t iter = fd_forest_conslist_iter_fwd_init( conslist, conspool );
+         !fd_forest_conslist_iter_done( iter, conslist, conspool );
+         iter = fd_forest_conslist_iter_fwd_next( iter, conslist, conspool ) ) {
+      fd_forest_cns_t const * ele = fd_forest_conslist_iter_ele_const( iter, conslist, conspool );
+      fd_forest_blk_t const * ele_ = fd_forest_pool_ele_const( pool, ele->forest_pool_idx );
+      if( ele_->slot > max_repaired_slot ) max_repaired_slot = ele_->slot;
+    }
+    ctx->metrics->repaired_slots = max_repaired_slot;
     return;
   }
 
@@ -875,6 +898,7 @@ after_credit( ctx_t *             ctx,
               int *               opt_poll_in FD_PARAM_UNUSED,
               int *               charge_busy ) {
   long now = fd_log_wallclock();
+
   *charge_busy = 1;
 
   /* Verify that there is at least one sign tile with available credits.
@@ -890,20 +914,6 @@ after_credit( ctx_t *             ctx,
     return;
   }
 
-  /* TODO make sure not to re-request inflights for stale slots */
-
-  if( FD_UNLIKELY( fd_inflights_should_drain( ctx->inflight, now ) ) ) {
-    ulong nonce; ulong slot; ulong shred_idx;
-    fd_inflights_request_pop( ctx->inflight, &nonce, &slot, &shred_idx );
-    fd_forest_blk_t * blk = fd_forest_query( ctx->forest, slot );
-    if( FD_UNLIKELY( !fd_forest_blk_idxs_test( blk->idxs, shred_idx ) ) ) {
-      fd_pubkey_t const * peer = fd_policy_peer_select( ctx->policy );
-      fd_repair_msg_t   * msg  = fd_repair_shred( ctx->protocol, peer, (ulong)((ulong)now / 1e6L), (uint)nonce, slot, shred_idx );
-      fd_repair_send_sign_request( ctx, sign_out, msg, NULL );
-      return;
-    }
-  }
-
   fd_repair_msg_t const * cout = fd_policy_next( ctx->policy, ctx->forest, ctx->protocol, now, ctx->metrics->current_slot );
   if( FD_UNLIKELY( !cout ) ) return;
 
@@ -1115,8 +1125,6 @@ populate_allowed_fds( fd_topo_t const *      topo FD_PARAM_UNUSED,
 
 static inline void
 metrics_write( ctx_t * ctx ) {
-  ctx->metrics->repaired_slots = fd_forest_highest_repaired_slot( ctx->forest );
-
   FD_MCNT_SET( REPAIR, CURRENT_SLOT,      ctx->metrics->current_slot );
   FD_MCNT_SET( REPAIR, REPAIRED_SLOTS,    ctx->metrics->repaired_slots );
   FD_MCNT_SET( REPAIR, REQUEST_PEERS,     fd_peer_pool_used( ctx->policy->peers.pool ) );