update

wangzhengquan

2021-01-27 cffa543dbf014dedc120114e13110a0e3b6e5b57

update

 src/net/net_mod_server_socket.cpp

@@ -171,13 +171,13 @@
      timeout.tv_sec = request_head.timeout / 1000;
      timeout.tv_nsec = (request_head.timeout - timeout.tv_sec * 1000) * 10e6;
      // printf(" timeout.tv_sec = %d,  timeout.tv_nsec=%ld\n",  timeout.tv_sec,  timeout.tv_nsec );
      ret = shmModSocket.sendandrecv_unsafe(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size, &timeout, BUS_TIMEOUT_FLAG);
      ret = shmModSocket.sendandrecv(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size, &timeout, BUS_TIMEOUT_FLAG);
    }
    else if(request_head.timeout == 0) {
      ret = shmModSocket.sendandrecv_unsafe(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size, NULL, BUS_NOWAIT_FLAG);
      ret = shmModSocket.sendandrecv(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size, NULL, BUS_NOWAIT_FLAG);
    }
    else if(request_head.timeout == -1) {
      ret = shmModSocket.sendandrecv_unsafe(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size);
      ret = shmModSocket.sendandrecv(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size);
    }

    if( ret != 0) {

 src/socket/bus_server_socket.cpp

@@ -122,7 +122,7 @@
    void *buf;
    int size = ShmModSocket::get_bus_sendbuf(head, NULL, 0, NULL,  0, &buf);
    if(size > 0) {
        ret = client.sendandrecv_unsafe( buf, size, shm_socket->key, NULL, NULL);
        ret = client.sendandrecv( buf, size, shm_socket->key, NULL, NULL);
        free(buf);
        return ret;
    } else {

 src/socket/shm_mod_socket.cpp

@@ -82,11 +82,11 @@
    return shm_sendandrecv(shm_socket, send_buf, send_size, send_key, recv_buf, recv_size, timeout, flag);
}
 
// 超时返回。 @sec 秒 ， @nsec 纳秒
int ShmModSocket::sendandrecv_unsafe(const void *send_buf, const int send_size, const int send_key, 
    void **recv_buf, int *recv_size, const struct timespec *timeout, int flag){
    return shm_sendandrecv_unsafe(shm_socket, send_buf, send_size, send_key, recv_buf, recv_size, timeout, flag);
}
// // 超时返回。 @sec 秒 ， @nsec 纳秒
// int ShmModSocket::sendandrecv_unsafe(const void *send_buf, const int send_size, const int send_key, 
//     void **recv_buf, int *recv_size, const struct timespec *timeout, int flag){
//     return shm_sendandrecv_unsafe(shm_socket, send_buf, send_size, send_key, recv_buf, recv_size, timeout, flag);
// }
 
/**
 * 订阅指定主题

 src/socket/shm_socket.cpp

@@ -553,27 +553,27 @@
 
}

int shm_sendandrecv_unsafe(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) {
  if (socket->socket_type != SHM_SOCKET_DGRAM) {
    logger->error( "shm_socket.shm_sendandrecv_unsafe : Can't invoke shm_sendandrecv method in a %d type socket  "
                "which is not a SHM_SOCKET_DGRAM socket ",
             socket->socket_type);
    exit(1);
  }
  int recv_key;
  int rv;
// int shm_sendandrecv_unsafe(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) {
//   if (socket->socket_type != SHM_SOCKET_DGRAM) {
//     logger->error( "shm_socket.shm_sendandrecv_unsafe : Can't invoke shm_sendandrecv method in a %d type socket  "
//                 "which is not a SHM_SOCKET_DGRAM socket ",
//              socket->socket_type);
//     exit(1);
//   }
//   int recv_key;
//   int rv;

 
  if ((rv = shm_sendto(socket, send_buf, send_size, send_key, timeout, flags)) == 0) {
    rv = shm_recvfrom(socket, recv_buf, recv_size, &recv_key, timeout, flags);
    return rv;
  } else {
    return rv;
  }
  return -1;
}
//   if ((rv = shm_sendto(socket, send_buf, send_size, send_key, timeout, flags)) == 0) {
//     rv = shm_recvfrom(socket, recv_buf, recv_size, &recv_key, timeout, flags);
//     return rv;
//   } else {
//     return rv;
//   }
//   return -1;
// }

int shm_sendandrecv(shm_socket_t *socket, const void *send_buf,
                    const int send_size, const int send_key, void **recv_buf,

 test_net_socket/net_mod_socket.sh

@@ -27,7 +27,7 @@
# 无限循环send
function send() {
    ./test_net_mod_socket --fun="test_net_sendandrecv" \
     --sendlist="localhost:5000:100, localhost:5000:100"
     --sendlist=" :5000:100,  :5000:100"
     
}
# 多线程send

 test_net_socket/test_net_mod_socket.cpp

@@ -346,7 +346,7 @@
    n = net_mod_socket_sendandrecv_timeout(client, node_arr, node_arr_size, content, strlen(content), &recv_arr, &recv_arr_size, 1000);
    printf(" %d nodes reply\n", n);
    for(i=0; i<recv_arr_size; i++) {
      LoggerFactory::getLogger()->debug("reply from (host:%s, port: %d, key:%d) >> %s\n", 
      LoggerFactory::getLogger()->debug("%ld received reply from (host:%s, port: %d, key:%d) >> %s\n", (long)getpid(),
        recv_arr[i].host,
        recv_arr[i].port,
        recv_arr[i].key,