test for perf

zhangmeng

2024-04-09 2561a007b8d8999a4750046d0cfb3b1ad5af50ac

 src/net/net_mod_socket.cpp

@@ -15,39 +15,20 @@

NetModSocket::NetModSocket() 
{
  int s;
  if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR)
      logger->error(errno, "NetModSocket::NetModSocket signal");

  gpool = new NetConnPool();

  pthread_mutexattr_t mtxAttr;
  s = pthread_mutexattr_init(&mtxAttr);
  if (s != 0)
    err_exit(s, "pthread_mutexattr_init");
  s = pthread_mutexattr_settype(&mtxAttr, PTHREAD_MUTEX_ERRORCHECK);
  if (s != 0)
    err_exit(s, "pthread_mutexattr_settype");
  s = pthread_mutex_init(&sendMutex, &mtxAttr);
  if (s != 0)
    err_exit(s, "pthread_mutex_init");

  s = pthread_mutexattr_destroy(&mtxAttr);
  if (s != 0)
    err_exit(s, "pthread_mutexattr_destroy");
    logger->error(errno, "NetModSocket::NetModSocket signal");

}


NetModSocket::~NetModSocket() {
  int s;
  delete gpool;
  s =  pthread_mutex_destroy(&sendMutex);
  if(s != 0) {
    err_exit(s, "shm_close_socket");
  }

}


int NetModSocket::stop() {
  return shmModSocket.stop();
}

/**
 * 绑定端口到socket, 如果不绑定则系统自动分配一个
@@ -65,19 +46,40 @@
  return shmModSocket.force_bind(key);
}

int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
  return _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, -1);
int NetModSocket::reg(void *pData, int len, void **buf, int *size, const int timeout_ms, int flag) {
  
  return shmModSocket.reg(pData, len, buf, size, timeout_ms, flag);
}
int NetModSocket::sendandrecv_timeout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int  msec) {
  return _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, msec);
}
int NetModSocket::sendandrecv_nowait(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
  return _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, 0);

void NetModSocket::int_set(int data) {
  int_val = data;
}

void NetModSocket::svr_set(int data) {
  svr_val = data;
}

int NetModSocket::int_get(void) {
  return int_val;
}

int NetModSocket::svr_get(void) {
  return svr_val;
}

// int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
//   net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
//   return _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, -1);
// }
// int NetModSocket::sendandrecv_timeout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
//   net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int  msec) {
//   return _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, msec);
// }
// int NetModSocket::sendandrecv_nowait(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
//   net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
//   return _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, 0);

// }


/* Free thread-specific data buffer */
@@ -86,7 +88,7 @@

  NetConnPool *mpool = (NetConnPool *)_pool;
  delete mpool;
  logger->debug("destory connPool");

}

 /* One-time key creation function */
@@ -119,7 +121,6 @@
  if (mpool == NULL)
  {
    /* If first call from this thread, allocate buffer for thread, and save its location */
    logger->debug("Create connPool");
    mpool = new NetConnPool();
    if (mpool == NULL) {
      LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ malloc");
@@ -141,20 +142,24 @@
}


int NetModSocket::_sendandrecv_(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int  msec ) {

  int i, n, recv_size, connfd;

int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, 
  net_mod_err_t ** _err_arr, int *_err_arr_size, int  msec ) {

  int i,  recv_size, connfd;
  net_node_t *node;
  void *recv_buf = NULL;
  struct timespec timeout;
  int ret;
  int n_req = 0, n_recv_suc = 0, n_resp =0;
  int n_req = 0, n_recv_suc = 0, n_err = 0, n_resp =0;
  
  net_mod_request_head_t request_head = {};
   
  net_mod_recv_msg_t *ret_arr = (net_mod_recv_msg_t *)calloc(arrlen, sizeof(net_mod_recv_msg_t));

  net_mod_err_t *err_arr = (net_mod_err_t *)calloc(arrlen, sizeof(net_mod_err_t));
 
  NetConnPool *mpool = _get_pool();

@@ -165,14 +170,15 @@
      // 本地发送
     
      if(msec == 0) {
        ret = shmModSocket.sendandrecv_nowait(send_buf, send_size, node->key, &recv_buf, &recv_size);
        ret = shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size, NULL, BUS_NOWAIT_FLAG);
      } else if(msec > 0){
        timeout.tv_sec = msec / 1000;
        timeout.tv_nsec = (msec - timeout.tv_sec * 1000) * 10e6;
        ret = shmModSocket.sendandrecv_timeout(send_buf, send_size, node->key, &recv_buf, &recv_size, &timeout);
        ret = shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size, &timeout, BUS_TIMEOUT_FLAG);
      } else {
        ret = shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size);
      }

      if( ret == 0) {
        strcpy( ret_arr[n_recv_suc].host, "");
        ret_arr[n_recv_suc].port = 0;
@@ -181,13 +187,11 @@
        ret_arr[n_recv_suc].content_length = recv_size;
        n_recv_suc++;
      } else {
        if(ret > EBUS_BASE) {
          // bus_errno = EBUS_TIMEOUT;
          logger->debug("NetModSocket:: %d _sendandrecv_ to key %d failed, %s", get_key(), node->key, bus_strerror(ret));
           
        } else {
          logger->error(ret, "NetModSocket:: %d _sendandrecv_ to key %d failed", get_key(),  node->key);
        }
        err_arr[n_err].port = 0;
        err_arr[n_err].key = node->key;
        err_arr[n_err].code = ret;
        n_err++;
        // logger->error("NetModSocket::  _sendandrecv_ to key %d failed. %s",  node->key, bus_strerror(ret));
      }

     
@@ -195,6 +199,11 @@
    }

    if( (connfd = mpool->getConn(node->host, node->port)) < 0 ) {
      memcpy(err_arr[n_err].host, node->host, sizeof(err_arr[n_err].host));
      err_arr[n_err].port =  node->port;
      err_arr[n_err].key = node->key;
      err_arr[n_err].code = EBUS_NET;
      n_err++;
      continue;
    }

@@ -209,6 +218,11 @@
 // 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);
      memcpy(err_arr[n_err].host, node->host, sizeof(err_arr[n_err].host));
      err_arr[n_err].port =  node->port;
      err_arr[n_err].key = node->key;
      err_arr[n_err].code = EBUS_NET;
      n_err++;
      mpool->closeConn( connfd);
    } else {
      n_req++;
@@ -234,19 +248,25 @@
        {
          mpool->nready--;
// printf("POLLIN %d\n", connfd);
          if( (n = read_response(connfd, ret_arr+n_recv_suc)) == 0) {
          if( (ret = read_response(connfd, ret_arr+n_recv_suc, err_arr + n_err)) == 0) {
            n_recv_suc++;
            // 成功收到返回消息，清空读入位
            mpool->conns[i].fd = -1;
            
          }
          else if(n == EBUS_NET)  {
          else if(ret == EBUS_NET)  {
            // 网络错误

            logger->error("NetModSocket::_sendandrecv_ read_response key = %d , %s", get_key(),  bus_strerror(ret));
            mpool->closeConn( connfd);
            n_err++;
            // mpool->conns[i].fd = -1;
          } else {
            // 代理服务没有转发成功
             mpool->conns[i].fd = -1;
             
            logger->error("NetModSocket::_sendandrecv_ read_response key = %d , %s", get_key(),  bus_strerror(ret));
            mpool->conns[i].fd = -1;
            n_err++;
          }

          n_resp++;
@@ -278,19 +298,71 @@

  mpool->maxi = -1;


  if(recv_arr != NULL) {
    *recv_arr = ret_arr;
   
    if(n_recv_suc > 0)  {
     *recv_arr = ret_arr;
     
    } else {
      free_recv_msg_arr(ret_arr, n_recv_suc);
    }
    
  } else {
    free_recv_msg_arr(ret_arr, n_recv_suc);
  }
  

  if(recv_arr_size != NULL) {
    *recv_arr_size = n_recv_suc;
     *recv_arr_size = n_recv_suc;
  }
  

  if(_err_arr != NULL) {
     
    if(n_err > 0) {
      *_err_arr = err_arr;
     

    } else {
      *_err_arr = NULL;
      *_err_arr_size = 0;
      free(err_arr);
    }
   
  } else {
    free(err_arr);
  }

  if(_err_arr_size != NULL) {
    *_err_arr_size = n_err;
  }
  return n_recv_suc;
     
}

void NetModSocket::buf_data_set(std::string str, int val) {
  recvbuf.insert({str, val}); 
}

int NetModSocket::buf_data_get(std::string str) {
  
  int i;
  int val = 0;
  std::map<std::string, int>::iterator recvIter;

  recvIter = recvbuf.find(str);
  if(recvIter != recvbuf.end()) {
    
    val = recvIter->second;

  }

  return val;
}

void NetModSocket::buf_data_del(std::string str) {
  recvbuf.erase(str); 
}

void NetModSocket::free_recv_msg_arr(net_mod_recv_msg_t * arr, size_t size) {

@@ -302,22 +374,19 @@
}


int NetModSocket::pub(net_node_t *node_arr, int arrlen, char *topic, int topic_size, void *content, int content_size) {
int NetModSocket::pub(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content, int content_size) {
  return _pub_(node_arr, arrlen, topic, topic_size, content,   content_size, -1);
}

int NetModSocket::pub_nowait(net_node_t *node_arr, int arrlen, char *topic, int topic_size, void *content, int content_size) {
int NetModSocket::pub_nowait(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content, int content_size) {
  return _pub_(node_arr, arrlen, topic, topic_size, content,   content_size, 0);
}

int NetModSocket::pub_timeout(net_node_t *node_arr, int arrlen, char *topic, int topic_size, void *content, int content_size, int  msec ) {
int NetModSocket::pub_timeout(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content, int content_size, int  msec ) {
  return _pub_(node_arr, arrlen, topic, topic_size, content, content_size, msec);
}


// int  pub(char *topic, int topic_size, void *content, int content_size, int port);

int NetModSocket::_pub_(net_node_t *node_arr, int arrlen, char *topic, int topic_size, void *content,
int NetModSocket::_pub_(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content,
 int content_size, int  msec) {
  int i, connfd;
  net_node_t *node;
@@ -331,20 +400,23 @@
  int ret;
  NetConnPool *mpool = _get_pool();

  net_mod_err_t err_msg;

  // 本地发送
  if(node_arr == NULL || arrlen == 0) {
  if ((node_arr == NULL) || (arrlen == 0)) {
    if(msec == 0) {
      ret = shmModSocket.pub_nowait(topic, topic_size, content, content_size, SHM_BUS_KEY);
      ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY,  NULL, BUS_NOWAIT_FLAG);
    } else if(msec > 0) {
      timeout.tv_sec = msec / 1000;
      timeout.tv_nsec = (msec - timeout.tv_sec * 1000) * 10e6;
      ret = shmModSocket.pub_timeout(topic, topic_size, content, content_size, SHM_BUS_KEY, &timeout);
      ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, &timeout, BUS_TIMEOUT_FLAG);
    } else {
      ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY);
    }
    if(ret == 0 ) {
      n_pub_suc++;
    }
    return n_pub_suc;
  }

  for (i = 0; i < arrlen; i++) {
@@ -353,11 +425,11 @@
    if(node->host == NULL) {
      // 本地发送
      if(msec == 0) {
        ret = shmModSocket.pub_nowait(topic, topic_size, content, content_size, SHM_BUS_KEY);
        ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, NULL, BUS_NOWAIT_FLAG);
      } else if(msec > 0) {
        timeout.tv_sec = msec / 1000;
        timeout.tv_nsec = (msec - timeout.tv_sec * 1000) * 10e6;
        ret = shmModSocket.pub_timeout(topic, topic_size, content, content_size, SHM_BUS_KEY, &timeout);
        ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, &timeout, BUS_TIMEOUT_FLAG);
      } else {
        ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY);
      }
@@ -374,7 +446,7 @@
      }
      request_head.mod = BUS;
      memcpy(request_head.host, node->host, sizeof(request_head.host));
      request_head.key = node->key;
      request_head.key = SHM_BUS_KEY;
      request_head.content_length = content_size;
      request_head.topic_length = strlen(topic) + 1;
      request_head.timeout = msec;
@@ -403,7 +475,7 @@
        {
          mpool->nready--;
// printf("POLLIN %d\n", connfd);
          if( (ret = read_response(connfd, &recv_msg)) == 0) {
          if( (ret = read_response(connfd, &recv_msg, &err_msg)) == 0) {
            
            // 成功收到返回消息，清空读入位
            mpool->conns[i].fd = -1;
@@ -455,56 +527,21 @@
 * @key 发送给谁
 * @return 0 成功， 其他值 失败的错误码
 */
int NetModSocket::sendto(const void *buf, const int size, const int key){
  int rv = shmModSocket.sendto(buf, size, key);
  if(rv == 0) {
    logger->debug("NetModSocket::sendto: %d sendto %d success.\n", get_key(), key);
    return 0;
  }

  if(rv > EBUS_BASE) {
    // bus_errno = EBUS_TIMEOUT;
    logger->debug("NetModSocket::sendto: %d sendto  %d failed %s", get_key(), key, bus_strerror(rv));
  } else {
    logger->error(rv, "NetModSocket::sendto : %d sendto  %d failed", get_key(), key);
  }
  return rv;
int NetModSocket::sendto(const void *buf, const int size, const int key, int reset, int data_set){
  return shmModSocket.sendto(buf, size, key, 0, 0, reset, data_set);
}

// 发送信息超时返回。 @sec 秒 ， @nsec 纳秒
int NetModSocket::sendto_timeout(const void *buf, const int size, const int key, int sec, int nsec){
int NetModSocket::sendto_timeout(const void *buf, const int size, const int key, int sec, int nsec, int reset, int data_set){
  struct timespec timeout = {sec, nsec};
  int rv = shmModSocket.sendto_timeout(buf, size, key, &timeout);
  if(rv == 0) {
    logger->debug("NetModSocket::sendto_timeout: %d sendto %d success.\n", get_key(), key);
    return 0;
  }

  if(rv > EBUS_BASE) {
    // bus_errno = EBUS_TIMEOUT;
    logger->debug("NetModSocket::sendto_timeout : %d sendto  %d failed %s", get_key(),  key, bus_strerror(rv));
  } else {
    logger->error(rv, "NetModSocket::sendto_timeout:  %d sendto  %d failed", get_key(),  key);
  }
  return rv;
  return shmModSocket.sendto(buf, size, key, &timeout, BUS_TIMEOUT_FLAG, reset, data_set);
   
}

// 发送信息立刻返回。
int NetModSocket::sendto_nowait(const void *buf, const int size, const int key){
  int rv = shmModSocket.sendto_nowait(buf, size, key);
  if(rv == 0) {
    logger->debug("NetModSocket::sendto_nowait: %d sendto %d success.\n", get_key(), key);
    return 0;
  }

  if(rv > EBUS_BASE) {
    // bus_errno = EBUS_TIMEOUT;
    logger->debug("NetModSocket::sendto_nowait %d sendto  %d failed %s", get_key(), key, bus_strerror(rv));
     
  } else {
    logger->error(rv, "NetModSocket::sendto_nowait %d sendto  %d failed", get_key(), key);
  }
  return rv;
int NetModSocket::sendto_nowait(const void *buf, const int size, const int key, int reset, int data_set){
  return shmModSocket.sendto(buf, size, key, NULL, BUS_NOWAIT_FLAG, reset, data_set);
  
}

/**
@@ -512,55 +549,29 @@
 * @key 从谁哪里收到的信息
 * @return 0 成功， 其他值 失败的错误码
*/
int NetModSocket::recvfrom(void **buf, int *size, int *key) {
  int rv = shmModSocket.recvfrom(buf, size, key);
int NetModSocket::recvfrom(void **buf, int *size, int *key, int reset, int data_set) {

  if(rv == 0) {
    logger->debug("NetModSocket::recvfrom:  %d recvfrom %d success.\n", get_key(), *key);
    return 0;
  }

  if(rv > EBUS_BASE) {
    logger->debug("NetModSocket::recvfrom: socket %d recvfrom failed %s", get_key(), bus_strerror(rv));
  } else {
    logger->error(rv, "NetModSocket::recvfrom: socket %d recvfrom failed",  get_key());
  }
  return rv;
  return shmModSocket.recvfrom(buf, size, key, 0, 0, reset, data_set);
 
}

// 接受信息超时返回。 @sec 秒 ， @nsec 纳秒
int NetModSocket::recvfrom_timeout(void **buf, int *size, int *key, int sec, int nsec){
int NetModSocket::recvfrom_timeout(void **buf, int *size, int *key, int sec, int nsec, int reset, int data_set){
  struct timespec timeout = {sec, nsec};
  int rv = shmModSocket.recvfrom_timeout(buf, size, key, &timeout);
  if(rv == 0) {
    logger->debug("NetModSocket::recvfrom:  %d recvfrom %d success.\n", get_key(), *key);
    return 0;
  }

  if(rv > EBUS_BASE) {
    // bus_errno = EBUS_TIMEOUT;
    logger->debug("NetModSocket::recvfrom_timeout:  %d recvfrom failed %s", get_key(), bus_strerror(rv));
  } else {
    logger->error(rv, "NetModSocket::recvfrom_timeout:  %d recvfrom failed",  get_key());
  }
  return rv;
  return shmModSocket.recvfrom(buf, size, key, &timeout, BUS_TIMEOUT_FLAG, reset, data_set);
  
}

int NetModSocket::recvfrom_nowait(void **buf, int *size, int *key){
  int rv = shmModSocket.recvfrom_nowait(buf, size, key);
  if(rv == 0) {
    logger->debug("NetModSocket::recvfrom:  %d recvfrom %d success.\n", get_key(), *key);
    return 0;
  }

  if(rv > EBUS_BASE) {
    // bus_errno = EBUS_TIMEOUT;
    logger->debug("NetModSocket::recvfrom_nowait:  %d recvfrom failed %s", get_key(), bus_strerror(rv));
  } else {
    logger->error(rv, "NetModSocket::recvfrom_nowait:  %d recvfrom failed",  get_key());
  }
  return rv;
int NetModSocket::recvfrom_nowait(void **buf, int *size, int *key, int reset, int data_set){
  return shmModSocket.recvfrom(buf, size, key, NULL, BUS_NOWAIT_FLAG, reset, data_set);
}

int NetModSocket::recvandsend(recvandsend_callback_fn callback,
                              const struct timespec *timeout , int flag, void * user_data ) {

  return shmModSocket.recvandsend(callback, timeout, flag, user_data);
}
 

/**
 * 发送请求信息并等待接收应答
@@ -573,10 +584,10 @@
// 超时返回。 @sec 秒 ， @nsec 纳秒
int NetModSocket::sendandrecv_timeout( const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size, int sec, int nsec){
  struct timespec timeout = {sec, nsec};
  return shmModSocket.sendandrecv_timeout(send_buf, send_size, key, recv_buf, recv_size, &timeout);
  return shmModSocket.sendandrecv(send_buf, send_size, key, recv_buf, recv_size, &timeout, BUS_TIMEOUT_FLAG);
}
int NetModSocket::sendandrecv_nowait( const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size) {
  return shmModSocket.sendandrecv_nowait(send_buf, send_size, key, recv_buf, recv_size);
  return shmModSocket.sendandrecv(send_buf, send_size, key, recv_buf, recv_size, NULL, BUS_NOWAIT_FLAG);
}


@@ -592,10 +603,10 @@
// 超时返回。 @sec 秒 ， @nsec 纳秒
int  NetModSocket::sub_timeout( void *topic, int size, int key, int sec, int nsec){
  struct timespec timeout = {sec, nsec};
  return shmModSocket.sub_timeout((char *)topic,  size,  key, &timeout);
  return shmModSocket.sub((char *)topic,  size,  key, &timeout, BUS_TIMEOUT_FLAG);
}
int  NetModSocket::sub_nowait( void *topic, int size, int key){
  return shmModSocket.sub_nowait((char *)topic,  size,  key);
  return shmModSocket.sub((char *)topic,  size,  key, NULL, BUS_NOWAIT_FLAG);
}


@@ -612,10 +623,10 @@
// 超时返回。 @sec 秒 ， @nsec 纳秒
int  NetModSocket::desub_timeout( void *topic, int size, int key, int sec, int nsec){
  struct timespec timeout = {sec, nsec};
  return shmModSocket.desub_timeout((char *)topic,  size,  key, &timeout);
  return shmModSocket.desub((char *)topic,  size,  key, &timeout, BUS_TIMEOUT_FLAG);
}
int  NetModSocket::desub_nowait( void *topic, int size, int key){
  return shmModSocket.desub_nowait((char *)topic,  size,  key);
  return shmModSocket.desub((char *)topic,  size,  key, NULL, BUS_NOWAIT_FLAG);
}


@@ -632,10 +643,10 @@
//  超时返回。 @sec 秒 ， @nsec 纳秒
int  NetModSocket::pub_timeout( char *topic, int topic_size, void *content, int content_size, int key, int sec, int nsec){
  struct timespec timeout = {sec, nsec};
  return shmModSocket.pub_timeout(topic, topic_size, content, content_size, key, &timeout);
  return shmModSocket.pub(topic, topic_size, content, content_size, key, &timeout, BUS_TIMEOUT_FLAG);
}
int  NetModSocket::pub_nowait( char *topic, int topic_size, void *content, int content_size, int key){
  return shmModSocket.pub_nowait(topic, topic_size, content, content_size, key);
  return shmModSocket.pub(topic, topic_size, content, content_size, key, NULL, BUS_NOWAIT_FLAG);
}


@@ -653,7 +664,7 @@
//======================================================================================

int NetModSocket::write_request(int clientfd, net_mod_request_head_t &request_head, 
  void *content_buf, int content_size, void *topic_buf, int topic_size) {
  const void *content_buf, int content_size, const void *topic_buf, int topic_size) {
 
  int buf_size;
  char *buf;
@@ -694,7 +705,7 @@
 * @return 0 成功,   EBUS_NET 网络错误， 其他值 代理服务没有转发成功。
 *
 */
int NetModSocket::read_response(int connfd, net_mod_recv_msg_t *recv_msg) {
int NetModSocket::read_response(int connfd, net_mod_recv_msg_t *recv_msg, net_mod_err_t *err_arr) {
  int recv_size;
  void *recv_buf;
  char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH];
@@ -702,14 +713,21 @@
  net_mod_response_head_t response_head;
  if ( rio_readn(connfd, response_head_bs, NET_MODE_RESPONSE_HEAD_LENGTH) !=  NET_MODE_RESPONSE_HEAD_LENGTH) {
    LoggerFactory::getLogger()->error(errno, "NetModSocket::read_response  rio_readnb response_head");
    
    return -1;
    memcpy(err_arr->host, "unkown", sizeof(err_arr->host));
    err_arr->port =  0;
    err_arr->key = 0;
    err_arr->code = EBUS_NET;
    return EBUS_NET;
  }

  response_head =  NetModSocket::decode_response_head(response_head_bs);
// printf(">>>> read_response %s\n", response_head.host);
  if(response_head.code != 0) {
    // 代理服务没能成功发送给对应的key
    memcpy(err_arr->host, response_head.host, sizeof(err_arr->host));
    err_arr->port =  response_head.port;
    err_arr->key = response_head.key;
    err_arr->code = response_head.code;
    return response_head.code;
  }

@@ -718,7 +736,13 @@
    LoggerFactory::getLogger()->error(errno, "NetModSocket::send malloc recv_buf");
    exit(1);
  }

  if ( (recv_size = rio_readn(connfd, recv_buf, response_head.content_length) ) !=  response_head.content_length) {

    memcpy(err_arr->host, response_head.host, sizeof(err_arr->host));
    err_arr->port =  response_head.port;
    err_arr->key = response_head.key;
    err_arr->code = EBUS_NET;
    LoggerFactory::getLogger()->error(errno, "NetModSocket::read_response  rio_readnb recv_buf");
    //网络错误
    return EBUS_NET;
@@ -779,23 +803,23 @@

  head.mod = ntohl(GET(tmp_ptr));

  tmp_ptr += 4;
  tmp_ptr += sizeof(uint32_t);
  memcpy(head.host, tmp_ptr, sizeof(head.host));

 
  tmp_ptr += sizeof(head.host);
  head.port = ntohl(GET(tmp_ptr));

  tmp_ptr += 4;
  tmp_ptr += sizeof(uint32_t);
  head.key = ntohl(GET(tmp_ptr));

  tmp_ptr += 4;
  tmp_ptr += sizeof(uint32_t);
  head.content_length = ntohl(GET(tmp_ptr));

  tmp_ptr += 4;
  tmp_ptr += sizeof(uint32_t);
  head.topic_length = ntohl(GET(tmp_ptr));

  tmp_ptr += 4;
  tmp_ptr += sizeof(uint32_t);
  head.timeout = ntohl(GET_INT32(tmp_ptr));
 
  return head;