Support 16-node distributed Soft-RoCE routing

include/distributed_server.h

@@ -316,8 +316,8 @@ typedef struct {
 #include <unordered_map>
 
 /* TCP transport support */
-#include "tcp_communication.h"
 #include "shared_memory_manager.h"
+#include "tcp_communication.h"
 
 /* RDMA transport support */
 #include "rdma_communication.h"
@@ -402,6 +402,8 @@ struct RDMACalibrationResult {
 /* Per-node RDMA connection state */
 struct RDMANodeConnection {
     std::unique_ptr<RDMAClient> client;
+    std::string endpoint_addr;
+    uint16_t endpoint_port;
     // Outgoing RDMA connection
     uint64_t remote_addr;
     // Remote base address
@@ -412,7 +414,7 @@ struct RDMANodeConnection {
     RDMACalibrationResult calibration;
     // Per-node calibration data
 
-    RDMANodeConnection() : remote_addr(0), remote_buffer_size(0), connected(false) {}
+    RDMANodeConnection() : endpoint_port(0), remote_addr(0), remote_buffer_size(0), connected(false) {}
 };
 
 /* Message handler callback type */
@@ -646,6 +648,9 @@ class DistributedRDMATransport {
     // RDMA server for incoming connections
     std::unique_ptr<RDMAServer> server_;
     std::thread accept_thread_;
+    std::vector<std::shared_ptr<RDMAConnection>> incoming_connections_;
+    std::vector<std::thread> incoming_threads_;
+    std::mutex incoming_mutex_;
     std::atomic<bool> running_;
 
     // Calibration results per node
@@ -660,8 +665,18 @@ class DistributedRDMATransport {
     bool initialize();
     void shutdown();
 
+    // Route incoming two-sided RDMA requests to the distributed server's
+    // local-memory implementation. Without this, RDMAServer's default
+    // handler only returns a synthetic success response.
+    void set_message_handler(RDMAConnection::MessageHandler handler) {
+        if (server_) {
+            server_->set_message_handler(handler);
+        }
+    }
+
     // Connection management
-    bool connect_to_node(uint32_t node_id, const std::string &addr, uint16_t port);
+    bool connect_to_node(uint32_t node_id, const std::string &addr, uint16_t port, uint64_t remote_addr,
+                         size_t remote_buffer_size);
     void disconnect_node(uint32_t node_id);
     bool is_connected(uint32_t node_id) const;
     std::vector<uint32_t> get_connected_nodes() const;
@@ -761,6 +776,8 @@ class DistributedMemoryServer {
     /* Memory operations (may be forwarded to remote nodes) */
     int read(uint64_t addr, void *data, size_t size, uint64_t *latency_ns);
     int write(uint64_t addr, const void *data, size_t size, uint64_t *latency_ns);
+    int read_bulk(uint64_t addr, void *data, size_t size, uint64_t *latency_ns);
+    int write_bulk(uint64_t addr, const void *data, size_t size, uint64_t *latency_ns);
     int atomic_faa(uint64_t addr, uint64_t value, uint64_t *old_value);
     int atomic_cas(uint64_t addr, uint64_t expected, uint64_t desired, uint64_t *old_value);
     void fence();

include/rdma_communication.h

@@ -4,6 +4,7 @@
 #include <atomic>
 #include <cstdint>
 #include <functional>
+#include <map>
 #include <memory>
 #include <string>
 
@@ -12,7 +13,7 @@
 #include <rdma/rdma_cma.h>
 #endif
 
-#define RDMA_BUFFER_SIZE 4096
+#define RDMA_BUFFER_SIZE (64 * 1024)
 #define RDMA_CQ_SIZE 1024
 #define RDMA_MAX_WR 512
 #define RDMA_CACHELINE_SIZE 64
@@ -26,14 +27,14 @@ struct RDMARequest {
     uint64_t timestamp;
     uint8_t host_id;
     uint64_t virtual_addr;
-    uint8_t data[RDMA_CACHELINE_SIZE];
+    uint8_t data[RDMA_BUFFER_SIZE];
 } __attribute__((packed));
 
 struct RDMAResponse {
     uint8_t status;
     uint64_t latency_ns;
     uint8_t cache_state;
-    uint8_t data[RDMA_CACHELINE_SIZE];
+    uint8_t data[RDMA_BUFFER_SIZE];
 } __attribute__((packed));
 
 struct RDMAMessage {
@@ -83,9 +84,13 @@ class RDMAConnection {
     RDMAConnection();
     virtual ~RDMAConnection();
 
+#ifdef HAS_RDMA
+    int accept_cm_id(struct rdma_cm_id *id);
+#endif
+    void mark_connected();
     void set_message_handler(MessageHandler handler) { message_handler_ = handler; }
     int send_message(const RDMAMessage &msg);
-    int receive_message(RDMAMessage &msg);
+    int receive_message(RDMAMessage &msg, int timeout_ms = -1);
     bool is_connected() const { return connected_.load(); }
     void disconnect();
 };
@@ -96,15 +101,16 @@ class RDMAServer : public RDMAConnection {
     uint16_t port_;
 #ifdef HAS_RDMA
     struct rdma_cm_id *listen_id_;
+    std::map<struct rdma_cm_id *, std::shared_ptr<RDMAConnection>> pending_connections_;
 #endif
 
 public:
     RDMAServer(const std::string &addr, uint16_t port);
     ~RDMAServer();
 
     int start();
-    int accept_connection();
-    void handle_client();
+    std::shared_ptr<RDMAConnection> accept_connection();
+    void handle_client(const std::shared_ptr<RDMAConnection> &client);
     void stop();
 };
 

include/shared_memory_manager.h

@@ -143,6 +143,8 @@ class SharedMemoryManager {
     uint8_t *get_cacheline_data(uint64_t cacheline_addr);
     bool read_cacheline(uint64_t addr, uint8_t *buffer, size_t size);
     bool write_cacheline(uint64_t addr, const uint8_t *data, size_t size);
+    bool read_range(uint64_t addr, uint8_t *buffer, size_t size);
+    bool write_range(uint64_t addr, const uint8_t *data, size_t size);
     bool atomic_fetch_add_uint64(uint64_t addr, uint64_t value, uint64_t *old_value);
     bool atomic_compare_exchange_uint64(uint64_t addr, uint64_t expected, uint64_t desired, uint64_t *old_value);
     bool flush();