flamenco, gossip: use vector as push queue; order-preserving cleanup

Author: ravyu-jump

src/flamenco/gossip/fd_gossip.c

@@ -174,8 +174,8 @@ void fd_hash_copy( fd_hash_t * keyd, const fd_hash_t * keys ) {
    larger size footprint (100x). Decoupling the two
    allows for retaining more metadata without significantly
    increasing memory footprint and iteration overhead.
-   
-      
+
+
    An entry in the vector must have a corresponding entry
    in the metadata map, while an entry in the metadata
    map may not have an entry in the vector (denoted by a
@@ -189,13 +189,14 @@ struct fd_value {
   fd_pubkey_t origin; /* Where did this value originate */
   uchar data[PACKET_DATA_SIZE]; /* Serialized form of value (bincode) including signature */
   ulong datalen;
+  ulong del; /* Set to queue for deletion in fd_gossip_cleanup */
 };
 
 typedef struct fd_value fd_value_t;
 
 /* Value vector that:
-   - backs the values pointed by fd_value_meta_t->value 
-   - is used in generating push and pullresp 
+   - backs the values pointed by fd_value_meta_t->value
+   - is used in generating push and pullresp
      messages */
 #define VEC_NAME fd_value_vec
 #define VEC_T    fd_value_t
@@ -405,10 +406,8 @@ struct fd_gossip {
     fd_push_state_t * push_states[FD_PUSH_LIST_MAX];
     ulong push_states_cnt;
     fd_push_state_t * push_states_pool;
-    /* Queue of values vec ptrs that need pushing */
-    fd_value_t * * need_push;
+    /* Index into values vector */
     ulong need_push_head;
-    ulong need_push_cnt;
 
     /* Table of receive statistics */
     fd_stats_elem_t * stats;
@@ -459,7 +458,6 @@ fd_gossip_footprint( void ) {
   l = FD_LAYOUT_APPEND( l, fd_peer_table_align(), fd_peer_table_footprint(FD_PEER_KEY_MAX) );
   l = FD_LAYOUT_APPEND( l, fd_active_table_align(), fd_active_table_footprint(FD_ACTIVE_KEY_MAX) );
   l = FD_LAYOUT_APPEND( l, alignof(fd_gossip_peer_addr_t), INACTIVES_MAX*sizeof(fd_gossip_peer_addr_t) );
-  l = FD_LAYOUT_APPEND( l, alignof(fd_value_t *), FD_NEED_PUSH_MAX*sizeof(fd_value_t *) );
   l = FD_LAYOUT_APPEND( l, fd_value_meta_map_align(), fd_value_meta_map_footprint( FD_VALUE_KEY_MAX ) );
   l = FD_LAYOUT_APPEND( l, fd_value_vec_align(), fd_value_vec_footprint( FD_VALUE_DATA_MAX ) );
   l = FD_LAYOUT_APPEND( l, fd_pending_heap_align(), fd_pending_heap_footprint(FD_PENDING_MAX) );
@@ -487,13 +485,13 @@ fd_gossip_new ( void * shmem, ulong seed ) {
   glob->actives = fd_active_table_join(fd_active_table_new(shm, FD_ACTIVE_KEY_MAX, seed));
 
   glob->inactives = (fd_gossip_peer_addr_t*)FD_SCRATCH_ALLOC_APPEND(l, alignof(fd_gossip_peer_addr_t), INACTIVES_MAX*sizeof(fd_gossip_peer_addr_t));
-  glob->need_push = (fd_value_t * *)FD_SCRATCH_ALLOC_APPEND(l, alignof(fd_value_t *), FD_NEED_PUSH_MAX*sizeof(fd_value_t *));
 
   shm = FD_SCRATCH_ALLOC_APPEND( l, fd_value_meta_map_align(), fd_value_meta_map_footprint( FD_VALUE_KEY_MAX ) );
   glob->value_metas = fd_value_meta_map_join( fd_value_meta_map_new( shm, FD_VALUE_KEY_MAX, seed ) );
 
   shm = FD_SCRATCH_ALLOC_APPEND( l, fd_value_vec_align(), fd_value_vec_footprint( FD_VALUE_DATA_MAX ) );
   glob->values = fd_value_vec_join( fd_value_vec_new( shm, FD_VALUE_DATA_MAX ) );
+  glob->need_push_head = 0; // point to start of values
 
   glob->last_contact_time = 0;
 
@@ -1235,8 +1233,8 @@ fd_gossip_random_pull( fd_gossip_t * glob, fd_pending_event_arg_t * arg ) {
     fd_hash_t * hash = &(ele->key);
 
     /* Purge expired value's data entry */
-    if (ele->wallclock < expire && ele->value != NULL) {
-      fd_value_vec_remove_ele( glob->values, ele->value );
+    if( ele->wallclock<expire && ele->value!=NULL ) {
+      ele->value->del = 1; // Mark for deletion
       ele->value = NULL;
       continue;
     }
@@ -1511,6 +1509,8 @@ fd_gossip_recv_crds_array( fd_gossip_t * glob, const fd_gossip_peer_addr_t * fro
     fd_crds_value_t * crd = &crds[ i ];
     retained_crds[ num_retained_crds ] = crd; /* for use in insert pass */
 
+    val->del = 0UL;
+
     /* Setup fd_crds_value_processed_t entry */
     switch( crd->data.discriminant ) {
     case fd_crds_data_enum_contact_info_v1:
@@ -1648,15 +1648,6 @@ fd_gossip_recv_crds_array( fd_gossip_t * glob, const fd_gossip_peer_addr_t * fro
     ele->value      = val;
     ele->wallclock  = val->wallclock;
 
-    if ( FD_UNLIKELY( glob->need_push_cnt < FD_NEED_PUSH_MAX ) ) {
-      /* Remember that I need to push this value */
-      ulong i = ((glob->need_push_head + (glob->need_push_cnt++)) & (FD_NEED_PUSH_MAX-1U));
-      /* TODO: change this to vector idx/ptrs? */
-      glob->need_push[ i ] = val;
-      glob->metrics.push_crds_queue_cnt = glob->need_push_cnt;
-    } else {
-      INC_RECV_CRDS_DROP_METRIC( PUSH_QUEUE_FULL );
-    }
 
     fd_crds_value_t * crd = retained_crds[ i ];
 
@@ -1829,10 +1820,10 @@ fd_gossip_push_updated_contact(fd_gossip_t * glob) {
   if( (glob->now - glob->last_contact_time)<(long)1e9 )
     return;
 
-  if (glob->last_contact_time != 0) {
-    fd_value_meta_t * ele = fd_value_meta_map_query(glob->value_metas, &glob->last_contact_info_v2_key, NULL);
-    if (ele != NULL) {
-      fd_value_vec_remove_ele( glob->values, ele->value ); /* Remove the old value from the vector */
+  if( glob->last_contact_time!=0 ) {
+    fd_value_meta_t * ele = fd_value_meta_map_query( glob->value_metas, &glob->last_contact_info_v2_key, NULL );
+    if( ele!=NULL ) {
+      ele->value->del = 1UL;
       fd_value_meta_map_remove( glob->value_metas, &glob->last_contact_info_v2_key );
     }
   }
@@ -2112,11 +2103,14 @@ fd_gossip_push( fd_gossip_t * glob, fd_pending_event_arg_t * arg ) {
   /* Push an updated version of my contact info into values */
   fd_gossip_push_updated_contact(glob);
 
+  ulong pending_values_cnt = fd_value_vec_cnt( glob->values ) - glob->need_push_head;
+  ulong need_push_cnt = fd_ulong_if( pending_values_cnt < FD_NEED_PUSH_MAX, pending_values_cnt, FD_NEED_PUSH_MAX );
+
   /* Iterate across recent values */
-  ulong expire = FD_NANOSEC_TO_MILLI(glob->now) - FD_GOSSIP_PULL_TIMEOUT;
-  while (glob->need_push_cnt > 0) {
-    fd_value_t * msg = glob->need_push[ ((glob->need_push_head++) & (FD_NEED_PUSH_MAX-1)) ];
-    glob->need_push_cnt--;
+  ulong expire = FD_NANOSEC_TO_MILLI( glob->now ) - FD_GOSSIP_PULL_TIMEOUT;
+  while( need_push_cnt>0 ) {
+    fd_value_t * msg = &glob->values[ glob->need_push_head++ ];
+    need_push_cnt--;
 
     if( msg->wallclock<expire )
       continue;
@@ -2239,20 +2233,13 @@ fd_gossip_push_value_nolock( fd_gossip_t * glob, fd_crds_data_t * data, fd_hash_
   val->key        = key;
   val->wallclock  = ele->wallclock;
   val->origin     = *glob->public_key;
+  val->del        = 0UL;
 
   /* We store the serialized form for convenience */
   fd_memcpy( val->data, buf, datalen );
   val->datalen = datalen;
 
-  if (glob->need_push_cnt < FD_NEED_PUSH_MAX) {
-    /* Remember that I need to push this value */
-    ulong i = ((glob->need_push_head + (glob->need_push_cnt++)) & (FD_NEED_PUSH_MAX-1U));
-    glob->need_push[ i ] = val; /* Store the pointer to the value element */
-  } else {
-    INC_PUSH_CRDS_DROP_METRIC( PUSH_QUEUE_FULL );
-  }
-
-  glob->metrics.push_crds_queue_cnt = glob->need_push_cnt;
+  glob->metrics.push_crds_queue_cnt = fd_value_vec_cnt( glob->values ) - glob->need_push_head;
   glob->metrics.push_crds[ data->discriminant ] += 1UL;
   return 0;
 
@@ -2407,25 +2394,73 @@ fd_gossip_gettime( fd_gossip_t * glob ) {
   return glob->now;
 }
 
+/* Single pass values vector compaction. This
+   preserves ordering, allowing us to use the
+   vector as a push queue.
+   See fd_gossip_cleanup_values for an example */
+static ulong
+fd_gossip_compact_values( fd_gossip_t * glob ) {
+  fd_value_t * vec = glob->values;
+
+  ulong start = 0;
+  ulong cur_count = fd_value_vec_cnt( vec );
+  /* find first element to delete */
+  for( ; start<cur_count ; start++ ) if( FD_UNLIKELY( vec[start].del ) ) break;
+
+  ulong next = start + 1;
+
+  ulong num_deleted = 0UL;
+  ulong push_head_snapshot = ULONG_MAX;
+
+  while( next<cur_count ) {
+    if( FD_UNLIKELY( vec[next].del ) ) {
+      if( next>(start + 1) ) {
+        /* move all values between start and next */
+        memmove( &vec[start - num_deleted],
+                  &vec[start + 1],
+                  (next - start - 1) * sizeof(fd_value_t) );
+      }
+      start = next;
+      next = start + 1;
+      num_deleted++;
+      /* Need to adjust push queue */
+      if( FD_UNLIKELY( glob->need_push_head > start &&
+                       glob->need_push_head <= next ) ) {
+        push_head_snapshot = num_deleted;
+      }
+    } else {
+      next++;
+    }
+  }
+
+  glob->need_push_head -= fd_ulong_if( push_head_snapshot != ULONG_MAX, push_head_snapshot, num_deleted );
+  fd_value_vec_contract( glob->values, num_deleted );
+  glob->metrics.value_vec_cnt = fd_value_vec_cnt( glob->values );
+  FD_LOG_NOTICE(( "GOSSIP compacted %lu values", num_deleted ));
+  return num_deleted;
+}
+
 static void
 fd_gossip_cleanup_values( fd_gossip_t * glob,
                           fd_pending_event_arg_t * arg ) {
   (void)arg;
-  fd_gossip_add_pending( glob, fd_gossip_cleanup_values, fd_pending_event_arg_null(), glob->now + (long)3600e9 );
+  fd_gossip_add_pending( glob, fd_gossip_cleanup_values, fd_pending_event_arg_null(), glob->now + (long)15e9 );
 
   ulong value_expire = FD_NANOSEC_TO_MILLI(glob->now) - FD_GOSSIP_VALUE_EXPIRE;
-  // ulong value_purge = value_expire - FD_GOSSIP_VALUE_EXPIRE/4U; // for purging, which removes data entries but retains the key
   for( fd_value_meta_map_iter_t iter = fd_value_meta_map_iter_init( glob->value_metas );
        !fd_value_meta_map_iter_done( glob->value_metas, iter );
        iter = fd_value_meta_map_iter_next( glob->value_metas, iter ) ) {
     fd_value_meta_t * ele = fd_value_meta_map_iter_ele( glob->value_metas, iter );
-    if ( ele->wallclock < value_expire) {
-      /* This value has expired, remove it from the value set and vector */
-      if( ele->value != NULL )
-        fd_value_vec_remove_ele( glob->values, ele->value ); /* Remove the old value from the vector */
+    if ( ele->wallclock<value_expire ) {
+      /* This value has expired, mark it for deletion in the value vector and remove from map */
+      if( ele->value!=NULL ){
+        ele->value->del = 1UL;
+      }
       fd_value_meta_map_remove( glob->value_metas, &ele->key ); /* Remove from the value set */
     }
   }
+
+  fd_gossip_compact_values( glob );
   glob->metrics.value_map_cnt = fd_value_meta_map_key_cnt( glob->value_metas );
   glob->metrics.value_vec_cnt = fd_value_vec_cnt( glob->values );
 }
@@ -2440,7 +2475,7 @@ fd_gossip_start( fd_gossip_t * glob ) {
   fd_gossip_add_pending( glob, fd_gossip_refresh_push_states, fd_pending_event_arg_null(), glob->now + (long)  20e9 );
   fd_gossip_add_pending( glob, fd_gossip_push,                fd_pending_event_arg_null(), glob->now + (long)   1e8 );
   fd_gossip_add_pending( glob, fd_gossip_make_prune,          fd_pending_event_arg_null(), glob->now + (long)  30e9 );
-  fd_gossip_add_pending( glob, fd_gossip_cleanup_values,      fd_pending_event_arg_null(), glob->now + (long)3600e9 );
+  fd_gossip_add_pending( glob, fd_gossip_cleanup_values,      fd_pending_event_arg_null(), glob->now + (long)  15e9 );
   fd_gossip_unlock( glob );
   return 0;
 }