fix(ipc): auto-claim shm client slots

Author: charlielye

ipc-codegen/src/typescript_package_codegen.ts

@@ -311,7 +311,9 @@ async function connectClient(
       }
 ${supportsShm ? `      if (transport === 'shm') {
         return createNapiShmAsyncClient(ipcPath.replace(/\\.shm$/, ''), {
-          clientId: options.clientId ?? 0,
+          // Pass clientId through as-is: when unset, the client self-allocates a
+          // free producer slot (don't default to 0, which aliases every client).
+          clientId: options.clientId,
           customAddonPath: options.napiPath,
         });
       }

ipc-runtime/cpp/ipc_runtime/ipc_client.hpp

@@ -2,13 +2,19 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <limits>
 #include <memory>
 #include <span>
 #include <string>
 #include <sys/types.h>
 
 namespace ipc {
 
+// Passed as the MPSC-SHM client id to mean "atomically claim a free slot on
+// connect" rather than pinning a specific one. This is the default for
+// make_client / create_mpsc_shm; an explicit id is only used by tests.
+inline constexpr std::size_t kAutoClientId = std::numeric_limits<std::size_t>::max();
+
 /**
  * @brief Abstract interface for IPC client
  *
@@ -84,7 +90,7 @@ class IpcClient {
     static std::unique_ptr<IpcClient> create_shm(const std::string& base_name);
     // Multi-producer SHM: one request ring per client slot and one response
     // ring per client slot. This is what make_client("*.shm") selects.
-    static std::unique_ptr<IpcClient> create_mpsc_shm(const std::string& base_name, size_t client_id);
+    static std::unique_ptr<IpcClient> create_mpsc_shm(const std::string& base_name, size_t client_id = kAutoClientId);
 };
 
 /**
@@ -95,12 +101,12 @@ class IpcClient {
  *  - "*.shm"  → IpcClient::create_mpsc_shm(<basename>, client_id)
  *
  * Returns nullptr if the suffix is not recognised. `shm_client_id` is only
- * consulted for the SHM path; for MPSC-SHM, each connecting client picks a
- * distinct slot (0..max_clients-1).
+ * consulted for the SHM path; it defaults to kAutoClientId, so each connecting
+ * client atomically claims a distinct free slot (0..max_clients-1) on connect.
  *
  * @param input_path Path passed by the caller (often a CLI flag).
- * @param shm_client_id Client slot to claim in MPSC-SHM mode. Ignored for UDS.
+ * @param shm_client_id MPSC-SHM slot to pin, or kAutoClientId to self-allocate. Ignored for UDS.
  */
-std::unique_ptr<IpcClient> make_client(const std::string& input_path, std::size_t shm_client_id = 0);
+std::unique_ptr<IpcClient> make_client(const std::string& input_path, std::size_t shm_client_id = kAutoClientId);
 
 } // namespace ipc

ipc-runtime/cpp/ipc_runtime/mpsc_shm_client.hpp

@@ -24,8 +24,13 @@ namespace ipc {
  */
 class MpscShmClient : public IpcClient {
   public:
+    // client_id == kAutoClientId (the default for make_client / create_mpsc_shm)
+    // makes connect() atomically claim a free producer slot from the server's
+    // shared slot table, so callers don't have to coordinate unique ids. An
+    // explicit id pins a specific slot (used by tests).
     MpscShmClient(std::string base_name, size_t client_id)
         : base_name_(std::move(base_name))
+        , auto_claim_(client_id == kAutoClientId)
         , client_id_(client_id)
     {}
 
@@ -48,6 +53,14 @@ class MpscShmClient : public IpcClient {
 
         for (size_t attempt = 0; attempt < max_attempts; ++attempt) {
             try {
+                // Self-allocate a producer slot when no explicit id was given.
+                // The claim is held for the connection's lifetime and released by
+                // close()/destruction so the slot can be reused.
+                if (auto_claim_) {
+                    slot_claim_ = MpscSlotClaim::claim(base_name_ + "_req");
+                    client_id_ = slot_claim_->id();
+                }
+
                 // Connect as producer to the MPSC request system
                 producer_ = MpscProducer::connect(base_name_ + "_req", client_id_);
 
@@ -59,6 +72,7 @@ class MpscShmClient : public IpcClient {
             } catch (...) {
                 producer_.reset();
                 response_ring_.reset();
+                slot_claim_.reset();
                 if (attempt + 1 == max_attempts) {
                     return false;
                 }
@@ -112,6 +126,7 @@ class MpscShmClient : public IpcClient {
     {
         producer_.reset();
         response_ring_.reset();
+        slot_claim_.reset(); // release our producer slot for reuse
     }
 
     void wakeup() override
@@ -123,7 +138,9 @@ class MpscShmClient : public IpcClient {
 
   private:
     std::string base_name_;
+    bool auto_claim_;
     size_t client_id_;
+    std::optional<MpscSlotClaim> slot_claim_;
     std::optional<MpscProducer> producer_;
     std::optional<SpscShm> response_ring_;
 };

ipc-runtime/cpp/ipc_runtime/shm.test.cpp

@@ -505,4 +505,218 @@ TEST(ShmTest, MpscReactorPipelinedConcurrencyAndOrder)
     server->close();
 }
 
+TEST(ShmTest, MpscReactorMultiClientResponseRouting)
+{
+    constexpr uint32_t NUM_CLIENTS = 4;
+    constexpr uint32_t K = 128;
+    constexpr size_t RING_SIZE = 64UL * 1024;
+
+    std::string base_name = "shm_mpsc_multi_" + std::to_string(getpid());
+    auto server = IpcServer::create_mpsc_shm(base_name, NUM_CLIENTS, RING_SIZE, RING_SIZE);
+    ASSERT_TRUE(server->listen()) << "MPSC multi-client server failed to listen";
+
+    ReactorTestPool pool(8);
+    std::thread server_thread([&]() {
+        server->run_reactor([&pool](int, std::span<const uint8_t> req, IpcServer::Respond respond) {
+            std::vector<uint8_t> r(req.begin(), req.end());
+            pool.enqueue([r = std::move(r), respond = std::move(respond)]() mutable {
+                uint32_t seq = 0;
+                std::memcpy(&seq, r.data() + sizeof(uint32_t), sizeof(uint32_t));
+                std::this_thread::sleep_for(std::chrono::microseconds(20 * (seq % 8)));
+                respond(std::move(r));
+            });
+        });
+    });
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    std::atomic<int> wrong_client{ 0 };
+    std::atomic<int> out_of_order{ 0 };
+    std::atomic<int> stalls{ 0 };
+
+    auto run_client = [&](uint32_t c) {
+        auto client = IpcClient::create_mpsc_shm(base_name, c);
+        if (!client->connect()) {
+            stalls++;
+            return;
+        }
+        for (uint32_t s = 0; s < K; s++) {
+            uint32_t msg[2] = { c, s };
+            while (!client->send(msg, sizeof(msg), 100'000'000ULL)) {
+            }
+        }
+        for (uint32_t s = 0; s < K; s++) {
+            std::span<const uint8_t> resp;
+            size_t empties = 0;
+            while ((resp = client->receive(100'000'000ULL)).empty()) {
+                if (++empties > 50) {
+                    stalls++;
+                    break;
+                }
+            }
+            if (resp.empty()) {
+                break;
+            }
+            uint32_t got[2] = { 0, 0 };
+            std::memcpy(got, resp.data(), sizeof(got));
+            if (got[0] != c) {
+                wrong_client++;
+            }
+            if (got[1] != s) {
+                out_of_order++;
+            }
+            client->release(resp.size());
+        }
+        client->close();
+    };
+
+    std::vector<std::thread> clients;
+    for (uint32_t c = 0; c < NUM_CLIENTS; c++) {
+        clients.emplace_back(run_client, c);
+    }
+    for (auto& t : clients) {
+        t.join();
+    }
+
+    server->request_shutdown();
+    server_thread.join();
+    server->close();
+
+    EXPECT_EQ(wrong_client.load(), 0) << "responses were routed to the wrong client";
+    EXPECT_EQ(out_of_order.load(), 0) << "responses arrived out of order within a client";
+    EXPECT_EQ(stalls.load(), 0) << "a client stalled waiting for its responses";
+}
+
+TEST(ShmTest, MpscReactorAutoClaimMultiClient)
+{
+    constexpr uint32_t NUM_CLIENTS = 4;
+    constexpr uint32_t K = 128;
+    constexpr size_t RING_SIZE = 64UL * 1024;
+
+    std::string base_name = "shm_mpsc_autoclaim_" + std::to_string(getpid());
+    auto server = IpcServer::create_mpsc_shm(base_name, NUM_CLIENTS, RING_SIZE, RING_SIZE);
+    ASSERT_TRUE(server->listen()) << "MPSC auto-claim server failed to listen";
+
+    ReactorTestPool pool(8);
+    std::thread server_thread([&]() {
+        server->run_reactor([&pool](int, std::span<const uint8_t> req, IpcServer::Respond respond) {
+            std::vector<uint8_t> r(req.begin(), req.end());
+            pool.enqueue([r = std::move(r), respond = std::move(respond)]() mutable {
+                uint32_t seq = 0;
+                std::memcpy(&seq, r.data() + sizeof(uint32_t), sizeof(uint32_t));
+                std::this_thread::sleep_for(std::chrono::microseconds(20 * (seq % 8)));
+                respond(std::move(r));
+            });
+        });
+    });
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    std::atomic<int> wrong_client{ 0 };
+    std::atomic<int> out_of_order{ 0 };
+    std::atomic<int> stalls{ 0 };
+
+    auto run_client = [&](uint32_t c) {
+        auto client = IpcClient::create_mpsc_shm(base_name);
+        if (!client->connect()) {
+            stalls++;
+            return;
+        }
+        for (uint32_t s = 0; s < K; s++) {
+            uint32_t msg[2] = { c, s };
+            while (!client->send(msg, sizeof(msg), 100'000'000ULL)) {
+            }
+        }
+        for (uint32_t s = 0; s < K; s++) {
+            std::span<const uint8_t> resp;
+            size_t empties = 0;
+            while ((resp = client->receive(100'000'000ULL)).empty()) {
+                if (++empties > 50) {
+                    stalls++;
+                    break;
+                }
+            }
+            if (resp.empty()) {
+                break;
+            }
+            uint32_t got[2] = { 0, 0 };
+            std::memcpy(got, resp.data(), sizeof(got));
+            if (got[0] != c) {
+                wrong_client++;
+            }
+            if (got[1] != s) {
+                out_of_order++;
+            }
+            client->release(resp.size());
+        }
+        client->close();
+    };
+
+    std::vector<std::thread> clients;
+    for (uint32_t c = 0; c < NUM_CLIENTS; c++) {
+        clients.emplace_back(run_client, c);
+    }
+    for (auto& t : clients) {
+        t.join();
+    }
+
+    server->request_shutdown();
+    server_thread.join();
+    server->close();
+
+    EXPECT_EQ(wrong_client.load(), 0) << "self-allocated clients aliased onto a shared slot";
+    EXPECT_EQ(out_of_order.load(), 0) << "responses arrived out of order within a client";
+    EXPECT_EQ(stalls.load(), 0) << "a client stalled";
+}
+
+TEST(ShmTest, MpscSlotFreedOnDisconnectAndReclaimed)
+{
+    constexpr size_t NUM_CLIENTS = 1;
+    constexpr size_t RING_SIZE = 16UL * 1024;
+
+    std::string base_name = "shm_mpsc_reuse_" + std::to_string(getpid());
+    auto server = IpcServer::create_mpsc_shm(base_name, NUM_CLIENTS, RING_SIZE, RING_SIZE);
+    ASSERT_TRUE(server->listen()) << "server failed to listen";
+
+    ReactorTestPool pool(2);
+    std::thread server_thread([&]() {
+        server->run_reactor([&pool](int, std::span<const uint8_t> req, IpcServer::Respond respond) {
+            std::vector<uint8_t> r(req.begin(), req.end());
+            pool.enqueue([r = std::move(r), respond = std::move(respond)]() mutable { respond(std::move(r)); });
+        });
+    });
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    auto round_trip = [](IpcClient& client, uint32_t value) -> bool {
+        if (!client.send(&value, sizeof(value), 100'000'000ULL)) {
+            return false;
+        }
+        std::span<const uint8_t> resp;
+        size_t empties = 0;
+        while ((resp = client.receive(100'000'000ULL)).empty()) {
+            if (++empties > 50) {
+                return false;
+            }
+        }
+        uint32_t got = 0;
+        std::memcpy(&got, resp.data(), sizeof(got));
+        client.release(resp.size());
+        return got == value;
+    };
+
+    {
+        auto a = IpcClient::create_mpsc_shm(base_name);
+        ASSERT_TRUE(a->connect()) << "client A failed to connect";
+        EXPECT_TRUE(round_trip(*a, 0x1111)) << "client A round-trip failed";
+        a->close();
+    }
+
+    auto b = IpcClient::create_mpsc_shm(base_name);
+    ASSERT_TRUE(b->connect()) << "client B failed to reclaim the freed slot";
+    EXPECT_TRUE(round_trip(*b, 0x2222)) << "client B round-trip over the reused ring failed";
+    b->close();
+
+    server->request_shutdown();
+    server_thread.join();
+    server->close();
+}
+
 } // namespace