fd_mvcc from #512

fd_leader_schedule skeleton

update .gitignore for JetBrains

add the leader schedule to forward workspace

.github/workflows/functional_tests.yml

@@ -25,4 +25,4 @@ jobs:
 
       - name: Run functional tests
         run: |
-          ./src/test/frank-single-transaction.sh
+          ./src/test/functional-tests.sh

.gitignore

@@ -38,3 +38,6 @@ deps-bundle.tar.zst
 
 # Logs
 *.log
+
+# JetBrains
+.idea

ffi/rust/firedancer-sys/src/tango/mod.rs

@@ -3,13 +3,17 @@ mod dcache;
 mod fctl;
 mod fseq;
 mod mcache;
+mod mvcc;
 mod tcache;
+mod validator;
 mod xdp;
 
 pub use cnc::*;
 pub use dcache::*;
 pub use fctl::*;
 pub use fseq::*;
 pub use mcache::*;
+pub use mvcc::*;
 pub use tcache::*;
+pub use validator::*;
 pub use xdp::*;

ffi/rust/firedancer-sys/src/tango/mvcc.rs

@@ -0,0 +1,8 @@
+pub use crate::generated::{
+    fd_mvcc_version_laddr,
+    fd_mvcc_begin_write,
+    fd_mvcc_end_write,
+    fd_mvcc_begin_read,
+    fd_mvcc_end_read,
+    fd_mvcc_t,
+};

ffi/rust/firedancer-sys/src/tango/validator.rs

@@ -0,0 +1,11 @@
+pub use crate::generated::{
+    fd_leader_schedule_align,
+    fd_leader_schedule_footprint,
+    fd_leader_schedule_new,
+    fd_leader_schedule_delete,
+    fd_leader_schedule_join,
+    fd_leader_schedule_leave,
+    fd_leader_schedule_get,
+    fd_leader_schedule_t,
+    Pubkey
+};

src/app/fdctl/configure/workspace.c

@@ -95,6 +95,13 @@ static void xsk_aio( void * pod, char * fmt, ulong tx_depth, ulong batch_count,
             fd_xsk_aio_new      ( shmem,    tx_depth, batch_count ) );
 }
 
+static void leader_schedule( void * pod, char * fmt, ... ) {
+  INSERTER( fmt,
+            fd_leader_schedule_align    (         ),
+            fd_leader_schedule_footprint(         ),
+            fd_leader_schedule_new      ( shmem ) );
+}
+
 FD_FN_UNUSED static void alloc( void * pod, char * fmt, ulong align, ulong sz, ... ) {
   INSERTER( sz, align, sz, 1 );
 }
@@ -294,7 +301,8 @@ init( config_t * const config ) {
         cnc   ( pod, "cnc" );
         break;
       case wksp_forward:
-        cnc   ( pod, "cnc" );
+        cnc            ( pod, "cnc" );
+        leader_schedule( pod, "leader_schedule" );
         break;
     }
 

src/app/frank/fd_frank_forward.c

@@ -1,4 +1,5 @@
 #include "fd_frank.h"
+#include "../../tango/validator/fd_leader_schedule.h"
 #include <linux/unistd.h>
 
 static void
@@ -13,6 +14,11 @@ run( fd_frank_args_t * args ) {
   ulong * cnc_diag = (ulong *)fd_cnc_app_laddr( cnc );
   cnc_diag[ FD_FRANK_CNC_DIAG_PID ] = (ulong)args->pid;
 
+  FD_LOG_INFO(( "joining leader schedule" ));
+  fd_leader_schedule_new( fd_wksp_pod_map( args->tile_pod, "leader_schedule" ) );
+  fd_leader_schedule_t * leader_schedule = fd_leader_schedule_get( args->app_name );
+  if( FD_UNLIKELY( !leader_schedule ) ) FD_LOG_ERR(( "fd_leader_schedule_join failed" ));
+
   FD_LOG_INFO(( "joining mcache" ));
   fd_frag_meta_t const * mcache = fd_mcache_join( fd_wksp_pod_map( args->in_pod, "mcache" ) );
   if( FD_UNLIKELY( !mcache ) ) FD_LOG_ERR(( "fd_mcache_join failed" ));
@@ -65,6 +71,7 @@ run( fd_frank_args_t * args ) {
 
   long now            = fd_tickcount();
   long then           = now;            /* Do housekeeping on first iteration of run loop */
+  ulong version = 0;
   for(;;) {
 
     /* Do housekeeping at a low rate in the background */
@@ -98,6 +105,18 @@ run( fd_frank_args_t * args ) {
       then = now + (long)fd_tempo_async_reload( rng, async_min );
     }
 
+    ulong begin_version = fd_mvcc_begin_read( &leader_schedule->mvcc);
+    if( begin_version != version && begin_version % 2 == 0 ) {
+      ulong size = leader_schedule->size;
+      ulong end_version = fd_mvcc_end_read( &leader_schedule->mvcc);
+      if( begin_version == end_version ) {
+        FD_LOG_NOTICE(( "schedule clean read! length %lu", size ));
+      } else {
+        FD_LOG_NOTICE(( "schedule bad read! %lu %lu", begin_version, end_version ));
+      }
+      version = begin_version;
+    }
+
     /* See if there are any transactions waiting to be forwarded */
     ulong seq_found = fd_frag_meta_seq_query( mline );
     long  diff      = fd_seq_diff( seq_found, seq );

src/tango/fd_tango.h

@@ -1,15 +1,16 @@
 #ifndef HEADER_fd_src_tango_fd_tango_h
 #define HEADER_fd_src_tango_fd_tango_h
 
-//#include "fd_tango_base.h"  /* Includes ../util/fd_util.h */
-#include "tempo/fd_tempo.h"   /* Includes fd_tango_base.h */
-#include "cnc/fd_cnc.h"       /* Includes fd_tango_base.h */
-#include "fseq/fd_fseq.h"     /* Includes fd_tango_base.h */
-#include "fctl/fd_fctl.h"     /* Includes fd_tango_base.h */
-#include "mcache/fd_mcache.h" /* Includes fd_tango_base.h */
-#include "dcache/fd_dcache.h" /* Includes fd_tango_base.h */
-#include "tcache/fd_tcache.h" /* Includes fd_tango_base.h */
-#include "aio/fd_aio.h"       /* Includes fd_tango_base.h */
+//#include "fd_tango_base.h"                    /* Includes ../util/fd_util.h */
+#include "tempo/fd_tempo.h"                     /* Includes fd_tango_base.h */
+#include "cnc/fd_cnc.h"                         /* Includes fd_tango_base.h */
+#include "fseq/fd_fseq.h"                       /* Includes fd_tango_base.h */
+#include "fctl/fd_fctl.h"                       /* Includes fd_tango_base.h */
+#include "mvcc/fd_mvcc.h"
+#include "mcache/fd_mcache.h"                   /* Includes fd_tango_base.h */
+#include "dcache/fd_dcache.h"                   /* Includes fd_tango_base.h */
+#include "tcache/fd_tcache.h"                   /* Includes fd_tango_base.h */
+#include "aio/fd_aio.h"                         /* Includes fd_tango_base.h */
+#include "validator/fd_leader_schedule.h"       /* Includes fd_tango_base.h */
 
 #endif /* HEADER_fd_src_tango_fd_tango_h */
-

src/tango/mvcc/Local.mk

@@ -0,0 +1,4 @@
+$(call add-hdrs,fd_mvcc.h)
+$(call add-objs,fd_mvcc,fd_tango)
+$(call make-unit-test,test_mvcc,test_mvcc,fd_tango fd_util)
+$(call run-unit-test,test_mvcc,)

src/tango/mvcc/fd_mvcc.c

@@ -0,0 +1,38 @@
+#include "../../util/fd_util.h"
+#include "fd_mvcc.h"
+
+ulong *
+fd_mvcc_version_laddr( fd_mvcc_t * mvcc ) {
+  return &mvcc->version;
+}
+
+ulong
+fd_mvcc_begin_write( fd_mvcc_t * mvcc ) {
+  ulong version = FD_ATOMIC_FETCH_AND_ADD( fd_mvcc_version_laddr( mvcc ), 1 );
+  FD_COMPILER_MFENCE();
+  return version;
+}
+
+ulong
+fd_mvcc_end_write( fd_mvcc_t * mvcc ) {
+  FD_COMPILER_MFENCE();
+  return FD_ATOMIC_FETCH_AND_ADD( fd_mvcc_version_laddr( mvcc ), 1 );
+}
+
+ulong
+fd_mvcc_read( fd_mvcc_t * mvcc ) {
+  FD_COMPILER_MFENCE();
+  ulong version = FD_VOLATILE_CONST( mvcc->version );
+  FD_COMPILER_MFENCE();
+  return version;
+}
+
+ulong
+fd_mvcc_begin_read( fd_mvcc_t * mvcc ) {
+  return fd_mvcc_read( mvcc );
+}
+
+ulong
+fd_mvcc_end_read( fd_mvcc_t * mvcc ) {
+  return fd_mvcc_read( mvcc );
+}

src/tango/mvcc/fd_mvcc.h

@@ -0,0 +1,67 @@
+#ifndef HEADER_fd_src_tango_mvcc_fd_mvcc_h
+#define HEADER_fd_src_tango_mvcc_fd_mvcc_h
+
+#include "../../util/fd_util.h"
+
+/* fd_mvcc ("Multiversion Concurrency Control") is a simple primitive for lock-free synchronization
+   of concurrent readers and writers. It is strictly less general than the MVCC used in various
+   DBMS [https://dl.acm.org/doi/pdf/10.1145/356842.356846], but it is conceptually similar in that
+   it uses a version number to detect conflicts.
+
+   Usage:
+   - Writer increments version number
+   - Writer does update
+   - Writer increments version number
+   - Therefore, if the version number is odd, a write is in progress.
+
+   - Reader reads version number
+   - Reader reads data
+   - Reader reads version number
+   - Therefore, if the version number has changed, the read is invalid.
+
+   fd_mvcc_begin_write()  // release-store
+   ... write ...
+   fd_mvcc_end_write()    // acquire-load
+
+   ulong begin = fd_mvcc_begin_read()  // acquire-load
+   ulong end = fd_mvcc_end_read()      // acquire-load
+   if (end != begin) {
+     ... read is invalid ...
+   }
+
+   Note this is similar to how producers / consumers synchronize across mcache / dcache.
+
+   TODO hardware fencing */
+
+struct fd_mvcc {
+  ulong version;
+};
+typedef struct fd_mvcc fd_mvcc_t;
+
+/* fd_mvcc_version_laddr returns a local pointer to the version number for the current joined
+ * process. Caller is responsible for fencing the dereference if necessary. */
+ulong *
+fd_mvcc_version_laddr( fd_mvcc_t * mvcc ); 
+
+/* fd_mvcc_begin_write increments then returns the version number, fencing preceding memory
+ * accesses. Corresponds to C++ memory_order_release. */
+ulong
+fd_mvcc_begin_write( fd_mvcc_t * mvcc ); 
+
+/* fd_mvcc_begin_write increments then returns the version number, fencing subsequent memory
+ * accesses. Corresponds to C++ memory_order_acquire. */
+ulong
+fd_mvcc_end_write( fd_mvcc_t * mvcc ); 
+
+/* fd_mvcc_{begin,end}_read are convenience exports for code readability assisting with
+   remembering to read back the version. */
+ulong
+fd_mvcc_begin_read( fd_mvcc_t * mvcc ); 
+
+ulong
+fd_mvcc_end_read( fd_mvcc_t * mvcc );
+
+ulong
+fd_mvcc_read( fd_mvcc_t * mvcc );
+
+#endif /* HEADER_fd_src_tango_mvcc_fd_mvcc_h */

src/tango/mvcc/test_mvcc.c

@@ -0,0 +1,25 @@
+#include "../../util/fd_util.h"
+#include "fd_mvcc.h"
+
+int
+main( int argc, char ** argv ) {
+  fd_boot( &argc, &argv );
+
+  fd_mvcc_t mvcc = { .version = 0 };
+  FD_TEST( fd_mvcc_begin_read( &mvcc ) == 0 );
+  FD_TEST( fd_mvcc_end_read( &mvcc ) == 0 );
+
+  FD_TEST( fd_mvcc_begin_write( &mvcc ) == 0 );
+  FD_TEST( fd_mvcc_begin_read( &mvcc ) == 1 );
+  FD_TEST( fd_mvcc_end_read( &mvcc ) == 1 );
+  FD_TEST( fd_mvcc_end_write( &mvcc ) == 1 );
+
+  FD_TEST( fd_mvcc_begin_read( &mvcc ) == 2 );
+  FD_TEST( fd_mvcc_begin_write( &mvcc ) == 2 );
+  FD_TEST( fd_mvcc_end_read( &mvcc ) == 3 );
+  FD_TEST( fd_mvcc_end_write( &mvcc ) == 3 );
+
+  FD_LOG_NOTICE( ( "pass" ) );
+  fd_halt();
+  return 0;
+}

src/tango/validator/Local.mk

@@ -0,0 +1,3 @@
+$(call add-hdrs,fd_leader_schedule.h)
+$(call add-objs,fd_leader_schedule,fd_tango)
+$(call make-unit-test,test_leader_schedule,test_leader_schedule,fd_tango fd_util)

src/tango/validator/fd_leader_schedule.c

@@ -0,0 +1,32 @@
+#include "fd_leader_schedule.h"
+#include "../../util/wksp/fd_wksp_private.h"
+#include <stdio.h>
+
+void *
+fd_leader_schedule_new( void * mem )
+{
+  fd_leader_schedule_t * schedule = ( fd_leader_schedule_t * )mem;
+  fd_memset( schedule, 0, fd_leader_schedule_footprint() );
+  return (void *)schedule;
+}
+
+/* TODO: LML same functionality as run.c:workspace_pod_join we should refactor */
+fd_leader_schedule_t *
+fd_leader_schedule_get( char const * app_name ) {
+  char name[ FD_WKSP_CSTR_MAX ];
+  snprintf( name, FD_WKSP_CSTR_MAX, "%s_forward0.wksp", app_name );
+
+  fd_wksp_t * wksp = fd_wksp_attach( name );
+  if( FD_UNLIKELY( !wksp ) ) FD_LOG_ERR(( "could not attach to workspace `%s`", name ));
+
+  void * laddr = fd_wksp_laddr( wksp, wksp->gaddr_lo );
+  if( FD_UNLIKELY( !laddr ) ) FD_LOG_ERR(( "could not get gaddr_low from workspace `%s`", name ));
+
+  uchar const * pod = fd_pod_join( laddr );
+  if( FD_UNLIKELY( !pod ) ) FD_LOG_ERR(( "fd_pod_join to pod at gaddr_lo failed" ));
+
+  fd_leader_schedule_t  * leader_schedule = (fd_leader_schedule_t *)fd_wksp_pod_map( pod, "leader_schedule" );
+  if( FD_UNLIKELY( !leader_schedule ) ) FD_LOG_ERR(( "fd_wksp_pod_map failed" ));
+
+  return leader_schedule;
+}

src/tango/validator/fd_leader_schedule.h

@@ -0,0 +1,33 @@
+#ifndef HEADER_fd_src_tango_validators_fd_leader_schedule_h
+#define HEADER_fd_src_tango_validators_fd_leader_schedule_h
+
+#include "../fd_tango_base.h"
+#include "../mvcc/fd_mvcc.h"
+
+#define DEFAULT_SLOTS_PER_EPOCH 432000
+
+typedef uchar Pubkey[32];
+
+typedef struct {
+  ulong size;
+  fd_mvcc_t mvcc;
+  Pubkey schedule[DEFAULT_SLOTS_PER_EPOCH];
+} fd_leader_schedule_t;
+
+FD_FN_CONST static inline ulong fd_leader_schedule_align     ( void ) { return alignof( fd_leader_schedule_t ); }
+FD_FN_CONST static inline ulong fd_leader_schedule_footprint ( void ) { return sizeof ( fd_leader_schedule_t ); }
+
+void * fd_leader_schedule_new( void * mem );
+
+FD_FN_UNUSED static inline void *
+fd_leader_schedule_delete( void                 * _leader_schedule ) { return (void                 *)_leader_schedule; }
+
+FD_FN_UNUSED static inline fd_leader_schedule_t *
+fd_leader_schedule_join  ( void                 * _leader_schedule ) { return (fd_leader_schedule_t *)_leader_schedule; }
+
+fd_leader_schedule_t * fd_leader_schedule_get( char const * app_name );
+
+FD_FN_UNUSED static inline void *
+fd_leader_schedule_leave ( fd_leader_schedule_t * leader_schedule  ) { return (void                 *) leader_schedule; }
+
+#endif /* HEADER_fd_src_tango_validators_fd_leader_schedule_h */

src/tango/validator/test_leader_schedule.c

@@ -0,0 +1,13 @@
+#include "../fd_tango.h"
+
+int
+main( int     argc,
+      char ** argv ) {
+  fd_boot( &argc, &argv );
+
+  // TODO: LML write a test that confirms the schedule is
+  FD_LOG_NOTICE(( "pass" ));
+
+  fd_halt();
+  return 0;
+}

src/test/frank-leader-schedule.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+# bash strict mode
+set -euo pipefail
+IFS=$'\n\t'
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+cd "${SCRIPT_DIR}/../../"
+
+LOG_PATH=${LOG_PATH:-~/log}
+
+# frank-single-transaction.sh needs to be run first so that a log is generated
+grep -qE 'schedule clean read! length [1-9]' "${LOG_PATH}"