flamenco, gossip: two-tiered value table for fast iteration in push and pullresp loops

Author: ravyu-jump

book/api/metrics-generated.md

@@ -376,6 +376,8 @@
 | gossip_​peer_​counts_​repair | `gauge` | Number of peers of each type (Repair) |
 | gossip_​peer_​counts_​voter | `gauge` | Number of peers of each type (Voter) |
 | gossip_​shred_​version_​zero | `counter` | Shred version zero |
+| gossip_​value_​map_​size | `gauge` | Current size of the CRDS value map |
+| gossip_​value_​vec_​size | `gauge` | Current size of the CRDS value vector |
 | gossip_​received_​packets | `counter` | Number of all gossip packets received |
 | gossip_​corrupted_​messages | `counter` | Number of corrupted gossip messages received |
 | gossip_​received_​gossip_​messages_​pull_​request | `counter` | Number of gossip messages received (Pull Request) |

src/disco/metrics/generated/fd_metrics_gossip.c

@@ -14,6 +14,8 @@ const fd_metrics_meta_t FD_METRICS_GOSSIP[FD_METRICS_GOSSIP_TOTAL] = {
     DECLARE_METRIC_ENUM( GOSSIP_PEER_COUNTS, GAUGE, PEER_TYPES, REPAIR ),
     DECLARE_METRIC_ENUM( GOSSIP_PEER_COUNTS, GAUGE, PEER_TYPES, VOTER ),
     DECLARE_METRIC( GOSSIP_SHRED_VERSION_ZERO, COUNTER ),
+    DECLARE_METRIC( GOSSIP_VALUE_MAP_SIZE, GAUGE ),
+    DECLARE_METRIC( GOSSIP_VALUE_VEC_SIZE, GAUGE ),
     DECLARE_METRIC( GOSSIP_RECEIVED_PACKETS, COUNTER ),
     DECLARE_METRIC( GOSSIP_CORRUPTED_MESSAGES, COUNTER ),
     DECLARE_METRIC_ENUM( GOSSIP_RECEIVED_GOSSIP_MESSAGES, COUNTER, GOSSIP_MESSAGE, PULL_REQUEST ),

src/disco/metrics/generated/fd_metrics_gossip.h

@@ -637,6 +637,9 @@ metric introduced.
     <gauge   name="PeerCounts"                                      enum="PeerTypes"            clickhouse_exclude="true" summary="Number of peers of each type" />
     <counter name="ShredVersionZero"                                                            clickhouse_exclude="true" summary="Shred version zero" />
 
+    <gauge   name="ValueMapSize"                                                                 clickhouse_exclude="true" summary="Current size of the CRDS value map" />
+    <gauge   name="ValueVecSize"                                                             clickhouse_exclude="true" summary="Current size of the CRDS value vector" />
+
     <counter name="ReceivedPackets"                                 clickhouse_exclude="true" summary="Number of all gossip packets received" />
     <counter name="CorruptedMessages"                               clickhouse_exclude="true" summary="Number of corrupted gossip messages received" />
     <counter name="ReceivedGossipMessages" enum="GossipMessage"     clickhouse_exclude="true" summary="Number of gossip messages received" />

src/disco/metrics/metrics.xml

@@ -1072,6 +1072,9 @@ fd_gossip_update_gossip_metrics( fd_gossip_metrics_t * metrics ) {
   FD_MCNT_ENUM_COPY( GOSSIP, PUSH_CRDS_DROP, metrics->push_crds_drop_reason );
   FD_MGAUGE_SET( GOSSIP, PUSH_CRDS_QUEUE_COUNT, metrics->push_crds_queue_cnt );
 
+  FD_MGAUGE_SET( GOSSIP, VALUE_MAP_SIZE, metrics->value_map_cnt );
+  FD_MGAUGE_SET( GOSSIP, VALUE_VEC_SIZE, metrics->value_vec_cnt );
+
   FD_MGAUGE_SET( GOSSIP, ACTIVE_PUSH_DESTINATIONS, metrics->active_push_destinations );
   FD_MCNT_SET( GOSSIP, REFRESH_PUSH_STATES_FAIL_COUNT, metrics->refresh_push_states_failcnt );
 

src/discof/gossip/fd_gossip_tile.c

@@ -176,6 +176,10 @@ struct fd_gossip_metrics {
   ulong push_crds_drop_reason[FD_METRICS_COUNTER_GOSSIP_PUSH_CRDS_DROP_CNT];
   ulong push_crds_queue_cnt;
 
+  /* Value DS sizes */
+  ulong value_map_cnt;
+  ulong value_vec_cnt;
+
   /* Active Push Destinations */
   ulong active_push_destinations;
   ulong refresh_push_states_failcnt;

src/flamenco/gossip/fd_gossip.c

@@ -161,7 +161,7 @@ VEC_(new)( void * shmem,
   return shmem;
 }
 
-FD_FN_CONST static inline VEC_T *
+static inline VEC_T *
 VEC_(join)( void * shvec ) {
 
   if( FD_UNLIKELY( !shvec ) ) return NULL;
@@ -179,7 +179,7 @@ VEC_(leave)( VEC_T * join ) {
   return (void *)(((ulong)join) - VEC_(private_meta_footprint)());
 }
 
-FD_FN_CONST static inline void *
+static inline void *
 VEC_(delete)( void * shvec ) {
 
   if( FD_UNLIKELY( !shvec ) ) return NULL;
@@ -223,8 +223,8 @@ VEC_(contract)( VEC_T * join,
 }
 
 static inline VEC_T *
-VEC_(remove)( VEC_T * join,
-              ulong   idx ) {
+VEC_(remove_idx)( VEC_T * join,
+                  ulong   idx ) {
   VEC_(private_t) * vec = VEC_(private)( join );
   ulong cnt = vec->cnt - 1UL;
   join[idx] = join[cnt]; /* TODO: Consider letting user decide if self copy is cheaper than testing */
@@ -233,8 +233,14 @@ VEC_(remove)( VEC_T * join,
 }
 
 static inline VEC_T *
-VEC_(remove_compact)( VEC_T * join,
-                      ulong   idx ) {
+VEC_(remove_ele)( VEC_T * join,
+                  VEC_T * ele){
+  return VEC_(remove_idx)( join, (ulong)(ele-join) );
+}
+
+static inline VEC_T *
+VEC_(remove_compact_idx)( VEC_T * join,
+                          ulong   idx ) {
   VEC_(private_t) * vec = VEC_(private)( join );
   ulong cnt = vec->cnt - 1UL;
   for( ; idx<cnt; idx++ ) join[idx] = join[idx+1UL];

src/flamenco/gossip/fd_gossip.h

@@ -52,7 +52,7 @@ main( int     argc,
     FD_TEST( !memcmp( vec, ref, cnt*sizeof(MYVEC_T) ) );
 
     uint r  = fd_rng_uint( rng );
-    int  op = (int)(r & 3U); r>>=2;
+    int  op = (int)(r & 7U); r>>=3;
 
     switch( op ) {
     default:
@@ -78,19 +78,28 @@ main( int     argc,
       break;
     }
 
-    case 2: { /* remove with backfill */
+    case 2: { /* remove with backfill via idx */
       if( !cnt ) break;
       ulong idx = fd_rng_ulong_roll( rng, cnt );
-      FD_TEST( myvec_remove( vec, idx )==vec );
+      FD_TEST( myvec_remove_idx( vec, idx )==vec );
       cnt--;
       ref[idx] = ref[cnt];
       break;
     }
 
-    case 3: { /* remove with compaction */
+    case 3: { /* remove with backfill via ptr */
       if( !cnt ) break;
       ulong idx = fd_rng_ulong_roll( rng, cnt );
-      FD_TEST( myvec_remove_compact( vec, idx )==vec );
+      FD_TEST( myvec_remove_ele( vec, &vec[idx] )==vec );
+      cnt--;
+      ref[idx] = ref[cnt];
+      break;
+    }
+
+    case 4: { /* remove with compaction */
+      if( !cnt ) break;
+      ulong idx = fd_rng_ulong_roll( rng, cnt );
+      FD_TEST( myvec_remove_compact_idx( vec, idx )==vec );
       cnt--;
       for( ; idx<cnt; idx++ ) ref[idx] = ref[idx+1UL];
       break;