From 8c42a659c0cc9178d1f1305acb41dfbf4a8697ef Mon Sep 17 00:00:00 2001
From: wangzhengquan <wangzhengquan85@126.com>
Date: 星期四, 22 十月 2020 16:20:26 +0800
Subject: [PATCH] update
---
src/socket/shm_socket.c | 77 ++++++++++++++++++++++++++++++--------
1 files changed, 60 insertions(+), 17 deletions(-)
diff --git a/src/socket/shm_socket.c b/src/socket/shm_socket.c
index f95379a..f07d8a6 100644
--- a/src/socket/shm_socket.c
+++ b/src/socket/shm_socket.c
@@ -3,7 +3,7 @@
#include "logger_factory.h"
#include <map>
-static Logger logger = LoggerFactory::getLogger();
+static Logger *logger = LoggerFactory::getLogger();
@@ -30,6 +30,12 @@
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;
@@ -37,20 +43,26 @@
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) {
+
+ int ret;
switch (socket->socket_type) {
- case SHM_SOCKET_STREAM:
- return _shm_close_stream_socket(socket, true);
- case SHM_SOCKET_DGRAM:
- return _shm_close_dgram_socket(socket);
- default:
- return -1;
+ 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;
}
- return -1;
+ SemUtil::remove(socket->mutex);
+ free(socket);
+ return ret;
}
int shm_socket_bind(shm_socket_t *socket, int port) {
@@ -252,6 +264,7 @@
}
hashtable_t *hashtable = mm_get_hashtable();
+ SemUtil::dec(socket->mutex);
if (socket->queue == NULL) {
if (socket->port == -1) {
socket->port = hashtable_alloc_key(hashtable);
@@ -262,9 +275,17 @@
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;
@@ -274,11 +295,6 @@
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, the other end has been closed, or has not been opened!");
- return -1;
- }
// printf("shm_sendto push before\n");
bool rv;
if(flags & SHM_MSG_NOWAIT != 0) {
@@ -295,6 +311,7 @@
return 0;
} else {
delete remoteQueue;
+ mm_free(dest.buf);
err_msg(errno, "sendto port %d failed!", port);
return -1;
}
@@ -308,6 +325,7 @@
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);
@@ -318,6 +336,7 @@
socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
}
+ SemUtil::inc(socket->mutex);
shm_msg_t src;
// printf("shm_recvfrom pop before\n");
@@ -356,9 +375,33 @@
int rv;
shm_socket_t *tmp_socket = shm_open_socket(SHM_SOCKET_DGRAM);
- if (shm_sendto(tmp_socket, send_buf, send_size, send_port, timeout, flags) == 0) {
+ 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;
+}
+
+int shm_sendandrecv_unsafe(shm_socket_t *socket, const void *send_buf,
+ const int send_size, const int send_port, void **recv_buf,
+ 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 ",
+ socket->socket_type);
+ }
+ int recv_port;
+ int rv;
+
+
+ if ((rv = shm_sendto(socket, send_buf, send_size, send_port, timeout, flags)) == 0) {
+ rv = shm_recvfrom(socket, recv_buf, recv_size, &recv_port, timeout, flags);
+ return rv;
+ } else {
return rv;
}
return -1;
@@ -515,7 +558,7 @@
if (socket->dispatch_thread != 0)
pthread_cancel(socket->dispatch_thread);
- free(socket);
+
return 0;
}
@@ -525,7 +568,7 @@
delete socket->queue;
socket->queue = NULL;
}
- free(socket);
+
return 0;
}
--
Gitblit v1.8.0