From 68d23225a38a35f1325eb39fa4ed5a005d5de473 Mon Sep 17 00:00:00 2001 From: fujuntang <fujuntang@aiot.com> Date: 星期三, 11 八月 2021 09:50:20 +0800 Subject: [PATCH] fix from 3.1 first commit --- src/socket/shm_socket.cpp | 340 +++++++++++++++++++++++++------------------------------- 1 files changed, 154 insertions(+), 186 deletions(-) diff --git a/src/socket/shm_socket.cpp b/src/socket/shm_socket.cpp index f55a111..978eda9 100644 --- a/src/socket/shm_socket.cpp +++ b/src/socket/shm_socket.cpp @@ -5,29 +5,31 @@ #include <cassert> #include "bus_error.h" #include "sole.h" +#include "shm_mm.h" +#include "key_def.h" static Logger *logger = LoggerFactory::getLogger(); - +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); } static pthread_once_t _once_ = PTHREAD_ONCE_INIT; -static pthread_key_t _perthread_socket_key_; +static pthread_key_t _localthread_socket_key_; -static void _destrory_socket_perthread(void *tmp_socket); -static void _create_socket_key_perthread(void); +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_sendpakto(shm_socket_t *sockt, const shm_packet_t *sendpak, +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_sendandrecv_use_uuid(shm_socket_t *sockt, const void *send_buf, +static int _shm_sendandrecv_uuid(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); @@ -39,21 +41,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>(32); 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); return NULL; } @@ -65,7 +62,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; } @@ -74,58 +70,11 @@ return queue; } -//鍒犻櫎鍖呭惈鍦╧eys鍐呯殑queue -size_t shm_socket_remove_keys(int keys[], size_t length) { - hashtable_t *hashtable = mm_get_hashtable(); - LockFreeQueue<shm_packet_t> *mqueue; - size_t count = 0; - for(int i = 0; i< length; i++) { - // 閿�姣佸叡浜唴瀛樼殑queue - mqueue = (LockFreeQueue<shm_packet_t> *)hashtable_get(hashtable, keys[i]); - delete mqueue; - hashtable_remove(hashtable, keys[i]); - count++; - } - return count; -} - - -// 鍒犻櫎涓嶅湪keys鍐呯殑queue -size_t shm_socket_remove_keys_exclude(int keys[], size_t length) { - hashtable_t *hashtable = mm_get_hashtable(); - std::set<int> *keyset = hashtable_keyset(hashtable); - std::set<int>::iterator keyItr; - LockFreeQueue<shm_packet_t> *mqueue; - bool found; - size_t count = 0; - for (keyItr = keyset->begin(); keyItr != keyset->end(); keyItr++) { - found = false; - for (size_t i = 0; i < length; i++) { - if (*keyItr == keys[i]) { - found = true; - break; - } - } - // 100鍐呯殑鏄痓us鍐呴儴鑷繁鐢ㄧ殑 - if (!found && *keyItr > 100) { - // 閿�姣佸叡浜唴瀛樼殑queue - mqueue = (LockFreeQueue<shm_packet_t> *)hashtable_get(hashtable, *keyItr); - delete mqueue; - hashtable_remove(hashtable, *keyItr); - count++; - } - } - delete keyset; - return count; -} - - - -shm_socket_t *shm_open_socket(shm_socket_type_t socket_type) { +shm_socket_t *shm_socket_open(shm_socket_type_t socket_type) { int s, type; pthread_mutexattr_t mtxAttr; - logger->debug("shm_open_socket\n"); + 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; @@ -148,25 +97,55 @@ if (s != 0) err_exit(s, "pthread_mutexattr_destroy"); + + shmQueueStMap = shm_mm_attach<ShmQueueStMap>(SHM_QUEUE_ST_KEY); + return sockt; } -int shm_close_socket(shm_socket_t *sockt) { + +static int _shm_socket_close_(shm_socket_t *sockt) { - int rv; - logger->debug("shm_close_socket\n"); + 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); + // if(it != shmQueueStMap->end()) { + // it->second.status = SHM_QUEUE_ST_CLOSED; + // it->second.closeTime = time(NULL); + // } + // } + + + if(sockt->queue != NULL) { - delete sockt->queue; - 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); - rv = pthread_mutex_destroy(&(sockt->mutex) ); - if(rv != 0) { - err_exit(rv, "shm_close_socket"); + hashtable_remove(hashtable, sockt->key); + // sockt->queue = NULL; } - + + pthread_mutex_destroy(&(sockt->mutex) ); free(sockt); return 0; +} + + +int shm_socket_close(shm_socket_t *sockt) { + + shm_socket_t * threadlocal_socket = (shm_socket_t *)pthread_getspecific(_localthread_socket_key_); + if(threadlocal_socket != NULL) { + _destrory_threadlocal_socket_(threadlocal_socket); + } + return _shm_socket_close_(sockt); } @@ -194,8 +173,6 @@ return sockt->key; } - - // 鐭繛鎺ユ柟寮忓彂閫� int shm_sendto(shm_socket_t *sockt, const void *buf, const int size, const int key, const struct timespec *timeout, const int flag) { @@ -217,6 +194,7 @@ 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); } @@ -258,7 +236,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); @@ -283,20 +261,19 @@ if (rv != 0) { - if(rv == ETIMEDOUT) - return EBUS_TIMEOUT; - else { - logger->debug("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(rv)); - return rv; - } + logger->debug("%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) @@ -305,6 +282,9 @@ if(key != NULL) *key = recvpak.key; + if(recvpak.key == 0) { + err_exit(0, "key = %d, pid= %d, recvpak.key == 0", shm_socket_get_key(sockt), getpid()); + } mm_free(recvpak.buf); return 0; } @@ -313,15 +293,17 @@ // ================================================================================================= /* Free thread-specific data buffer */ -static void _destrory_socket_perthread(void *tmp_socket) +static void _destrory_threadlocal_socket_(void *tmp_socket) { int rv; + + logger->debug("%lu destroy threadlocal socket\n", pthread_self()); + if(tmp_socket == NULL) return; - - logger->debug("%d destroy tmp socket\n", pthread_self()); - shm_close_socket((shm_socket_t *)tmp_socket); - rv = pthread_setspecific(_perthread_socket_key_, NULL); + + _shm_socket_close_((shm_socket_t *)tmp_socket); + rv = pthread_setspecific(_localthread_socket_key_, NULL); if ( rv != 0) { logger->error(rv, "shm_sendandrecv : pthread_setspecific"); exit(1); @@ -329,14 +311,14 @@ } /* One-time key creation function */ -static void _create_socket_key_perthread(void) +static void _create_threadlocal_socket_key_(void) { int s; /* Allocate a unique thread-specific data key and save the address of the destructor for thread-specific data buffers */ - s = pthread_key_create(&_perthread_socket_key_, _destrory_socket_perthread); - //s = pthread_key_create(&_perthread_socket_key_, NULL); + s = pthread_key_create(&_localthread_socket_key_, _destrory_threadlocal_socket_); + //s = pthread_key_create(&_localthread_socket_key_, NULL); if (s != 0) { logger->error(s, "pthread_key_create"); exit(1); @@ -344,11 +326,11 @@ } -static int _shm_sendandrecv_use_uuid(shm_socket_t *sockt, const void *send_buf, +static int _shm_sendandrecv_uuid(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) { - int rv, tryn = 3; + int rv, tryn = 16; shm_packet_t sendpak; shm_packet_t recvpak; std::map<std::string, shm_packet_t>::iterator recvbufIter; @@ -395,7 +377,7 @@ 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 { @@ -407,24 +389,19 @@ LABLE_FAIL: return EBUS_RECVFROM_WRONG_END; - // return rv; LABLE_SUC: if(recv_buf != NULL) { void *_buf = malloc(recvpak.size); memcpy(_buf, recvpak.buf, recvpak.size); *recv_buf = _buf; + mm_free(recvpak.buf); } if(recv_size != NULL) *recv_size = recvpak.size; - mm_free(recvpak.buf); - return 0; - - - } // use thread local @@ -433,66 +410,34 @@ int *recv_size, const struct timespec *timeout, int flags) { - int rv, tryn = 3; + int rv = 0, tryn = 16; + static int Counter_suc = 0; + static int Counter_fail = 0; shm_packet_t sendpak; shm_packet_t recvpak; std::map<int, shm_packet_t>::iterator recvbufIter; // 鐢╰hread local 淇濊瘉姣忎釜绾跨▼鐢ㄤ竴涓嫭鍗犵殑socket鎺ュ彈瀵规柟杩斿洖鐨勪俊鎭� - shm_socket_t *tmp_socket; + shm_socket_t *tmp_socket = NULL; - /* If first call from this thread, allocate buffer for thread, and save its location */ - // logger->debug("%d create tmp socket\n", pthread_self() ); - rv = pthread_once(&_once_, _create_socket_key_perthread); + rv = pthread_once(&_once_, _create_threadlocal_socket_key_); if (rv != 0) { logger->error(rv, "shm_sendandrecv pthread_once"); exit(1); } - tmp_socket = (shm_socket_t *)pthread_getspecific(_perthread_socket_key_); + 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 */ - logger->debug("%ld create tmp socket\n", (long)pthread_self() ); - tmp_socket = shm_open_socket(SHM_SOCKET_DGRAM); + tmp_socket = shm_socket_open(SHM_SOCKET_DGRAM); - rv = pthread_setspecific(_perthread_socket_key_, tmp_socket); + rv = pthread_setspecific(_localthread_socket_key_, tmp_socket); if ( rv != 0) { logger->error(rv, "shm_sendandrecv : pthread_setspecific"); exit(1); } } - // int rv; - // int tryn = 0; - // int recv_key; - // rv = shm_sendto(tmp_socket, send_buf, send_size, send_key, timeout, flags); - - // if (() == 0) { - - // while(tryn < 3) { - // tryn++; - // rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags); - // if(rv != 0) { - // logger->error("_shm_sendandrecv_thread_local : %s\n", bus_strerror(rv)); - // return rv; - // } - - // // 瓒呮椂瀵艰嚧鎺ュ彂閫佸璞★紝涓庤繑鍥炲璞′笉瀵瑰簲鐨勬儏鍐� - // 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); - // // exit(1); - // continue; - // // return EBUS_RECVFROM_WRONG_END; - // } - - // return 0; - // } - - // return EBUS_RECVFROM_WRONG_END; - // } - - sendpak.key = tmp_socket->key; sendpak.size = send_size; if(send_buf != NULL) { @@ -505,34 +450,24 @@ return rv; } - if(rv != 0) { - return rv; - } - while(tryn > 0) { tryn--; recvbufIter = tmp_socket->recvbuf2.find(key); if(recvbufIter != tmp_socket->recvbuf2.end()) { - // 鍦ㄧ紦瀛橀噷鏌ュ埌浜哢UID鍖归厤鎴愬姛鐨� -// logger->debug("get from recvbuf: %s", uuid.c_str()); + // 鍦ㄧ紦瀛橀噷鏌ュ埌浜唊ey鍖归厤鎴愬姛鐨� recvpak = recvbufIter->second; - sockt->recvbuf2.erase(recvbufIter); + tmp_socket->recvbuf2.erase(recvbufIter); goto LABLE_SUC; } rv = shm_recvpakfrom(tmp_socket, &recvpak, timeout, flags); if (rv != 0) { - - if(rv == ETIMEDOUT) { - return EBUS_TIMEOUT; - } - logger->debug("%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 { @@ -550,12 +485,13 @@ void *_buf = malloc(recvpak.size); memcpy(_buf, recvpak.buf, recvpak.size); *recv_buf = _buf; + mm_free(recvpak.buf); } if(recv_size != NULL) *recv_size = recvpak.size; - mm_free(recvpak.buf); + return 0; } @@ -569,7 +505,7 @@ shm_socket_t *tmp_socket; - tmp_socket = shm_open_socket(SHM_SOCKET_DGRAM); + tmp_socket = shm_socket_open(SHM_SOCKET_DGRAM); if ((rv = shm_sendto(tmp_socket, send_buf, send_size, send_key, timeout, flags)) == 0) { rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags); @@ -578,7 +514,7 @@ tryn++; rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags); if(rv != 0) { - logger->error("_shm_sendandrecv_thread_local : %s\n", bus_strerror(rv)); + logger->error("_shm_sendandrecv_alloc_new : %s\n", bus_strerror(rv)); return rv; } @@ -595,25 +531,27 @@ return EBUS_RECVFROM_WRONG_END; } - shm_close_socket(tmp_socket); + _shm_socket_close_(tmp_socket); return rv; } -static int shm_sendpakto(shm_socket_t *sockt, const shm_packet_t *sendpak, +static int shm_sendpakto(shm_socket_t *sockt, shm_packet_t *sendpak, const int key, const struct timespec *timeout, const int flag) { int rv; + shm_queue_status_t stRecord; + LockFreeQueue<shm_packet_t> *remoteQueue; hashtable_t *hashtable = mm_get_hashtable(); if( sockt->queue != NULL) goto LABEL_PUSH; - if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) { - return EBUS_EXCEED_LIMIT; - } + // if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) { + // return EBUS_EXCEED_LIMIT; + // } { if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0) @@ -626,8 +564,16 @@ 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_sendto : pthread_mutex_unlock"); return EBUS_KEY_INUSED; } + + // 鏍囪key瀵瑰簲鐨勭姸鎬� 锛屼负opened + stRecord.status = SHM_QUEUE_ST_OPENED; + stRecord.createTime = time(NULL); + shmQueueStMap->insert({sockt->key, stRecord}); + } if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) @@ -642,33 +588,53 @@ return EBUS_SENDTO_SELF; } - LockFreeQueue<shm_packet_t> *remoteQueue; - if ((remoteQueue = shm_socket_attach_queue(key)) == NULL) { - bus_errno = EBUS_CLOSED; - logger->error("sendto key %d failed, %s", key, bus_strerror(bus_errno)); - return EBUS_CLOSED; + // 妫�鏌ey鏍囪鐨勭姸鎬� + 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; rv = remoteQueue->push(*sendpak, timeout, flag); + if(rv != 0) { + mm_free(sendpak->buf); + } + if(rv == ETIMEDOUT) { + return EBUS_TIMEOUT; + } return rv; + +ERR_CLOSED: + logger->error("sendto key %d failed, %s", key, bus_strerror(EBUS_CLOSED)); + return EBUS_CLOSED; + } // 鐭繛鎺ユ柟寮忔帴鍙� static int shm_recvpakfrom(shm_socket_t *sockt, shm_packet_t *_recvpak , const struct timespec *timeout, int flag) { int rv; - + shm_queue_status_t stRecord; hashtable_t *hashtable = mm_get_hashtable(); shm_packet_t recvpak; if( sockt->queue != NULL) goto LABEL_POP; - if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) { - return EBUS_EXCEED_LIMIT; - } + // if(hashtable_get_queue_count(hashtable) > QUEUE_COUNT_LIMIT) { + // return EBUS_EXCEED_LIMIT; + // } { if ((rv = pthread_mutex_lock(&(sockt->mutex))) != 0) @@ -681,9 +647,15 @@ 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}); if ((rv = pthread_mutex_unlock(&(sockt->mutex))) != 0) err_exit(rv, "shm_recvfrom : pthread_mutex_unlock"); @@ -692,24 +664,20 @@ LABEL_POP: - // - // printf("%p start recv.....\n", sockt); rv = sockt->queue->pop(recvpak, timeout, flag); - if(rv != 0) + if(rv != 0) { + if(rv == ETIMEDOUT) { + return EBUS_TIMEOUT; + } return rv; + } if(recvpak.action == BUS_ACTION_STOP) { return EBUS_STOPED; } *_recvpak = recvpak; - return rv; + return 0; } -// int shm_sendandrecv(shm_socket_t *sockt, const void *send_buf, -// const int send_size, const int send_key, void **recv_buf, -// int *recv_size, const struct timespec *timeout, int flags) { - -// struct timespec tm = {10, 0}; -// return _shm_sendandrecv_thread_local(sockt, send_buf, send_size, send_key,recv_buf, recv_size, &tm, flags); -// } + -- Gitblit v1.8.0