work

Author: asiegel-jt

src/discof/rpcserver/fd_rpc_history.c

@@ -77,26 +77,12 @@ 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_reasm_map_elem {
-  union {
-    ulong next;
-    struct {
-      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_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_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 ele_cnt;  /* The number of shreds received in this column */    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];
+    fd_reasm_fec_t ele[FD_REASM_MAP_COL_HEIGHT];
   } cols[FD_REASM_MAP_COL_CNT];
   ulong head; /* Next open column */
   ulong tail; /* Oldest column */
@@ -111,7 +97,6 @@ struct fd_rpc_history {
   fd_rpc_acct_map_t * acct_map;
   fd_rpc_acct_map_elem_t * acct_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;
@@ -140,9 +125,6 @@ fd_rpc_history_create(fd_rpcserver_args_t * args) {
   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 ));
@@ -259,24 +241,30 @@ 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_reasm_map_column * col, fd_reasm_fec_t * fec) {
+fd_rpc_history_process_column(fd_rpc_history_t * hist, struct fd_rpc_reasm_map_column * col, fd_store_t * store, fd_reasm_fec_t * fec) {
   FD_SPAD_FRAME_BEGIN( hist->spad ) {
 
     FD_LOG_NOTICE(( "assembling slot %lu block", fec->slot ));
 
     /* Assemble the block */
+    fd_store_fec_t * list[FD_REASM_MAP_COL_HEIGHT];
     ulong slot = fec->slot;
     ulong blk_sz = 0;
     for( ulong i = 0; i < col->ele_cnt; i++ ) {
-      fd_rpc_reasm_map_elem_t * ele = col->ele[i];
-      blk_sz += ele->data.data_sz;
+      fd_reasm_fec_t * ele = &col->ele[i];
+      fd_store_fec_t * fec_p = list[i] = fd_store_query( store, &ele->key );
+      if( !fec_p ) {
+        FD_LOG_WARNING(( "missing fec" ));
+        return;
+      }
+      blk_sz += fec_p->data_sz;
     }
     uchar * blk_data = fd_spad_alloc( hist->spad, alignof(ulong), blk_sz );
     ulong blk_off = 0;
     for( ulong i = 0; i < col->ele_cnt; i++ ) {
-      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_store_fec_t * fec_p = list[i];
+      fd_memcpy( blk_data + blk_off, fec_p->data, fec_p->data_sz );
+      blk_off += fec_p->data_sz;
     }
     FD_TEST( blk_off == blk_sz );
 
@@ -299,16 +287,9 @@ fd_rpc_history_process_column(fd_rpc_history_t * hist, struct fd_rpc_reasm_map_c
 }
 
 static void
-fd_rpc_history_discard_column(fd_rpc_reasm_map_t * reasm_map, fd_rpc_reasm_map_elem_t * reasm_pool, ulong slot) {
+fd_rpc_history_discard_column(fd_rpc_reasm_map_t * reasm_map, 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->ele_cnt; i++ ) {
-    fd_rpc_reasm_map_elem_t * ele = col->ele[i];
-    if( ele ) {
-      fd_rpc_reasm_pool_ele_release( reasm_pool, ele );
-      col->ele[i] = NULL;
-    }
-  }
   col->ele_cnt = 0;
 }
 
@@ -318,7 +299,6 @@ fd_rpc_history_save_fec(fd_rpc_history_t * hist, fd_store_t * store, fd_reasm_fe
   if( !fec_p ) return;
 
   fd_rpc_reasm_map_t * reasm_map = hist->reasm_map;
-  fd_rpc_reasm_map_elem_t * reasm_pool = hist->reasm_pool;
 
   if( reasm_map->head == 0UL ) {
     reasm_map->head = fec_msg->slot+1;
@@ -327,7 +307,7 @@ fd_rpc_history_save_fec(fd_rpc_history_t * hist, fd_store_t * store, fd_reasm_fe
   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++ );
+    fd_rpc_history_discard_column( reasm_map, reasm_map->tail++ );
   }
   while( fec_msg->slot >= reasm_map->head ) {
     ulong col_idx = (reasm_map->head++) & (FD_REASM_MAP_COL_CNT - 1);
@@ -339,25 +319,19 @@ fd_rpc_history_save_fec(fd_rpc_history_t * hist, fd_store_t * store, fd_reasm_fe
   ulong col_idx = fec_msg->slot & (FD_REASM_MAP_COL_CNT - 1);
   struct fd_rpc_reasm_map_column * col = &reasm_map->cols[col_idx];
 
-  FD_TEST( col->ele_cnt < FD_REASM_MAP_COL_HEIGHT );
+  if( col->ele_cnt == 0 && fec_msg->fec_set_idx != 0 ) {
+    FD_LOG_WARNING(( "fec_set_idx %u is not 0 but this is the first fec for slot %lu", fec_msg->fec_set_idx, fec_msg->slot ));
+    return;
+  }
 
-  /* Acquire space for the fec data. If we've run out of space, discard the oldest column */
-  fd_rpc_reasm_map_elem_t * ele = NULL;
-  do {
-    ele = fd_rpc_reasm_pool_ele_acquire( reasm_pool );
-    if( ele != NULL ) break;
-    FD_TEST( reasm_map->tail < reasm_map->head );
-    fd_rpc_history_discard_column( reasm_map, reasm_pool, reasm_map->tail++ );
-  } while( 1 );
+  FD_TEST( col->ele_cnt < FD_REASM_MAP_COL_HEIGHT );
 
-  ulong data_sz = ele->data.data_sz = fec_p->data_sz;
-  fd_memcpy( ele->data.data, fec_p->data, data_sz );
-  col->ele[col->ele_cnt++] = ele;
+  col->ele[col->ele_cnt++] = *fec_msg;
 
   if( fec_msg->slot_complete ) {
     /* We've received all the shreds for this slot. Process it. */
-    fd_rpc_history_process_column( hist, col, fec_msg );
-    fd_rpc_history_discard_column( reasm_map, reasm_pool, fec_msg->slot );
+    fd_rpc_history_process_column( hist, col, store, fec_msg );
+    fd_rpc_history_discard_column( reasm_map, fec_msg->slot );
   }
 }