func_client

Author: QuarkContainer

qlib/bytestream.rs

@@ -107,7 +107,7 @@ impl SocketBufIovs {
             let dst = buf.as_mut_ptr();
             unsafe {
                 ptr::copy_nonoverlapping(src, dst, size);
-            }
+            }            
         } else {
             let src = self.iovs[0].start as * const u8;
             let dst = buf.as_mut_ptr();
@@ -121,12 +121,14 @@ impl SocketBufIovs {
                 ptr::copy_nonoverlapping(src, dst, size - self.iovs[0].len);
             }
         }
-
+        
+        self.Consume(size);
         return Ok(buf)
     }  
 
     pub fn ReadString(&mut self, size: usize) -> Result<String> {
         let buf = self.ReadVec(size)?;
+        self.Consume(size);
         match String::from_utf8(buf) {
             Err(_) => return Err(Error::SysError(SysErr::EINVAL)),
             Ok(s) => return Ok(s),
@@ -142,12 +144,14 @@ impl SocketBufIovs {
 
         if self.iovs[0].len >= size {
             let data = unsafe {*(self.iovs[0].start as * const T)};
+            self.Consume(size);
             return Ok(data)
         } else {
             let buf = self.ReadVec(size);
             match buf {
                 Ok(b) => {
                     let data = unsafe {*(&b[0] as * const _ as u64 as * const T)};
+                    self.Consume(size);
                     return Ok(data)
                 } 
                 Err(e) => {

rdma_cli/Cargo.lock

@@ -59,6 +59,7 @@ tokio-eventfd = "0.2.0"
 tokio-io = "0.1"
 futures-lite = "1.11"
 bytes = "1.3.0"
+futures = "0.3.25"
 
 [dependencies.lazy_static]
 version = "1.4"

rdma_cli/Cargo.toml

@@ -109,6 +109,8 @@ use rdma_ingress_informer::RdmaIngressInformer;
 use service_informer::ServiceInformer;
 use crate::constants::*;
 
+use crate::funclib::func_client::*;
+
 pub static GLOBAL_ALLOCATOR: HostAllocator = HostAllocator::New();
 
 lazy_static! {
@@ -118,6 +120,12 @@ lazy_static! {
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    //not_semaphore_reads_and_resets().await?;
+    println!("main ....");
+    let client1 = FuncClient::NewTestTCPClient(1).await?;
+    let client2 = FuncClient::NewTestTCPClient(1).await?;
+    tokio::select! {
+        _ = client1.msgStream.ReadMsg() => {},
+        _ = client2.msgStream.ReadMsg() => {},
+    }
     Ok(())
 }

rdma_cli/src/async_test.rs

@@ -109,6 +109,9 @@ use rdma_ingress_informer::RdmaIngressInformer;
 use service_informer::ServiceInformer;
 use crate::constants::*;
 
+use crate::funclib::func_agent::*;
+use crate::funclib::agent::*;
+
 pub static GLOBAL_ALLOCATOR: HostAllocator = HostAllocator::New();
 
 lazy_static! {
@@ -118,5 +121,10 @@ lazy_static! {
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    return Ok(());
+    tokio::select! {
+        _a = Execution() => {}
+        _b = ChannelProcess() => {}
+    };
+
+    return Ok(())
 }