repair: queue-based repair replacement for tree-traversal

* handling tricky edge cases with dropping peers, and being at head of turbine, invalidating iterator on publish

Author: emwang-jump

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 ) {
+fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t const * pubkey, ulong slot, ulong shred_idx ) {
   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,8 +62,10 @@ 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 );
 }
 
@@ -84,6 +86,16 @@ 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

@@ -4,13 +4,18 @@
 #include "../../flamenco/types/fd_types.h"
 
 /* fd_inflights tracks repair requests that are inflight to other
-   validators.  This module is not necessary for the repair protocol and
-   strategy, but is useful for metrics and reporting.  Incorrect updates
-   and removals from this module are non-critical.  Requests are key-ed
-   by nonce as in the current strategy (see fd_policy.h), all requests
-   have a unique nonce.  The chances that an inflight request does not
-   get a response are non-negligible due to shred tile upstream deduping
-   duplicates. */
+   validators.  This module is useful for metrics and reporting.
+   In-exact updates of orphan requests and highest window requests from
+   this module are non-critical, but exact updates of shred requests are
+   critical. Repair tile relies on this module to be able to re-request
+   any shreds that it has sent, because policy next does not request any
+   shred twice.
+   (TODO should this be rolled into policy.h?)
+
+   Requests are key-ed by nonce as in the current strategy (see
+   fd_policy.h), all requests have a unique nonce.  The chances that an
+   inflight request does not get a response are non-negligible due to
+   shred tile upstream deduping duplicates. */
 
 /* Max number of pending requests */
 #define FD_INFLIGHT_REQ_MAX (1<<20)
@@ -21,6 +26,9 @@ 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 */
@@ -74,11 +82,30 @@ fd_inflights_t *
 fd_inflights_join( void * shmem );
 
 void
-fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t const * pubkey );
+fd_inflights_request_insert( fd_inflights_t * table, ulong nonce, fd_pubkey_t const * pubkey, ulong slot, ulong shred_idx );
 
 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 + 90e6L < 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,7 +41,6 @@ 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;
@@ -165,6 +164,8 @@ fd_policy_peer_select( fd_policy_t * policy ) {
   fd_peer_dlist_t * worst_dlist = policy->peers.slow;
   fd_peer_t       * pool        = policy->peers.pool;
 
+  if( FD_UNLIKELY( fd_peer_pool_used( policy->peers.pool ) == 0 ) ) return NULL;
+
   fd_peer_dlist_t * dlist = bucket_stages[policy->peers.select.stage] == FD_POLICY_LATENCY_FAST ? best_dlist : worst_dlist;
 
   while( FD_UNLIKELY( fd_peer_dlist_iter_done( policy->peers.select.iter, dlist, pool ) ) ) {
@@ -202,18 +203,13 @@ fd_policy_next( fd_policy_t * policy, fd_forest_t * forest, fd_repair_t * repair
     }
   }
 
-  /* 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. */
+  /**********************/
+  /* ADVANCE ITERATOR   */
+  /**********************/
 
-  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.
+  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.
     return NULL;
   }
 
@@ -231,53 +227,34 @@ 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;
-    }
 
-    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;
-      }
-    }
-
-    /* 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. */
+  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;
+    /* TODO: Heinous... I'm sorry. Easiest way to ensure this slot gets added back to the requests deque.
+       but maybe there should be an explicit API for it. */
+    return NULL;
+  }
 
-    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;
+  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;
     }
+  } 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;
@@ -325,7 +302,7 @@ fd_policy_peer_remove( fd_policy_t * policy, fd_pubkey_t const * key ) {
 
   if( FD_UNLIKELY( policy->peers.select.iter == fd_peer_pool_idx( policy->peers.pool, peer_ele ) ) ) {
     /* In general removal during iteration is safe, except when the iterator is on the peer to be removed. */
-    fd_peer_dlist_t * dlist = policy->peers.select.stage == FD_POLICY_LATENCY_FAST ? policy->peers.fast : policy->peers.slow;
+    fd_peer_dlist_t * dlist = bucket_stages[policy->peers.select.stage] == FD_POLICY_LATENCY_FAST ? policy->peers.fast : policy->peers.slow;
     policy->peers.select.iter = fd_peer_dlist_iter_fwd_next( policy->peers.select.iter, dlist, policy->peers.pool );
   }
 
@@ -365,12 +342,6 @@ 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,7 +180,6 @@ 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;
@@ -287,7 +286,4 @@ 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 */
+#endif /* HEADER_fd_src_choreo_policy_fd_policy_h */

src/discof/repair/fd_inflight.h

@@ -18,7 +18,7 @@ struct fd_slot_metrics {
 };
 typedef struct fd_slot_metrics fd_slot_metrics_t;
 
-#define FD_CATCHUP_METRICS_MAX 256
+#define FD_CATCHUP_METRICS_MAX 16384
 
 struct fd_repair_metrics_t {
   fd_slot_metrics_t slots[ FD_CATCHUP_METRICS_MAX ];