feat(ipc): MPSC shared-memory transport with multi-client echo test

Adds a multi-producer, single-consumer shared-memory IPC transport. Internally this is **N SPSC rings (one per producer) + a shared doorbell futex** — there is no shared producer ring. Each client writes only to its own SPSC ring; producers ring the doorbell to wake the consumer, which then checks which ring has data and reads from it. Each client also has its own SPSC response ring.

This is the same pattern that existed in earlier IPC iterations and is what \`aztec-wsdb\` will could use when serving both the TS layer and a C++ AVM worker pool from the same world state if we need the performance.

Right now though we'll use UDS. This is just if needed in future.

Author: charlielye

barretenberg/cpp/src/CMakeLists.txt

@@ -127,6 +127,8 @@ if(NOT FUZZING AND NOT WASM AND NOT BB_LITE)
     # NOTE: Do not conditionally base this on the AVM flag as it defines a necessary vm2_sim library.
     add_subdirectory(barretenberg/vm2)
     add_subdirectory(barretenberg/ipc)
+    add_subdirectory(barretenberg/wsdb)
+    add_subdirectory(barretenberg/wsdb_client)
     add_subdirectory(barretenberg/nodejs_module)
 endif()
 

barretenberg/cpp/src/barretenberg/ipc/ipc_client.cpp

@@ -1,4 +1,5 @@
 #include "barretenberg/ipc/ipc_client.hpp"
+#include "barretenberg/ipc/mpsc_shm_client.hpp"
 #include "barretenberg/ipc/shm_client.hpp"
 #include "barretenberg/ipc/socket_client.hpp"
 #include <cstddef>
@@ -17,4 +18,9 @@ std::unique_ptr<IpcClient> IpcClient::create_shm(const std::string& base_name)
     return std::make_unique<ShmClient>(base_name);
 }
 
+std::unique_ptr<IpcClient> IpcClient::create_mpsc_shm(const std::string& base_name, size_t client_id)
+{
+    return std::make_unique<MpscShmClient>(base_name, client_id);
+}
+
 } // namespace bb::ipc

barretenberg/cpp/src/barretenberg/ipc/ipc_client.hpp

@@ -72,6 +72,7 @@ class IpcClient {
     // Factory methods
     static std::unique_ptr<IpcClient> create_socket(const std::string& socket_path);
     static std::unique_ptr<IpcClient> create_shm(const std::string& base_name);
+    static std::unique_ptr<IpcClient> create_mpsc_shm(const std::string& base_name, size_t client_id);
 };
 
 } // namespace bb::ipc

barretenberg/cpp/src/barretenberg/ipc/ipc_server.cpp

@@ -1,4 +1,5 @@
 #include "barretenberg/ipc/ipc_server.hpp"
+#include "barretenberg/ipc/mpsc_shm_server.hpp"
 #include "barretenberg/ipc/shm_server.hpp"
 #include "barretenberg/ipc/socket_server.hpp"
 #include <cstddef>
@@ -19,4 +20,12 @@ std::unique_ptr<IpcServer> IpcServer::create_shm(const std::string& base_name,
     return std::make_unique<ShmServer>(base_name, request_ring_size, response_ring_size);
 }
 
+std::unique_ptr<IpcServer> IpcServer::create_mpsc_shm(const std::string& base_name,
+                                                      size_t max_clients,
+                                                      size_t request_ring_size,
+                                                      size_t response_ring_size)
+{
+    return std::make_unique<MpscShmServer>(base_name, max_clients, request_ring_size, response_ring_size);
+}
+
 } // namespace bb::ipc

barretenberg/cpp/src/barretenberg/ipc/ipc_server.hpp

@@ -194,6 +194,10 @@ class IpcServer {
     static std::unique_ptr<IpcServer> create_shm(const std::string& base_name,
                                                  size_t request_ring_size = static_cast<size_t>(1024 * 1024),
                                                  size_t response_ring_size = static_cast<size_t>(1024 * 1024));
+    static std::unique_ptr<IpcServer> create_mpsc_shm(const std::string& base_name,
+                                                      size_t max_clients,
+                                                      size_t request_ring_size = static_cast<size_t>(1024 * 1024),
+                                                      size_t response_ring_size = static_cast<size_t>(1024 * 1024));
 
   protected:
     std::atomic<bool> shutdown_requested_{ false };

barretenberg/cpp/src/barretenberg/ipc/mpsc_shm_client.hpp

@@ -0,0 +1,111 @@
+#pragma once
+
+#include "ipc_client.hpp"
+#include "shm/mpsc_shm.hpp"
+#include "shm/spsc_shm.hpp"
+#include "shm_common.hpp"
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <optional>
+#include <string>
+#include <utility>
+
+namespace bb::ipc {
+
+/**
+ * @brief IPC client for multi-client shared memory server
+ *
+ * Uses MpscProducer for sending requests and a dedicated SPSC ring for
+ * receiving responses. Each client is assigned a unique client_id.
+ */
+class MpscShmClient : public IpcClient {
+  public:
+    MpscShmClient(std::string base_name, size_t client_id)
+        : base_name_(std::move(base_name))
+        , client_id_(client_id)
+    {}
+
+    ~MpscShmClient() override = default;
+
+    // Non-copyable, non-movable
+    MpscShmClient(const MpscShmClient&) = delete;
+    MpscShmClient& operator=(const MpscShmClient&) = delete;
+    MpscShmClient(MpscShmClient&&) = delete;
+    MpscShmClient& operator=(MpscShmClient&&) = delete;
+
+    bool connect() override
+    {
+        if (producer_.has_value()) {
+            return true; // Already connected
+        }
+
+        try {
+            // Connect as producer to the MPSC request system
+            producer_ = MpscProducer::connect(base_name_ + "_req", client_id_);
+
+            // Connect to our dedicated SPSC response ring
+            std::string resp_name = base_name_ + "_resp_" + std::to_string(client_id_);
+            response_ring_ = SpscShm::connect(resp_name);
+
+            return true;
+        } catch (...) {
+            producer_.reset();
+            response_ring_.reset();
+            return false;
+        }
+    }
+
+    bool send(const void* data, size_t len, uint64_t timeout_ns) override
+    {
+        if (!producer_.has_value()) {
+            return false;
+        }
+
+        // Claim space for length prefix + data
+        size_t total_size = sizeof(uint32_t) + len;
+        void* buf = producer_->claim(total_size, static_cast<uint32_t>(timeout_ns));
+        if (buf == nullptr) {
+            return false;
+        }
+
+        // Write length prefix + data
+        auto len_u32 = static_cast<uint32_t>(len);
+        std::memcpy(buf, &len_u32, sizeof(uint32_t));
+        std::memcpy(static_cast<uint8_t*>(buf) + sizeof(uint32_t), data, len);
+
+        // Publish (rings doorbell to wake server)
+        producer_->publish(total_size);
+        return true;
+    }
+
+    std::span<const uint8_t> receive(uint64_t timeout_ns) override
+    {
+        if (!response_ring_.has_value()) {
+            return {};
+        }
+        return ring_receive_msg(response_ring_.value(), timeout_ns);
+    }
+
+    void release(size_t message_size) override
+    {
+        if (!response_ring_.has_value()) {
+            return;
+        }
+        response_ring_->release(sizeof(uint32_t) + message_size);
+    }
+
+    void close() override
+    {
+        producer_.reset();
+        response_ring_.reset();
+    }
+
+  private:
+    std::string base_name_;
+    size_t client_id_;
+    std::optional<MpscProducer> producer_;
+    std::optional<SpscShm> response_ring_;
+};
+
+} // namespace bb::ipc

barretenberg/cpp/src/barretenberg/ipc/mpsc_shm_server.hpp

@@ -0,0 +1,154 @@
+#pragma once
+
+#include "ipc_server.hpp"
+#include "shm/mpsc_shm.hpp"
+#include "shm/spsc_shm.hpp"
+#include "shm_common.hpp"
+#include <cstdint>
+#include <iostream>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace bb::ipc {
+
+/**
+ * @brief IPC server implementation using shared memory with multi-client support
+ *
+ * Uses MPSC (multi-producer single-consumer) for requests and per-client SPSC
+ * rings for responses. Supports up to max_clients concurrent clients.
+ *
+ * Shared memory layout:
+ * - Request: MPSC consumer with one SPSC ring per client (client writes, server reads)
+ * - Response: Separate SPSC ring per client (server writes, client reads)
+ */
+class MpscShmServer : public IpcServer {
+  public:
+    static constexpr size_t DEFAULT_RING_SIZE = 1 << 20; // 1MB
+
+    MpscShmServer(std::string base_name,
+                  size_t max_clients,
+                  size_t request_ring_size = DEFAULT_RING_SIZE,
+                  size_t response_ring_size = DEFAULT_RING_SIZE)
+        : base_name_(std::move(base_name))
+        , max_clients_(max_clients)
+        , request_ring_size_(request_ring_size)
+        , response_ring_size_(response_ring_size)
+    {}
+
+    ~MpscShmServer() override { close(); }
+
+    // Non-copyable, non-movable
+    MpscShmServer(const MpscShmServer&) = delete;
+    MpscShmServer& operator=(const MpscShmServer&) = delete;
+    MpscShmServer(MpscShmServer&&) = delete;
+    MpscShmServer& operator=(MpscShmServer&&) = delete;
+
+    bool listen() override
+    {
+        if (request_consumer_.has_value()) {
+            return true; // Already listening
+        }
+
+        // Clean up any leftover shared memory
+        MpscConsumer::unlink(base_name_ + "_req", max_clients_);
+        for (size_t i = 0; i < max_clients_; i++) {
+            SpscShm::unlink(base_name_ + "_resp_" + std::to_string(i));
+        }
+
+        try {
+            // Create MPSC consumer for requests (one ring per client)
+            request_consumer_ = MpscConsumer::create(base_name_ + "_req", max_clients_, request_ring_size_);
+
+            // Create per-client SPSC response rings
+            response_rings_.reserve(max_clients_);
+            for (size_t i = 0; i < max_clients_; i++) {
+                std::string resp_name = base_name_ + "_resp_" + std::to_string(i);
+                response_rings_.push_back(SpscShm::create(resp_name, response_ring_size_));
+            }
+
+            return true;
+        } catch (...) {
+            close();
+            return false;
+        }
+    }
+
+    int wait_for_data(uint64_t timeout_ns) override
+    {
+        if (!request_consumer_.has_value()) {
+            return -1;
+        }
+        // MpscConsumer::wait_for_data returns ring index = client_id
+        return request_consumer_->wait_for_data(static_cast<uint32_t>(timeout_ns));
+    }
+
+    std::span<const uint8_t> receive(int client_id) override
+    {
+        if (!request_consumer_.has_value() || client_id < 0 || static_cast<size_t>(client_id) >= max_clients_) {
+            return {};
+        }
+        // Peek on the specific client's request ring via MpscConsumer
+        void* len_ptr = request_consumer_->peek(static_cast<size_t>(client_id), sizeof(uint32_t), 100000000);
+        if (len_ptr == nullptr) {
+            return {};
+        }
+        uint32_t msg_len = 0;
+        std::memcpy(&msg_len, len_ptr, sizeof(uint32_t));
+
+        void* msg_ptr = request_consumer_->peek(static_cast<size_t>(client_id), sizeof(uint32_t) + msg_len, 100000000);
+        if (msg_ptr == nullptr) {
+            return {};
+        }
+        return std::span<const uint8_t>(static_cast<const uint8_t*>(msg_ptr) + sizeof(uint32_t), msg_len);
+    }
+
+    void release(int client_id, size_t message_size) override
+    {
+        if (!request_consumer_.has_value() || client_id < 0 || static_cast<size_t>(client_id) >= max_clients_) {
+            return;
+        }
+        request_consumer_->release(static_cast<size_t>(client_id), sizeof(uint32_t) + message_size);
+    }
+
+    bool send(int client_id, const void* data, size_t len) override
+    {
+        if (client_id < 0 || static_cast<size_t>(client_id) >= response_rings_.size()) {
+            return false;
+        }
+        return ring_send_msg(response_rings_[static_cast<size_t>(client_id)], data, len, 100000000);
+    }
+
+    void close() override
+    {
+        request_consumer_.reset();
+        response_rings_.clear();
+
+        // Clean up shared memory
+        MpscConsumer::unlink(base_name_ + "_req", max_clients_);
+        for (size_t i = 0; i < max_clients_; i++) {
+            SpscShm::unlink(base_name_ + "_resp_" + std::to_string(i));
+        }
+    }
+
+    void wakeup_all() override
+    {
+        if (request_consumer_.has_value()) {
+            request_consumer_->wakeup_all();
+        }
+        for (auto& ring : response_rings_) {
+            ring.wakeup_all();
+        }
+    }
+
+  private:
+    std::string base_name_;
+    size_t max_clients_;
+    size_t request_ring_size_;
+    size_t response_ring_size_;
+    std::optional<MpscConsumer> request_consumer_;
+    std::vector<SpscShm> response_rings_;
+};
+
+} // namespace bb::ipc