update

wangzhengquan

2021-02-22 044e10574fa4e007be408d991861d34ecf22622a

 src/net/net_mod_socket.cpp

@@ -150,17 +150,18 @@
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 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_recv_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_recv_err_t err_arr[arrlen];
 
  NetConnPool *mpool = _get_pool();

@@ -179,6 +180,7 @@
      } 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;
@@ -187,13 +189,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_recv_err].port = 0;
        err_arr[n_recv_err].key = node->key;
        err_arr[n_recv_err].code = ret;
        n_recv_err++;
        logger->error("NetModSocket:: %d _sendandrecv_ to key %d failed, %s", get_key(), node->key, bus_strerror(ret));
      }

     
@@ -201,6 +201,11 @@
    }

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

@@ -215,6 +220,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_recv_err].host, node->host, sizeof(err_arr[n_recv_err].host));
      err_arr[n_recv_err].port =  node->port;
      err_arr[n_recv_err].key = node->key;
      err_arr[n_recv_err].code = EBUS_NET;
      n_recv_err++;
      mpool->closeConn( connfd);
    } else {
      n_req++;
@@ -240,19 +250,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_recv_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_recv_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_recv_err++;
          }

          n_resp++;
@@ -337,6 +353,8 @@
  int ret;
  NetConnPool *mpool = _get_pool();

  net_mod_recv_err_t err_msg;

  // 本地发送
  if(node_arr == NULL || arrlen == 0) {
    if(msec == 0) {
@@ -410,7 +428,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;
@@ -641,7 +659,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_recv_err_t *err_arr) {
  int recv_size;
  void *recv_buf;
  char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH];
@@ -649,14 +667,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;
  }

@@ -666,6 +691,11 @@
    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;