Clean up C API

libs/walproposer/build.rs

@@ -1,7 +1,5 @@
-use std::{path::PathBuf, env, process::Command};
-
-use anyhow::{anyhow, Context};
-use bindgen::{callbacks::ParseCallbacks, CargoCallbacks};
+use std::{env, path::PathBuf};
+use bindgen::CargoCallbacks;
 
 extern crate bindgen;
 
@@ -16,13 +14,7 @@ fn main() -> anyhow::Result<()> {
         .write_to_file("rust_bindings.h");
 
     // Tell cargo to invalidate the built crate whenever the wrapper changes
-    println!("cargo:rerun-if-changed=bindgen_deps.h,walproposer.c,walproposer.h,test.c,libpostgres.a,../../pgxn/neon/walproposer.c,build.sh");
-    // println!("cargo:rustc-link-lib=walproposer");
-    // println!("cargo:rustc-link-lib=ext");
-    // println!("cargo:rustc-link-lib=pgport_srv");
-    // println!("cargo:rustc-link-lib=postgres");
-    // println!("cargo:rustc-link-lib=pgcommon_srv");
-    // println!("cargo:rustc-link-lib=pgport_srv");
+    println!("cargo:rerun-if-changed=bindgen_deps.h,test.c,../../pgxn/neon/walproposer.c,build.sh");
     println!("cargo:rustc-link-arg=-Wl,--start-group");
     println!("cargo:rustc-link-arg=-lsim");
     println!("cargo:rustc-link-arg=-lpgport_srv");
@@ -37,17 +29,6 @@ fn main() -> anyhow::Result<()> {
     println!("cargo:rustc-link-arg=-lm");
     println!("cargo:rustc-link-arg=-lwalproposer");
     println!("cargo:rustc-link-arg=-Wl,--end-group");
-    // println!("cargo:rustc-flags=-C default-linker-libraries=y");
-
-    // echo -lseccomp -lssl -lcrypto -lz -lpthread -lrt -ldl -lm
-
-    // println!("cargo:rustc-link-lib=ssl");
-    // println!("cargo:rustc-link-lib=crypto");
-    // println!("cargo:rustc-link-lib=walproposer2");
-    // println!("cargo:rustc-link-lib=postgres");
-    // println!("cargo:rustc-link-lib=pq");
-    // println!("cargo:rustc-link-lib=ssl");
-    // println!("cargo:rustc-link-lib=crypto");
     println!("cargo:rustc-link-search=/home/admin/simulator/libs/walproposer");
     // disable fPIE
     println!("cargo:rustc-link-arg=-no-pie");
@@ -62,10 +43,6 @@ fn main() -> anyhow::Result<()> {
         panic!("could not compile object file");
     }
 
-    // println!("cargo:rustc-link-lib=dylib=neon");
-    // println!("cargo:rustc-link-search=/Users/arthur/zen/zenith/pg_install/build/neon-v15");
-    // println!("cargo:rustc-link-arg=-Wl,-rpath,/Users/arthur/zen/zenith/pg_install/build/neon-v15");
-
     // // Finding the location of C headers for the Postgres server:
     // // - if POSTGRES_INSTALL_DIR is set look into it, otherwise look into `<project_root>/pg_install`
     // // - if there's a `bin/pg_config` file use it for getting include server, otherwise use `<project_root>/pg_install/{PG_MAJORVERSION}/include/postgresql/server`
@@ -113,8 +90,6 @@ fn main() -> anyhow::Result<()> {
 
     // let inc_pgxn_path = "/Users/arthur/zen/zenith/pgxn/neon";
 
-    // TODO: build a libwalproposer.a
-
     // The bindgen::Builder is the main entry point
     // to bindgen, and lets you build up options for
     // the resulting bindings.
@@ -128,7 +103,6 @@ fn main() -> anyhow::Result<()> {
         .allowlist_function("TestFunc")
         .allowlist_function("RunClientC")
         .allowlist_function("WalProposerRust")
-        // .allowlist_function("WalProposerRust")
         // .clang_arg(format!("-I{inc_server_path}"))
         // .clang_arg(format!("-I{inc_pgxn_path}"))
         // Finish the builder and generate the bindings.

libs/walproposer/rust_bindings.h

@@ -3,16 +3,34 @@
 #include <stdint.h>
 #include <stdlib.h>
 
-typedef struct ReplCell {
-  uint32_t value;
-  uint32_t client_id;
-  uint32_t seqno;
-} ReplCell;
+/**
+ * List of all possible AnyMessage.
+ */
+enum AnyMessageTag {
+  None,
+  InternalConnect,
+  Just32,
+  ReplCell,
+};
+typedef uint8_t AnyMessageTag;
 
+/**
+ * List of all possible NodeEvent.
+ */
+enum EventTag {
+  Accept,
+  Closed,
+  Message,
+};
+typedef uint8_t EventTag;
+
+/**
+ * Event returned by epoll_recv.
+ */
 typedef struct Event {
+  EventTag tag;
   int64_t tcp;
-  uint32_t value;
-  uint32_t tag;
+  AnyMessageTag any_message;
 } Event;
 
 void rust_function(uint32_t a);
@@ -28,6 +46,19 @@ uint32_t sim_id(void);
 
 int64_t sim_open_tcp(uint32_t dst);
 
-void sim_tcp_send(int64_t tcp, struct ReplCell value);
+/**
+ * Send MESSAGE_BUF content to the given tcp.
+ */
+void sim_tcp_send(int64_t tcp);
 
 struct Event sim_epoll_rcv(void);
+
+/**
+ * Read AnyMessage::Just32 message.
+ */
+void sim_msg_get_just_u32(uint32_t *val);
+
+/**
+ * Write AnyMessage::ReplCell message.
+ */
+void sim_msg_set_repl_cell(uint32_t value, uint32_t client_id, uint32_t seqno);

libs/walproposer/src/lib.rs

@@ -13,6 +13,7 @@ pub extern "C" fn rust_function(a: u32) {
 }
 
 pub mod sim;
+pub mod sim_proto;
 
 #[cfg(test)]
 mod test;

libs/walproposer/src/sim.rs

@@ -1,6 +1,8 @@
 use std::{cell::RefCell, collections::HashMap};
 
-use safekeeper::simlib::{node_os::NodeOs, network::TCP, proto::AnyMessage, world::NodeEvent, self};
+use safekeeper::simlib::{network::TCP, node_os::NodeOs, proto::AnyMessage, world::NodeEvent};
+
+use crate::sim_proto::{AnyMessageTag, Event, EventTag, MESSAGE_BUF};
 
 thread_local! {
     static CURRENT_NODE_OS: RefCell<Option<NodeOs>> = RefCell::new(None);
@@ -9,9 +11,7 @@ thread_local! {
 
 /// Get the current node os.
 fn os() -> NodeOs {
-    CURRENT_NODE_OS.with(|cell| {
-        cell.borrow().clone().expect("no node os set")
-    })
+    CURRENT_NODE_OS.with(|cell| cell.borrow().clone().expect("no node os set"))
 }
 
 fn tcp_save(tcp: TCP) -> i64 {
@@ -63,13 +63,9 @@ pub extern "C" fn sim_open_tcp(dst: u32) -> i64 {
 }
 
 #[no_mangle]
-// TODO: custom types!!
-pub extern "C" fn sim_tcp_send(tcp: i64, value: ReplCell) {
-    tcp_load(tcp).send(AnyMessage::ReplCell(simlib::proto::ReplCell {
-        value: value.value,
-        client_id: value.client_id,
-        seqno: value.seqno,
-    }));
+/// Send MESSAGE_BUF content to the given tcp.
+pub extern "C" fn sim_tcp_send(tcp: i64) {
+    tcp_load(tcp).send(MESSAGE_BUF.with(|cell| cell.borrow().clone()));
 }
 
 #[no_mangle]
@@ -77,38 +73,31 @@ pub extern "C" fn sim_epoll_rcv() -> Event {
     let event = os().epoll().recv();
     match event {
         NodeEvent::Accept(tcp) => Event {
+            tag: EventTag::Accept,
             tcp: tcp_save(tcp),
-            value: 0,
-            tag: 1,
+            any_message: AnyMessageTag::None,
         },
         NodeEvent::Closed(tcp) => Event {
+            tag: EventTag::Closed,
             tcp: tcp_save(tcp),
-            value: 0,
-            tag: 2,
-        },
-        NodeEvent::Message((message, tcp)) => Event {
-            tcp: tcp_save(tcp),
-            value: match message {
-                AnyMessage::Just32(value) => value.into(),
-                AnyMessage::ReplCell(cell) => cell.value,
-                _ => 0,
-            },
-            tag: 3,
+            any_message: AnyMessageTag::None,
         },
+        NodeEvent::Message((message, tcp)) => {
+            // store message in thread local storage, C code should use
+            // sim_msg_* functions to access it.
+            MESSAGE_BUF.with(|cell| {
+                *cell.borrow_mut() = message.clone();
+            });
+            Event {
+                tag: EventTag::Message,
+                tcp: tcp_save(tcp),
+                any_message: match message {
+                    AnyMessage::None => AnyMessageTag::None,
+                    AnyMessage::InternalConnect => AnyMessageTag::InternalConnect,
+                    AnyMessage::Just32(_) => AnyMessageTag::Just32,
+                    AnyMessage::ReplCell(_) => AnyMessageTag::ReplCell,
+                },
+            }
+        }
     }
 }
-
-#[repr(C)]
-pub struct Event {
-    pub tcp: i64,
-    // TODO: !!!
-    pub value: u32,
-    pub tag: u32,
-}
-
-#[repr(C)]
-pub struct ReplCell {
-    pub value: u32,
-    pub client_id: u32,
-    pub seqno: u32,
-}

libs/walproposer/src/sim_proto.rs

@@ -0,0 +1,54 @@
+use safekeeper::simlib::proto::{AnyMessage, ReplCell};
+use std::cell::RefCell;
+
+thread_local! {
+    pub static MESSAGE_BUF: RefCell<AnyMessage> = RefCell::new(AnyMessage::None);
+}
+
+#[no_mangle]
+/// Read AnyMessage::Just32 message.
+pub extern "C" fn sim_msg_get_just_u32(val: &mut u32) {
+    MESSAGE_BUF.with(|cell| match &*cell.borrow() {
+        AnyMessage::Just32(v) => {
+            *val = *v;
+        }
+        _ => panic!("expected Just32 message"),
+    });
+}
+
+#[no_mangle]
+/// Write AnyMessage::ReplCell message.
+pub extern "C" fn sim_msg_set_repl_cell(value: u32, client_id: u32, seqno: u32) {
+    MESSAGE_BUF.with(|cell| {
+        *cell.borrow_mut() = AnyMessage::ReplCell(ReplCell {
+            value,
+            client_id,
+            seqno,
+        });
+    });
+}
+
+#[repr(C)]
+/// Event returned by epoll_recv.
+pub struct Event {
+    pub tag: EventTag,
+    pub tcp: i64,
+    pub any_message: AnyMessageTag,
+}
+
+#[repr(u8)]
+/// List of all possible NodeEvent.
+pub enum EventTag {
+    Accept,
+    Closed,
+    Message,
+}
+
+#[repr(u8)]
+/// List of all possible AnyMessage.
+pub enum AnyMessageTag {
+    None,
+    InternalConnect,
+    Just32,
+    ReplCell,
+}

libs/walproposer/src/simtest/mod.rs

@@ -1,6 +1,7 @@
-use std::sync::Arc;
-
-use safekeeper::{simlib::{network::{NetworkOptions, Delay}, world::World}, simtest::{Options, start_simulation}};
+use safekeeper::{
+    simlib::network::{Delay, NetworkOptions},
+    simtest::{start_simulation, Options},
+};
 
 use crate::{bindings::RunClientC, sim::c_attach_node_os};
 
@@ -27,7 +28,9 @@ fn run_rust_c_test() {
         time_limit: 1_000_000,
         client_fn: Box::new(move |os, server_id| {
             c_attach_node_os(os);
-            unsafe { RunClientC(server_id); }
+            unsafe {
+                RunClientC(server_id);
+            }
         }),
         u32_data,
     });

libs/walproposer/src/test.rs

@@ -1,4 +1,4 @@
-use crate::bindings::{TestFunc, RunClientC, WalProposerRust};
+use crate::bindings::{TestFunc, WalProposerRust};
 
 #[test]
 fn test_rust_c_calls() {
@@ -9,5 +9,7 @@ fn test_rust_c_calls() {
 #[test]
 fn test_sim_bindings() {
     // unsafe { RunClientC(0); }
-    unsafe { WalProposerRust(); }
+    unsafe {
+        WalProposerRust();
+    }
 }

libs/walproposer/test.c

@@ -30,23 +30,28 @@ void RunClientC(uint32_t serverId) {
     int delivered = 0;
     int tcp = sim_open_tcp(serverId);
     while (delivered < data_len) {
-        ReplCell cell = {
-            .value = delivered+1,
-            .client_id = clientId,
-            .seqno = delivered,
-        };
-        sim_tcp_send(tcp, cell);
+        sim_msg_set_repl_cell(delivered+1, clientId, delivered);
+        sim_tcp_send(tcp);
 
         Event event = sim_epoll_rcv();
-        if (event.tag == 2) {
-            // closed
+        switch (event.tag)
+        {
+        case Closed:
             elog(LOG, "connection closed");
             tcp = sim_open_tcp(serverId);
-        } else if (event.tag == 3) {
-            // got message
-            if (event.value == delivered + 1) {
+            break;
+
+        case Message:
+            Assert(event.any_message == Just32);
+            uint32_t val;
+            sim_msg_get_just_u32(&val);
+            if (val == delivered + 1) {
                 delivered += 1;
             }
+            break;
+
+        default:
+            Assert(false);
         }
     }
 }

safekeeper/src/simlib/proto.rs

@@ -1,8 +1,9 @@
 /// All possible flavours of messages.
 /// Grouped by the receiver node.
 #[derive(Clone, Debug)]
-#[repr(C)]
 pub enum AnyMessage {
+    /// Not used, empty placeholder.
+    None,
     /// Used internally for notifying node about new incoming connection.
     InternalConnect,
     Just32(u32),
@@ -10,7 +11,6 @@ pub enum AnyMessage {
 }
 
 #[derive(Clone, Debug)]
-#[repr(C)]
 pub struct ReplCell {
     pub value: u32,
     pub client_id: u32,

safekeeper/src/simtest/mod.rs

@@ -6,41 +6,49 @@ use std::sync::Arc;
 
 use crate::{
     simlib::{
-        network::{Delay, NetworkOptions},
+        network::{NetworkOptions},
         proto::ReplCell,
         world::World, node_os::NodeOs,
     },
-    simtest::{client::run_client, disk::SharedStorage, server::run_server},
+    simtest::{disk::SharedStorage, server::run_server},
 };
 
-#[test]
-fn run_pure_rust_test() {
-    let delay = Delay {
-        min: 1,
-        max: 60,
-        fail_prob: 0.4,
-    };
+#[cfg(test)]
+mod tests {
+    use crate::simlib::network::{Delay, NetworkOptions};
 
-    let network = NetworkOptions {
-        timeout: Some(50),
-        connect_delay: delay.clone(),
-        send_delay: delay.clone(),
-    };
+    use super::{u32_to_cells, start_simulation, Options, client::run_client};
 
-    for seed in 0..2000 {
-        let u32_data: [u32; 5] = [1, 2, 3, 4, 5];
-        let data = u32_to_cells(&u32_data, 1);
+    #[test]
+    fn run_pure_rust_test() {
+        let delay = Delay {
+            min: 1,
+            max: 60,
+            fail_prob: 0.4,
+        };
 
-        start_simulation(Options {
-            seed,
-            network: network.clone(),
-            time_limit: 1_000_000,
-            client_fn: Box::new(move |os, server_id| {
-                run_client(os, &data, server_id)
-            }),
-            u32_data,
-        });
+        let network = NetworkOptions {
+            timeout: Some(50),
+            connect_delay: delay.clone(),
+            send_delay: delay.clone(),
+        };
+
+        for seed in 0..20 {
+            let u32_data: [u32; 5] = [1, 2, 3, 4, 5];
+            let data = u32_to_cells(&u32_data, 1);
+
+            start_simulation(Options {
+                seed,
+                network: network.clone(),
+                time_limit: 1_000_000,
+                client_fn: Box::new(move |os, server_id| {
+                    run_client(os, &data, server_id)
+                }),
+                u32_data,
+            });
+        }
     }
+
 }
 
 pub struct Options {
@@ -79,7 +87,7 @@ pub fn start_simulation(options: Options) {
     assert!(verify_data(&disk_data, &options.u32_data[..]));
 }
 
-fn u32_to_cells(data: &[u32], client_id: u32) -> Vec<ReplCell> {
+pub fn u32_to_cells(data: &[u32], client_id: u32) -> Vec<ReplCell> {
     let mut res = Vec::new();
     for i in 0..data.len() {
         res.push(ReplCell {