Add the heartbeat logic feature.

Fu Juntang

2021-09-17 5c912c70e9333298ff48f7ea15424f72ca977b99

 src/socket/shm_socket.cpp

@@ -46,7 +46,7 @@
  void *tmp_ptr = hashtable_get(hashtable, key);

  if (tmp_ptr == NULL || tmp_ptr == (void *)1  ) {
    queue = new LockFreeQueue<shm_packet_t>(32);
    queue = new LockFreeQueue<shm_packet_t>(LOCK_FREE_Q_DEFAULT_SIZE);
    hashtable_put(hashtable, key, (void *)queue);
    return queue;
  } else if(force) {
@@ -76,7 +76,6 @@
  int s, type;
  pthread_mutexattr_t mtxAttr;

  logger->debug("shm_socket_open\n");
  // shm_socket_t *socket = (shm_socket_t *)calloc(1, sizeof(shm_socket_t));
  shm_socket_t *sockt = new shm_socket_t;
  sockt->socket_type = socket_type;
@@ -231,7 +230,7 @@

  if (rv != 0) {
    if(rv == ETIMEDOUT){
      logger->debug("%d shm_recvandsend failed %s", shm_socket_get_key(sockt), bus_strerror(EBUS_TIMEOUT));
      logger->error("%d shm_recvandsend failed %s", shm_socket_get_key(sockt), bus_strerror(EBUS_TIMEOUT));
      return EBUS_TIMEOUT;
    }
    
@@ -275,7 +274,7 @@

  if (rv != 0) {

    logger->debug("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
    logger->error("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
    return rv;
   
  } 
@@ -368,7 +367,6 @@
    recvbufIter = sockt->recvbuf.find(uuid);
    if(recvbufIter != sockt->recvbuf.end()) {
      // 在缓存里查到了UUID匹配成功的
logger->debug("get from recvbuf: %s", uuid.c_str());
      recvpak = recvbufIter->second;
      sockt->recvbuf.erase(recvbufIter);
      goto LABLE_SUC;
@@ -382,11 +380,10 @@
        return EBUS_TIMEOUT;
      }

      logger->debug("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
      logger->error("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
      return rv;
    } 

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) {
@@ -474,7 +471,7 @@
    rv = shm_recvpakfrom(tmp_socket, &recvpak, timeout, flags);

    if (rv != 0) {
      logger->debug("%d shm_recvfrom failed %s", shm_socket_get_key(tmp_socket), bus_strerror(rv));
      logger->error("%d shm_recvfrom failed %s", shm_socket_get_key(tmp_socket), bus_strerror(rv));
      return rv;
    }