resize the buf allocation to dynamic buf.

fujuntang

2021-09-07 b984dac55bdd0a5b694837fb8fb2ceb07ce9808f

 src/net/net_mod_socket.cpp

@@ -22,7 +22,7 @@


NetModSocket::~NetModSocket() {
 

}


@@ -46,19 +46,28 @@
  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::bind_proc_id(char *buf, int len) {
  return shmModSocket.bind_proc_id(buf, len);
}
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);

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(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 */
@@ -67,7 +76,7 @@

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

}

 /* One-time key creation function */
@@ -124,21 +133,22 @@



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 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_recv_err = 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_recv_err_t err_arr[arrlen];
  net_mod_err_t *err_arr = (net_mod_err_t *)calloc(arrlen, sizeof(net_mod_err_t));
 
  NetConnPool *mpool = _get_pool();

@@ -166,11 +176,11 @@
        ret_arr[n_recv_suc].content_length = recv_size;
        n_recv_suc++;
      } else {
        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::  _sendandrecv_ to key %d failed. %s",  node->key, bus_strerror(ret));
        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));
      }

     
@@ -178,11 +188,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++;
      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;
    }

@@ -197,11 +207,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++;
      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++;
@@ -227,7 +237,7 @@
        {
          mpool->nready--;
// printf("POLLIN %d\n", connfd);
          if( (ret = read_response(connfd, ret_arr+n_recv_suc, err_arr + n_recv_err)) == 0) {
          if( (ret = read_response(connfd, ret_arr+n_recv_suc, err_arr + n_err)) == 0) {
            n_recv_suc++;
            // 成功收到返回消息，清空读入位
            mpool->conns[i].fd = -1;
@@ -238,14 +248,14 @@

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

          n_resp++;
@@ -277,14 +287,43 @@

  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;
     
@@ -313,9 +352,6 @@
  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, const char *topic, int topic_size, const void *content,
 int content_size, int  msec) {
  int i, connfd;
@@ -330,10 +366,10 @@
  int ret;
  NetConnPool *mpool = _get_pool();

  net_mod_recv_err_t err_msg;
  net_mod_err_t err_msg;

  // 本地发送
  if(node_arr == NULL || arrlen == 0) {
  if ((node_arr == NULL) || (arrlen == 0)) {
    if(msec == 0) {
      ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY,  NULL, BUS_NOWAIT_FLAG);
    } else if(msec > 0) {
@@ -496,7 +532,6 @@
  return shmModSocket.recvfrom(buf, size, key, NULL, BUS_NOWAIT_FLAG);
}


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

@@ -636,7 +671,7 @@
 * @return 0 成功,   EBUS_NET 网络错误， 其他值 代理服务没有转发成功。
 *
 */
int NetModSocket::read_response(int connfd, net_mod_recv_msg_t *recv_msg, net_mod_recv_err_t *err_arr) {
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];
@@ -667,6 +702,7 @@
    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));
@@ -733,23 +769,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;