From abc89da58e0a321fc4d27d0ca70620cd14141d3a Mon Sep 17 00:00:00 2001
From: wangzhengquan <wangzhengquan85@126.com>
Date: 星期二, 01 十二月 2020 15:34:35 +0800
Subject: [PATCH] thread local, thread pool , thread safe

---
 src/socket/net_mod_socket.c |  261 +++++++++++++++++---------------
 src/socket/net_mod_socket.h |   17 -
 lib/libusgcommon.so.bk      |    0 
 src/socket/net_conn_pool.h  |   37 ++++
 src/socket/net_conn_pool.c  |  109 +++++++++++++
 5 files changed, 289 insertions(+), 135 deletions(-)

diff --git a/lib/libusgcommon.so b/lib/libusgcommon.so.bk
similarity index 100%
rename from lib/libusgcommon.so
rename to lib/libusgcommon.so.bk
Binary files differ
diff --git a/src/socket/net_conn_pool.c b/src/socket/net_conn_pool.c
new file mode 100644
index 0000000..77e91d3
--- /dev/null
+++ b/src/socket/net_conn_pool.c
@@ -0,0 +1,109 @@
+#include "net_conn_pool.h"
+#include "socket_io.h"
+#include "logger_factory.h"
+
+NetConnPool::NetConnPool() {
+	int i;
+  maxi = -1; 
+  nready = 0 ;                  
+  for (i = 0; i < OPEN_MAX; i++) {
+    conns[i].fd = -1; 
+    conns[i].events = 0;      
+  }
+}
+
+NetConnPool::~NetConnPool() {
+	int connfd;
+  for (auto map_iter = connectionMap.begin(); map_iter != connectionMap.end(); map_iter++) {
+    connfd = map_iter->second;
+    Close(connfd);
+  }
+}
+
+
+int NetConnPool::getConn(const char *host, int port) {
+  std::map<std::string, int>::iterator mapIter;
+  int connfd;
+  int i;
+  char mapKey[ADDRSTRLEN];
+  char portstr[NI_MAXSERV];
+
+  sprintf(mapKey, "%s:%d", host, port);
+  mapIter =  connectionMap.find(mapKey);
+  if( mapIter != connectionMap.end()) {
+    connfd = mapIter->second;
+// printf("hit: %s\n", mapKey);
+  } else {
+// printf("mis: %s\n", mapKey);     
+    sprintf(portstr, "%d", port);
+// printf("open before: %s\n", mapKey); 
+    connfd = open_clientfd(host, portstr);
+// printf("open after: %s\n", mapKey); 
+    if(connfd < 0) {
+      LoggerFactory::getLogger()->error(errno, "NetModSocket::connect %s:%d ", host, port);
+      return -1;
+    }
+    connectionMap.insert({mapKey, connfd});
+  }
+
+  
+  for (i = 0; i < OPEN_MAX; i++) { /* Find an available slot */
+    if (conns[i].fd < 0)
+    {
+      /* Add connected descriptor to the req_resp_pool */
+      conns[i].fd = connfd;  
+               
+      conns[i].events = POLLIN;
+      /* Add the descriptor to descriptor set */
+      break;
+    }
+  }
+
+  if (i > maxi)      
+      maxi = i;   
+
+  if (i == OPEN_MAX) {
+    /* Couldn't find an empty slot */
+    LoggerFactory::getLogger()->error(errno, "add_client error: Too many clients");
+    return -1;
+  }
+
+  
+  return connfd;
+}
+
+void NetConnPool::putConn(int connfd) {
+	int i;
+	for (i = 0; i <= maxi; i++) {
+    if(conns[i].fd == connfd) {
+      conns[i].fd = -1;
+    }
+  }
+}
+
+void NetConnPool::closeConn(int connfd) {
+  int i;
+  std::map<std::string, int>::iterator map_iter;
+  if(close(connfd) != 0) {
+    LoggerFactory::getLogger()->error(errno, "NetModSocket::close_connect close");
+  }
+ 
+
+  for (i = 0; i <= maxi; i++) {
+    if(conns[i].fd == connfd) {
+      conns[i].fd = -1;
+    }
+  }
+
+  for ( map_iter = connectionMap.begin(); map_iter != connectionMap.end(); ) {
+    if(connfd == map_iter->second) {
+// std::cout << "map_iter->first==" << map_iter->first << std::endl;
+     map_iter = connectionMap.erase(map_iter);
+    } else {
+      ++map_iter;
+    }
+  }
+
+  // LoggerFactory::getLogger()->debug( "closed %d\n", connfd);
+
+}
\ No newline at end of file
diff --git a/src/socket/net_conn_pool.h b/src/socket/net_conn_pool.h
new file mode 100644
index 0000000..f8562c9
--- /dev/null
+++ b/src/socket/net_conn_pool.h
@@ -0,0 +1,37 @@
+#ifndef __NET_CONN_POOL_H__
+#define __NET_CONN_POOL_H__ 
+
+#include "usg_common.h"
+#include <poll.h>
+
+#define OPEN_MAX 1024
+
+#define ADDRSTRLEN (NI_MAXHOST + NI_MAXSERV + 10) 
+
+class NetConnPool{ /* Represents a pool of connected descriptors */ 
+public:
+  int nready;       /* Number of ready descriptors from select */
+  int maxi;         /* Highwater index into client array */
+  struct pollfd conns[OPEN_MAX];
+  std::map<std::string, int> connectionMap;
+
+	NetConnPool() ;
+	~NetConnPool() ;
+	/**
+	 * 鑾峰彇杩炴帴
+	*/
+	int getConn(const char *host, int port);
+	/**
+	 * 鏀惧洖杩炴帴
+	*/
+	void putConn(int connfd);
+	/**
+	 * 鍏抽棴杩炴帴
+	 */
+	void closeConn(int connfd);
+
+ 
+	 
+}; 
+
+#endif
\ No newline at end of file
diff --git a/src/socket/net_mod_socket.c b/src/socket/net_mod_socket.c
index 5e0b0b1..a3d4543 100644
--- a/src/socket/net_mod_socket.c
+++ b/src/socket/net_mod_socket.c
@@ -1,15 +1,21 @@
 #include "net_mod_socket.h"
 #include "socket_io.h"
 #include "net_mod_socket_io.h"
+#include "net_conn_pool.h"
 #include <sys/types.h>          /* See NOTES */
 #include <sys/socket.h>
+#include <pthread.h>
 
+static Logger *logger = LoggerFactory::getLogger();
+
+static pthread_once_t once = PTHREAD_ONCE_INIT;
+static pthread_key_t poolKey;
 
 
 
 NetModSocket::NetModSocket() 
 {
-    if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR)    err_msg(errno, "signal");
+  if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR)    err_msg(errno, "signal");
 }
 
 
@@ -49,64 +55,26 @@
 }
 
 
-
-int NetModSocket::connect( pool &mpool, net_node_t *node) {
-  
-  int connfd;
-  int i;
-  char portstr[32];
- 
-  // printf("mis: %s\n", mapKey);     
-  sprintf(portstr, "%d", node->port);
-// printf("open before: %s\n", mapKey); 
-  connfd = open_clientfd(node->host, portstr);
-// printf("open after: %s\n", mapKey); 
-  if(connfd < 0) {
-    LoggerFactory::getLogger()->error(errno, "NetModSocket::connect %s:%d ", node->host, node->port);
-    exit(1);
-    return -1;
-  }
-  
-  for (i = 0; i < OPEN_MAX; i++) { /* Find an available slot */
-    if (mpool.conns[i].fd < 0)
-    {
-      /* Add connected descriptor to the mpool */
-      mpool.conns[i].fd = connfd;  
-               
-      mpool.conns[i].events = POLLIN;
-      /* Add the descriptor to descriptor set */
-      break;
-    }
-  }
-
-  if (i > mpool.maxi)      
-      mpool.maxi = i;   
-
-  if (i == OPEN_MAX) {
-    /* Couldn't find an empty slot */
-    LoggerFactory::getLogger()->error(errno, "add_client error: Too many clients");
-    return -1;
-  }
-
-  return connfd;
+/* Free thread-specific data buffer */
+void NetModSocket::_destructor_(void *_pool)
+{
+  NetConnPool *mpool = (NetConnPool *)_pool;
+  delete mpool;
 }
 
+ /* One-time key creation function */
+void NetModSocket::_createKey_(void)
+{
+  int ret;
 
-void NetModSocket::close_connect(pool &mpool, int connfd) {
-  int i;
-  if(close(connfd) != 0) {
-    LoggerFactory::getLogger()->error(errno, "NetModSocket::close_connect close");
+  /* Allocate a unique thread-specific data key and save the address
+     of the destructor for thread-specific data buffers */
+
+  ret = pthread_key_create(&poolKey, _destructor_);
+  if (ret != 0) {
+    logger->error(ret, "pthread_key_create");
+    exit(1);
   }
- 
-
-  for (i = 0; i <= mpool.maxi; i++) {
-    if(mpool.conns[i].fd == connfd) {
-      mpool.conns[i].fd = -1;
-    }
-  }
-
-  // LoggerFactory::getLogger()->debug( "NetModSocket::close_connect %d\n", connfd);
-
 }
 
 int NetModSocket::_sendandrecv_(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
@@ -116,7 +84,7 @@
   net_node_t *node;
   void *recv_buf;
 
-  pool mpool;
+ 
   
   net_mod_request_head_t request_head = {};
  
@@ -124,8 +92,39 @@
 
    
   net_mod_recv_msg_t *ret_arr = (net_mod_recv_msg_t *)calloc(arrlen, sizeof(net_mod_recv_msg_t));
+
+  int ret;
+  NetConnPool *mpool;
+
+  /* Make first caller allocate key for thread-specific data */
+
+  ret = pthread_once(&once, _createKey_);
+  if (ret != 0) {
+    LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_once");
+    exit(1);
+  }
+
+  mpool = (NetConnPool *)pthread_getspecific(poolKey);
+  if (mpool == NULL)
+  {
+    /* If first call from this thread, allocate
+                                   buffer for thread, and save its location */
+    mpool = new NetConnPool();
+    if (mpool == NULL) {
+      LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ malloc");
+      exit(1);
+    }
+
+   
+
+    ret = pthread_setspecific(poolKey, mpool);
+    if (ret != 0) {
+      LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_setspecific");
+      exit(1);
+    }
+  }
   
-  init_conn_pool(mpool);
+ 
 
   for (i = 0; i< arrlen; i++) {
 
@@ -142,7 +141,7 @@
       continue;
     }
 
-    if( (connfd = connect(mpool, node)) < 0 ) {
+    if( (connfd = mpool->getConn(node->host, node->port)) < 0 ) {
       continue;
     }
 
@@ -157,73 +156,76 @@
  // printf("write_request %s:%d\n", request_head.host, request_head.port);
     if(write_request(connfd, request_head, send_buf, send_size, NULL, 0) != 0) {
       LoggerFactory::getLogger()->error("write_request failture %s:%d\n", node->host, node->port);
-      close_connect(mpool, connfd);
+      mpool->closeConn( connfd);
     } else {
       n_req++;
     }
   
   }
 
-// printf(" mpool.maxi = %d\n",  mpool.maxi);
+// printf(" mpool->maxi = %d\n",  mpool->maxi);
 // printf(" n_req = %d\n", n_req);
 
   while(n_resp < n_req)
   {
     /* Wait for listening/connected descriptor(s) to become ready */
-    if( (mpool.nready = poll(mpool.conns, mpool.maxi + 1, msec) ) <= 0) {
+    if( (mpool->nready = poll(mpool->conns, mpool->maxi + 1, msec) ) <= 0) {
        // wirite_set 鍜� read_set 鍦ㄦ寚瀹氭椂闂村唴閮芥病鍑嗗濂�
       break;
     }
-// printf("mpool.nready =%d\n", mpool.nready);
-    for (i = 0; (i <= mpool.maxi) && (mpool.nready > 0); i++) {
-      if ( (connfd = mpool.conns[i].fd) > 0 ) {
+// printf("mpool->nready =%d\n", mpool->nready);
+    for (i = 0; (i <= mpool->maxi) && (mpool->nready > 0); i++) {
+      if ( (connfd = mpool->conns[i].fd) > 0 ) {
         /* If the descriptor is ready, echo a text line from it */
-        if (mpool.conns[i].revents & POLLIN )
+        if (mpool->conns[i].revents & POLLIN )
         {
-          mpool.nready--;
+          mpool->nready--;
 // printf("POLLIN %d\n", connfd);
           if( (n = read_response(connfd, ret_arr+n_recv_suc)) == 0) {
             n_recv_suc++;
+            // 鎴愬姛鏀跺埌杩斿洖娑堟伅锛屾竻绌鸿鍏ヤ綅
+            mpool->conns[i].fd = -1;
             
           }
-          // else if(n == -1)  {
-          //   // 缃戠粶閿欒
-          // } else {
-          //   // 瀵规柟key鏄叧闂殑
-          // }
+          else if(n == -1)  {
+            // 缃戠粶閿欒
+            mpool->closeConn( connfd);
+            // mpool->conns[i].fd = -1;
+          } else {
+            // 瀵规柟key鏄叧闂殑
+             mpool->conns[i].fd = -1;
+          }
 
-          mpool.conns[i].fd = -1;
-          close_connect(mpool, connfd);
           n_resp++;
 // printf("read response %d\n", n);
           
         }
 
-        if (mpool.conns[i].revents & POLLOUT ) {
+        if (mpool->conns[i].revents & POLLOUT ) {
   // printf("poll POLLOUT %d\n", connfd);        
         }
 
-        if (mpool.conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) )
+        if (mpool->conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) )
         {
 // printf("poll POLLERR %d\n", connfd);
-          mpool.nready--;
-          close_connect(mpool, connfd);
-          mpool.conns[i].fd = -1;
+          mpool->nready--;
+          mpool->closeConn( connfd);
+          // mpool->conns[i].fd = -1;
         }
       }
     }
   }
 
    //瓒呮椂鍚庯紝鍏抽棴瓒呮椂杩炴帴
-  for (i = 0; i <= mpool.maxi; i++) {
-    if ( (connfd = mpool.conns[i].fd) > 0 ) {
+  for (i = 0; i <= mpool->maxi; i++) {
+    if ( (connfd = mpool->conns[i].fd) > 0 ) {
       // 鍏抽棴骞舵竻闄ゅ啓鍏ユ垨璇诲彇澶辫触鐨勮繛鎺�
-      close_connect(mpool, connfd);
-      mpool.conns[i].fd = -1;
+      mpool->closeConn( connfd);
+      // mpool->conns[i].fd = -1;
     }
   }
 
-  mpool.maxi = -1;
+  mpool->maxi = -1;
 
   *recv_arr = ret_arr;
   if(recv_arr_size != NULL) {
@@ -257,10 +259,33 @@
   net_mod_recv_msg_t recv_msg;
   char portstr[32];
   int n_req = 0, n_pub_suc = 0, n_resp = 0;
-  int ret;
 
-  pool mpool;
-  init_conn_pool(mpool);
+  int ret;
+  NetConnPool *mpool;
+
+  /* Make first caller allocate key for thread-specific data */
+  ret = pthread_once(&once, _createKey_);
+  if (ret != 0) {
+    LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_once");
+    exit(1);
+  }
+
+  mpool = (NetConnPool *)pthread_getspecific(poolKey);
+  if (mpool == NULL)
+  {
+    /* If first call from this thread, allocte buffer for thread, and save its location */
+    mpool = new NetConnPool();
+    if (mpool == NULL) {
+      LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ malloc");
+      exit(1);
+    }
+
+    ret = pthread_setspecific(poolKey, mpool);
+    if (ret != 0) {
+      LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_setspecific");
+      exit(1);
+    }
+  }
 
   
   for (i = 0; i < arrlen; i++) {
@@ -274,7 +299,7 @@
      
     } else {
       sprintf(portstr, "%d", node->port);
-      if( (connfd = connect(mpool, node)) < 0 ) {
+      if( (connfd = mpool->getConn(node->host, node->port)) < 0 ) {
         continue;
       }
       request_head.mod = BUS;
@@ -285,7 +310,7 @@
 
       if(write_request(connfd, request_head, content, content_size, topic, request_head.topic_length) != 0) {
         LoggerFactory::getLogger()->error(" NetModSocket::_pub_ write_request failture %s:%d\n", node->host, node->port);
-        close_connect(mpool, connfd);
+        mpool->closeConn( connfd);
       } else {
         n_req++;
       }
@@ -296,61 +321,60 @@
   while(n_resp < n_req)
   {
     /* Wait for listening/connected descriptor(s) to become ready */
-    if( (mpool.nready = poll(mpool.conns, mpool.maxi + 1, timeout) ) <= 0) {
+    if( (mpool->nready = poll(mpool->conns, mpool->maxi + 1, timeout) ) <= 0) {
        // wirite_set 鍜� read_set 鍦ㄦ寚瀹氭椂闂村唴閮芥病鍑嗗濂�
       break;
     }
-// printf("mpool.nready =%d\n", mpool.nready);
-    for (i = 0; (i <= mpool.maxi) && (mpool.nready > 0); i++) {
-      if ( (connfd = mpool.conns[i].fd) > 0 ) {
-        if (mpool.conns[i].revents & POLLIN )
+// printf("mpool->nready =%d\n", mpool->nready);
+    for (i = 0; (i <= mpool->maxi) && (mpool->nready > 0); i++) {
+      if ( (connfd = mpool->conns[i].fd) > 0 ) {
+        if (mpool->conns[i].revents & POLLIN )
         {
-          mpool.nready--;
+          mpool->nready--;
 // printf("POLLIN %d\n", connfd);
           if( (ret = read_response(connfd, &recv_msg)) == 0) {
             
             // 鎴愬姛鏀跺埌杩斿洖娑堟伅锛屾竻绌鸿鍏ヤ綅
-            mpool.conns[i].fd = -1;
+            mpool->conns[i].fd = -1;
             n_pub_suc++;
           } 
-          // else if(ret == -1)  {
-          //   // 缃戠粶杩炴帴閿欒
-          // } else {
-          //   // 瀵规柟鐨刱ey鏄叧闂殑
-          // }
-          mpool.conns[i].fd = -1;
-          close_connect(mpool, connfd);
+          else if(ret == -1)  {
+            // 缃戠粶杩炴帴閿欒
+             mpool->closeConn( connfd);
+          } else {
+            // 瀵规柟鐨刱ey鏄叧闂殑
+            mpool->conns[i].fd = -1;
+          }
           n_resp++;
 // printf("read response %d\n", n);
-          
         }
 
-        if (mpool.conns[i].revents & POLLOUT ) {
+        if (mpool->conns[i].revents & POLLOUT ) {
   // printf("poll POLLOUT %d\n", connfd);        
         }
 
-        if (mpool.conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) )
+        if (mpool->conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) )
         {
 // printf("poll POLLERR %d\n", connfd);
-          mpool.nready--;
+          mpool->nready--;
           
-          mpool.conns[i].fd = -1;
-          close_connect(mpool, connfd);
+          mpool->conns[i].fd = -1;
+          mpool->closeConn( connfd);
         }
       }
     }
   }
 
   //瓒呮椂鍚庯紝鍏抽棴瓒呮椂杩炴帴
-  for (i = 0; i <= mpool.maxi; i++) {
-    if ( (connfd = mpool.conns[i].fd) > 0 ) {
+  for (i = 0; i <= mpool->maxi; i++) {
+    if ( (connfd = mpool->conns[i].fd) > 0 ) {
       // 鍏抽棴骞舵竻闄ゅ啓鍏ユ垨璇诲彇澶辫触鐨勮繛鎺�
-      close_connect(mpool, connfd);
-      mpool.conns[i].fd = -1;
+      mpool->closeConn( connfd);
+      // mpool->conns[i].fd = -1;
     }
   }
 
-  mpool.maxi = -1;
+  mpool->maxi = -1;
    
   return n_pub_suc;
 }
@@ -706,17 +730,6 @@
 
 
 
-void  NetModSocket::init_conn_pool(pool& mpool)
-{
-  /* Initially, there are no connected descriptors */
-  int i;
-  mpool.maxi = -1;                   //line:conc:echoservers:beginempty
-  for (i = 0; i < OPEN_MAX; i++) {
-    mpool.conns[i].fd = -1; 
-    mpool.conns[i].events = 0;      
-  }
-
-}
 
 /**
   uint32_t mod;
diff --git a/src/socket/net_mod_socket.h b/src/socket/net_mod_socket.h
index 5e7b7b7..367dec8 100644
--- a/src/socket/net_mod_socket.h
+++ b/src/socket/net_mod_socket.h
@@ -5,12 +5,13 @@
 #include "socket_io.h"
 #include <poll.h>
 
-#define OPEN_MAX 1024
+
 #define GET(p)       (*(uint32_t *)(p))
 #define PUT(p, val)  (*(uint32_t *)(p) = (val))
 
 #define GET_INT32(p)       (*(int32_t *)(p))
 #define PUT_INT32(p, val)  (*(int32_t *)(p) = (val))
+
 
 class NetModServerSocket;
 
@@ -60,13 +61,7 @@
 };
 
 class NetModSocket {
-  struct pool{ /* Represents a pool of connected descriptors */ //line:conc:echoservers:beginpool
-
-    int nready;       /* Number of ready descriptors from select */
-    int maxi;         /* Highwater index into client array */
-    struct pollfd conns[OPEN_MAX];
-    // std::map<std::string, int> connectionMap;
-  } ; 
+ 
 
   friend class NetModServerSocket;
 private:
@@ -82,9 +77,9 @@
   static void * encode_response_head(net_mod_response_head_t & response);
   static net_mod_response_head_t  decode_response_head(void *_headbs);
 
-  void init_conn_pool( pool& mpool);
-  int connect(pool& mpool, net_node_t* node);
-  void close_connect(pool& mpool , int connfd);
+  static void _destructor_(void *_pool);
+  static void _createKey_(void);
+
   int read_response(int clientfd, net_mod_recv_msg_t *recv_msg);
   int write_request(int clientfd, net_mod_request_head_t &request_head, void *send_buf, int send_size, void *topic_buf, int topic_size);
 

--
Gitblit v1.8.0