From 0da7d09982d61ea6864cec53e2341b51b30cf208 Mon Sep 17 00:00:00 2001
From: wangzhengquan <wangzhengquan85@126.com>
Date: 星期二, 13 十月 2020 18:19:22 +0800
Subject: [PATCH] update
---
src/socket/shm_socket.c | 102 +++++++++++++++++++++++++++++++++++---------------
1 files changed, 71 insertions(+), 31 deletions(-)
diff --git a/src/socket/shm_socket.c b/src/socket/shm_socket.c
index 52251e6..362e5e8 100644
--- a/src/socket/shm_socket.c
+++ b/src/socket/shm_socket.c
@@ -3,29 +3,47 @@
#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 *_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);
-int _shm_close_dgram_socket(shm_socket_t *socket);
+static int _shm_close_dgram_socket(shm_socket_t *socket);
-int _shm_close_stream_socket(shm_socket_t *socket, bool notifyRemote);
+static int _shm_close_stream_socket(shm_socket_t *socket, bool notifyRemote);
+
+static inline int _shm_socket_check_key(shm_socket_t *socket) {
+ void *tmp_ptr = mm_get_by_key(socket->port);
+ if (tmp_ptr!= NULL && tmp_ptr != (void *)1 && !socket->force_bind ) {
+ err_exit(0, "key %d has already been in used!", socket->port);
+ return 0;
+ }
+ return 1;
+}
SHMQueue<shm_msg_t> *_attach_remote_queue(int port);
+
+
+
+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->port = -1;
+ socket->force_bind = false;
socket->dispatch_thread = 0;
socket->status = SHM_CONN_CLOSED;
-
+ socket->mutex = SemUtil::get(IPC_PRIVATE, 1);
return socket;
}
@@ -46,6 +64,12 @@
return 0;
}
+int shm_socket_force_bind(shm_socket_t *socket, int port) {
+ socket->force_bind = true;
+ socket->port = port;
+ return 0;
+}
+
int shm_listen(shm_socket_t *socket) {
if (socket->socket_type != SHM_SOCKET_STREAM) {
@@ -60,9 +84,7 @@
socket->port = port;
} else {
- if (hashtable_get(hashtable, socket->port) != NULL) {
- err_exit(0, "key %d has already been in used!", socket->port);
- }
+ _shm_socket_check_key(socket);
}
socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
@@ -130,6 +152,7 @@
return NULL;
}
+
int shm_connect(shm_socket_t *socket, int port) {
if (socket->socket_type != SHM_SOCKET_STREAM) {
err_exit(0, "can not invoke shm_connect method with a socket which is not "
@@ -143,10 +166,7 @@
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);
- }
+ _shm_socket_check_key(socket);
}
socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
@@ -227,9 +247,10 @@
}
}
+
// 鐭繛鎺ユ柟寮忓彂閫�
int shm_sendto(shm_socket_t *socket, const void *buf, const int size,
- const int port, const struct timespec *timeout) {
+ const int port, const struct timespec *timeout, const int flags) {
if (socket->socket_type != SHM_SOCKET_DGRAM) {
err_exit(0, "Can't invoke shm_sendto method in a %d type socket which is "
"not a SHM_SOCKET_DGRAM socket ",
@@ -237,21 +258,28 @@
}
hashtable_t *hashtable = mm_get_hashtable();
+ SemUtil::dec(socket->mutex);
if (socket->queue == NULL) {
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);
- }
+ _shm_socket_check_key(socket);
}
socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
}
+ SemUtil::inc(socket->mutex);
+
if (port == socket->port) {
err_msg(0, "can not send to your self!");
return -1;
+ }
+
+ SHMQueue<shm_msg_t> *remoteQueue;
+ if ((remoteQueue = _attach_remote_queue(port)) == NULL) {
+ err_msg(0, "shm_sendto failed, the other end has been closed, or has not been opened!");
+ return SHM_SOCKET_ECONNFAILED;
}
shm_msg_t dest;
@@ -261,14 +289,11 @@
dest.buf = mm_malloc(size);
memcpy(dest.buf, buf, size);
- SHMQueue<shm_msg_t> *remoteQueue;
- if ((remoteQueue = _attach_remote_queue(port)) == NULL) {
- err_msg(0, "shm_sendto failed, then other end has been closed!");
- return -1;
- }
// printf("shm_sendto push before\n");
bool rv;
- if(timeout != NULL) {
+ 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);
@@ -280,35 +305,45 @@
return 0;
} else {
delete remoteQueue;
+ mm_free(dest.buf);
err_msg(errno, "sendto port %d failed!", port);
return -1;
}
}
// 鐭繛鎺ユ柟寮忔帴鍙�
-int shm_recvfrom(shm_socket_t *socket, void **buf, int *size, int *port) {
+int shm_recvfrom(shm_socket_t *socket, void **buf, int *size, int *port, struct timespec *timeout, int flags) {
if (socket->socket_type != SHM_SOCKET_DGRAM) {
err_exit(0, "Can't invoke shm_recvfrom method in a %d type socket which "
"is not a SHM_SOCKET_DGRAM socket ",
socket->socket_type);
}
hashtable_t *hashtable = mm_get_hashtable();
+ SemUtil::dec(socket->mutex);
if (socket->queue == NULL) {
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);
- }
+ _shm_socket_check_key(socket);
}
socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
}
+ SemUtil::inc(socket->mutex);
shm_msg_t src;
// printf("shm_recvfrom pop before\n");
- if (socket->queue->pop(src)) {
+ 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;
@@ -324,7 +359,7 @@
int shm_sendandrecv(shm_socket_t *socket, const void *send_buf,
const int send_size, const int send_port, void **recv_buf,
- int *recv_size) {
+ int *recv_size, struct timespec *timeout, int flags) {
if (socket->socket_type != SHM_SOCKET_DGRAM) {
err_exit(0, "Can't invoke shm_sendandrecv method in a %d type socket "
"which is not a SHM_SOCKET_DGRAM socket ",
@@ -334,14 +369,19 @@
int rv;
shm_socket_t *tmp_socket = shm_open_socket(SHM_SOCKET_DGRAM);
- if (shm_sendto(tmp_socket, send_buf, send_size, send_port) == 0) {
- rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_port);
+ if ((rv = shm_sendto(tmp_socket, send_buf, send_size, send_port, timeout, flags)) == 0) {
+ rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_port, timeout, flags);
+ shm_close_socket(tmp_socket);
+ return rv;
+ } else {
shm_close_socket(tmp_socket);
return rv;
}
return -1;
}
+// ============================================================================================================
+
/**
* 缁戝畾key鍒伴槦鍒楋紝浣嗘槸骞朵笉浼氬垱寤洪槦鍒椼�傚鏋滄病鏈夊搴旀寚瀹歬ey鐨勯槦鍒楁彁绀洪敊璇苟閫�鍑�
*/
--
Gitblit v1.8.0