Fix the system hang issue when the app is killed contantly.

fujuntang

2021-11-10 c479ef57baaaa28964fc3ec8d80ff99dffa7d49f

 src/socket/shm_socket.cpp

@@ -2,10 +2,12 @@
#include "socket_def.h"
#include "hashtable.h"
#include "logger_factory.h"
#include "net_mod_socket_wrapper.h"
#include <map>
#include <cassert>
#include "bus_error.h"
#include "sole.h"
#include "bh_api.h"
#include "shm_mm.h"
#include "key_def.h"

@@ -68,7 +70,6 @@
  }

  queue = ( LockFreeQueue<shm_packet_t> *)tmp_ptr;
  // hashtable_unlock(hashtable);
  return queue;
}

@@ -106,9 +107,6 @@


static int _shm_socket_close_(shm_socket_t *sockt) {
  
  int rv, i;
  hashtable_t *hashtable = mm_get_hashtable();

  // if(sockt->key != 0) {
  //   auto it =  shmQueueStMap->find(sockt->key);
@@ -118,18 +116,6 @@
  //   }
  // }

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

    hashtable_remove(hashtable, sockt->key);
  //   sockt->queue = NULL;
  }
 
  pthread_mutex_destroy(&(sockt->mutex) );
  free(sockt);
  return 0;
@@ -166,20 +152,8 @@
  return 0;
}

int shm_socket_bind_proc_id(shm_socket_t *sockt, const char *buf, int len) {
  strncpy(sockt->proc_id, buf, len > MAX_STR_LEN ? MAX_STR_LEN : len);

  return 0;
}

int shm_socket_get_key(shm_socket_t *sockt){
  return sockt->key;
}

int shm_socket_get_procid(shm_socket_t *sockt, char *buf, int len) {
  strncpy(buf, sockt->proc_id, len);

  return 0;
}

// 短连接方式发送
@@ -207,7 +181,6 @@
int shm_sendandrecv(shm_socket_t *sockt, const void *send_buf,
                    const int send_size, const int key, void **recv_buf,
                    int *recv_size,  const struct timespec *timeout,  int flags) {
  // return _shm_sendandrecv_uuid(sockt, send_buf, send_size, key, recv_buf, recv_size, timeout, flags);
  return _shm_sendandrecv_thread_local(sockt, send_buf, send_size, key, recv_buf, recv_size, timeout, flags);
} 

@@ -418,14 +391,16 @@
                    const int send_size, const int key, void **recv_buf,
                    int *recv_size,  const struct timespec *timeout,  int flags) {
  
 
  int data;
  int timeout_ms;
  char data_buf[MAX_STR_LEN] = { 0x00 };
  int rv = 0, tryn = 16;
  shm_packet_t sendpak;
  shm_packet_t recvpak;
  std::map<int, shm_packet_t>::iterator recvbufIter;
  // 用thread local 保证每个线程用一个独占的socket接受对方返回的信息
  shm_socket_t *tmp_socket = NULL;
 
  hashtable_t *hashtable = mm_get_hashtable();

  rv = pthread_once(&_once_, _create_threadlocal_socket_key_);
  if (rv != 0) {
    logger->error(rv, "shm_sendandrecv pthread_once");
@@ -435,15 +410,21 @@
  tmp_socket = (shm_socket_t *)pthread_getspecific(_localthread_socket_key_);
  if (tmp_socket == NULL)
  {
    /* If first call from this thread, allocate buffer for thread, and save its location */
    tmp_socket = shm_socket_open(SHM_SOCKET_DGRAM);
    
  }
    
  rv =  pthread_setspecific(_localthread_socket_key_, tmp_socket);
  if ( rv != 0) {
    logger->error(rv, "shm_sendandrecv : pthread_setspecific");
    exit(1);
   
    tmp_socket->key = hashtable_alloc_key(hashtable);
    data = inter_key_get();
    timeout_ms = timeout->tv_sec * 1000 + 3000;
    sprintf(data_buf, "%d, %d", data, tmp_socket->key);
    if (socket_data_get() != NULL) {
      net_mod_socket_reg(socket_data_get(), data_buf, strlen(data_buf), NULL, 0, timeout_ms, PROC_REG_BUF);
    }

    rv =  pthread_setspecific(_localthread_socket_key_, tmp_socket);
    if ( rv != 0) {
      logger->error(rv, "shm_sendandrecv : pthread_setspecific");
      exit(1);
    }
  }
 
  sendpak.key = tmp_socket->key;
@@ -455,6 +436,10 @@
  rv = shm_sendpakto(tmp_socket, &sendpak, key, timeout, flags);

  if(rv != 0) {
    if(send_buf != NULL) {
      mm_free(sendpak.buf);
    }

    return rv;
  }

@@ -462,9 +447,8 @@
    tryn--;
    recvbufIter = tmp_socket->recvbuf2.find(key);
    if(recvbufIter != tmp_socket->recvbuf2.end()) {
      // 在缓存里查到了key匹配成功的
      recvpak = recvbufIter->second;
      tmp_socket->recvbuf2.erase(recvbufIter);
      tmp_socket->recvbuf2.erase(key);
      goto LABLE_SUC;
    }

@@ -476,12 +460,12 @@
    } 

    if (key == recvpak.key) {
      // 发送与接受的UUID匹配成功
      
      goto LABLE_SUC;

    } else {
      // 答非所问，放到缓存里
      
      tmp_socket->recvbuf2.insert({recvpak.key, recvpak});
      exit(0);
      continue;
    }
  }
@@ -578,6 +562,7 @@
      if (sockt->key == 0) {
        sockt->key = hashtable_alloc_key(hashtable);
      }

      sockt->queue = shm_socket_bind_queue( sockt->key, sockt->force_bind);
      if(sockt->queue  == NULL ) {
        logger->error("%s. key = %d", bus_strerror(EBUS_KEY_INUSED), sockt->key);
@@ -636,13 +621,6 @@
    sendpak->key = sockt->key;
  }
  rv = remoteQueue->push(*sendpak, timeout, flag);

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

ERR_CLOSED:
@@ -750,7 +728,7 @@
  count += strlen(ptr->int_info) + 1;
  memcpy(dst + count, ptr->svr_info, strlen(ptr->svr_info) + 1);
  count += strlen(ptr->svr_info) + 1;

  
  *counter = count;
}