fix

Author: topointon-jump

src/funkier/fd_funkier_rec.c

@@ -22,6 +22,9 @@
 #define MAP_IMPL_STYLE        2
 #include "../util/tmpl/fd_map_para.c"
 
+#include "../flamenco/fd_rwlock.h"
+static fd_rwlock_t _pool_lock = {0};
+
 fd_funkier_rec_t const *
 fd_funkier_rec_query_try( fd_funkier_t *               funk,
                           fd_funkier_txn_t const *     txn,
@@ -203,7 +206,9 @@ fd_funkier_rec_prepare( fd_funkier_t *               funk,
   prepare->funk = funk;
   prepare->wksp = fd_funkier_wksp( funk );
   fd_funkier_rec_pool_t rec_pool = fd_funkier_rec_pool( funk, prepare->wksp );
+  fd_rwlock_write(&_pool_lock);
   fd_funkier_rec_t * rec = prepare->rec = fd_funkier_rec_pool_acquire( &rec_pool, NULL, 1, opt_err );
+  fd_rwlock_unwrite(&_pool_lock);
   if( opt_err && *opt_err == FD_POOL_ERR_CORRUPT ) {
     FD_LOG_ERR(( "corrupt element returned from funkier rec pool" ));
   }
@@ -254,6 +259,7 @@ fd_funkier_rec_publish( fd_funkier_rec_prepare_t * prepare ) {
   /* We need to hold the pool lock whilst inserting and removing from the map, to make the updates
      to the prev_idx/next_idx pointers atomic. */
   fd_funkier_rec_pool_lock( &rec_pool, 1 );
+  fd_rwlock_write(&_pool_lock);
   rec->prev_idx = rec_prev_idx;
   if( fd_funkier_rec_idx_is_null( rec_prev_idx ) ) {
     *rec_head_idx = rec_idx;
@@ -264,14 +270,17 @@ fd_funkier_rec_publish( fd_funkier_rec_prepare_t * prepare ) {
   if( fd_funkier_rec_map_insert( &rec_map, rec, FD_MAP_FLAG_BLOCKING ) ) {
     FD_LOG_CRIT(( "fd_funkier_rec_map_insert failed" ));
   }
+  fd_rwlock_unwrite(&_pool_lock);
   fd_funkier_rec_pool_unlock( &rec_pool );
 }
 
 void
 fd_funkier_rec_cancel( fd_funkier_rec_prepare_t * prepare ) {
   fd_funkier_val_flush( prepare->rec, fd_funkier_alloc( prepare->funk, prepare->wksp ), prepare->wksp );
   fd_funkier_rec_pool_t rec_pool = fd_funkier_rec_pool( prepare->funk, prepare->wksp );
+  fd_rwlock_write(&_pool_lock);
   fd_funkier_rec_pool_release( &rec_pool, prepare->rec, 1 );
+  fd_rwlock_unwrite(&_pool_lock);
 }
 
 fd_funkier_rec_t *
@@ -335,13 +344,15 @@ fd_funkier_rec_hard_remove( fd_funkier_t *               funk,
   fd_funkier_rec_key_copy( pair->key, key );
 
   fd_funkier_rec_pool_lock( &rec_pool, 1 );
+  fd_rwlock_write(&_pool_lock);
 
   fd_funkier_rec_t * rec = NULL;
   for(;;) {
     fd_funkier_rec_map_query_t rec_query[1];
     int err = fd_funkier_rec_map_remove( &rec_map, pair, NULL, rec_query, FD_MAP_FLAG_BLOCKING );
     if( FD_UNLIKELY( err == FD_MAP_ERR_AGAIN ) ) continue;
     if( err == FD_MAP_ERR_KEY ) {
+      fd_rwlock_unwrite(&_pool_lock);
       fd_funkier_rec_pool_unlock( &rec_pool );
       return;
     }
@@ -363,10 +374,13 @@ fd_funkier_rec_hard_remove( fd_funkier_t *               funk,
     if( fd_funkier_rec_idx_is_null( next_idx ) ) txn->rec_tail_idx =                prev_idx;
     else                                         rec_pool.ele[ next_idx ].prev_idx = prev_idx;
   }
+  fd_rwlock_unwrite(&_pool_lock);
   fd_funkier_rec_pool_unlock( &rec_pool );
 
   fd_funkier_val_flush( rec, alloc, wksp );
+  fd_rwlock_write(&_pool_lock);
   fd_funkier_rec_pool_release( &rec_pool, rec, 1 );
+  fd_rwlock_unwrite(&_pool_lock);
 }
 
 int