update

wangzhengquan

2021-02-08 82841486c36288d73e95f3316e91dd7a522d8602

 src/socket/shm_socket.cpp

@@ -429,12 +429,14 @@

// use thread local
static int _shm_sendandrecv_thread_local(shm_socket_t *sockt, const void *send_buf,
                    const int send_size, const int send_key, void **recv_buf,
                    const int send_size, const int key, void **recv_buf,
                    int *recv_size,  const struct timespec *timeout,  int flags) {
  int recv_key;
  int rv;
  int tryn = 0;

  
 
  int rv, tryn = 3;
  shm_packet_t sendpak;
  shm_packet_t recvpak;
  std::map<int, shm_packet_t>::iterator recvbufIter;
  // 用thread local 保证每个线程用一个独占的socket接受对方返回的信息
  shm_socket_t *tmp_socket;
 
@@ -459,33 +461,102 @@
      exit(1);
    }
  }
   // int rv;
  // int tryn = 0;
  // int recv_key;
  // rv = shm_sendto(tmp_socket, send_buf, send_size, send_key, timeout, flags);

  if ((rv = shm_sendto(tmp_socket, send_buf, send_size, send_key, timeout, flags)) == 0) {
  // if (() == 0) {

    while(tryn < 3) {
      tryn++;
      rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags);
      if(rv != 0) {
        logger->error("_shm_sendandrecv_thread_local : %s\n", bus_strerror(rv));
        return rv;
      }
  //   while(tryn < 3) {
  //     tryn++;
  //     rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags);
  //     if(rv != 0) {
  //       logger->error("_shm_sendandrecv_thread_local : %s\n", bus_strerror(rv));
  //       return rv;
  //     }
     
       // 超时导致接发送对象，与返回对象不对应的情况
      if(send_key != recv_key) {
        logger->debug("======%d use tmp_socket %d, send to  %d, receive from  %d\n", shm_socket_get_key(sockt), shm_socket_get_key(tmp_socket), send_key, recv_key);
        // logger->error( "_shm_sendandrecv_alloc_new: send key expect to equal to recv key! send key =%d , recv key=%d", send_key, recv_key);
        // exit(1);
        continue;
        // return EBUS_RECVFROM_WRONG_END;
  //      // 超时导致接发送对象，与返回对象不对应的情况
  //     if(send_key != recv_key) {
  //       logger->debug("======%d use tmp_socket %d, send to  %d, receive from  %d\n", shm_socket_get_key(sockt), shm_socket_get_key(tmp_socket), send_key, recv_key);
  //       // logger->error( "_shm_sendandrecv_alloc_new: send key expect to equal to recv key! send key =%d , recv key=%d", send_key, recv_key);
  //       // exit(1);
  //       continue;
  //       // return EBUS_RECVFROM_WRONG_END;
  //     }

  //     return 0;
  //   }
    
  //   return EBUS_RECVFROM_WRONG_END;
  // }  

 

  sendpak.key = tmp_socket->key;
  sendpak.size = send_size;
  if(send_buf != NULL) {
    sendpak.buf = mm_malloc(send_size);
    memcpy(sendpak.buf, send_buf, send_size);
  }
  rv = shm_sendpakto(tmp_socket, &sendpak, key, timeout, flags);

  if(rv != 0) {
    return rv;
  }

  if(rv != 0) {
    return rv;
  }

  while(tryn > 0) {
    tryn--;
    recvbufIter = tmp_socket->recvbuf2.find(key);
    if(recvbufIter != tmp_socket->recvbuf2.end()) {
      // 在缓存里查到了UUID匹配成功的
// logger->debug("get from recvbuf: %s", uuid.c_str());
      recvpak = recvbufIter->second;
      sockt->recvbuf2.erase(recvbufIter);
      goto LABLE_SUC;
    }

    rv = shm_recvpakfrom(tmp_socket, &recvpak, timeout, flags);

    if (rv != 0) {

      if(rv == ETIMEDOUT) {
        return EBUS_TIMEOUT;
      }

      return 0;
    }
    
    return EBUS_RECVFROM_WRONG_END;
  }  
      logger->debug("%d shm_recvfrom failed %s", shm_socket_get_key(tmp_socket), bus_strerror(rv));
      return rv;
    } 

  return rv;
   if (key == recvpak.key) {
      // 发送与接受的UUID匹配成功
      goto LABLE_SUC;
    } else {
      // 答非所问，放到缓存里
      tmp_socket->recvbuf2.insert({recvpak.key, recvpak});
      continue;
    }
  }

LABLE_FAIL:
  return EBUS_RECVFROM_WRONG_END;
 
LABLE_SUC:
 if(recv_buf != NULL) {
    void *_buf = malloc(recvpak.size);
    memcpy(_buf, recvpak.buf, recvpak.size);
    *recv_buf = _buf; 
  }
 
  if(recv_size != NULL)
    *recv_size = recvpak.size;

  mm_free(recvpak.buf);
  return 0;
}

static int _shm_sendandrecv_alloc_new(shm_socket_t *sockt, const void *send_buf,