Add concurrent

Author: f0cii

.gitignore

@@ -9,4 +9,5 @@ Cargo.lock
 # These are backup files generated by rustfmt
 **/*.rs.bk
 
-.idea/
+/.idea/
+/.vscode/

README.md

@@ -1,2 +1,126 @@
 # erpc-rs
 eRPC library for Rust
+
+# Installing junction & turf
+```
+$ git clone https://github.com/preshing/junction.git
+$ git clone https://github.com/preshing/turf.git
+$ cd junction
+$ mkdir build
+$ cd build
+$ cmake -DCMAKE_INSTALL_PREFIX=~/junction-install -DJUNCTION_WITH_SAMPLES=OFF ..
+$ cmake --build . --target install --config RelWithDebInfo
+```
+
+# Installing eRPC
+```
+sudo apt-get install libboost-dev libboost-filesystem-dev libboost-thread-dev libboost-program-options-dev libboost-python-dev libboost-dev
+
+# Installing googletest
+sudo apt install libgtest-dev build-essential cmake
+cd /usr/src/googletest
+sudo cmake .
+sudo cmake --build . --target install
+
+cd ~/
+git clone https://github.com/erpc-io/eRPC.git
+cd eRPC/
+cmake . -DPERF=OFF -DTRANSPORT=infiniband -DROCE=on; make -j;
+```
+
+# Server
+```rust,editable
+use erpc_sys::ffi;
+use std::os::raw::{c_int, c_void};
+use erpc_rs::context::AppContext;
+use erpc_rs::nexus::Nexus;
+use erpc_rs::reqhandle::ReqHandle;
+use erpc_rs::rpc::Rpc;
+
+extern fn req_handler(req_handle: *mut ffi::ReqHandle, context: *mut c_void) -> () {
+    println!("req_handler start");
+    let req_handle = ReqHandle::from_raw(req_handle);
+    let s = req_handle.get_req_msgbuf();
+    println!("req: {}", String::from_utf8(s).expect(""));
+
+    let c = AppContext::from_raw(context);
+    let r = Rpc::from_context(&c);
+    let s = "world".to_string().into_bytes();
+    r.enqueue_response(&req_handle, s);
+    println!("req_handler end");
+}
+
+extern fn sm_handler(_session_num: c_int, _sm_event_type: ffi::SmEventType, _sm_err_type: ffi::SmErrType, _context: *mut c_void) {
+    println!("sm_handler");
+}
+
+fn main() {
+    // sudo rxe_cfg start
+    // sudo rxe_cfg status
+    let context = AppContext::new();
+    let nexus = Nexus::new("127.0.0.1:31850".to_string(), 0, 0);
+
+    nexus.register_req_func(1, req_handler, 0);
+
+    let rpc = Rpc::new(&context, &nexus, 0, sm_handler, 0);
+
+    loop {
+        rpc.run_event_loop(1000);
+    }
+}
+```
+
+# Client
+```rust,editable
+use erpc_sys::ffi;
+use std::os::raw::{c_int, c_void};
+use erpc_rs::context::AppContext;
+use erpc_rs::msgbuffs;
+use erpc_rs::rpc::Rpc;
+use erpc_rs::nexus::Nexus;
+
+use msgbuffs::MsgBuffers;
+
+extern fn sm_handler(session_num: c_int, sm_event_type: ffi::SmEventType, sm_err_type: ffi::SmErrType, context: *mut c_void) {
+    println!("sm_handler session_num: {} sm_event_type: {} sm_err_type: {}", session_num, sm_event_type, sm_err_type);
+    let _ctx: *mut ffi::AppContext = context as *mut ffi::AppContext;
+}
+
+extern fn cont_func(_context: *mut c_void, tag: *mut c_void) {
+    let context = AppContext::from_raw(_context);
+    let tag = tag as usize;
+
+    let msg_buffs = MsgBuffers::from_context(&context, tag);
+    let s = msg_buffs.get_resp_msgbuf();
+    //let s = context.get_resp_msgbuf(tag);
+    let s = String::from_utf8(s).expect("");
+    println!("cont_func tag: {} resp: {}", tag, s);
+
+    let session_num = context.get_session_num();
+    let rpc = Rpc::from_context(&context);
+    let s = "hello".to_string().into_bytes();
+    rpc.enqueue_request(&context, session_num, 1, s, cont_func, 1000, 0);
+}
+
+fn main() {
+    let context = AppContext::new();
+    let nexus = Nexus::new("127.0.0.1:31851".to_string(), 0, 0);
+    let rpc = Rpc::new(&context, &nexus, 0, sm_handler, 0);
+
+    let session_num = context.connect_session("127.0.0.1:31850".to_string(), 0);
+
+    println!("session_num: {}", session_num);
+
+    while !rpc.is_connected(session_num) {
+        rpc.run_event_loop_once();
+    }
+
+    println!("connected");
+
+    let s = "hello".to_string().into_bytes();
+    rpc.enqueue_request(&context, session_num, 1, s, cont_func, 1000, 0);
+    rpc.run_event_loop(1000*5);
+
+    println!("OK");
+}
+```

erpc-examples/src/client.rs

@@ -1,9 +1,12 @@
 use erpc_sys::ffi;
 use std::os::raw::{c_int, c_void};
 use erpc_rs::context::AppContext;
+use erpc_rs::msgbuffs;
 use erpc_rs::rpc::Rpc;
 use erpc_rs::nexus::Nexus;
 
+use msgbuffs::MsgBuffers;
+
 extern fn sm_handler(session_num: c_int, sm_event_type: ffi::SmEventType, sm_err_type: ffi::SmErrType, context: *mut c_void) {
     println!("sm_handler session_num: {} sm_event_type: {} sm_err_type: {}", session_num, sm_event_type, sm_err_type);
     let _ctx: *mut ffi::AppContext = context as *mut ffi::AppContext;
@@ -13,7 +16,9 @@ extern fn cont_func(_context: *mut c_void, tag: *mut c_void) {
     let context = AppContext::from_raw(_context);
     let tag = tag as usize;
 
-    let s = context.get_resp_msgbuf(tag);
+    let msg_buffs = MsgBuffers::from_context(&context, tag);
+    let s = msg_buffs.get_resp_msgbuf();
+    //let s = context.get_resp_msgbuf(tag);
     let s = String::from_utf8(s).expect("");
     println!("cont_func tag: {} resp: {}", tag, s);
 

erpc-examples/src/client_raw.rs

@@ -15,11 +15,13 @@ extern fn cont_func(_context: *mut c_void, _tag: *mut c_void) {
 
     let data: *mut u8;
     let data_size : size_t = 0;
-    unsafe { data = ffi::erpc_get_resp_msgbuf(ctx, &data_size) };
-    //println!("data_size: {:?} {}", data, data_size);
+    let msgbufs = unsafe { ffi::erpc_msgbuffs_get_by_tag(ctx, tag) };
+    data = unsafe { ffi::erpc_msgbuffs_resp_msgbuf(msgbufs, &data_size) };
 
     let s = unsafe { String::from_raw_parts(data, data_size, 0) };
     println!("cont_func tag: {} resp: {}", tag, s);
+
+    unsafe { ffi::erpc_msgbuffs_destroy(msgbufs) };
 }
 
 fn main() {

erpc-sys/build.rs

@@ -12,6 +12,7 @@ fn main() {
         .flag("-std=c++11") // -std=c++1y
         .flag("-DERPC_INFINIBAND=true")
         .include(Path::new( &home_dir ).join("eRPC/src"))
+        .include(Path::new( &home_dir ).join("junction-install/include"))
         .include("src")
         .file("src/ffi.cpp")
         .file("src/server.cpp")
@@ -20,13 +21,17 @@ fn main() {
         .compile("erpc_ffi");
 
     println!("cargo:rustc-link-search=native={}", Path::new(&home_dir).join("eRPC/build").display());
+    println!("cargo:rustc-link-search=native={}", Path::new(&home_dir).join("junction-install/lib").display());
 
     println!("cargo:rustc-link-lib=erpc");
     println!("cargo:rustc-link-lib=pthread");
     println!("cargo:rustc-link-lib=numa");
     println!("cargo:rustc-link-lib=dl");
     println!("cargo:rustc-link-lib=ibverbs");
 
+    println!("cargo:rustc-link-lib=junction");
+    println!("cargo:rustc-link-lib=turf");
+
     println!("cargo:rerun-if-changed=src/*");
     println!("cargo:rerun-if-changed=build.rs");
 }

erpc-sys/src/ffi.cpp

@@ -5,15 +5,31 @@
 #include "util/numautils.h"
 #include "transport_impl/eth_common.h"
 #include "rpc_types.h"
+#include <junction/ConcurrentMap_Leapfrog.h>
+
+class MsgBuffers {
+public:
+    erpc::MsgBuffer req_msgbuf;
+    erpc::MsgBuffer resp_msgbuf;
+};
+
+typedef junction::ConcurrentMap_Leapfrog<size_t, MsgBuffers*> ConcurrentMap;
 
 class AppContext {
 public:
     erpc::Rpc<erpc::CTransport> *rpc = nullptr;
     int session_num = 0;
-    erpc::MsgBuffer req_msgbuf;
-    erpc::MsgBuffer resp_msgbuf;
+    ConcurrentMap msgbufs;
 
     ~AppContext() {}
+
+    inline MsgBuffers *alloc_msg_buffer_or_die(size_t req_id, size_t req_max_data_size, size_t resp_max_data_size) {
+        MsgBuffers *buffs = new MsgBuffers();
+        buffs->req_msgbuf = rpc->alloc_msg_buffer_or_die(req_max_data_size);
+        buffs->resp_msgbuf = rpc->alloc_msg_buffer_or_die(resp_max_data_size);
+        msgbufs.assign(req_id, buffs);
+        return buffs;
+    }
 };
 
 extern "C" {
@@ -97,13 +113,12 @@ uint8_t *erpc_get_req_msgbuf(erpc::ReqHandle *req_handle, size_t &data_size) {
 
 void erpc_enqueue_request(AppContext *context, erpc::Rpc<erpc::CTransport> *rpc, int session_num, uint8_t req_type,
         const uint8_t *data, size_t data_size, erpc::erpc_cont_func_t cont_func, size_t tag, size_t cont_etid) {
-    context->req_msgbuf = rpc->alloc_msg_buffer_or_die(data_size);
-    context->resp_msgbuf = rpc->alloc_msg_buffer_or_die(256); // rpc->get_max_msg_size()
+    auto buffs = context->alloc_msg_buffer_or_die(tag, data_size, 1024);
 
-    memcpy(context->req_msgbuf.buf, data, data_size);
+    memcpy(buffs->req_msgbuf.buf, data, data_size);
 
-    rpc->enqueue_request(session_num, 1, &context->req_msgbuf,
-                           &context->resp_msgbuf, cont_func, reinterpret_cast<void *>(tag)); // nullptr
+    rpc->enqueue_request(session_num, 1, &buffs->req_msgbuf,
+                           &buffs->resp_msgbuf, cont_func, reinterpret_cast<void *>(tag)); // nullptr
 }
 
 void erpc_enqueue_response(erpc::Rpc<erpc::CTransport> *rpc, erpc::ReqHandle *req_handle, const uint8_t *data,
@@ -115,9 +130,30 @@ void erpc_enqueue_response(erpc::Rpc<erpc::CTransport> *rpc, erpc::ReqHandle *re
     rpc->enqueue_response(req_handle, &resp);
 }
 
-uint8_t *erpc_get_resp_msgbuf(AppContext *context, size_t &data_size) {
-    data_size = context->resp_msgbuf.get_data_size();
-    return context->resp_msgbuf.buf;
+MsgBuffers *erpc_msgbuffs_get_by_tag(AppContext *context, size_t tag) {
+    auto item = context->msgbufs.get(tag);
+    if (item == nullptr) {
+        return nullptr;
+    }
+    context->msgbufs.erase(tag);
+    return item;
+}
+
+void erpc_msgbuffs_destroy(MsgBuffers *buffs) {
+    if (buffs != nullptr) {
+        delete buffs;
+        buffs = nullptr;
+    }
+}
+
+uint8_t *erpc_msgbuffs_req_msgbuf(MsgBuffers *buffs, size_t &data_size) {
+    data_size = buffs->req_msgbuf.get_data_size();
+    return buffs->req_msgbuf.buf;
+}
+
+uint8_t *erpc_msgbuffs_resp_msgbuf(MsgBuffers *buffs, size_t &data_size) {
+    data_size = buffs->resp_msgbuf.get_data_size();
+    return buffs->resp_msgbuf.buf;
 }
 
 }

erpc-sys/src/ffi.rs

@@ -4,6 +4,7 @@ use libc::{size_t};
 pub enum Nexus {} // erpc::Nexus
 pub enum ReqHandle {} // erpc::ReqHandle
 pub enum AppContext {} // AppContext
+pub enum MsgBuffers {} // MsgBuffers
 pub enum Rpc {} // erpc::Rpc<erpc::CTransport>
 
 unsafe impl Send for Rpc {}
@@ -81,7 +82,11 @@ extern "C" {
                                 data_size: size_t, cont_func: extern fn(*mut c_void, *mut c_void),
                                 tag: size_t, cont_etid: size_t) -> ();
     pub fn erpc_enqueue_response(rpc: *mut Rpc, req_handle: *mut ReqHandle, data: *const u8, data_size: size_t) -> ();
-    pub fn erpc_get_resp_msgbuf(context: *mut AppContext, data_size: &size_t) -> *mut u8;
+
+    pub fn erpc_msgbuffs_get_by_tag(context: *mut AppContext, tag: size_t) -> *mut MsgBuffers;
+    pub fn erpc_msgbuffs_destroy(buffs: *mut MsgBuffers) -> ();
+    pub fn erpc_msgbuffs_req_msgbuf(buffs: *mut MsgBuffers, data_size: &size_t) -> *mut u8;
+    pub fn erpc_msgbuffs_resp_msgbuf(buffs: *mut MsgBuffers, data_size: &size_t) -> *mut u8;
 
     pub fn server_test() -> c_int;
     pub fn client_test() -> c_int;

src/context.rs

@@ -32,15 +32,15 @@ impl AppContext {
         unsafe { ffi::erpc_connect_session(self.inner, server_uri.as_ptr(), rem_rpc_id) }
     }
 
-    pub fn get_resp_msgbuf(&self, _tag: usize) -> Vec<u8> {
-        let data: *mut u8;
-        let data_size : usize = 0;
-        unsafe { data = ffi::erpc_get_resp_msgbuf(self.inner, &data_size) };
-        //println!("data_size: {:?} {}", data, data_size);
-
-        let s = unsafe { String::from_raw_parts(data, data_size, 0) };
-        s.into_bytes()
-    }
+    // pub fn get_resp_msgbuf(&self, tag: usize) -> Vec<u8> {
+    //     let data: *mut u8;
+    //     let data_size : usize = 0;
+    //     unsafe { data = ffi::erpc_get_resp_msgbuf(self.inner, tag, &data_size) };
+    //     //println!("data_size: {:?} {}", data, data_size);
+
+    //     let s = unsafe { String::from_raw_parts(data, data_size, 0) };
+    //     s.into_bytes()
+    // }
 }
 
 impl Drop for AppContext {

src/lib.rs

@@ -1,4 +1,5 @@
 pub mod context;
+pub mod msgbuffs;
 pub mod nexus;
 pub mod reqhandle;
 pub mod rpc;

src/msgbuffs.rs

@@ -0,0 +1,39 @@
+use erpc_sys::ffi::{self};
+use crate::context::AppContext;
+
+pub struct MsgBuffers {
+    pub inner: *mut ffi::MsgBuffers,
+}
+
+impl MsgBuffers {
+    pub fn from_context(context: &AppContext, tag: usize) -> Self {
+        let msg_buffs = unsafe { ffi::erpc_msgbuffs_get_by_tag(context.inner, tag) };
+        MsgBuffers {
+            inner: msg_buffs,
+        }
+    }
+
+    pub fn get_req_msgbuf(&self) -> Vec<u8> {
+        let data: *mut u8;
+        let data_size : usize = 0;
+        unsafe { data = ffi::erpc_msgbuffs_req_msgbuf(self.inner, &data_size) };
+
+        let s = unsafe { String::from_raw_parts(data, data_size, 0) };
+        s.into_bytes()
+    }
+
+    pub fn get_resp_msgbuf(&self) -> Vec<u8> {
+        let data: *mut u8;
+        let data_size : usize = 0;
+        unsafe { data = ffi::erpc_msgbuffs_resp_msgbuf(self.inner, &data_size) };
+
+        let s = unsafe { String::from_raw_parts(data, data_size, 0) };
+        s.into_bytes()
+    }
+}
+
+impl Drop for MsgBuffers {
+    fn drop(&mut self) {
+        unsafe { ffi::erpc_msgbuffs_destroy(self.inner) }
+    }
+}