fjt fix req-rep error

looper3339

2021-08-09 567e7bba26bb8fe485be6c2ddfaaf8ab77eb6141

 src/socket/shm_socket.cpp

@@ -41,21 +41,16 @@
static LockFreeQueue<shm_packet_t> * shm_socket_bind_queue(int key, bool force) {
  hashtable_t *hashtable = mm_get_hashtable();
  LockFreeQueue<shm_packet_t> *queue;
  hashtable_lock(hashtable);
  void *tmp_ptr = hashtable_get(hashtable, key);


  if (tmp_ptr == NULL || tmp_ptr == (void *)1  ) {
    queue = new LockFreeQueue<shm_packet_t>(32);
    hashtable_put(hashtable, key, (void *)queue);
    hashtable_unlock(hashtable);
    return queue;
  } else if(force) {
    hashtable_unlock(hashtable);
    return (LockFreeQueue<shm_packet_t> *) tmp_ptr;
  }
  
  hashtable_unlock(hashtable);
  return NULL;
}

@@ -67,7 +62,6 @@
  hashtable_t *hashtable = mm_get_hashtable();
  void *tmp_ptr = hashtable_get(hashtable, key);
  if (tmp_ptr == NULL || tmp_ptr == (void *)1) {
    //logger->error("shm_socket._remote_queue_attach：connet at key %d  failed!", key);
    return NULL;
  }

@@ -126,7 +120,6 @@
  // }


  // printf("====sockt->queue addr = %p\n", sockt->queue);

  if(sockt->queue != NULL) {
    sockt->queue->close();
@@ -243,7 +236,7 @@
  shm_packet_t sendpak = {0};
  sendpak.key = sockt->key;
  sendpak.size = sendsize;
  memcpy(sendpak.uuid, recvpak.uuid, sizeof sendpak.uuid);
  memcpy(sendpak.uuid, recvpak.uuid, sizeof(sendpak.uuid));
  if(sendbuf !=NULL && sendsize > 0) {
    sendpak.buf = mm_malloc(sendsize);
    memcpy(sendpak.buf, sendbuf, sendsize);
@@ -273,11 +266,14 @@
   
  } 


 if(buf != NULL && recvpak.buf != NULL) {
    void *_buf = malloc(recvpak.size);
    memcpy(_buf, recvpak.buf, recvpak.size);
    *buf = _buf; 
  if(recvpak.buf != NULL) {
    if (buf == NULL) {
      logger->warn("!!!Alert: buf should be not NULL!\n"); 
    } else {
      void *_buf = malloc(recvpak.size);
      memcpy(_buf, recvpak.buf, recvpak.size);
      *buf = _buf; 
    }
  }
 
  if(size != NULL)
@@ -381,7 +377,7 @@
logger->debug("send uuid:%s, recv uuid: %s", uuid.c_str(), recvpak.uuid);
    if(strlen(recvpak.uuid) == 0) {
      continue;
    } else if (strncmp(uuid.c_str(), recvpak.uuid, sizeof recvpak.uuid) == 0) {
    } else if (strncmp(uuid.c_str(), recvpak.uuid, sizeof(recvpak.uuid)) == 0) {
      // 发送与接受的UUID匹配成功
      goto LABLE_SUC;
    } else {
@@ -415,6 +411,8 @@
  
 
  int rv = 0, tryn = 16;
  static int Counter_suc = 0;
  static int Counter_fail = 0;
  shm_packet_t sendpak;
  shm_packet_t recvpak;
  std::map<int, shm_packet_t>::iterator recvbufIter;
@@ -431,7 +429,6 @@
  if (tmp_socket == NULL)
  {
    /* If first call from this thread, allocate buffer for thread, and save its location */
    logger->debug("%lu create threadlocal socket\n", (long)pthread_self() );
    tmp_socket = shm_socket_open(SHM_SOCKET_DGRAM);

    rv =  pthread_setspecific(_localthread_socket_key_, tmp_socket);
@@ -458,7 +455,6 @@
    recvbufIter = tmp_socket->recvbuf2.find(key);
    if(recvbufIter != tmp_socket->recvbuf2.end()) {
      // 在缓存里查到了key匹配成功的
 // logger->info("get from recvbuf: %d", key);
      recvpak = recvbufIter->second;
      tmp_socket->recvbuf2.erase(recvbufIter);
      goto LABLE_SUC;
@@ -471,7 +467,7 @@
      return rv;
    } 

   if (key == recvpak.key) {
    if (key == recvpak.key) {
      // 发送与接受的UUID匹配成功
      goto LABLE_SUC;
    } else {
@@ -485,7 +481,8 @@
  return EBUS_RECVFROM_WRONG_END;
 
LABLE_SUC:
 if(recv_buf != NULL) {
  sockt->key = tmp_socket->key;
  if(recv_buf != NULL) {
    void *_buf = malloc(recvpak.size);
    memcpy(_buf, recvpak.buf, recvpak.size);
    *recv_buf = _buf; 
@@ -494,7 +491,6 @@
 
  if(recv_size != NULL)
    *recv_size = recvpak.size;

  
  return 0;
}