From 5c912c70e9333298ff48f7ea15424f72ca977b99 Mon Sep 17 00:00:00 2001 From: Fu Juntang <StrongTiger_001@163.com> Date: 星期五, 17 九月 2021 09:43:55 +0800 Subject: [PATCH] Add the heartbeat logic feature. --- src/socket/shm_socket.cpp | 222 ++++++++++++++++++++++++++++++++++++------------------ 1 files changed, 147 insertions(+), 75 deletions(-) diff --git a/src/socket/shm_socket.cpp b/src/socket/shm_socket.cpp index 8025564..6705b96 100644 --- a/src/socket/shm_socket.cpp +++ b/src/socket/shm_socket.cpp @@ -1,4 +1,5 @@ #include "shm_socket.h" +#include "socket_def.h" #include "hashtable.h" #include "logger_factory.h" #include <map> @@ -22,11 +23,12 @@ static void _destrory_threadlocal_socket_(void *tmp_socket); static void _create_threadlocal_socket_key_(void); -static int shm_recvpakfrom(shm_socket_t *sockt, shm_packet_t *recvpak , const struct timespec *timeout, int flag); +static int shm_recvpakfrom(shm_socket_t *sockt, shm_packet_t *recvpak , const struct timespec *timeout, + int flag, int reset = 0, int data_set = 0); -static int shm_sendpakto(shm_socket_t *sockt, shm_packet_t *sendpak, - const int key, const struct timespec *timeout, const int flag); +static int shm_sendpakto(shm_socket_t *sockt, shm_packet_t *sendpak, const int key, const struct timespec *timeout, + const int flag, int reset = 0, int data_set = 0); static int _shm_sendandrecv_uuid(shm_socket_t *sockt, const void *send_buf, @@ -41,21 +43,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); + queue = new LockFreeQueue<shm_packet_t>(LOCK_FREE_Q_DEFAULT_SIZE); 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 +64,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锛歝onnet at key %d failed!", key); return NULL; } @@ -80,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; @@ -114,8 +109,6 @@ int rv, i; hashtable_t *hashtable = mm_get_hashtable(); - logger->debug("shm_socket_close\n"); - // if(sockt->key != 0) { // auto it = shmQueueStMap->find(sockt->key); @@ -124,9 +117,6 @@ // it->second.closeTime = time(NULL); // } // } - - - // printf("====sockt->queue addr = %p\n", sockt->queue); if(sockt->queue != NULL) { sockt->queue->close(); @@ -176,25 +166,41 @@ 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; +} + // 鐭繛鎺ユ柟寮忓彂閫� -int shm_sendto(shm_socket_t *sockt, const void *buf, const int size, - const int key, const struct timespec *timeout, const int flag) { +int shm_sendto(shm_socket_t *sockt, const void *buf, const int size, const int key, const struct timespec *timeout, + const int flag, int reset, int data_set) { int rv; shm_packet_t sendpak = {0}; - sendpak.key = sockt->key; + if (reset == 0) { + sendpak.key = sockt->key; + } else { + sendpak.key = data_set; + } sendpak.size = size; if(buf != NULL) { sendpak.buf = mm_malloc(size); memcpy(sendpak.buf, buf, size); } - rv = shm_sendpakto(sockt, &sendpak, key, timeout, flag); + rv = shm_sendpakto(sockt, &sendpak, key, timeout, flag, reset, data_set); return rv; } @@ -224,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; } @@ -243,7 +249,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); @@ -260,24 +266,27 @@ } // 鐭繛鎺ユ柟寮忔帴鍙� -int shm_recvfrom(shm_socket_t *sockt, void **buf, int *size, int *key, const struct timespec *timeout, int flag) { +int shm_recvfrom(shm_socket_t *sockt, void **buf, int *size, int *key, const struct timespec *timeout, int flag, int reset, int data_set) { int rv; shm_packet_t recvpak; - rv = shm_recvpakfrom(sockt , &recvpak, timeout, flag); + rv = shm_recvpakfrom(sockt , &recvpak, timeout, flag, reset, data_set); 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; } - - 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) @@ -300,8 +309,6 @@ static void _destrory_threadlocal_socket_(void *tmp_socket) { int rv; - - logger->debug("%lu destroy threadlocal socket\n", pthread_self()); if(tmp_socket == NULL) return; @@ -360,7 +367,6 @@ recvbufIter = sockt->recvbuf.find(uuid); if(recvbufIter != sockt->recvbuf.end()) { // 鍦ㄧ紦瀛橀噷鏌ュ埌浜哢UID鍖归厤鎴愬姛鐨� -logger->debug("get from recvbuf: %s", uuid.c_str()); recvpak = recvbufIter->second; sockt->recvbuf.erase(recvbufIter); goto LABLE_SUC; @@ -374,14 +380,13 @@ 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) { + } else if (strncmp(uuid.c_str(), recvpak.uuid, sizeof(recvpak.uuid)) == 0) { // 鍙戦�佷笌鎺ュ彈鐨刄UID鍖归厤鎴愬姛 goto LABLE_SUC; } else { @@ -431,14 +436,14 @@ 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); - if ( rv != 0) { - logger->error(rv, "shm_sendandrecv : pthread_setspecific"); - exit(1); - } + + } + + 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; @@ -458,7 +463,6 @@ recvbufIter = tmp_socket->recvbuf2.find(key); if(recvbufIter != tmp_socket->recvbuf2.end()) { // 鍦ㄧ紦瀛橀噷鏌ュ埌浜唊ey鍖归厤鎴愬姛鐨� - // logger->info("get from recvbuf: %d", key); recvpak = recvbufIter->second; tmp_socket->recvbuf2.erase(recvbufIter); goto LABLE_SUC; @@ -467,16 +471,17 @@ 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; } - if (key == recvpak.key) { + if (key == recvpak.key) { // 鍙戦�佷笌鎺ュ彈鐨刄UID鍖归厤鎴愬姛 goto LABLE_SUC; } else { // 绛旈潪鎵�闂紝鏀惧埌缂撳瓨閲� tmp_socket->recvbuf2.insert({recvpak.key, recvpak}); + exit(0); continue; } } @@ -485,7 +490,7 @@ return EBUS_RECVFROM_WRONG_END; LABLE_SUC: - if(recv_buf != NULL) { + if(recv_buf != NULL) { void *_buf = malloc(recvpak.size); memcpy(_buf, recvpak.buf, recvpak.size); *recv_buf = _buf; @@ -494,7 +499,6 @@ if(recv_size != NULL) *recv_size = recvpak.size; - return 0; } @@ -524,8 +528,8 @@ // 瓒呮椂瀵艰嚧鎺ュ彂閫佸璞★紝涓庤繑鍥炲璞′笉瀵瑰簲鐨勬儏鍐� 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); + + logger->error( "_shm_sendandrecv_alloc_new: send key expect to equal to recv key! send key =%d , recv key=%d", send_key, recv_key); continue; } @@ -542,15 +546,24 @@ -static int shm_sendpakto(shm_socket_t *sockt, shm_packet_t *sendpak, - const int key, const struct timespec *timeout, const int flag) { +static int shm_sendpakto(shm_socket_t *sockt, shm_packet_t *sendpak, const int key, const struct timespec *timeout, + const int flag, int reset, int data_set) { int rv; shm_queue_status_t stRecord; LockFreeQueue<shm_packet_t> *remoteQueue; + LockFreeQueue<shm_packet_t> *fixedQueue; hashtable_t *hashtable = mm_get_hashtable(); - if( sockt->queue != NULL) + if ((reset != 0) && (data_set == 0)) { + return EBUS_KEY_INUSED; + } + + if (reset != 0) { + fixedQueue = shm_socket_attach_queue(data_set); + } + + if (((reset == 0) && (sockt->queue != NULL)) || ((reset != 0) && (fixedQueue != NULL))) goto LABEL_PUSH; // if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) { @@ -561,7 +574,7 @@ if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0) err_exit(rv, "shm_sendto : pthread_mutex_lock"); - if (sockt->queue == NULL) { + if ((sockt->queue == NULL) && (reset == 0)) { if (sockt->key == 0) { sockt->key = hashtable_alloc_key(hashtable); } @@ -578,6 +591,16 @@ // stRecord.createTime = time(NULL); // shmQueueStMap->insert({sockt->key, stRecord}); + } + + if ((fixedQueue == NULL) && (reset != 0)) { + fixedQueue = shm_socket_bind_queue(data_set, false); + if (fixedQueue == NULL ) { + logger->error("%s. key = %d", bus_strerror(EBUS_KEY_INUSED), sockt->key); + if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) + err_exit(rv, "shm_sendto : pthread_mutex_unlock"); + return EBUS_KEY_INUSED; + } } if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) @@ -609,7 +632,9 @@ goto ERR_CLOSED; } - sendpak->key = sockt->key; + if (reset == 0) { + sendpak->key = sockt->key; + } rv = remoteQueue->push(*sendpak, timeout, flag); if(rv != 0) { @@ -627,13 +652,23 @@ } // 鐭繛鎺ユ柟寮忔帴鍙� -static int shm_recvpakfrom(shm_socket_t *sockt, shm_packet_t *_recvpak , const struct timespec *timeout, int flag) { +static int shm_recvpakfrom(shm_socket_t *sockt, shm_packet_t *_recvpak , const struct timespec *timeout, + int flag, int reset, int data_set) { int rv; shm_queue_status_t stRecord; + LockFreeQueue<shm_packet_t> *fixedQueue; hashtable_t *hashtable = mm_get_hashtable(); shm_packet_t recvpak; - if( sockt->queue != NULL) + if ((reset != 0) && (data_set == 0)) { + return EBUS_KEY_INUSED; + } + + if (reset != 0) { + fixedQueue = shm_socket_attach_queue(data_set); + } + + if (((sockt->queue != NULL) && (reset == 0)) || ((reset != 0) && (fixedQueue != NULL))) goto LABEL_POP; // if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) { @@ -644,22 +679,33 @@ if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0) err_exit(rv, "shm_recvfrom : pthread_mutex_lock"); - - 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); - if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) - err_exit(rv, "shm_recvfrom : pthread_mutex_unlock"); - return EBUS_KEY_INUSED; + if ((sockt->queue == NULL) && (reset == 0)) { + 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); + if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) + err_exit(rv, "shm_recvfrom : pthread_mutex_unlock"); + return EBUS_KEY_INUSED; + } + + // 鏍囪key瀵瑰簲鐨勭姸鎬� 锛屼负opened + // stRecord.status = SHM_QUEUE_ST_OPENED; + // stRecord.createTime = time(NULL); + // shmQueueStMap->insert({sockt->key, stRecord}); } - - // 鏍囪key瀵瑰簲鐨勭姸鎬� 锛屼负opened - // stRecord.status = SHM_QUEUE_ST_OPENED; - // stRecord.createTime = time(NULL); - // shmQueueStMap->insert({sockt->key, stRecord}); + + if ((fixedQueue == NULL) && (reset != 0)) { + fixedQueue = shm_socket_bind_queue(data_set, false); + if (fixedQueue == NULL ) { + logger->error("%s. key = %d", bus_strerror(EBUS_KEY_INUSED), sockt->key); + if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) + err_exit(rv, "shm_sendto : pthread_mutex_unlock"); + return EBUS_KEY_INUSED; + } + } if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) err_exit(rv, "shm_recvfrom : pthread_mutex_unlock"); @@ -668,8 +714,11 @@ LABEL_POP: - - rv = sockt->queue->pop(recvpak, timeout, flag); + if (reset == 0) { + rv = sockt->queue->pop(recvpak, timeout, flag); + } else { + rv = fixedQueue->pop(recvpak, timeout, flag); + } if(rv != 0) { if(rv == ETIMEDOUT) { return EBUS_TIMEOUT; @@ -684,4 +733,27 @@ *_recvpak = recvpak; return 0; } - + +void proc_copy(char *dst, void *src, int *counter) { + int count = 0; + ProcInfo *ptr = static_cast<ProcInfo *>(src); + + memcpy(dst, ptr->proc_id, strlen(ptr->proc_id) + 1); + count = strlen(ptr->proc_id) + 1; + memcpy(dst + count, ptr->name, strlen(ptr->name) + 1); + count += strlen(ptr->name) + 1; + memcpy(dst + count, ptr->public_info, strlen(ptr->public_info) + 1); + count += strlen(ptr->public_info) + 1; + memcpy(dst + count, ptr->private_info, strlen(ptr->private_info) + 1); + count += strlen(ptr->private_info) + 1; + memcpy(dst + count, ptr->int_info, strlen(ptr->int_info) + 1); + 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; +} + + + + -- Gitblit v1.8.0