update

wangzhengquan

2021-03-13 a38304f7f6b91aaa1b0aa76cc9d3e5b6aef1f85f

 src/socket/shm_socket.cpp

@@ -46,13 +46,13 @@


  if (tmp_ptr == NULL || tmp_ptr == (void *)1  ) {
    queue = new LockFreeQueue<shm_packet_t>(16);
    queue = new LockFreeQueue<shm_packet_t>(1024);
    hashtable_put(hashtable, key, (void *)queue);
    hashtable_unlock(hashtable);
    return queue;
  } else if(force) {
    hashtable_unlock(hashtable);
    return (LockFreeQueue<shm_packet_t> *) queue;
    return (LockFreeQueue<shm_packet_t> *) tmp_ptr;
  }
  
  hashtable_unlock(hashtable);
@@ -112,12 +112,9 @@

static int _shm_socket_close_(shm_socket_t *sockt) {
  
  int rv;
  int rv, i;
  logger->debug("shm_socket_close\n");
  // hashtable_remove(hashtable, mkey);
  // if(sockt->queue != NULL) {
  //   sockt->queue = NULL;
  // }
 

  if(sockt->key != 0) {
    auto it =  shmQueueStMap->find(sockt->key);
@@ -127,6 +124,24 @@
    }
  }


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

  if(sockt->queue != NULL) {
    for( i = 0; i < sockt->queue->size(); i++) {
      mm_free((*(sockt->queue))[i].buf);
      logger->info("======= %d free queue element buf\n", sockt->key);
    }

    // hashtable_remove(hashtable, mkey);
  //   sockt->queue = NULL;
  }
 

  // hashtable_remove(hashtable, mkey);
  // if(sockt->queue != NULL) {
  //   sockt->queue = NULL;
  // }
 

  pthread_mutex_destroy(&(sockt->mutex) );
@@ -168,8 +183,6 @@
int shm_socket_get_key(shm_socket_t *sockt){
  return sockt->key;
}



// 短连接方式发送
int shm_sendto(shm_socket_t *sockt, const void *buf, const int size,
@@ -292,7 +305,7 @@
{
  int rv;

  logger->debug("%d destroy threadlocal socket\n", pthread_self()); 
  logger->debug("%lu destroy threadlocal socket\n", pthread_self()); 

  if(tmp_socket == NULL)
    return;
@@ -325,7 +338,7 @@
                    const int send_size, const int key, void **recv_buf,
                    int *recv_size,  const struct timespec *timeout,  int flags) {

  int rv, tryn = 6;
  int rv, tryn = 16;
  shm_packet_t sendpak;
  shm_packet_t recvpak;
  std::map<std::string, shm_packet_t>::iterator recvbufIter;
@@ -600,6 +613,10 @@

  sendpak->key = sockt->key;
  rv = remoteQueue->push(*sendpak, timeout, flag);

  if(rv != 0) {
    mm_free(sendpak->buf);
  }
  if(rv == ETIMEDOUT) {
    return EBUS_TIMEOUT;
  }