From aa2f3b2a9968bb4928463bdae05fb026d16b60bb Mon Sep 17 00:00:00 2001
From: wangzhengquan <wangzhengquan85@126.com>
Date: 星期五, 04 十二月 2020 19:07:01 +0800
Subject: [PATCH] 固定bus key
---
src/socket/shm_socket.c | 902 ++++++++++++++++++++++++++++++++++++-------------------
1 files changed, 588 insertions(+), 314 deletions(-)
diff --git a/src/socket/shm_socket.c b/src/socket/shm_socket.c
index 4fb90cf..0ac71cb 100644
--- a/src/socket/shm_socket.c
+++ b/src/socket/shm_socket.c
@@ -1,380 +1,654 @@
#include "shm_socket.h"
+#include "hashtable.h"
#include "logger_factory.h"
#include <map>
+static Logger *logger = LoggerFactory::getLogger();
-static Logger logger = LoggerFactory::getLogger();
-void print_msg(char *head, shm_msg_t& msg) {
- //err_msg(0, "%s: port=%d, type=%d\n", head, msg.port, msg.type);
+void print_msg(char *head, shm_msg_t &msg) {
+ // err_msg(0, "%s: key=%d, type=%d\n", head, msg.key, msg.type);
}
-void * _server_run_msg_rev(void* _socket);
+static void *_server_run_msg_rev(void *_socket);
-void * _client_run_msg_rev(void* _socket);
+static void *_client_run_msg_rev(void *_socket);
-SHMQueue<shm_msg_t> * _attach_remote_queue(int port) ;
+static int _shm_close_dgram_socket(shm_socket_t *socket);
-shm_socket_t *shm_open_socket() {
- shm_socket_t *socket = (shm_socket_t *)calloc(1, sizeof(shm_socket_t));
-
- socket->port = -1;
- socket->dispatch_thread = 0;
- socket->status=SHM_CONN_CLOSED;
-
- return socket;
+static int _shm_close_stream_socket(shm_socket_t *socket, bool notifyRemote);
+
+// 妫�鏌ey鏄惁宸茬粡琚娇鐢紝鏄繑鍥�0, 鍚﹁繑鍥�1
+static inline int _shm_socket_check_key(shm_socket_t *socket) {
+ void *tmp_ptr = mm_get_by_key(socket->key);
+ if (tmp_ptr!= NULL && tmp_ptr != (void *)1 && !socket->force_bind ) {
+ logger->error("key %d has already been in used!", socket->key);
+ return 0;
+ }
+ return 1;
}
-
-int _shm_close_socket(shm_socket_t *socket, bool notifyRemote) {
- socket->status = SHM_CONN_CLOSED;
- //缁欏鏂瑰彂閫佷竴涓叧闂繛鎺ョ殑娑堟伅
- struct timespec timeout = {1, 0};
- shm_msg_t close_msg;
-
- close_msg.port = socket->port;
- close_msg.size = 0;
- close_msg.type=SHM_SOCKET_CLOSE;
- if(notifyRemote && socket->remoteQueue != NULL) {
- socket->remoteQueue->push_timeout(close_msg, &timeout);
- }
-
- if(socket->queue != NULL) {
- delete socket->queue;
- socket->queue = NULL;
- }
-
- if(socket->remoteQueue != NULL) {
- delete socket->remoteQueue;
- socket->remoteQueue = NULL;
- }
-
- if(socket->messageQueue != NULL) {
- delete socket->messageQueue;
- socket->messageQueue = NULL;
- }
-
- if(socket->acceptQueue != NULL) {
- delete socket->acceptQueue;
- socket->acceptQueue = NULL;
- }
-
- if(socket->clientSocketMap != NULL) {
- shm_socket_t *client_socket;
- for(auto iter = socket->clientSocketMap->begin(); iter != socket->clientSocketMap->end(); iter++) {
- client_socket= iter->second;
-
- client_socket->remoteQueue->push_timeout(close_msg, &timeout);
- delete client_socket->remoteQueue;
- client_socket->remoteQueue=NULL;
-
- delete client_socket->messageQueue;
- client_socket->messageQueue=NULL;
- socket->clientSocketMap->erase(iter);
- free((void *)client_socket);
- }
- delete socket->clientSocketMap;
- }
+SHMQueue<shm_msg_t> *_attach_remote_queue(int key);
- if(socket->dispatch_thread != 0)
- pthread_cancel(socket->dispatch_thread);
-
- free(socket);
- return 0;
+size_t shm_socket_remove_keys(int keys[], size_t length) {
+ return SHMQueue<shm_msg_t>::remove_queues(keys, length);
+}
+
+shm_socket_t *shm_open_socket(shm_socket_type_t socket_type) {
+ shm_socket_t *socket = (shm_socket_t *)calloc(1, sizeof(shm_socket_t));
+ socket->socket_type = socket_type;
+ socket->key = -1;
+ socket->force_bind = false;
+ socket->dispatch_thread = 0;
+ socket->status = SHM_CONN_CLOSED;
+ socket->mutex = SemUtil::get(IPC_PRIVATE, 1);
+ return socket;
}
int shm_close_socket(shm_socket_t *socket) {
- return _shm_close_socket(socket, true);
+
+ int ret;
+ switch (socket->socket_type) {
+ case SHM_SOCKET_STREAM:
+ ret = _shm_close_stream_socket(socket, true);
+ break;
+ case SHM_SOCKET_DGRAM:
+ ret = _shm_close_dgram_socket(socket);
+ break;
+ default:
+ break;
+ }
+ SemUtil::remove(socket->mutex);
+ free(socket);
+ return ret;
}
-int shm_socket_bind(shm_socket_t * socket, int port) {
- shm_socket_t * _socket = (shm_socket_t *) socket;
- _socket -> port = port;
- return 0;
+int shm_socket_bind(shm_socket_t *socket, int key) {
+ socket->key = key;
+ return 0;
}
-int shm_listen(shm_socket_t* socket) {
- int port;
- hashtable_t *hashtable = mm_get_hashtable();
- if(socket -> port == -1) {
- port = hashtable_alloc_key(hashtable);
- socket -> port = port;
- } else {
-
- if(hashtable_get(hashtable, socket->port)!= NULL) {
- err_exit(0, "key %d has already been in used!", socket->port);
- }
- }
-
- socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
- socket->acceptQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
- socket->clientSocketMap = new std::map<int, shm_socket_t* >;
- socket->status = SHM_CONN_LISTEN;
- pthread_create(&(socket->dispatch_thread), NULL, _server_run_msg_rev , (void *)socket);
-
-
- return 0;
+int shm_socket_force_bind(shm_socket_t *socket, int key) {
+ socket->force_bind = true;
+ socket->key = key;
+ return 0;
}
-void _server_close_conn_to_client(shm_socket_t* socket, int port) {
- shm_socket_t *client_socket;
- auto iter = socket->clientSocketMap->find(port);
- if( iter != socket->clientSocketMap->end() ) {
- socket->clientSocketMap->erase(iter);
- }
- //free((void *)client_socket);
+int shm_listen(shm_socket_t *socket) {
+ if (socket->socket_type != SHM_SOCKET_STREAM) {
+ logger->error("can not invoke shm_listen method with a socket which is not a "
+ "SHM_SOCKET_STREAM socket");
+ exit(1);
+ }
+
+ int key;
+ hashtable_t *hashtable = mm_get_hashtable();
+ if (socket->key == -1) {
+ key = hashtable_alloc_key(hashtable);
+ socket->key = key;
+ } else {
+
+ if(!_shm_socket_check_key(socket)) {
+ return -1;
+ }
+ }
+
+ socket->queue = new SHMQueue<shm_msg_t>(socket->key, 16);
+ socket->acceptQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
+ socket->clientSocketMap = new std::map<int, shm_socket_t *>;
+ socket->status = SHM_CONN_LISTEN;
+ pthread_create(&(socket->dispatch_thread), NULL, _server_run_msg_rev,
+ (void *)socket);
+
+ return 0;
}
-
-/**
- * server绔悇绉嶇被鍨嬫秷鎭紙锛夊湪杩欓噷杩涚▼鍒嗘嫞
- */
-void * _server_run_msg_rev(void* _socket) {
- pthread_detach(pthread_self());
- shm_socket_t* socket = (shm_socket_t*) _socket;
- struct timespec timeout = {1, 0};
- shm_msg_t src;
- shm_socket_t *client_socket;
- std::map<int, shm_socket_t* >::iterator iter;
-
- while(socket->queue->pop(src)) {
-
- switch (src.type) {
- case SHM_SOCKET_OPEN :
- socket->acceptQueue->push_timeout(src, &timeout);
- break;
- case SHM_SOCKET_CLOSE :
- _server_close_conn_to_client(socket, src.port);
- break;
- case SHM_COMMON_MSG :
-
- iter = socket->clientSocketMap->find(src.port);
- if( iter != socket->clientSocketMap->end()) {
- client_socket= iter->second;
- // print_msg("_server_run_msg_rev push before", src);
- client_socket->messageQueue->push_timeout(src, &timeout);
- // print_msg("_server_run_msg_rev push after", src);
- }
-
- break;
-
- default:
- err_msg(0, "socket.__shm_rev__: undefined message type.");
- }
- }
-
- return NULL;
-}
-
-
-
/**
* 鎺ュ彈瀹㈡埛绔缓绔嬫柊杩炴帴鐨勮姹�
*
*/
+shm_socket_t *shm_accept(shm_socket_t *socket) {
+ if (socket->socket_type != SHM_SOCKET_STREAM) {
+ logger->error("can not invoke shm_accept method with a socket which is not a "
+ "SHM_SOCKET_STREAM socket");
+ exit(1);
+ }
+ hashtable_t *hashtable = mm_get_hashtable();
+ int client_key;
+ shm_socket_t *client_socket;
+ shm_msg_t src;
-shm_socket_t* shm_accept(shm_socket_t* socket) {
- hashtable_t *hashtable = mm_get_hashtable();
- int client_port;
- shm_socket_t *client_socket;
- shm_msg_t src;
+ if (socket->acceptQueue->pop(src)) {
- if (socket->acceptQueue->pop(src) ) {
+ // print_msg("===accept:", src);
+ client_key = src.key;
+ // client_socket = (shm_socket_t *)malloc(sizeof(shm_socket_t));
+ client_socket = shm_open_socket(socket->socket_type);
+ client_socket->key = socket->key;
+ // client_socket->queue= socket->queue;
+ //鍒濆鍖栨秷鎭痲ueue
+ client_socket->messageQueue =
+ new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
+ //杩炴帴鍒板鏂筿ueue
+ client_socket->remoteQueue = _attach_remote_queue(client_key);
-// print_msg("===accept:", src);
- client_port = src.port;
- client_socket = (shm_socket_t *)malloc(sizeof(shm_socket_t));
- client_socket->port = socket->port;
- // client_socket->queue= socket->queue;
- //鍒濆鍖栨秷鎭痲ueue
- client_socket->messageQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
- //杩炴帴鍒板鏂筿ueue
- client_socket->remoteQueue = _attach_remote_queue(client_port);
+ socket->clientSocketMap->insert({client_key, client_socket});
- socket->clientSocketMap->insert({client_port, client_socket});
+ /*
+* shm_accept 鐢ㄦ埛鎵ц鐨勬柟娉�
+* 涓巁server_run_msg_rev鍦ㄤ袱涓笉鍚岀殑闄愬埗宸ヤ綔,accept瑕佷繚璇佸湪瀹㈡埛鐨勫彂閫佹秷鎭箣鍓嶅畬鎴愯祫婧愮殑鍑嗗宸ヤ綔锛屼互閬垮厤鍑虹幇绔炴�侀棶棰�
+ */
+ //鍙戦�乷pen_reply,鍥炲簲瀹㈡埛绔殑connect璇锋眰
+ struct timespec timeout = {1, 0};
+ shm_msg_t msg;
+ msg.key = socket->key;
+ msg.size = 0;
+ msg.type = SHM_SOCKET_OPEN_REPLY;
- /*
- * shm_accept 鐢ㄦ埛鎵ц鐨勬柟娉� 涓巁server_run_msg_rev鍦ㄤ袱涓笉鍚岀殑闄愬埗宸ヤ綔,accept瑕佷繚璇佸湪瀹㈡埛鐨勫彂閫佹秷鎭箣鍓嶅畬鎴愯祫婧愮殑鍑嗗宸ヤ綔锛屼互閬垮厤鍑虹幇绔炴�侀棶棰�
- */
- //鍙戦�乷pen_reply,鍥炲簲瀹㈡埛绔殑connect璇锋眰
- struct timespec timeout = {1, 0};
- shm_msg_t msg;
- msg.port = socket->port;
- msg.size = 0;
- msg.type = SHM_SOCKET_OPEN_REPLY;
+ if (client_socket->remoteQueue->push_timeout(msg, &timeout)) {
+ client_socket->status = SHM_CONN_ESTABLISHED;
+ return client_socket;
+ } else {
+ logger->error( "shm_accept: 鍙戦�乷pen_reply澶辫触");
+ return NULL;
+ }
- if (client_socket->remoteQueue->push_timeout(msg, &timeout) )
- {
- client_socket->status = SHM_CONN_ESTABLISHED;
- return client_socket;
- } else {
- err_msg(0, "shm_accept: 鍙戦�乷pen_reply澶辫触");
- return NULL;
- }
-
-
- } else {
- err_exit(errno, "shm_accept");
- }
- return NULL;
-
-}
-
-
-int shm_connect(shm_socket_t* socket, int port) {
- hashtable_t *hashtable = mm_get_hashtable();
- if(hashtable_get(hashtable, port)== NULL) {
- err_exit(0, "shm_connect锛歝onnect at port %d failed!", port);
- }
- if(socket->port == -1) {
- socket->port = hashtable_alloc_key(hashtable);
- } else {
-
- if(hashtable_get(hashtable, socket->port)!= NULL) {
- err_exit(0, "key %d has already been in used!", socket->port);
- }
- }
-
- socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
- socket->remoteQueue = _attach_remote_queue(port);
- socket->messageQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
-
-
- //鍙戦�乷pen璇锋眰
- struct timespec timeout = {1, 0};
- shm_msg_t msg;
- msg.port = socket->port;
- msg.size = 0;
- msg.type=SHM_SOCKET_OPEN;
- socket->remoteQueue->push_timeout(msg, &timeout);
-
- //鎺ュ彈open reply
- if(socket->queue->pop(msg)) {
- // 鍦ㄨ繖閲宻erver绔凡缁忓噯澶囧ソ鎺ュ彈瀹㈡埛绔彂閫佽姹備簡,瀹屾垚涓庢湇鍔＄鐨勮繛鎺�
- if(msg.type == SHM_SOCKET_OPEN_REPLY) {
- socket->status = SHM_CONN_ESTABLISHED;
- pthread_create(&(socket->dispatch_thread), NULL, _client_run_msg_rev , (void *)socket);
- } else {
- err_exit(0, "shm_connect: 涓嶅尮閰嶇殑搴旂瓟淇℃伅!");
- }
-
- } else {
- err_exit(0, "connect failted!");
- }
-
- return 0;
-}
-
-void _client_close_conn_to_server(shm_socket_t* socket) {
-
- _shm_close_socket(socket, false);
+ } else {
+ err_exit(errno, "shm_accept");
+ }
+ return NULL;
}
/**
- * client绔殑鍚勭绫诲瀷娑堟伅锛堬級鍦ㄨ繖閲岃繘绋嬪垎鎷�
+ * @return 0鎴愬姛. 鍏朵粬鍊煎け璐�
*/
-void * _client_run_msg_rev(void* _socket) {
- pthread_detach(pthread_self());
- shm_socket_t* socket = (shm_socket_t*) _socket;
- struct timespec timeout = {1, 0};
- shm_msg_t src;
-
- while(socket->queue->pop(src)) {
- switch (src.type) {
-
- case SHM_SOCKET_CLOSE :
- _client_close_conn_to_server(socket);
- break;
- case SHM_COMMON_MSG :
- socket->messageQueue->push_timeout(src, &timeout);
- break;
- default:
- err_msg(0, "socket.__shm_rev__: undefined message type.");
- }
+int shm_connect(shm_socket_t *socket, int key) {
+ if (socket->socket_type != SHM_SOCKET_STREAM) {
+ logger->error( "can not invoke shm_connect method with a socket which is not "
+ "a SHM_SOCKET_STREAM socket");
+ exit(1);
+ }
+ hashtable_t *hashtable = mm_get_hashtable();
+ if (hashtable_get(hashtable, key) == NULL) {
+ logger->error("shm_connect锛歝onnect at key %d failed!", key);
+ return -1;
+ }
+
+ if (socket->key == -1) {
+ socket->key = hashtable_alloc_key(hashtable);
+ } else {
+ if(!_shm_socket_check_key(socket)) {
+ return -1;
}
-
- return NULL;
+ }
+
+ socket->queue = new SHMQueue<shm_msg_t>(socket->key, 16);
+
+ if ((socket->remoteQueue = _attach_remote_queue(key)) == NULL) {
+ logger->error("connect to %d failted", key);
+ return -1;
+ }
+ socket->messageQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
+
+ //鍙戦�乷pen璇锋眰
+ struct timespec timeout = {1, 0};
+ shm_msg_t msg;
+ msg.key = socket->key;
+ msg.size = 0;
+ msg.type = SHM_SOCKET_OPEN;
+ socket->remoteQueue->push_timeout(msg, &timeout);
+
+ //鎺ュ彈open reply
+ if (socket->queue->pop(msg)) {
+ // 鍦ㄨ繖閲宻erver绔凡缁忓噯澶囧ソ鎺ュ彈瀹㈡埛绔彂閫佽姹備簡,瀹屾垚涓庢湇鍔＄鐨勮繛鎺�
+ if (msg.type == SHM_SOCKET_OPEN_REPLY) {
+ socket->status = SHM_CONN_ESTABLISHED;
+ pthread_create(&(socket->dispatch_thread), NULL, _client_run_msg_rev,
+ (void *)socket);
+ } else {
+ logger->error( "shm_connect: 涓嶅尮閰嶇殑搴旂瓟淇℃伅!");
+ exit(1);
+ }
+
+ } else {
+ logger->error( "connect failted!");
+ return -1;
+ }
+
+ return 0;
}
-
-
-
-
int shm_send(shm_socket_t *socket, const void *buf, const int size) {
- // hashtable_t *hashtable = mm_get_hashtable();
- // if(socket->remoteQueue == NULL) {
- // err_msg(errno, "褰撳墠瀹㈡埛绔棤杩炴帴!");
- // return -1;
- // }
- shm_msg_t dest;
- dest.type=SHM_COMMON_MSG;
- dest.port = socket->port;
- dest.size = size;
- dest.buf = mm_malloc(size);
- memcpy(dest.buf, buf, size);
+ if (socket->socket_type != SHM_SOCKET_STREAM) {
+ logger->error("shm_socket.shm_send: can not invoke shm_send method with a socket which is not a "
+ "SHM_SOCKET_STREAM socket");
+ exit(1);
+ }
+ // hashtable_t *hashtable = mm_get_hashtable();
+ // if(socket->remoteQueue == NULL) {
+ // err_msg(errno, "褰撳墠瀹㈡埛绔棤杩炴帴!");
+ // return -1;
+ // }
+ shm_msg_t dest;
+ dest.type = SHM_COMMON_MSG;
+ dest.key = socket->key;
+ dest.size = size;
+ dest.buf = mm_malloc(size);
+ memcpy(dest.buf, buf, size);
- // struct timeval time;
- // gettimeofday(&time, NULL);
-//err_msg(0, "%d %d=======>push befor %d", time.tv_sec, time.tv_usec, socket->port);
- if(socket->remoteQueue->push(dest)) {
-
- //gettimeofday(&time, NULL);
-//err_msg(0, "%d %d=======>push after %d", time.tv_sec, time.tv_usec, socket->port);
- return 0;
- } else {
- err_msg(errno, "connection has been closed!");
- return -1;
- }
-
-
+ if (socket->remoteQueue->push(dest)) {
+ return 0;
+ } else {
+ logger->error(errno, "connection has been closed!");
+ return -1;
+ }
}
-int shm_recv(shm_socket_t* socket, void **buf, int *size) {
- shm_msg_t src;
+int shm_recv(shm_socket_t *socket, void **buf, int *size) {
+ if (socket->socket_type != SHM_SOCKET_STREAM) {
+ logger->error( "shm_socket.shm_recv: can not invoke shm_recv method in a %d type socket which is "
+ "not a SHM_SOCKET_STREAM socket ",
+ socket->socket_type);
+ exit(1);
+ }
+ shm_msg_t src;
-// struct timeval time;
-// gettimeofday(&time, NULL);
-// err_msg(0, "%d %d=======>pop befor %d", time.tv_sec, time.tv_usec, socket->port);
- if (socket->messageQueue->pop(src)) {
-// gettimeofday(&time, NULL);
-// err_msg(0, "%d %d=======>pop after %d", time.tv_sec, time.tv_usec, socket->port);
- void * _buf = malloc(src.size);
- memcpy(_buf, src.buf, src.size);
- *buf = _buf;
- *size = src.size;
- mm_free(src.buf);
- return 0;
- } else {
- return -1;
- }
-
-
-
+ if (socket->messageQueue->pop(src)) {
+ void *_buf = malloc(src.size);
+ memcpy(_buf, src.buf, src.size);
+ *buf = _buf;
+ *size = src.size;
+ mm_free(src.buf);
+ return 0;
+ } else {
+ return -1;
+ }
}
+
+// 鐭繛鎺ユ柟寮忓彂閫�
+int shm_sendto(shm_socket_t *socket, const void *buf, const int size,
+ const int key, const struct timespec *timeout, const int flags) {
+ if (socket->socket_type != SHM_SOCKET_DGRAM) {
+ logger->error( "shm_socket.shm_sendto: Can't invoke shm_sendto method in a %d type socket which is "
+ "not a SHM_SOCKET_DGRAM socket ",
+ socket->socket_type);
+ exit(0);
+ }
+ hashtable_t *hashtable = mm_get_hashtable();
+
+ SemUtil::dec(socket->mutex);
+ if (socket->queue == NULL) {
+ if (socket->key == -1) {
+ socket->key = hashtable_alloc_key(hashtable);
+ } else {
+
+ if(!_shm_socket_check_key(socket)) {
+ return -1;
+ }
+
+ }
+
+ socket->queue = new SHMQueue<shm_msg_t>(socket->key, 16);
+ }
+ SemUtil::inc(socket->mutex);
+
+ if (key == socket->key) {
+ logger->error( "can not send to your self!");
+ return -1;
+ }
+
+ SHMQueue<shm_msg_t> *remoteQueue;
+ if ((remoteQueue = _attach_remote_queue(key)) == NULL) {
+ logger->error( "shm_sendto failed, the other end has been closed, or has not been opened!");
+ return SHM_SOCKET_ECONNFAILED;
+ }
+
+ shm_msg_t dest;
+ dest.type = SHM_COMMON_MSG;
+ dest.key = socket->key;
+ dest.size = size;
+ dest.buf = mm_malloc(size);
+ memcpy(dest.buf, buf, size);
+
+ // printf("shm_sendto push before\n");
+ bool rv;
+ if(flags & SHM_MSG_NOWAIT != 0) {
+ rv = remoteQueue->push_nowait(dest);
+ } else if(timeout != NULL) {
+ rv = remoteQueue->push_timeout(dest, timeout);
+ } else {
+ rv = remoteQueue->push(dest);
+ }
+
+ if (rv) {
+ // printf("shm_sendto push after\n");
+ delete remoteQueue;
+ return 0;
+ } else {
+ delete remoteQueue;
+ mm_free(dest.buf);
+ logger->error(errno, "sendto key %d failed!", key);
+ return -1;
+ }
+}
+
+// 鐭繛鎺ユ柟寮忔帴鍙�
+int shm_recvfrom(shm_socket_t *socket, void **buf, int *size, int *key, struct timespec *timeout, int flags) {
+ if (socket->socket_type != SHM_SOCKET_DGRAM) {
+ logger->error("shm_socket.shm_recvfrom: Can't invoke shm_recvfrom method in a %d type socket which "
+ "is not a SHM_SOCKET_DGRAM socket ",
+ socket->socket_type);
+ exit(1);
+ }
+ hashtable_t *hashtable = mm_get_hashtable();
+ SemUtil::dec(socket->mutex);
+ if (socket->queue == NULL) {
+ if (socket->key == -1) {
+ socket->key = hashtable_alloc_key(hashtable);
+ } else {
+
+ if(!_shm_socket_check_key(socket)) {
+ return -1;
+ }
+ }
+
+ socket->queue = new SHMQueue<shm_msg_t>(socket->key, 16);
+ }
+ SemUtil::inc(socket->mutex);
+
+ shm_msg_t src;
+ // printf("shm_recvfrom pop before\n");
+ bool rv;
+ if(flags & SHM_MSG_NOWAIT != 0) {
+ rv = socket->queue->pop_nowait(src);
+ } else if(timeout != NULL) {
+ rv = socket->queue->pop_timeout(src, timeout);
+ } else {
+ rv = socket->queue->pop(src);
+ }
+
+ if (rv) {
+ void *_buf = malloc(src.size);
+ memcpy(_buf, src.buf, src.size);
+ *buf = _buf;
+ *size = src.size;
+ *key = src.key;
+ mm_free(src.buf);
+ // printf("shm_recvfrom pop after\n");
+ return 0;
+ } else {
+ return -1;
+ }
+}
+
+static pthread_once_t _once_ = PTHREAD_ONCE_INIT;
+static pthread_key_t _tmp_recv_socket_key_;
+
+ /* Free thread-specific data buffer */
+static void _destrory_tmp_recv_socket_(void *tmp_socket)
+{
+ logger->debug("%d destroy tmp socket\n", pthread_self());
+ shm_close_socket((shm_socket_t *)tmp_socket);
+}
+
+/* One-time key creation function */
+static void _create_tmp_recv_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(&_tmp_recv_socket_key_, _destrory_tmp_recv_socket_);
+ if (s != 0) {
+ logger->error(s, "pthread_key_create");
+ abort(); /* dump core and terminate */
+ exit(1);
+ }
+}
+
+
+
+int shm_sendandrecv(shm_socket_t *socket, const void *send_buf,
+ const int send_size, const int send_key, void **recv_buf,
+ int *recv_size, struct timespec *timeout, int flags) {
+ int recv_key;
+ int rv;
+
+ // 鐢╰hread local 淇濊瘉姣忎釜绾跨▼鐢ㄤ竴涓嫭鍗犵殑socket鎺ュ彈瀵规柟杩斿洖鐨勪俊鎭�
+ shm_socket_t *tmp_socket;
+
+ if (socket->socket_type != SHM_SOCKET_DGRAM) {
+ logger->error( "shm_socket.shm_sendandrecv: Can't invoke shm_sendandrecv method in a %d type socket "
+ "which is not a SHM_SOCKET_DGRAM socket ",
+ socket->socket_type);
+ exit(1);
+ }
+
+
+ rv = pthread_once(&_once_, _create_tmp_recv_socket_key);
+ if (rv != 0) {
+ logger->error(rv, "shm_sendandrecv pthread_once");
+ exit(1);
+ }
+
+ tmp_socket = (shm_socket_t *)pthread_getspecific(_tmp_recv_socket_key_);
+ if (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() );
+ tmp_socket = shm_open_socket(SHM_SOCKET_DGRAM);
+
+ rv = pthread_setspecific(_tmp_recv_socket_key_, tmp_socket);
+ if (rv != 0) {
+ logger->error(rv, "shm_sendandrecv : pthread_setspecific");
+ exit(1);
+ }
+ }
+
+ 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);
+
+ return rv;
+ } else {
+
+ return rv;
+ }
+ return -1;
+}
+
+int shm_sendandrecv_unsafe(shm_socket_t *socket, const void *send_buf,
+ const int send_size, const int send_key, void **recv_buf,
+ int *recv_size, struct timespec *timeout, int flags) {
+ if (socket->socket_type != SHM_SOCKET_DGRAM) {
+ logger->error( "shm_socket.shm_sendandrecv_unsafe : Can't invoke shm_sendandrecv method in a %d type socket "
+ "which is not a SHM_SOCKET_DGRAM socket ",
+ socket->socket_type);
+ exit(1);
+ }
+ int recv_key;
+ int rv;
+
+
+ if ((rv = shm_sendto(socket, send_buf, send_size, send_key, timeout, flags)) == 0) {
+ rv = shm_recvfrom(socket, recv_buf, recv_size, &recv_key, timeout, flags);
+ return rv;
+ } else {
+ return rv;
+ }
+ return -1;
+}
+
+// ============================================================================================================
/**
* 缁戝畾key鍒伴槦鍒楋紝浣嗘槸骞朵笉浼氬垱寤洪槦鍒椼�傚鏋滄病鏈夊搴旀寚瀹歬ey鐨勯槦鍒楁彁绀洪敊璇苟閫�鍑�
*/
-SHMQueue<shm_msg_t> * _attach_remote_queue(int port) {
- hashtable_t *hashtable = mm_get_hashtable();
- if(hashtable_get(hashtable, port)== NULL) {
- err_exit(0, "_remote_queue_attach锛歝onnet at port %d failed!", port);
- return NULL;
- }
-
- SHMQueue<shm_msg_t> *queue = new SHMQueue<shm_msg_t>(port, 0);
- return queue;
+SHMQueue<shm_msg_t> *_attach_remote_queue(int key) {
+
+ hashtable_t *hashtable = mm_get_hashtable();
+ if (hashtable_get(hashtable, key) == NULL) {
+ logger->error("shm_socket._remote_queue_attach锛歝onnet at key %d failed!", key);
+ return NULL;
+ }
+
+ SHMQueue<shm_msg_t> *queue = new SHMQueue<shm_msg_t>(key, 0);
+ return queue;
}
+void _server_close_conn_to_client(shm_socket_t *socket, int key) {
+ shm_socket_t *client_socket;
+ std::map<int, shm_socket_t *>::iterator iter =
+ socket->clientSocketMap->find(key);
+ if (iter != socket->clientSocketMap->end()) {
+ client_socket = iter->second;
+ free((void *)client_socket);
+ socket->clientSocketMap->erase(iter);
+ }
+}
+/**
+ * server绔悇绉嶇被鍨嬫秷鎭紙锛夊湪杩欓噷杩涚▼鍒嗘嫞
+ */
+void *_server_run_msg_rev(void *_socket) {
+ pthread_detach(pthread_self());
+ shm_socket_t *socket = (shm_socket_t *)_socket;
+ struct timespec timeout = {1, 0};
+ shm_msg_t src;
+ shm_socket_t *client_socket;
+ std::map<int, shm_socket_t *>::iterator iter;
+ while (socket->queue->pop(src)) {
+ switch (src.type) {
+ case SHM_SOCKET_OPEN:
+ socket->acceptQueue->push_timeout(src, &timeout);
+ break;
+ case SHM_SOCKET_CLOSE:
+ _server_close_conn_to_client(socket, src.key);
+ break;
+ case SHM_COMMON_MSG:
+
+ iter = socket->clientSocketMap->find(src.key);
+ if (iter != socket->clientSocketMap->end()) {
+ client_socket = iter->second;
+ // print_msg("_server_run_msg_rev push before", src);
+ client_socket->messageQueue->push_timeout(src, &timeout);
+ // print_msg("_server_run_msg_rev push after", src);
+ }
+
+ break;
+
+ default:
+ logger->error("shm_socket._server_run_msg_rev: undefined message type.");
+ }
+ }
+
+ return NULL;
+}
+
+void _client_close_conn_to_server(shm_socket_t *socket) {
+
+ _shm_close_stream_socket(socket, false);
+}
+
+/**
+ * client绔殑鍚勭绫诲瀷娑堟伅锛堬級鍦ㄨ繖閲岃繘绋嬪垎鎷�
+ */
+void *_client_run_msg_rev(void *_socket) {
+ pthread_detach(pthread_self());
+ shm_socket_t *socket = (shm_socket_t *)_socket;
+ struct timespec timeout = {1, 0};
+ shm_msg_t src;
+
+ while (socket->queue->pop(src)) {
+ switch (src.type) {
+
+ case SHM_SOCKET_CLOSE:
+ _client_close_conn_to_server(socket);
+ break;
+ case SHM_COMMON_MSG:
+ socket->messageQueue->push_timeout(src, &timeout);
+ break;
+ default:
+ logger->error( "shm_socket._client_run_msg_rev: undefined message type.");
+ }
+ }
+
+ return NULL;
+}
+
+int _shm_close_stream_socket(shm_socket_t *socket, bool notifyRemote) {
+ socket->status = SHM_CONN_CLOSED;
+ //缁欏鏂瑰彂閫佷竴涓叧闂繛鎺ョ殑娑堟伅
+ struct timespec timeout = {1, 0};
+ shm_msg_t close_msg;
+
+ close_msg.key = socket->key;
+ close_msg.size = 0;
+ close_msg.type = SHM_SOCKET_CLOSE;
+ if (notifyRemote && socket->remoteQueue != NULL) {
+ socket->remoteQueue->push_timeout(close_msg, &timeout);
+ }
+
+ if (socket->queue != NULL) {
+ delete socket->queue;
+ socket->queue = NULL;
+ }
+
+ if (socket->remoteQueue != NULL) {
+ delete socket->remoteQueue;
+ socket->remoteQueue = NULL;
+ }
+
+ if (socket->messageQueue != NULL) {
+ delete socket->messageQueue;
+ socket->messageQueue = NULL;
+ }
+
+ if (socket->acceptQueue != NULL) {
+ delete socket->acceptQueue;
+ socket->acceptQueue = NULL;
+ }
+
+ if (socket->clientSocketMap != NULL) {
+ shm_socket_t *client_socket;
+ for (auto iter = socket->clientSocketMap->begin();
+ iter != socket->clientSocketMap->end(); iter++) {
+ client_socket = iter->second;
+
+ client_socket->remoteQueue->push_timeout(close_msg, &timeout);
+ delete client_socket->remoteQueue;
+ client_socket->remoteQueue = NULL;
+
+ delete client_socket->messageQueue;
+ client_socket->messageQueue = NULL;
+ free((void *)client_socket);
+ }
+ delete socket->clientSocketMap;
+ }
+
+ if (socket->dispatch_thread != 0)
+ pthread_cancel(socket->dispatch_thread);
+
+
+ return 0;
+
+}
+
+int _shm_close_dgram_socket(shm_socket_t *socket){
+ if(socket->queue != NULL) {
+ delete socket->queue;
+ socket->queue = NULL;
+ }
+
+ return 0;
+}
--
Gitblit v1.8.0