add txn lock

Author: topointon-jump

src/discof/replay/fd_replay_tile.c

@@ -560,10 +560,12 @@ checkpt( fd_replay_tile_ctx_t * ctx ) {
 
 static void
 funk_cancel( fd_replay_tile_ctx_t * ctx, ulong mismatch_slot ) {
+  fd_funkier_txn_start_write( ctx->funk );
   fd_funkier_txn_xid_t xid        = { .ul = { mismatch_slot, mismatch_slot } };
   fd_funkier_txn_map_t txn_map    = fd_funkier_txn_map( ctx->funk, fd_funkier_wksp( ctx->funk ) );
   fd_funkier_txn_t * mismatch_txn = fd_funkier_txn_query( &xid, &txn_map );
   FD_TEST( fd_funkier_txn_cancel( ctx->funk, mismatch_txn, 1 ) );
+  fd_funkier_txn_end_write( ctx->funk );
 }
 
 struct fd_status_check_ctx {
@@ -590,6 +592,8 @@ txncache_publish( fd_replay_tile_ctx_t * ctx,
     return;
   }
 
+  fd_funkier_txn_start_read( ctx->funk );
+
   fd_funkier_txn_t * txn = to_root_txn;
   fd_funkier_txn_pool_t txn_pool = fd_funkier_txn_pool( ctx->funk, fd_funkier_wksp( ctx->funk ) );
   while( txn!=rooted_txn ) {
@@ -603,6 +607,8 @@ txncache_publish( fd_replay_tile_ctx_t * ctx,
     }
     txn = fd_funkier_txn_parent( txn, &txn_pool );
   }
+
+  fd_funkier_txn_end_read( ctx->funk );
 }
 
 static void
@@ -679,6 +685,8 @@ funk_publish( fd_replay_tile_ctx_t * ctx,
               ulong                  wmk,
               uchar                  is_constipated ) {
 
+  fd_funkier_txn_start_write( ctx->funk );
+
   fd_epoch_bank_t * epoch_bank = fd_exec_epoch_ctx_epoch_bank( ctx->slot_ctx->epoch_ctx );
   fd_funkier_txn_pool_t txn_pool = fd_funkier_txn_pool( ctx->funk, fd_funkier_wksp( ctx->funk ) );
 
@@ -818,6 +826,8 @@ funk_publish( fd_replay_tile_ctx_t * ctx,
       }
     }
   }
+
+  fd_funkier_txn_end_write( ctx->funk );
 }
 
 static fd_funkier_txn_t*
@@ -982,7 +992,9 @@ funk_and_txncache_publish( fd_replay_tile_ctx_t * ctx, ulong wmk, fd_funkier_txn
 
   txncache_publish( ctx, to_root_txn, rooted_txn );
 
+  fd_funkier_txn_start_write( ctx->funk );
   funk_publish( ctx, to_root_txn, wmk, is_constipated );
+  fd_funkier_txn_end_write( ctx->funk );
 
   /* Update the snapshot state and determine if one is ready to be created. */
 
@@ -1260,7 +1272,9 @@ prepare_new_block_execution( fd_replay_tile_ctx_t * ctx,
   }
   xid.ul[0] = fork->slot_ctx.slot_bank.slot;
   /* push a new transaction on the stack */
+  fd_funkier_txn_start_write( ctx->funk );
   fork->slot_ctx.funk_txn = fd_funkier_txn_prepare(ctx->funk, fork->slot_ctx.funk_txn, &xid, 1);
+  fd_funkier_txn_end_write( ctx->funk );
 
   fd_runtime_block_pre_execute_process_new_epoch( &fork->slot_ctx,
                                                   ctx->tpool,

src/discof/restart/fd_restart.c

@@ -210,7 +210,9 @@ fd_restart_find_heaviest_fork_bank_hash( fd_restart_t * restart,
     *out_need_repair = 0;
   } else {
     /* Cancel any leftover in-preparation transactions from funk */
+    fd_funkier_txn_start_write( funk );
     fd_funkier_txn_cancel_all( funk, 1 );
+    fd_funkier_txn_end_write( funk );
 
     *out_need_repair = 1;
   }

src/discof/restart/fd_restart_tile.c

@@ -360,9 +360,11 @@ after_frag( fd_restart_tile_ctx_t * ctx,
     fd_funkier_rec_publish( prepare );
 
     /* Publish the txn in funk */
+    fd_funkier_txn_start_write( ctx->funk );
     if( FD_UNLIKELY( !fd_funkier_txn_publish( ctx->funk, funk_txn, 1 ) ) ) {
       FD_LOG_ERR(( "Wen-restart fails at funk txn publish" ));
     }
+    fd_funkier_txn_end_write( ctx->funk );
 
     /* Copy the bank hash of HeaviestForkSlot to fd_restart_t */
     fd_memcpy( &ctx->restart->heaviest_fork_bank_hash, &slot_bank.banks_hash, sizeof(fd_hash_t) );

src/flamenco/runtime/fd_runtime.c

@@ -2191,7 +2191,9 @@ fd_migrate_builtin_to_core_bpf( fd_exec_slot_ctx_t * slot_ctx,
   /* Start a funk write txn */
   fd_funkier_txn_t * parent_txn = slot_ctx->funk_txn;
   fd_funkier_txn_xid_t migration_xid = fd_funkier_generate_xid();
+  fd_funkier_txn_start_write( slot_ctx->acc_mgr->funk );
   slot_ctx->funk_txn = fd_funkier_txn_prepare( slot_ctx->acc_mgr->funk, slot_ctx->funk_txn, &migration_xid, 0UL );
+  fd_funkier_txn_end_write( slot_ctx->acc_mgr->funk );
 
   /* Attempt serialization of program account. If the program is stateless, we want to create the account. Otherwise,
      we want a writable handle to modify the existing account.
@@ -2270,13 +2272,17 @@ fd_migrate_builtin_to_core_bpf( fd_exec_slot_ctx_t * slot_ctx,
      a BPF cache entry here because the program is technically "delayed visibility", so the program
      should not be invokable until the next slot. The cache entry will be created at the end of the
      block as a part of the finalize routine. */
+  fd_funkier_txn_start_write( slot_ctx->acc_mgr->funk );
   fd_funkier_txn_publish_into_parent( slot_ctx->acc_mgr->funk, slot_ctx->funk_txn, 1 );
+  fd_funkier_txn_end_write( slot_ctx->acc_mgr->funk );
   slot_ctx->funk_txn = parent_txn;
   return;
 
 fail:
   /* Cancel the in-preparation transaction and discard any in-progress changes. */
+  fd_funkier_txn_start_write( slot_ctx->acc_mgr->funk );
   fd_funkier_txn_cancel( slot_ctx->acc_mgr->funk, slot_ctx->funk_txn, 0UL );
+  fd_funkier_txn_end_write( slot_ctx->acc_mgr->funk );
   slot_ctx->funk_txn = parent_txn;
 }
 
@@ -3603,13 +3609,15 @@ fd_runtime_block_verify_tpool( fd_exec_slot_ctx_t *    slot_ctx,
   } FD_SPAD_FRAME_END;
 }
 
+/* Should only be called in offline replay */
 static int
 fd_runtime_publish_old_txns( fd_exec_slot_ctx_t * slot_ctx,
                              fd_capture_ctx_t *   capture_ctx,
                              fd_tpool_t *         tpool,
                              fd_spad_t *          runtime_spad ) {
   /* Publish any transaction older than 31 slots */
   fd_funkier_t *       funk       = slot_ctx->acc_mgr->funk;
+  fd_funkier_txn_start_write( slot_ctx->acc_mgr->funk );
   fd_funkier_txn_pool_t txnpool   = fd_funkier_txn_pool( funk, fd_funkier_wksp( funk ) );
   fd_epoch_bank_t * epoch_bank = fd_exec_epoch_ctx_epoch_bank( slot_ctx->epoch_ctx );
 
@@ -3665,6 +3673,8 @@ fd_runtime_publish_old_txns( fd_exec_slot_ctx_t * slot_ctx,
     }
   }
 
+  fd_funkier_txn_end_write( slot_ctx->acc_mgr->funk );
+
   return 0;
 }
 
@@ -3819,7 +3829,9 @@ fd_runtime_block_eval_tpool( fd_exec_slot_ctx_t * slot_ctx,
     /* Start a new funk txn. */
 
     fd_funkier_txn_xid_t xid = { .ul = { slot_ctx->slot_bank.slot, slot_ctx->slot_bank.slot } };
+    fd_funkier_txn_start_write( funk );
     slot_ctx->funk_txn = fd_funkier_txn_prepare( funk, slot_ctx->funk_txn, &xid, 1 );
+    fd_funkier_txn_end_write( funk );
 
     /* Capturing block-agnostic state in preparation for the epoch boundary */
     uchar dump_block = capture_ctx && slot_ctx->slot_bank.slot >= capture_ctx->dump_proto_start_slot && capture_ctx->dump_block_to_pb;

src/flamenco/runtime/program/fd_bpf_program_util.c

@@ -357,15 +357,18 @@ fd_bpf_scan_and_create_bpf_program_cache_entry_tpool( fd_exec_slot_ctx_t * slot_
   /* Use random-ish xid to avoid concurrency issues */
   fd_funkier_txn_xid_t cache_xid = fd_funkier_generate_xid();
 
+  fd_funkier_txn_start_write( funk );
   fd_funkier_txn_t * cache_txn = fd_funkier_txn_prepare( funk, slot_ctx->funk_txn, &cache_xid, 1 );
   if( !cache_txn ) {
     FD_LOG_ERR(( "fd_funkier_txn_prepare() failed" ));
     return -1;
   }
+  fd_funkier_txn_end_write( funk );
 
   fd_funkier_txn_t * parent_txn = slot_ctx->funk_txn;
   slot_ctx->funk_txn = cache_txn;
 
+  fd_funkier_txn_start_read( funk );
   fd_funkier_rec_t const * rec = fd_funkier_txn_first_rec( funk, funk_txn );
   while( rec!=NULL ) {
     FD_SPAD_FRAME_BEGIN( runtime_spad ) {
@@ -402,11 +405,13 @@ fd_bpf_scan_and_create_bpf_program_cache_entry_tpool( fd_exec_slot_ctx_t * slot_
 
     } FD_SPAD_FRAME_END;
   }
+  fd_funkier_txn_start_read( funk );
 
+  fd_funkier_txn_start_write( funk );
   if( fd_funkier_txn_publish_into_parent( funk, cache_txn, 1 ) != FD_FUNKIER_SUCCESS ) {
     FD_LOG_ERR(( "fd_funkier_txn_publish_into_parent() failed" ));
-    return -1;
   }
+  fd_funkier_txn_end_write( funk );
 
   slot_ctx->funk_txn = parent_txn;
 
@@ -427,15 +432,18 @@ fd_bpf_scan_and_create_bpf_program_cache_entry( fd_exec_slot_ctx_t * slot_ctx,
   /* Use random-ish xid to avoid concurrency issues */
   fd_funkier_txn_xid_t cache_xid = fd_funkier_generate_xid();
 
+  fd_funkier_txn_start_write( funk );
   fd_funkier_txn_t * cache_txn = fd_funkier_txn_prepare( funk, slot_ctx->funk_txn, &cache_xid, 1 );
   if( !cache_txn ) {
     FD_LOG_ERR(( "fd_funkier_txn_prepare() failed" ));
     return -1;
   }
+  fd_funkier_txn_end_write( funk );
 
   fd_funkier_txn_t * parent_txn = slot_ctx->funk_txn;
   slot_ctx->funk_txn = cache_txn;
 
+  fd_funkier_txn_start_read( funk );
   for (fd_funkier_rec_t const *rec = fd_funkier_txn_first_rec( funk, funk_txn );
        NULL != rec;
        rec = fd_funkier_txn_next_rec( funk, rec )) {
@@ -454,13 +462,16 @@ fd_bpf_scan_and_create_bpf_program_cache_entry( fd_exec_slot_ctx_t * slot_ctx,
       cnt++;
     }
   }
+  fd_funkier_txn_end_read( funk );
 
   FD_LOG_DEBUG(( "loaded program cache: %lu", cnt));
 
+  fd_funkier_txn_start_write( funk );
   if( fd_funkier_txn_publish_into_parent( funk, cache_txn, 1 ) != FD_FUNKIER_SUCCESS ) {
     FD_LOG_ERR(( "fd_funkier_txn_publish_into_parent() failed" ));
     return -1;
   }
+  fd_funkier_txn_end_write( funk );
 
   slot_ctx->funk_txn = parent_txn;
   return 0;

src/flamenco/runtime/tests/fd_exec_instr_test.c

@@ -371,7 +371,9 @@ fd_exec_test_instr_context_create( fd_exec_instr_test_runner_t *        runner,
 
   /* Create temporary funk transaction and txn / slot / epoch contexts */
 
+  fd_funkier_txn_start_write( funk );
   fd_funkier_txn_t * funk_txn = fd_funkier_txn_prepare( funk, NULL, xid, 1 );
+  fd_funkier_txn_end_write( funk );
 
   ulong vote_acct_max = MAX_TX_ACCOUNT_LOCKS;
 
@@ -732,7 +734,9 @@ _txn_context_create_and_exec( fd_exec_instr_test_runner_t *      runner,
 
   /* Create temporary funk transaction and spad contexts */
 
+  fd_funkier_txn_start_write( funk );
   fd_funkier_txn_t * funk_txn = fd_funkier_txn_prepare( funk, NULL, xid, 1 );
+  fd_funkier_txn_end_write( funk );
 
   ulong vote_acct_max = MAX_TX_ACCOUNT_LOCKS;
 
@@ -1001,7 +1005,9 @@ _block_context_create_and_exec( fd_exec_instr_test_runner_t *        runner,
   xid[0] = fd_funkier_generate_xid();
 
   /* Create temporary funk transaction and slot / epoch contexts */
+  fd_funkier_txn_start_write( funk );
   fd_funkier_txn_t * funk_txn = fd_funkier_txn_prepare( funk, NULL, xid, 1 );
+  fd_funkier_txn_end_write( funk );
 
   /* Allocate contexts */
   ulong                 vote_acct_max = fd_ulong_max( 128UL,
@@ -1270,7 +1276,9 @@ _block_context_create_and_exec( fd_exec_instr_test_runner_t *        runner,
   /* Make a new funk transaction since we're done loading in accounts for context */
   fd_funkier_txn_xid_t fork_xid[1] = {0};
   fork_xid[0] = fd_funkier_generate_xid();
+  fd_funkier_txn_start_write( funk );
   slot_ctx->funk_txn = fd_funkier_txn_prepare( funk, slot_ctx->funk_txn, fork_xid, 1 );
+  fd_funkier_txn_end_write( funk );
 
   /* Calculate epoch account hash values. This sets epoch_bank.eah_{start_slot, stop_slot, interval} */
   fd_calculate_epoch_accounts_hash_values( slot_ctx );

src/flamenco/snapshot/fd_snapshot.c

@@ -144,7 +144,9 @@ fd_snapshot_load_init( fd_snapshot_load_ctx_t * ctx ) {
                                              incremental_snapshot_only_incremental_hash_calculation ) ) {
     fd_funkier_txn_xid_t xid;
     memset( &xid, 0xc3, sizeof(xid) );
+    fd_funkier_txn_start_write( ctx->slot_ctx->acc_mgr->funk );
     ctx->child_txn = fd_funkier_txn_prepare( ctx->slot_ctx->acc_mgr->funk, ctx->child_txn, &xid, 0 );
+    fd_funkier_txn_end_write( ctx->slot_ctx->acc_mgr->funk );
     ctx->slot_ctx->funk_txn = ctx->child_txn;
     }
 }
@@ -287,7 +289,9 @@ fd_snapshot_load_fini( fd_snapshot_load_ctx_t * ctx ) {
   }
 
   if( ctx->child_txn != ctx->par_txn ) {
+    fd_funkier_txn_start_write( ctx->slot_ctx->acc_mgr->funk );
     fd_funkier_txn_publish( ctx->slot_ctx->acc_mgr->funk, ctx->child_txn, 0 );
+    fd_funkier_txn_end_write( ctx->slot_ctx->acc_mgr->funk );
     ctx->slot_ctx->funk_txn = ctx->par_txn;
   }
 

src/funkier/fd_funkier_txn.c

@@ -21,6 +21,30 @@
 #define MAP_IMPL_STYLE        2
 #include "../util/tmpl/fd_map_para.c"
 
+/* TODO: remove this lock */
+#include "../flamenco/fd_rwlock.h"
+static fd_rwlock_t funkier_txn_lock[ 1 ] = {0};
+
+void
+fd_funkier_txn_start_read( fd_funkier_t * funk ) {
+  fd_rwlock_read( funkier_txn_lock );
+}
+
+void
+fd_funkier_txn_end_read( fd_funkier_t * funk ) {
+  fd_rwlock_unread( funkier_txn_lock );
+}
+
+void
+fd_funkier_txn_start_write( fd_funkier_t * funk ) {
+  fd_rwlock_write( funkier_txn_lock );
+}
+
+void
+fd_funkier_txn_end_write( fd_funkier_t * funk ) {
+  fd_rwlock_unwrite( funkier_txn_lock );
+}
+
 fd_funkier_txn_t *
 fd_funkier_txn_prepare( fd_funkier_t *               funk,
                         fd_funkier_txn_t *           parent,

src/funkier/fd_funkier_txn.h

@@ -89,6 +89,41 @@ static inline int fd_funkier_txn_idx_is_null( ulong idx ) { return idx==FD_FUNKI
 /* Generate a globally unique pseudo-random xid */
 fd_funkier_txn_xid_t fd_funkier_generate_xid(void);
 
+/* Concurrency control */
+
+/* APIs for marking the start and end of operations that read or write to
+   the Funkier transactions.
+
+   IMPORTANT SAFETY TIP
+
+   The following APIs need the write lock:
+   - fd_funkier_txn_prepare
+   - fd_funkier_txn_publish
+   - fd_funkier_txn_publish_into_parent
+   - fd_funkier_txn_cancel
+   - fd_funkier_txn_cancel_siblings
+   - fd_funkier_txn_cancel_children
+
+   The following APIs need the read lock:
+   - fd_funkier_txn_ancestor
+   - fd_funkier_txn_descendant
+   - fd_funkier_txn_all_iter
+
+   TODO: in future we may be able to make these lock-free, but they are called
+   infrequently so not sure how much of a gain this would be.
+   */
+void
+fd_funkier_txn_start_read( fd_funkier_t * funk );
+
+void
+fd_funkier_txn_end_read( fd_funkier_t * funk );
+
+void
+fd_funkier_txn_start_write( fd_funkier_t * funk );
+
+void
+fd_funkier_txn_end_write( fd_funkier_t * funk );
+
 /* Accessors */
 
 /* fd_funkier_txn_query returns a pointer to an in-preparation transaction
@@ -107,8 +142,8 @@ fd_funkier_txn_xid_t fd_funkier_generate_xid(void);
 FD_FN_PURE static inline fd_funkier_txn_t *
 fd_funkier_txn_query( fd_funkier_txn_xid_t const * xid,
                       fd_funkier_txn_map_t * map ) {
-  fd_funkier_txn_map_query_t query[1];
   do {
+    fd_funkier_txn_map_query_t query[1];
     if( FD_UNLIKELY( fd_funkier_txn_map_query_try( map, xid, NULL, query ) ) ) return NULL;
     fd_funkier_txn_t * ele = fd_funkier_txn_map_query_ele( query );
     if( FD_LIKELY( !fd_funkier_txn_map_query_test( query ) ) ) return ele;