From 14c345b38d57fd814f217eb8465963a08ca79f7e Mon Sep 17 00:00:00 2001
From: wangzhengquan <wangzhengquan85@126.com>
Date: 星期五, 05 二月 2021 17:41:09 +0800
Subject: [PATCH] update

---
 src/socket/shm_socket.cpp |  547 +++++++++++++++++++++++++++++++++++++++---------------
 1 files changed, 392 insertions(+), 155 deletions(-)

diff --git a/src/socket/shm_socket.cpp b/src/socket/shm_socket.cpp
index a76cb4b..14274ca 100644
--- a/src/socket/shm_socket.cpp
+++ b/src/socket/shm_socket.cpp
@@ -4,6 +4,7 @@
 #include <map>
 #include <cassert>
 #include "bus_error.h"
+#include "sole.h"
 
 static Logger *logger = LoggerFactory::getLogger();
 
@@ -18,6 +19,12 @@
 
 static void _destrory_socket_perthread(void *tmp_socket);
 static void _create_socket_key_perthread(void);
+
+static int shm_recvpakfrom(shm_socket_t *sockt, shm_packet_t *recvpak ,  const struct timespec *timeout,  int flag);
+
+   
+static int shm_sendpakto(shm_socket_t *sockt, const shm_packet_t *sendpak,
+               const int key, const struct timespec *timeout, const int flag);
 
 //  妫�鏌ey鏄惁宸茬粡琚娇鐢紝 鏈浣跨敤鍒欑粦瀹歬ey
 static LockFreeQueue<shm_packet_t> * shm_socket_bind_queue(int key, bool force) {
@@ -58,7 +65,7 @@
   return queue;
 }
 
-
+//鍒犻櫎鍖呭惈鍦╧eys鍐呯殑queue
 size_t shm_socket_remove_keys(int keys[], size_t length) {
   hashtable_t *hashtable = mm_get_hashtable();
   LockFreeQueue<shm_packet_t> *mqueue;
@@ -73,16 +80,49 @@
   return count;
 }
 
+ 
+// 鍒犻櫎涓嶅湪keys鍐呯殑queue 
+size_t shm_socket_remove_keys_exclude(int keys[], size_t length) {
+  hashtable_t *hashtable = mm_get_hashtable();
+  std::set<int> *keyset = hashtable_keyset(hashtable);
+  std::set<int>::iterator keyItr;
+  LockFreeQueue<shm_packet_t> *mqueue;
+  bool found;
+  size_t count = 0;
+  for (keyItr = keyset->begin(); keyItr != keyset->end(); keyItr++) {
+    found = false;
+    for (size_t i = 0; i < length; i++) {
+      if (*keyItr == keys[i]) {
+        found = true;
+        break;
+      }
+    }
+    // 100鍐呯殑鏄痓us鍐呴儴鑷繁鐢ㄧ殑
+    if (!found && *keyItr > 100) {
+      // 閿�姣佸叡浜唴瀛樼殑queue
+      mqueue = (LockFreeQueue<shm_packet_t> *)hashtable_get(hashtable, *keyItr);
+      delete mqueue;
+      hashtable_remove(hashtable, *keyItr);
+      count++;
+    }
+  }
+  delete keyset;
+  return count;
+}
+
+
+
 shm_socket_t *shm_open_socket(shm_socket_type_t socket_type) {
   int s, type;
   pthread_mutexattr_t mtxAttr;
 
   logger->debug("shm_open_socket\n");
-  shm_socket_t *socket = (shm_socket_t *)calloc(1, sizeof(shm_socket_t));
-  socket->socket_type = socket_type;
-  socket->key = 0;
-  socket->force_bind = false;
-  socket->queue = NULL;
+  // shm_socket_t *socket = (shm_socket_t *)calloc(1, sizeof(shm_socket_t));
+  shm_socket_t *sockt = new shm_socket_t;
+  sockt->socket_type = socket_type;
+  sockt->key = 0;
+  sockt->force_bind = false;
+  sockt->queue = NULL;
 
 
   s = pthread_mutexattr_init(&mtxAttr);
@@ -91,7 +131,7 @@
   s = pthread_mutexattr_settype(&mtxAttr, PTHREAD_MUTEX_ERRORCHECK);
   if (s != 0)
     err_exit(s, "pthread_mutexattr_settype");
-  s = pthread_mutex_init(&(socket->mutex), &mtxAttr);
+  s = pthread_mutex_init(&(sockt->mutex), &mtxAttr);
   if (s != 0)
     err_exit(s, "pthread_mutex_init");
 
@@ -99,45 +139,52 @@
   if (s != 0)
     err_exit(s, "pthread_mutexattr_destroy");
 
-  return socket;
+  return sockt;
 }
 
-int shm_close_socket(shm_socket_t *socket) {
+int shm_close_socket(shm_socket_t *sockt) {
   
-  int s;
-  
+  int rv;
   logger->debug("shm_close_socket\n");
-  if(socket->queue != NULL) {
-    delete socket->queue;
-    socket->queue = NULL;
+  if(sockt->queue != NULL) {
+    delete sockt->queue;
+    sockt->queue = NULL;
   }
 
-  s =  pthread_mutex_destroy(&(socket->mutex) );
-  if(s != 0) {
-    err_exit(s, "shm_close_socket");
+  rv =  pthread_mutex_destroy(&(sockt->mutex) );
+  if(rv != 0) {
+    err_exit(rv, "shm_close_socket");
   }
 
-  free(socket);
+  free(sockt);
   return 0;
 }
 
 
-int shm_socket_bind(shm_socket_t *socket, int key) {
-  socket->key = key;
+int shm_socket_stop(shm_socket_t *sockt) {
+  struct timespec timeout = {5, 0};
+  shm_packet_t sendpak = {0};
+  sendpak.key = sockt->key;
+  sendpak.action = BUS_ACTION_STOP;
+  sendpak.size = 0;
+  return shm_sendpakto(sockt, &sendpak, sockt->key, &timeout, BUS_TIMEOUT_FLAG);
+}
+
+int shm_socket_bind(shm_socket_t *sockt, int key) {
+  sockt->key = key;
   return 0;
 }
 
-int shm_socket_force_bind(shm_socket_t *socket, int key) {
-  socket->force_bind = true;
-  socket->key = key;
+int shm_socket_force_bind(shm_socket_t *sockt, int key) {
+  sockt->force_bind = true;
+  sockt->key = key;
   return 0;
 }
 
-int shm_socket_get_key(shm_socket_t *sk){
-  return sk->key;
+int shm_socket_get_key(shm_socket_t *sockt){
+  return sockt->key;
 }
 
-  
 
 
 // 鐭繛鎺ユ柟寮忓彂閫�
@@ -145,109 +192,166 @@
                const int key, const struct timespec *timeout, const int flag) {
 
   int rv;
-
-  hashtable_t *hashtable = mm_get_hashtable();
-
  
-  if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0)
-    err_exit(rv, "shm_sendto : pthread_mutex_lock");
-
-  if (sockt->queue == NULL) {
-    if (sockt->key == 0) {
-      sockt->key = hashtable_alloc_key(hashtable);
-    }
-    sockt->queue = shm_socket_bind_queue( sockt->key, sockt->force_bind);
-    if(sockt->queue  == NULL ) {
-      logger->error("%s. key = %d", bus_strerror(EBUS_KEY_INUSED), sockt->key);
-      return EBUS_KEY_INUSED;
-    }
+  shm_packet_t sendpak = {0};
+  sendpak.key = sockt->key;
+  sendpak.size = size;
+  if(buf != NULL) {
+    sendpak.buf = mm_malloc(size);
+    memcpy(sendpak.buf, buf, size);
   }
-
-  if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0)
-    err_exit(rv, "shm_sendto : pthread_mutex_unlock");
-  
-  // There is some case where a sockt need to send to himeself, for example when bus server need to stop, he need to send himself 
-  // a top message.
-  
-  if (key == sockt->key) {
-    logger->error( "can not send to your self!");
-    return EBUS_SENDTO_SELF;
-  }
-
-  LockFreeQueue<shm_packet_t> *remoteQueue;
-  if ((remoteQueue = shm_socket_attach_queue(key)) == NULL) {
-    bus_errno = EBUS_CLOSED;
-    logger->error("sendto key %d failed, %s", key, bus_strerror(bus_errno));
-    return EBUS_CLOSED;
-  }
-
-  shm_packet_t dest;
-  dest.type = SHM_COMMON_MSG;
-  dest.key = sockt->key;
-  dest.size = size;
-  dest.buf = mm_malloc(size);
-  memcpy(dest.buf, buf, size);
-
-  rv = remoteQueue->push(dest, timeout, flag);
-
-  if (rv == 0) {
-    printf("%d sendto %d suc.\n", shm_socket_get_key(sockt), key);
-    return 0;
-  } else {
-    mm_free(dest.buf);
-    if(rv == ETIMEDOUT)
-      return EBUS_TIMEOUT;
-    else {
-      logger->debug("====%d sendto key %d failed %s", shm_socket_get_key(sockt), key, bus_strerror(rv));
-      return rv;
-    }
-  }
+ 
+  rv = shm_sendpakto(sockt, &sendpak, key, timeout, flag);
+  return rv;
 }
 
+ 
+
+int shm_sendandrecv(shm_socket_t *sockt, const void *send_buf,
+                    const int send_size, const int key, void **recv_buf,
+                    int *recv_size,  const struct timespec *timeout,  int flags) {
+
+  int rv, tryn = 3;
+  shm_packet_t sendpak;
+  shm_packet_t recvpak;
+  std::map<std::string, shm_packet_t>::iterator recvbufIter;
+
+  std::string uuid = sole::uuid4().str();
+  
+  sendpak.key = sockt->key;
+  sendpak.size = send_size;
+  if(send_buf != NULL) {
+    sendpak.buf = mm_malloc(send_size);
+    memcpy(sendpak.buf, send_buf, send_size);
+  }
+  memcpy(sendpak.uuid, uuid.c_str(), uuid.length() + 1);
+  // uuid.copy(sendpak.uuid, sizeof sendpak.uuid);
+  rv = shm_sendpakto(sockt, &sendpak, key, timeout, flags);
+
+  if(rv != 0) {
+    return rv;
+  }
+
+  while(tryn > 0) {
+    tryn--;
+    recvbufIter = sockt->recvbuf.find(uuid);
+    if(recvbufIter != sockt->recvbuf.end()) {
+      // 鍦ㄧ紦瀛橀噷鏌ュ埌浜哢UID鍖归厤鎴愬姛鐨�
+logger->debug("get from recvbuf: %s", uuid.c_str());
+      recvpak = recvbufIter->second;
+      sockt->recvbuf.erase(recvbufIter);
+      goto LABLE_SUC;
+    }
+
+    rv = shm_recvpakfrom(sockt, &recvpak, timeout, flags);
+
+    if (rv != 0) {
+
+      if(rv == ETIMEDOUT) {
+        return EBUS_TIMEOUT;
+      }
+
+      logger->debug("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
+      return rv;
+    } 
+
+logger->debug("send uuid:%s, recv uuid: %s", uuid.c_str(), recvpak.uuid);
+    if(strlen(recvpak.uuid) == 0) {
+      continue;
+    } else if (strncmp(uuid.c_str(), recvpak.uuid, sizeof recvpak.uuid) == 0) {
+      // 鍙戦�佷笌鎺ュ彈鐨刄UID鍖归厤鎴愬姛
+      goto LABLE_SUC;
+    } else {
+      // 绛旈潪鎵�闂紝鏀惧埌缂撳瓨閲�
+      sockt->recvbuf.insert({recvpak.uuid, recvpak});
+      continue;
+    }
+  }
+
+LABLE_FAIL:
+  return EBUS_RECVFROM_WRONG_END;
+  // return rv;
+ 
+LABLE_SUC:
+ if(recv_buf != NULL) {
+    void *_buf = malloc(recvpak.size);
+    memcpy(_buf, recvpak.buf, recvpak.size);
+    *recv_buf = _buf; 
+  }
+ 
+  if(recv_size != NULL)
+    *recv_size = recvpak.size;
+
+  mm_free(recvpak.buf);
+
+  return 0;
+
+
+
+}
+
+/**
+ * @callback  void (*recvandsend_callback_fn)(void *recvbuf, int recvsize, int key, void **sendbuf, int *sendsize) 
+ *            sendbuf 鍜� sendsize鏄痗allbak_fn鍥炶皟鍑芥暟鐨勮繑鍥炲��, 鍒嗗埆琛ㄧず鍙戦�佹暟鎹紝鍜屽彂閫佹暟鎹殑澶у皬銆�
+ * 
+ */
+int shm_recvandsend(shm_socket_t *sockt,  recvandsend_callback_fn callback, 
+  const struct timespec *timeout, int flag, void *user_data) {
+  
+  int rv;
+ 
+  void *sendbuf, *recvbuf = NULL;
+  int sendsize = 0;
+  shm_packet_t recvpak = {0};
+
+  rv = shm_recvpakfrom(sockt , &recvpak, timeout, flag);
+
+  if (rv != 0) {
+    if(rv == ETIMEDOUT){
+      logger->debug("%d shm_recvandsend failed %s", shm_socket_get_key(sockt), bus_strerror(EBUS_TIMEOUT));
+      return EBUS_TIMEOUT;
+    }
+    
+    logger->error("%d shm_recvandsend failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
+    return rv;
+  }
+   
+  if(recvpak.buf != NULL) {
+    recvbuf = malloc(recvpak.size);
+    memcpy(recvbuf, recvpak.buf, recvpak.size);
+    mm_free(recvpak.buf);
+  }
+ 
+  callback(recvbuf, recvpak.size, recvpak.key, &sendbuf, &sendsize, user_data);
+
+  shm_packet_t sendpak = {0};
+  sendpak.key = sockt->key;
+  sendpak.size = sendsize;
+  memcpy(sendpak.uuid, recvpak.uuid, sizeof sendpak.uuid);
+  if(sendbuf !=NULL && sendsize > 0) {
+    sendpak.buf = mm_malloc(sendsize);
+    memcpy(sendpak.buf, sendbuf, sendsize);
+  } else {
+
+    logger->warn("%d shm_recvandsend : sendbuf is null", shm_socket_get_key(sockt));
+    // return -1;
+  }
+ 
+  rv = shm_sendpakto(sockt, &sendpak, recvpak.key, timeout, flag);
+
+  return 0;
+  
+}
 
 // 鐭繛鎺ユ柟寮忔帴鍙�
 int shm_recvfrom(shm_socket_t *sockt, void **buf, int *size, int *key,  const struct timespec *timeout,  int flag) {
   int rv;
   
-  hashtable_t *hashtable = mm_get_hashtable();
+  shm_packet_t recvpak;
+  rv = shm_recvpakfrom(sockt , &recvpak, timeout, flag);
 
-  if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0)
-    err_exit(rv, "shm_recvfrom : pthread_mutex_lock");
- 
-  if (sockt->queue == NULL) {
-    if (sockt->key == 0) {
-      sockt->key = hashtable_alloc_key(hashtable);
-    }  
-    sockt->queue = shm_socket_bind_queue( sockt->key, sockt->force_bind);
-    if(sockt->queue  == NULL ) {
-      logger->error("%s. key = %d", bus_strerror(EBUS_KEY_INUSED), sockt->key);
-      return EBUS_KEY_INUSED;
-    }
-  }
-  
-  if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0)
-    err_exit(rv, "shm_recvfrom : pthread_mutex_unlock");
+  if (rv != 0) {
 
-  shm_packet_t src;
- 
-  rv = sockt->queue->pop(src, timeout, flag);
-
-  if (rv == 0) {
-    if(buf != NULL) {
-      void *_buf = malloc(src.size);
-      memcpy(_buf, src.buf, src.size);
-      *buf = _buf; 
-    }
-   
-    if(size != NULL)
-      *size = src.size;
-
-    if(key != NULL)
-      *key = src.key;
-
-    mm_free(src.buf);
-    return 0;
-  } else {
     if(rv == ETIMEDOUT)
       return EBUS_TIMEOUT;
     else {
@@ -255,10 +359,27 @@
       return rv;
     }
    
+  } 
 
+
+ if(buf != NULL && recvpak.buf != NULL) {
+    void *_buf = malloc(recvpak.size);
+    memcpy(_buf, recvpak.buf, recvpak.size);
+    *buf = _buf; 
   }
+ 
+  if(size != NULL)
+    *size = recvpak.size;
+
+  if(key != NULL)
+    *key = recvpak.key;
+
+  mm_free(recvpak.buf);
+  return 0;
 }
 
+
+// =================================================================================================
 
  /* Free thread-specific data buffer */
 static void _destrory_socket_perthread(void *tmp_socket)
@@ -287,7 +408,6 @@
   //s = pthread_key_create(&_perthread_socket_key_, NULL);
   if (s != 0) {
      logger->error(s, "pthread_key_create");
-     abort(); /* dump core and terminate */ 
      exit(1);
   }
 }
@@ -299,11 +419,13 @@
                     int *recv_size,  const struct timespec *timeout,  int flags) {
   int recv_key;
   int rv;
+  int tryn = 0;
 
   // 鐢╰hread local 淇濊瘉姣忎釜绾跨▼鐢ㄤ竴涓嫭鍗犵殑socket鎺ュ彈瀵规柟杩斿洖鐨勪俊鎭�
   shm_socket_t *tmp_socket;
  
-
+  /* If first call from this thread, allocate buffer for thread, and save its location */
+  // logger->debug("%d create tmp socket\n", pthread_self() );
   rv = pthread_once(&_once_, _create_socket_key_perthread);
   if (rv != 0) {
     logger->error(rv, "shm_sendandrecv pthread_once");
@@ -325,26 +447,31 @@
   }
 
   if ((rv = shm_sendto(tmp_socket, send_buf, send_size, send_key, timeout, flags)) == 0) {
-    rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags);
-    if(rv != 0) {
-      logger->error("_shm_sendandrecv_thread_local : %s\n", bus_strerror(rv));
-    }
-    else if(rv == 0 ) {
-      logger->debug("======%d use tmp_socket %d, send to  %d, receive from  %d\n", shm_socket_get_key(sockt), shm_socket_get_key(tmp_socket), send_key, recv_key);
-    
-      if(recv_key == shm_socket_get_key(sockt)) {
-        logger->debug("=====鏀跺埌浜嗚嚜宸卞彂缁欒嚜宸辩殑娑堟伅\n");
+
+    while(tryn < 3) {
+      tryn++;
+      rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags);
+      if(rv != 0) {
+        logger->error("_shm_sendandrecv_thread_local : %s\n", bus_strerror(rv));
+        return rv;
       }
-      assert( send_key == recv_key);
+     
+       // 瓒呮椂瀵艰嚧鎺ュ彂閫佸璞★紝涓庤繑鍥炲璞′笉瀵瑰簲鐨勬儏鍐�
       if(send_key != recv_key) {
-        logger->error( "_shm_sendandrecv_alloc_new: send key expect to equal to recv key! send key =%d , recv key=%d", send_key, recv_key);
-        exit(1);
+        logger->debug("======%d use tmp_socket %d, send to  %d, receive from  %d\n", shm_socket_get_key(sockt), shm_socket_get_key(tmp_socket), send_key, recv_key);
+        // logger->error( "_shm_sendandrecv_alloc_new: send key expect to equal to recv key! send key =%d , recv key=%d", send_key, recv_key);
+        // exit(1);
+        continue;
+        // return EBUS_RECVFROM_WRONG_END;
       }
+
+      return 0;
     }
-    return rv;
-  } else {
-    return rv;
-  }
+    
+    return EBUS_RECVFROM_WRONG_END;
+  }  
+
+  return rv;
 }
 
 int _shm_sendandrecv_alloc_new(shm_socket_t *sockt, const void *send_buf,
@@ -353,31 +480,34 @@
   int recv_key;
   int rv;
 
-  // 鐢╰hread local 淇濊瘉姣忎釜绾跨▼鐢ㄤ竴涓嫭鍗犵殑socket鎺ュ彈瀵规柟杩斿洖鐨勪俊鎭�
+  int tryn = 0;
   shm_socket_t *tmp_socket;
  
-  /* If first call from this thread, allocate buffer for thread, and save its location */
-  // logger->debug("%d create tmp socket\n", pthread_self() );
+ 
   tmp_socket = shm_open_socket(SHM_SOCKET_DGRAM);
 
   if ((rv = shm_sendto(tmp_socket, send_buf, send_size, send_key, timeout, flags)) == 0) {
     rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags);
    
-    if(rv != 0) {
-      printf("_shm_sendandrecv_alloc_new : %s\n", bus_strerror(rv));
-    }
-    else if(rv == 0 ) {
-      printf("======%d use tmp_socket %d, send to  %d, receive from  %d\n", shm_socket_get_key(sockt), shm_socket_get_key(tmp_socket), send_key, recv_key);
-    
-      if(recv_key == shm_socket_get_key(sockt)) {
-        printf("=====鏀跺埌浜嗚嚜宸卞彂缁欒嚜宸辩殑娑堟伅\n");
-      }
-      assert( send_key == recv_key);
-      if(send_key != recv_key) {
-         err_exit(0, "_shm_sendandrecv_alloc_new: send key expect to equal to recv key! send key =%d , recv key=%d", send_key, recv_key);
+    while(tryn < 3) {
+      tryn++;
+      rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags);
+      if(rv != 0) {
+        logger->error("_shm_sendandrecv_thread_local : %s\n", bus_strerror(rv));
+        return rv;
       }
      
+      // 瓒呮椂瀵艰嚧鎺ュ彂閫佸璞★紝涓庤繑鍥炲璞′笉瀵瑰簲鐨勬儏鍐�
+      if(send_key != recv_key) {
+        // logger->debug("======%d use tmp_socket %d, send to  %d, receive from  %d\n", shm_socket_get_key(sockt), shm_socket_get_key(tmp_socket), send_key, recv_key);
+        // logger->error( "_shm_sendandrecv_alloc_new: send key expect to equal to recv key! send key =%d , recv key=%d", send_key, recv_key);
+        
+        continue;
+      }
+      return 0;
     }
+    
+    return EBUS_RECVFROM_WRONG_END;
   } 
    
   shm_close_socket(tmp_socket);  
@@ -385,9 +515,116 @@
  
 }
  
-int shm_sendandrecv(shm_socket_t *socket, const void *send_buf,
-                    const int send_size, const int send_key, void **recv_buf,
-                    int *recv_size,  const struct timespec *timeout,  int flags) {
 
-  return  _shm_sendandrecv_thread_local(socket, send_buf, send_size, send_key,recv_buf, recv_size, timeout,  flags);
+   
+static int shm_sendpakto(shm_socket_t *sockt, const shm_packet_t *sendpak,
+               const int key, const struct timespec *timeout, const int flag) {
+
+  int rv;
+  hashtable_t *hashtable = mm_get_hashtable();
+
+  if( sockt->queue != NULL) 
+    goto LABEL_PUSH;
+
+  if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) {
+    return EBUS_EXCEED_LIMIT;
+  }
+ 
+  {
+    if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0)
+    err_exit(rv, "shm_sendto : pthread_mutex_lock");
+    
+    if (sockt->queue == NULL) {
+      if (sockt->key == 0) {
+        sockt->key = hashtable_alloc_key(hashtable);
+      }
+      sockt->queue = shm_socket_bind_queue( sockt->key, sockt->force_bind);
+      if(sockt->queue  == NULL ) {
+        logger->error("%s. key = %d", bus_strerror(EBUS_KEY_INUSED), sockt->key);
+        return EBUS_KEY_INUSED;
+      }
+    }
+
+    if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0)
+      err_exit(rv, "shm_sendto : pthread_mutex_unlock");
+  
+  }
+  
+ 
+ LABEL_PUSH: 
+  if (sendpak->action != BUS_ACTION_STOP && key == sockt->key) {
+    logger->error( "can not send to your self!");
+    return EBUS_SENDTO_SELF;
+  }
+
+  LockFreeQueue<shm_packet_t> *remoteQueue;
+  if ((remoteQueue = shm_socket_attach_queue(key)) == NULL) {
+    bus_errno = EBUS_CLOSED;
+    logger->error("sendto key %d failed, %s", key, bus_strerror(bus_errno));
+    return EBUS_CLOSED;
+  }
+
+  
+
+  rv = remoteQueue->push(*sendpak, timeout, flag);
+
+  return rv;
 }
+
+// 鐭繛鎺ユ柟寮忔帴鍙�
+static int shm_recvpakfrom(shm_socket_t *sockt, shm_packet_t *_recvpak ,  const struct timespec *timeout,  int flag) {
+  int rv;
+  
+  hashtable_t *hashtable = mm_get_hashtable();
+  shm_packet_t recvpak;
+
+  if( sockt->queue != NULL) 
+    goto LABEL_POP;
+
+  if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) {
+    return EBUS_EXCEED_LIMIT;
+  }
+
+  {
+    if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0)
+      err_exit(rv, "shm_recvfrom : pthread_mutex_lock");
+ 
+   
+    if (sockt->key == 0) {
+      sockt->key = hashtable_alloc_key(hashtable);
+    }  
+    sockt->queue = shm_socket_bind_queue( sockt->key, sockt->force_bind);
+    if(sockt->queue  == NULL ) {
+      logger->error("%s. key = %d", bus_strerror(EBUS_KEY_INUSED), sockt->key);
+      return EBUS_KEY_INUSED;
+    }
+    
+    
+    if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0)
+      err_exit(rv, "shm_recvfrom : pthread_mutex_unlock");
+
+  }
+  
+LABEL_POP:
+
+ // 
+  // printf("%p start recv.....\n", sockt);
+ 
+  rv = sockt->queue->pop(recvpak, timeout, flag);
+  if(rv != 0) 
+    return rv;
+
+  
+  if(recvpak.action == BUS_ACTION_STOP) {
+    return EBUS_STOPED;
+  }
+  *_recvpak = recvpak;
+  return rv;
+}
+// int shm_sendandrecv(shm_socket_t *sockt, const void *send_buf,
+//                     const int send_size, const int send_key, void **recv_buf,
+//                     int *recv_size,  const struct timespec *timeout,  int flags) {
+
+//   struct timespec tm = {10, 0};
+//   return  _shm_sendandrecv_thread_local(sockt, send_buf, send_size, send_key,recv_buf, recv_size, &tm, flags);
+// }

--
Gitblit v1.8.0