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 | 277 ++++++++++++++++++++++++++++++++++++------------------- 1 files changed, 182 insertions(+), 95 deletions(-) diff --git a/src/socket/shm_socket.cpp b/src/socket/shm_socket.cpp index 2b3e626..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> @@ -10,7 +11,7 @@ static Logger *logger = LoggerFactory::getLogger(); -ShmQueueStMap * shmQueueStMap ; +// ShmQueueStMap * shmQueueStMap ; static void print_msg(char *head, shm_packet_t &msg) { // err_msg(0, "%s: key=%d, type=%d\n", head, msg.key, msg.type); @@ -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>(16); + 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; @@ -104,7 +99,7 @@ err_exit(s, "pthread_mutexattr_destroy"); - shmQueueStMap = shm_mm_attach<ShmQueueStMap>(SHM_QUEUE_ST_KEY); + // shmQueueStMap = shm_mm_attach<ShmQueueStMap>(SHM_QUEUE_ST_KEY); return sockt; } @@ -112,23 +107,29 @@ static int _shm_socket_close_(shm_socket_t *sockt) { - int rv; - logger->debug("shm_socket_close\n"); - // hashtable_remove(hashtable, mkey); - // if(sockt->queue != NULL) { - // sockt->queue = NULL; + int rv, i; + hashtable_t *hashtable = mm_get_hashtable(); + + // if(sockt->key != 0) { + // auto it = shmQueueStMap->find(sockt->key); + // if(it != shmQueueStMap->end()) { + // it->second.status = SHM_QUEUE_ST_CLOSED; + // it->second.closeTime = time(NULL); + // } // } - if(sockt->key != 0) { - auto it = shmQueueStMap->find(sockt->key); - if(it != shmQueueStMap->end()) { - it->second.status = SHM_QUEUE_ST_CLOSED; - it->second.closeTime = time(NULL); + 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; @@ -165,27 +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; } @@ -215,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; } @@ -234,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); @@ -251,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) @@ -291,8 +309,6 @@ static void _destrory_threadlocal_socket_(void *tmp_socket) { int rv; - - logger->debug("%d destroy threadlocal socket\n", pthread_self()); if(tmp_socket == NULL) return; @@ -325,7 +341,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; @@ -351,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; @@ -365,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 { @@ -422,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; @@ -449,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; @@ -458,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; } } @@ -476,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; @@ -485,7 +499,6 @@ if(recv_size != NULL) *recv_size = recvpak.size; - return 0; } @@ -515,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; } @@ -533,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) { @@ -552,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); } @@ -565,10 +587,20 @@ } // 鏍囪key瀵瑰簲鐨勭姸鎬� 锛屼负opened - stRecord.status = SHM_QUEUE_ST_OPENED; - stRecord.createTime = time(NULL); - shmQueueStMap->insert({sockt->key, stRecord}); + // 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) @@ -584,22 +616,30 @@ } // 妫�鏌ey鏍囪鐨勭姸鎬� - auto it = shmQueueStMap->find(key); - if(it != shmQueueStMap->end()) { - if(it->second.status == SHM_QUEUE_ST_CLOSED) { - // key瀵瑰簲鐨勭姸鎬佹槸鍏抽棴鐨� - goto ERR_CLOSED; - } - } + // auto it = shmQueueStMap->find(key); + // if(it != shmQueueStMap->end()) { + // if(it->second.status == SHM_QUEUE_ST_CLOSED) { + // // key瀵瑰簲鐨勭姸鎬佹槸鍏抽棴鐨� + // goto ERR_CLOSED; + // } + // } remoteQueue = shm_socket_attach_queue(key); if (remoteQueue == NULL ) { goto ERR_CLOSED; + } else if(remoteQueue->isClosed()) { + goto ERR_CLOSED; } - sendpak->key = sockt->key; + if (reset == 0) { + sendpak->key = sockt->key; + } rv = remoteQueue->push(*sendpak, timeout, flag); + + if(rv != 0) { + mm_free(sendpak->buf); + } if(rv == ETIMEDOUT) { return EBUS_TIMEOUT; } @@ -612,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) { @@ -629,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"); @@ -653,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; @@ -669,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