work

Author: asiegel-jt

src/discof/rpcserver/fd_rpc_history.c

@@ -77,32 +77,33 @@ typedef struct fd_rpc_acct_map_elem fd_rpc_acct_map_elem_t;
 #define POOL_T    fd_rpc_acct_map_elem_t
 #include "../../util/tmpl/fd_pool.c"
 
-struct fd_rpc_shred_map_elem {
+struct fd_rpc_reasm_map_elem {
   union {
     ulong next;
     struct {
-      ulong payload_sz;
-      uchar payload[FD_SHRED_DATA_PAYLOAD_MAX];
+      ulong data_sz;                 /* TODO fixed-32. sz of the FEC set payload, guaranteed < FD_STORE_DATA_MAX */
+      uchar data[FD_STORE_DATA_MAX]; /* FEC set payload = coalesced data shreds (byte array) */
     } data;
   };
 };
-typedef struct fd_rpc_shred_map_elem fd_rpc_shred_map_elem_t;
-#define POOL_NAME fd_rpc_shred_pool
-#define POOL_T    fd_rpc_shred_map_elem_t
+typedef struct fd_rpc_reasm_map_elem fd_rpc_reasm_map_elem_t;
+#define POOL_NAME fd_rpc_reasm_pool
+#define POOL_T    fd_rpc_reasm_map_elem_t
 #include "../../util/tmpl/fd_pool.c"
 
-#define FD_SHRED_MAP_COL_CNT (1UL<<10)
-struct fd_rpc_shred_map {
-  struct fd_rpc_shred_map_column {
+#define FD_REASM_MAP_COL_CNT (1UL<<10)
+#define FD_REASM_MAP_COL_HEIGHT (128UL)
+struct fd_rpc_reasm_map {
+  struct fd_rpc_reasm_map_column {
     ulong max_idx;   /* The max shred index set in this column + 1 */
     ulong used_cnt;  /* The number of shreds received in this column */
-    uchar end_found; /* Whether the last shred of the slot has been found */
-    fd_rpc_shred_map_elem_t * ele[FD_SHRED_BLK_MAX];
-  } cols[FD_SHRED_MAP_COL_CNT];
+    uchar end_found; /* Whether the last slice of the slot has been found */
+    fd_rpc_reasm_map_elem_t * ele[FD_REASM_MAP_COL_HEIGHT];
+  } cols[FD_REASM_MAP_COL_CNT];
   ulong head; /* Next open column */
   ulong tail; /* Oldest column */
 };
-typedef struct fd_rpc_shred_map fd_rpc_shred_map_t;
+typedef struct fd_rpc_reasm_map fd_rpc_reasm_map_t;
 
 struct fd_rpc_history {
   fd_spad_t * spad;
@@ -111,8 +112,8 @@ struct fd_rpc_history {
   fd_rpc_txn_t * txn_map;
   fd_rpc_acct_map_t * acct_map;
   fd_rpc_acct_map_elem_t * acct_pool;
-  fd_rpc_shred_map_t * shred_map;
-  fd_rpc_shred_map_elem_t * shred_pool;
+  fd_rpc_reasm_map_t * reasm_map;
+  fd_rpc_reasm_map_elem_t * reasm_pool;
   ulong first_slot;
   ulong latest_slot;
   int file_fd;
@@ -138,12 +139,12 @@ fd_rpc_history_create(fd_rpcserver_args_t * args) {
   mem = fd_spad_alloc( spad, fd_rpc_acct_map_pool_align(), fd_rpc_acct_map_pool_footprint( args->acct_index_max ) );
   hist->acct_pool = fd_rpc_acct_map_pool_join( fd_rpc_acct_map_pool_new( mem, args->acct_index_max ) );
 
-  mem = fd_spad_alloc( spad, alignof(fd_rpc_shred_map_t), sizeof(fd_rpc_shred_map_t) );
-  memset(mem, 0, sizeof(fd_rpc_shred_map_t));
-  hist->shred_map = (fd_rpc_shred_map_t *)mem;
-  static const ulong shred_pool_sz = 1024UL * 256UL;
-  mem = fd_spad_alloc( spad, fd_rpc_shred_pool_align(), fd_rpc_shred_pool_footprint( shred_pool_sz ) );
-  hist->shred_pool = fd_rpc_shred_pool_join( fd_rpc_shred_pool_new( mem, shred_pool_sz ) );
+  mem = fd_spad_alloc( spad, alignof(fd_rpc_reasm_map_t), sizeof(fd_rpc_reasm_map_t) );
+  memset(mem, 0, sizeof(fd_rpc_reasm_map_t));
+  hist->reasm_map = (fd_rpc_reasm_map_t *)mem;
+  static const ulong reasm_pool_sz = 1024UL;
+  mem = fd_spad_alloc( spad, fd_rpc_reasm_pool_align(), fd_rpc_reasm_pool_footprint( reasm_pool_sz ) );
+  hist->reasm_pool = fd_rpc_reasm_pool_join( fd_rpc_reasm_pool_new( mem, reasm_pool_sz ) );
 
   hist->file_fd = open( args->history_file, O_CREAT | O_RDWR | O_TRUNC, 0644 );
   if( hist->file_fd == -1 ) FD_LOG_ERR(( "unable to open rpc history file: %s", args->history_file ));
@@ -179,16 +180,16 @@ fd_rpc_history_alloc_block(fd_rpc_history_t * hist, ulong slot) {
 
 void
 fd_rpc_history_debug(fd_rpc_history_t * hist) {
-  fd_rpc_shred_map_t * shred_map = hist->shred_map;
+  fd_rpc_reasm_map_t * reasm_map = hist->reasm_map;
   ulong tot_cnt = 0;
-  for( ulong slot = shred_map->tail; slot < shred_map->head; slot++ ) {
-    ulong col_idx = slot & (FD_SHRED_MAP_COL_CNT - 1);
-    struct fd_rpc_shred_map_column * col = &shred_map->cols[col_idx];
-    FD_LOG_NOTICE(( "slot %lu: %lu shreds, %lu max, %d end_found", slot, col->used_cnt, col->max_idx, (int)col->end_found ));
+  for( ulong slot = reasm_map->tail; slot < reasm_map->head; slot++ ) {
+    ulong col_idx = slot & (FD_REASM_MAP_COL_CNT - 1);
+    struct fd_rpc_reasm_map_column * col = &reasm_map->cols[col_idx];
+    FD_LOG_NOTICE(( "slot %lu: %lu fecs, %lu max, %d end_found", slot, col->used_cnt, col->max_idx, (int)col->end_found ));
     tot_cnt += col->used_cnt;
   }
-  FD_LOG_NOTICE(( "%lu head, %lu tail, %lu total shreds, %lu total blocks",
-                  shred_map->head, shred_map->tail, tot_cnt, shred_map->head - shred_map->tail ));
+  FD_LOG_NOTICE(( "%lu head, %lu tail, %lu total fecs, %lu total blocks",
+                  reasm_map->head, reasm_map->tail, tot_cnt, reasm_map->head - reasm_map->tail ));
 }
 
 void
@@ -260,24 +261,24 @@ fd_rpc_history_scan_block(fd_rpc_history_t * hist, ulong slot, ulong file_offset
 }
 
 void
-fd_rpc_history_process_column(fd_rpc_history_t * hist, struct fd_rpc_shred_map_column * col, fd_shred_t * shred) {
+fd_rpc_history_process_column(fd_rpc_history_t * hist, struct fd_rpc_reasm_map_column * col, fd_reasm_fec_t * fec) {
   FD_SPAD_FRAME_BEGIN( hist->spad ) {
 
-    FD_LOG_NOTICE(( "assembling slot %lu block", shred->slot ));
+    FD_LOG_NOTICE(( "assembling slot %lu block", fec->slot ));
 
     /* Assemble the block */
-    ulong slot = shred->slot;
+    ulong slot = fec->slot;
     ulong blk_sz = 0;
     for( ulong i = 0; i < col->max_idx; i++ ) {
-      fd_rpc_shred_map_elem_t * ele = col->ele[i];
-      blk_sz += ele->data.payload_sz;
+      fd_rpc_reasm_map_elem_t * ele = col->ele[i];
+      blk_sz += ele->data.data_sz;
     }
     uchar * blk_data = fd_spad_alloc( hist->spad, alignof(ulong), blk_sz );
     ulong blk_off = 0;
     for( ulong i = 0; i < col->max_idx; i++ ) {
-      fd_rpc_shred_map_elem_t * ele = col->ele[i];
-      fd_memcpy( blk_data + blk_off, ele->data.payload, ele->data.payload_sz );
-      blk_off += ele->data.payload_sz;
+      fd_rpc_reasm_map_elem_t * ele = col->ele[i];
+      fd_memcpy( blk_data + blk_off, ele->data.data, ele->data.data_sz );
+      blk_off += ele->data.data_sz;
     }
     FD_TEST( blk_off == blk_sz );
 
@@ -300,13 +301,13 @@ fd_rpc_history_process_column(fd_rpc_history_t * hist, struct fd_rpc_shred_map_c
 }
 
 static void
-fd_rpc_history_discard_column(fd_rpc_shred_map_t * shred_map, fd_rpc_shred_map_elem_t * shred_pool, ulong slot) {
-  ulong col_idx = slot & (FD_SHRED_MAP_COL_CNT - 1);
-  struct fd_rpc_shred_map_column * col = &shred_map->cols[col_idx];
+fd_rpc_history_discard_column(fd_rpc_reasm_map_t * reasm_map, fd_rpc_reasm_map_elem_t * reasm_pool, ulong slot) {
+  ulong col_idx = slot & (FD_REASM_MAP_COL_CNT - 1);
+  struct fd_rpc_reasm_map_column * col = &reasm_map->cols[col_idx];
   for( ulong i = 0; i < col->max_idx; i++ ) {
-    fd_rpc_shred_map_elem_t * ele = col->ele[i];
+    fd_rpc_reasm_map_elem_t * ele = col->ele[i];
     if( ele ) {
-      fd_rpc_shred_pool_ele_release( shred_pool, ele );
+      fd_rpc_reasm_pool_ele_release( reasm_pool, ele );
       col->ele[i] = NULL;
     }
   }
@@ -316,59 +317,61 @@ fd_rpc_history_discard_column(fd_rpc_shred_map_t * shred_map, fd_rpc_shred_map_e
 }
 
 void
-fd_rpc_history_save_shred(fd_rpc_history_t * hist, fd_shred_t * shred) {
-  if( !fd_shred_is_data( shred->variant ) ) return;
-  if( shred->idx >= FD_SHRED_BLK_MAX ) return;
+fd_rpc_history_save_fec(fd_rpc_history_t * hist, fd_store_t * store, fd_reasm_fec_t * fec_msg ) {
+  fd_store_fec_t * fec_p = fd_store_query( store, &fec_msg->key );
+  if( !fec_p ) return;
 
-  fd_rpc_shred_map_t * shred_map = hist->shred_map;
-  fd_rpc_shred_map_elem_t * shred_pool = hist->shred_pool;
+  fd_rpc_reasm_map_t * reasm_map = hist->reasm_map;
+  fd_rpc_reasm_map_elem_t * reasm_pool = hist->reasm_pool;
 
-  if( shred_map->head == 0UL ) {
-    shred_map->head = shred->slot+1;
-    shred_map->tail = shred->slot;
+  if( reasm_map->head == 0UL ) {
+    reasm_map->head = fec_msg->slot+1;
+    reasm_map->tail = fec_msg->slot;
   }
-  if( shred->slot < shred_map->tail ) return; /* Do not go backwards */
-  while( shred->slot >= shred_map->tail + FD_SHRED_MAP_COL_CNT ) {
-    FD_TEST( shred_map->tail < shred_map->head );
-    fd_rpc_history_discard_column( shred_map, shred_pool, shred_map->tail++ );
+  if( fec_msg->slot < reasm_map->tail ) return; /* Do not go backwards */
+  while( fec_msg->slot >= reasm_map->tail + FD_REASM_MAP_COL_CNT ) {
+    FD_TEST( reasm_map->tail < reasm_map->head );
+    fd_rpc_history_discard_column( reasm_map, reasm_pool, reasm_map->tail++ );
   }
-  while( shred->slot >= shred_map->head ) {
-    ulong col_idx = (shred_map->head++) & (FD_SHRED_MAP_COL_CNT - 1);
-    struct fd_rpc_shred_map_column * col = &shred_map->cols[col_idx];
+  while( fec_msg->slot >= reasm_map->head ) {
+    ulong col_idx = (reasm_map->head++) & (FD_REASM_MAP_COL_CNT - 1);
+    struct fd_rpc_reasm_map_column * col = &reasm_map->cols[col_idx];
     col->max_idx = 0;
     col->used_cnt = 0;
     col->end_found = 0;
   }
-  FD_TEST( shred->slot >= shred_map->tail && shred->slot < shred_map->head && shred_map->head - shred_map->tail <= FD_SHRED_MAP_COL_CNT );
+  FD_TEST( fec_msg->slot >= reasm_map->tail && fec_msg->slot < reasm_map->head && reasm_map->head - reasm_map->tail <= FD_REASM_MAP_COL_CNT );
 
-  ulong col_idx = shred->slot & (FD_SHRED_MAP_COL_CNT - 1);
-  struct fd_rpc_shred_map_column * col = &shred_map->cols[col_idx];
+  ulong col_idx = fec_msg->slot & (FD_REASM_MAP_COL_CNT - 1);
+  struct fd_rpc_reasm_map_column * col = &reasm_map->cols[col_idx];
 
-  /* See if we've already received this shred */
-  if( col->max_idx > shred->idx && col->ele[shred->idx] != NULL ) return;
-  while( col->max_idx <= shred->idx ) {
+  FD_TEST( fec_msg->fec_set_idx < FD_REASM_MAP_COL_HEIGHT );
+
+  /* See if we've already received this fec */
+  if( col->max_idx > fec_msg->fec_set_idx && col->ele[fec_msg->fec_set_idx] != NULL ) return;
+  while( col->max_idx <= fec_msg->fec_set_idx ) {
     col->ele[col->max_idx++] = NULL;
   }
 
   /* Acquire space for the shred. If we've run out of space, discard the oldest column */
-  fd_rpc_shred_map_elem_t * ele = NULL;
+  fd_rpc_reasm_map_elem_t * ele = NULL;
   do {
-    ele = fd_rpc_shred_pool_ele_acquire( shred_pool );
+    ele = fd_rpc_reasm_pool_ele_acquire( reasm_pool );
     if( ele != NULL ) break;
-    FD_TEST( shred_map->tail < shred_map->head );
-    fd_rpc_history_discard_column( shred_map, shred_pool, shred_map->tail++ );
+    FD_TEST( reasm_map->tail < reasm_map->head );
+    fd_rpc_history_discard_column( reasm_map, reasm_pool, reasm_map->tail++ );
   } while( 1 );
 
-  ulong payload_sz = ele->data.payload_sz = fd_shred_payload_sz( shred );
-  fd_memcpy( ele->data.payload, fd_shred_data_payload( shred ), payload_sz );
-  col->ele[shred->idx] = ele;
+  ulong data_sz = ele->data.data_sz = fec_p->data_sz;
+  fd_memcpy( ele->data.data, fec_p->data, data_sz );
+  col->ele[fec_msg->fec_set_idx] = ele;
   col->used_cnt++;
-  if( FD_UNLIKELY( shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE ) ) col->end_found = 1;
+  if( FD_UNLIKELY( fec_msg->slot_complete ) ) col->end_found = 1;
 
   if( col->end_found && col->used_cnt == col->max_idx ) {
     /* We've received all the shreds for this slot. Process it. */
-    fd_rpc_history_process_column( hist, col, shred );
-    fd_rpc_history_discard_column( shred_map, shred_pool, shred->slot );
+    fd_rpc_history_process_column( hist, col, fec_msg );
+    fd_rpc_history_discard_column( reasm_map, reasm_pool, fec_msg->slot );
   }
 }
 

src/discof/rpcserver/fd_rpc_history.h

@@ -16,7 +16,7 @@ fd_rpc_history_t * fd_rpc_history_create(fd_rpcserver_args_t * args);
 
 void fd_rpc_history_save_info(fd_rpc_history_t * hist, fd_replay_notif_msg_t * msg);
 
-void fd_rpc_history_save_shred(fd_rpc_history_t * hist, fd_shred_t * shred);
+void fd_rpc_history_save_fec(fd_rpc_history_t * hist, fd_store_t * store, fd_reasm_fec_t * fec);
 
 ulong fd_rpc_history_first_slot(fd_rpc_history_t * hist);
 

src/discof/rpcserver/fd_rpc_service.c

@@ -85,6 +85,7 @@ struct fd_rpc_global_ctx {
   fd_spad_t * spad;
   fd_webserver_t ws;
   fd_funk_t * funk;
+  fd_store_t * store;
   struct fd_ws_subscription sub_list[FD_WS_MAX_SUBS];
   ulong sub_cnt;
   ulong last_subsc_id;
@@ -2340,8 +2341,8 @@ fd_rpc_create_ctx(fd_rpcserver_args_t * args, fd_rpc_ctx_t ** ctx_p) {
 void
 fd_rpc_start_service(fd_rpcserver_args_t * args, fd_rpc_ctx_t * ctx) {
   fd_rpc_global_ctx_t * gctx = ctx->global;
-
   gctx->funk = args->funk;
+  gctx->store = args->store;
 }
 
 int
@@ -2440,13 +2441,13 @@ fd_rpc_stake_after_frag(fd_rpc_ctx_t * ctx) {
 
 void
 fd_rpc_repair_during_frag(fd_rpc_ctx_t * ctx, void const * msg, int sz) {
-  (void)sz;
+  FD_TEST( sz==(int)sizeof(fd_reasm_fec_t) );
   fd_memcpy(ctx->global->buffer, msg, sizeof(fd_reasm_fec_t));
 }
 
 void
 fd_rpc_repair_after_frag(fd_rpc_ctx_t * ctx) {
   fd_rpc_global_ctx_t * subs = ctx->global;
   fd_reasm_fec_t * fec_p = (fd_reasm_fec_t *)subs->buffer;
-  (void)fec_p;
+  fd_rpc_history_save_fec( subs->history, subs->store, fec_p );
 }

src/discof/rpcserver/fd_rpcserv_tile.c

@@ -128,17 +128,17 @@ after_frag( fd_rpcserv_tile_ctx_t * ctx,
   (void)stem;
 
   ulong * last_seq = &ctx->in_links[ in_idx ].last_seq;
-  if( seq != *last_seq + 1 ) {
+  if( FD_UNLIKELY( seq != *last_seq + 1 && seq != 0 ) ) {
     FD_LOG_NOTICE(( "seq jump from %lu to %lu on input %lu", *last_seq, seq, in_idx ));
   }
   *last_seq = seq;
 
   uchar kind = ctx->in_kind[ in_idx ];
-  if( FD_UNLIKELY( kind==IN_KIND_REPLAY_NOTIF ) ) {
+  if( kind==IN_KIND_REPLAY_NOTIF ) {
     fd_rpc_replay_after_frag( ctx->ctx );
-  } else if( FD_UNLIKELY( kind == IN_KIND_STAKE_OUT ) ) {
+  } else if( kind == IN_KIND_STAKE_OUT ) {
     fd_rpc_stake_after_frag( ctx->ctx );
-  } else if( FD_UNLIKELY( kind == IN_KIND_REPAIR_REPLAY ) ) {
+  } else if( kind == IN_KIND_REPAIR_REPLAY ) {
     fd_rpc_repair_after_frag( ctx->ctx );
   } else {
     FD_LOG_ERR(("Unknown in_idx %lu for rpc", in_idx));