You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celix.apache.org by rb...@apache.org on 2020/04/29 19:58:58 UTC
[celix] branch
feature/add_msg_segemenation_to_tcp_admin_with_wire_v2 updated (6fb06e1 ->
209a081)
This is an automated email from the ASF dual-hosted git repository.
rbulter pushed a change to branch feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
in repository https://gitbox.apache.org/repos/asf/celix.git.
from 6fb06e1 Merge branch 'develop' into feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
add 79884e3 Fix TCP endpoint handling
add 7a46f24 Avoid segfault when stopping bundle
add 23f7d3d Fix test for OSX
add 57c3f5a Fix printf error
new 543a12f Merge branch 'feature/fix_tcp_endpoint_handling' into feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
new cfc7373 Merge Develop
add 9910215 Fix review comment
new 075ead4 Merge remote-tracking branch 'remotes/origin/feature/fix_tcp_endpoint_handling' into feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
new 7b8608b Merge
new 209a081 Fix wire 2
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../src/pubsub_psa_tcp_constants.h | 5 +-
.../pubsub_admin_tcp/src/pubsub_tcp_handler.c | 94 ++++++++++++++--------
.../pubsub_admin_tcp/src/pubsub_tcp_handler.h | 6 +-
.../src/pubsub_tcp_topic_receiver.c | 14 ++--
.../pubsub_admin_tcp/src/pubsub_tcp_topic_sender.c | 2 +
.../pubsub/pubsub_protocol_wire_v1/CMakeLists.txt | 9 ++-
.../pubsub/pubsub_protocol_wire_v2/CMakeLists.txt | 12 +--
.../pubsub_protocol_wire_v2/gtest/CMakeLists.txt | 2 +-
.../src/pubsub_wire_v2_protocol_impl.h | 2 +-
libs/framework/src/framework.c | 5 ++
libs/utils/private/test/celix_threads_test.cpp | 1 +
libs/utils/src/utils.c | 4 +-
12 files changed, 93 insertions(+), 63 deletions(-)
[celix] 03/05: Merge remote-tracking branch
'remotes/origin/feature/fix_tcp_endpoint_handling' into
feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
Posted by rb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rbulter pushed a commit to branch feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
in repository https://gitbox.apache.org/repos/asf/celix.git
commit 075ead4374f678580f0d2ce66dcb6524493257e9
Merge: cfc7373 9910215
Author: Roy Bulter <ro...@gmail.com>
AuthorDate: Wed Apr 29 21:08:03 2020 +0200
Merge remote-tracking branch 'remotes/origin/feature/fix_tcp_endpoint_handling' into feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
# Conflicts:
# bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
.../pubsub_admin_tcp/src/pubsub_tcp_handler.c | 527 ++++++++++-----------
.../src/pubsub_tcp_topic_receiver.c | 10 +-
2 files changed, 263 insertions(+), 274 deletions(-)
diff --cc bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
index 31d9d7f,9c469ce..6e2072e
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
@@@ -70,57 -70,56 +70,58 @@@
// Entry administration
//
typedef struct psa_tcp_connection_entry {
- char *interface_url;
- char *url;
- int fd;
- struct sockaddr_in addr;
- socklen_t len;
- bool connected;
- pubsub_protocol_message_t header;
- unsigned int maxMsgSize;
- unsigned int syncSize;
- unsigned int headerSize;
- unsigned int headerBufferSize; // Size of headerBuffer, size = 0, no headerBuffer -> included in payload
- void *headerBuffer;
- unsigned int bufferSize;
- void *buffer;
- unsigned int metaBufferSize;
- void *metaBuffer;
- struct msghdr msg;
- size_t msg_iovlen; /* Number of elements in the vector. */
- unsigned int retryCount;
+ char *interface_url;
+ char *url;
+ int fd;
+ struct sockaddr_in addr;
+ socklen_t len;
+ bool connected;
+ pubsub_protocol_message_t header;
++ unsigned int maxMsgSize;
+ unsigned int syncSize;
+ unsigned int headerSize;
+ unsigned int headerBufferSize; // Size of headerBuffer, size = 0, no headerBuffer -> included in payload
+ void *headerBuffer;
+ unsigned int bufferSize;
+ void *buffer;
+ unsigned int metaBufferSize;
+ void *metaBuffer;
+ struct msghdr msg;
+ size_t msg_iovlen; /* Number of elements in the vector. */
+ unsigned int retryCount;
} psa_tcp_connection_entry_t;
//
// Handle administration
//
struct pubsub_tcpHandler {
- celix_thread_rwlock_t dbLock;
- unsigned int timeout;
- hash_map_t *connection_url_map;
- hash_map_t *connection_fd_map;
- hash_map_t *interface_url_map;
- hash_map_t *interface_fd_map;
- int efd;
- pubsub_tcpHandler_receiverConnectMessage_callback_t receiverConnectMessageCallback;
- pubsub_tcpHandler_receiverConnectMessage_callback_t receiverDisconnectMessageCallback;
- void *receiverConnectPayload;
- pubsub_tcpHandler_acceptConnectMessage_callback_t acceptConnectMessageCallback;
- pubsub_tcpHandler_acceptConnectMessage_callback_t acceptDisconnectMessageCallback;
- void *acceptConnectPayload;
- pubsub_tcpHandler_processMessage_callback_t processMessageCallback;
- void *processMessagePayload;
- log_helper_t *logHelper;
- pubsub_protocol_service_t *protocol;
- unsigned int bufferSize;
- unsigned int maxNofBuffer;
- unsigned int maxMsgSize;
- unsigned int maxSendRetryCount;
- unsigned int maxRcvRetryCount;
- double sendTimeout;
- double rcvTimeout;
- celix_thread_t thread;
- bool running;
+ unsigned int readSeqNr;
+ celix_thread_rwlock_t dbLock;
+ unsigned int timeout;
+ hash_map_t *connection_url_map;
+ hash_map_t *connection_fd_map;
+ hash_map_t *interface_url_map;
+ hash_map_t *interface_fd_map;
+ int efd;
+ pubsub_tcpHandler_receiverConnectMessage_callback_t receiverConnectMessageCallback;
+ pubsub_tcpHandler_receiverConnectMessage_callback_t receiverDisconnectMessageCallback;
+ void *receiverConnectPayload;
+ pubsub_tcpHandler_acceptConnectMessage_callback_t acceptConnectMessageCallback;
+ pubsub_tcpHandler_acceptConnectMessage_callback_t acceptDisconnectMessageCallback;
+ void *acceptConnectPayload;
+ pubsub_tcpHandler_processMessage_callback_t processMessageCallback;
+ void *processMessagePayload;
+ log_helper_t *logHelper;
+ pubsub_protocol_service_t *protocol;
+ unsigned int bufferSize;
+ unsigned int maxNofBuffer;
++ unsigned int maxMsgSize;
+ unsigned int maxSendRetryCount;
+ unsigned int maxRcvRetryCount;
+ double sendTimeout;
+ double rcvTimeout;
+ celix_thread_t thread;
+ bool running;
};
static inline int
[celix] 04/05: Merge
Posted by rb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rbulter pushed a commit to branch feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
in repository https://gitbox.apache.org/repos/asf/celix.git
commit 7b8608b30f97b87d4df76a9e3d2ca7a4bb02d424
Author: Roy Bulter <ro...@gmail.com>
AuthorDate: Wed Apr 29 21:25:31 2020 +0200
Merge
---
.../src/pubsub_psa_tcp_constants.h | 5 ++---
.../pubsub_admin_tcp/src/pubsub_tcp_handler.c | 26 ++++++++++++++++------
.../pubsub_admin_tcp/src/pubsub_tcp_handler.h | 6 ++---
.../src/pubsub_tcp_topic_receiver.c | 7 ++----
.../pubsub_admin_tcp/src/pubsub_tcp_topic_sender.c | 2 ++
5 files changed, 27 insertions(+), 19 deletions(-)
diff --git a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_psa_tcp_constants.h b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_psa_tcp_constants.h
index 3e7a7b3..1ca4ff1 100644
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_psa_tcp_constants.h
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_psa_tcp_constants.h
@@ -23,7 +23,7 @@
#define PSA_TCP_BASE_PORT "PSA_TCP_BASE_PORT"
#define PSA_TCP_MAX_PORT "PSA_TCP_MAX_PORT"
-#define PSA_TCP_MAX_RECV_SESSIONS "PSA_TCP_MAX_RECV_SESSIONS"
+#define PSA_TCP_MAX_MESSAGE_SIZE "PSA_TCP_MAX_MESSAGE_SIZE"
#define PSA_TCP_RECV_BUFFER_SIZE "PSA_TCP_RECV_BUFFER_SIZE"
#define PSA_TCP_TIMEOUT "PSA_TCP_TIMEOUT"
#define PSA_TCP_SUBSCRIBER_CONNECTION_TIMEOUT "PSA_TCP_SUBSCRIBER_CONNECTION_TIMEOUT"
@@ -31,8 +31,7 @@
#define PSA_TCP_DEFAULT_BASE_PORT 5501
#define PSA_TCP_DEFAULT_MAX_PORT 6000
-#define PSA_TCP_DEFAULT_MAX_RECV_SESSIONS 1
-
+#define PSA_TCP_DEFAULT_MAX_MESSAGE_SIZE 0
#define PSA_TCP_DEFAULT_RECV_BUFFER_SIZE 65 * 1024
#define PSA_TCP_DEFAULT_TIMEOUT 2000 // 2 seconds
#define PSA_TCP_SUBSCRIBER_CONNECTION_DEFAULT_TIMEOUT 250 // 250 ms
diff --git a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
index 6e2072e..b63e45b 100644
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
@@ -330,6 +330,7 @@ pubsub_tcpHandler_createEntry(pubsub_tcpHandler_t *handle, int fd, char *url, ch
entry->headerBufferSize = size;
handle->protocol->getSyncHeaderSize(handle->protocol->handle, &size);
entry->syncSize = size;
+ entry->maxMsgSize = (handle->maxMsgSize) ? handle->maxMsgSize : UINT32_MAX;
entry->bufferSize = handle->bufferSize;
entry->connected = false;
entry->msg.msg_iov = calloc(sizeof(struct iovec), IOV_MAX);
@@ -628,22 +629,32 @@ int pubsub_tcpHandler_listen(pubsub_tcpHandler_t *handle, char *url) {
}
//
-// Setup buffer sizes
+// Setup receive buffer size
//
-int pubsub_tcpHandler_createReceiveBufferStore(pubsub_tcpHandler_t *handle,
- unsigned int maxNofBuffers
- __attribute__((__unused__)),
- unsigned int bufferSize) {
+int pubsub_tcpHandler_setReceiveBufferSize(pubsub_tcpHandler_t *handle, unsigned int size) {
if (handle != NULL) {
celixThreadRwlock_writeLock(&handle->dbLock);
- handle->bufferSize = bufferSize;
- handle->maxNofBuffer = maxNofBuffers;
+ handle->bufferSize = size;
celixThreadRwlock_unlock(&handle->dbLock);
}
return 0;
}
//
+// Setup receive buffer size
+//
+int pubsub_tcpHandler_setMaxMsgSize(pubsub_tcpHandler_t *handle, unsigned int size) {
+ if (handle != NULL) {
+ celixThreadRwlock_writeLock(&handle->dbLock);
+ handle->maxMsgSize = size;
+ celixThreadRwlock_unlock(&handle->dbLock);
+ }
+ return 0;
+}
+
+
+int pubsub_tcpHandler_setMaxMsgSize(pubsub_tcpHandler_t *handle, unsigned int maxMsgSize);
+//
// Setup thread timeout
//
void pubsub_tcpHandler_setTimeout(pubsub_tcpHandler_t *handle,
@@ -959,6 +970,7 @@ int pubsub_tcpHandler_write(pubsub_tcpHandler_t *handle, pubsub_protocol_message
int nofConnToClose = 0;
if (handle) {
hash_map_iterator_t iter = hashMapIterator_construct(handle->connection_fd_map);
+ size_t max_msg_iov_len = IOV_MAX - 2;
while (hashMapIterator_hasNext(&iter)) {
psa_tcp_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
void *payloadData = NULL;
diff --git a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.h b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.h
index 6f69cd2..34d1838 100644
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.h
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.h
@@ -58,10 +58,8 @@ int pubsub_tcpHandler_close(pubsub_tcpHandler_t *handle, int fd);
int pubsub_tcpHandler_connect(pubsub_tcpHandler_t *handle, char *url);
int pubsub_tcpHandler_disconnect(pubsub_tcpHandler_t *handle, char *url);
int pubsub_tcpHandler_listen(pubsub_tcpHandler_t *handle, char *url);
-
-int pubsub_tcpHandler_createReceiveBufferStore(pubsub_tcpHandler_t *handle,
- unsigned int maxNofBuffers,
- unsigned int bufferSize);
+int pubsub_tcpHandler_setReceiveBufferSize(pubsub_tcpHandler_t *handle, unsigned int size);
+int pubsub_tcpHandler_setMaxMsgSize(pubsub_tcpHandler_t *handle, unsigned int size);
void pubsub_tcpHandler_setTimeout(pubsub_tcpHandler_t *handle, unsigned int timeout);
void pubsub_tcpHandler_setSendRetryCnt(pubsub_tcpHandler_t *handle, unsigned int count);
void pubsub_tcpHandler_setReceiveRetryCnt(pubsub_tcpHandler_t *handle, unsigned int count);
diff --git a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_receiver.c b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_receiver.c
index fca41d2..18d17f4 100644
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_receiver.c
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_receiver.c
@@ -209,14 +209,11 @@ pubsub_tcp_topic_receiver_t *pubsub_tcpTopicReceiver_create(celix_bundle_context
PUBSUB_TCP_SUBSCRIBER_RETRY_CNT_DEFAULT);
double rcvTimeout = celix_properties_getAsDouble(topicProperties, PUBSUB_TCP_SUBSCRIBER_RCVTIMEO_KEY,
PUBSUB_TCP_SUBSCRIBER_RCVTIMEO_DEFAULT);
- long sessions = celix_bundleContext_getPropertyAsLong(ctx, PSA_TCP_MAX_RECV_SESSIONS,
- PSA_TCP_DEFAULT_MAX_RECV_SESSIONS);
- long buffer_size = celix_bundleContext_getPropertyAsLong(ctx, PSA_TCP_RECV_BUFFER_SIZE,
+ long bufferSize = celix_properties_getAsLong(topicProperties, PSA_TCP_RECV_BUFFER_SIZE,
PSA_TCP_DEFAULT_RECV_BUFFER_SIZE);
long timeout = celix_bundleContext_getPropertyAsLong(ctx, PSA_TCP_TIMEOUT, PSA_TCP_DEFAULT_TIMEOUT);
pubsub_tcpHandler_setThreadName(receiver->socketHandler, topic, scope);
- pubsub_tcpHandler_createReceiveBufferStore(receiver->socketHandler, (unsigned int) sessions,
- (unsigned int) buffer_size);
+ pubsub_tcpHandler_setReceiveBufferSize(receiver->socketHandler, (unsigned int) bufferSize);
pubsub_tcpHandler_setTimeout(receiver->socketHandler, (unsigned int) timeout);
pubsub_tcpHandler_addMessageHandler(receiver->socketHandler, receiver, processMsg);
pubsub_tcpHandler_addReceiverConnectionCallback(receiver->socketHandler, receiver, psa_tcp_connectHandler,
diff --git a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_sender.c b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_sender.c
index 7d0f5f7..0f3a83f 100644
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_sender.c
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_topic_sender.c
@@ -191,10 +191,12 @@ pubsub_tcp_topic_sender_t *pubsub_tcpTopicSender_create(
PUBSUB_TCP_PUBLISHER_RETRY_CNT_DEFAULT);
double timeout = celix_properties_getAsDouble(topicProperties, PUBSUB_TCP_PUBLISHER_SNDTIMEO_KEY,
PUBSUB_TCP_PUBLISHER_SNDTIMEO_DEFAULT);
+ long maxMsgSize = celix_properties_getAsLong(topicProperties, PSA_TCP_MAX_MESSAGE_SIZE, PSA_TCP_DEFAULT_MAX_MESSAGE_SIZE);
pubsub_tcpHandler_setThreadName(sender->socketHandler, topic, scope);
pubsub_tcpHandler_setThreadPriority(sender->socketHandler, prio, sched);
pubsub_tcpHandler_setSendRetryCnt(sender->socketHandler, (unsigned int) retryCnt);
pubsub_tcpHandler_setSendTimeOut(sender->socketHandler, timeout);
+ pubsub_tcpHandler_setReceiveBufferSize(sender->socketHandler, (unsigned int) maxMsgSize);
}
//setting up tcp socket for TCP TopicSender
[celix] 02/05: Merge Develop
Posted by rb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rbulter pushed a commit to branch feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
in repository https://gitbox.apache.org/repos/asf/celix.git
commit cfc7373e06ccf12ed689f9df64d6d3541f7c4168
Author: Roy Bulter <ro...@gmail.com>
AuthorDate: Wed Apr 29 20:43:52 2020 +0200
Merge Develop
---
.../pubsub_admin_tcp/src/pubsub_tcp_handler.c | 662 +++++++++++----------
.../pubsub/pubsub_protocol_wire_v1/CMakeLists.txt | 9 +-
.../pubsub/pubsub_protocol_wire_v2/CMakeLists.txt | 12 +-
.../pubsub_protocol_wire_v2/gtest/CMakeLists.txt | 2 +-
bundles/pubsub/test/CMakeLists.txt | 2 +-
5 files changed, 354 insertions(+), 333 deletions(-)
diff --git a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
index 99a5653..31d9d7f 100644
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
@@ -70,56 +70,57 @@
// Entry administration
//
typedef struct psa_tcp_connection_entry {
- char *interface_url;
- char *url;
- int fd;
- struct sockaddr_in addr;
- socklen_t len;
- bool connected;
- pubsub_protocol_message_t header;
- unsigned int syncSize;
- unsigned int headerSize;
- unsigned int headerBufferSize; // Size of headerBuffer, size = 0, no headerBuffer -> included in payload
- void *headerBuffer;
- unsigned int bufferSize;
- void *buffer;
- unsigned int metaBufferSize;
- void *metaBuffer;
- struct msghdr msg;
- size_t msg_iovlen; /* Number of elements in the vector. */
- unsigned int retryCount;
+ char *interface_url;
+ char *url;
+ int fd;
+ struct sockaddr_in addr;
+ socklen_t len;
+ bool connected;
+ pubsub_protocol_message_t header;
+ unsigned int maxMsgSize;
+ unsigned int syncSize;
+ unsigned int headerSize;
+ unsigned int headerBufferSize; // Size of headerBuffer, size = 0, no headerBuffer -> included in payload
+ void *headerBuffer;
+ unsigned int bufferSize;
+ void *buffer;
+ unsigned int metaBufferSize;
+ void *metaBuffer;
+ struct msghdr msg;
+ size_t msg_iovlen; /* Number of elements in the vector. */
+ unsigned int retryCount;
} psa_tcp_connection_entry_t;
//
// Handle administration
//
struct pubsub_tcpHandler {
- unsigned int readSeqNr;
- celix_thread_rwlock_t dbLock;
- unsigned int timeout;
- hash_map_t *connection_url_map;
- hash_map_t *connection_fd_map;
- hash_map_t *interface_url_map;
- hash_map_t *interface_fd_map;
- int efd;
- pubsub_tcpHandler_receiverConnectMessage_callback_t receiverConnectMessageCallback;
- pubsub_tcpHandler_receiverConnectMessage_callback_t receiverDisconnectMessageCallback;
- void *receiverConnectPayload;
- pubsub_tcpHandler_acceptConnectMessage_callback_t acceptConnectMessageCallback;
- pubsub_tcpHandler_acceptConnectMessage_callback_t acceptDisconnectMessageCallback;
- void *acceptConnectPayload;
- pubsub_tcpHandler_processMessage_callback_t processMessageCallback;
- void *processMessagePayload;
- log_helper_t *logHelper;
- pubsub_protocol_service_t *protocol;
- unsigned int bufferSize;
- unsigned int maxNofBuffer;
- unsigned int maxSendRetryCount;
- unsigned int maxRcvRetryCount;
- double sendTimeout;
- double rcvTimeout;
- celix_thread_t thread;
- bool running;
+ celix_thread_rwlock_t dbLock;
+ unsigned int timeout;
+ hash_map_t *connection_url_map;
+ hash_map_t *connection_fd_map;
+ hash_map_t *interface_url_map;
+ hash_map_t *interface_fd_map;
+ int efd;
+ pubsub_tcpHandler_receiverConnectMessage_callback_t receiverConnectMessageCallback;
+ pubsub_tcpHandler_receiverConnectMessage_callback_t receiverDisconnectMessageCallback;
+ void *receiverConnectPayload;
+ pubsub_tcpHandler_acceptConnectMessage_callback_t acceptConnectMessageCallback;
+ pubsub_tcpHandler_acceptConnectMessage_callback_t acceptDisconnectMessageCallback;
+ void *acceptConnectPayload;
+ pubsub_tcpHandler_processMessage_callback_t processMessageCallback;
+ void *processMessagePayload;
+ log_helper_t *logHelper;
+ pubsub_protocol_service_t *protocol;
+ unsigned int bufferSize;
+ unsigned int maxNofBuffer;
+ unsigned int maxMsgSize;
+ unsigned int maxSendRetryCount;
+ unsigned int maxRcvRetryCount;
+ double sendTimeout;
+ double rcvTimeout;
+ celix_thread_t thread;
+ bool running;
};
static inline int
@@ -330,6 +331,7 @@ pubsub_tcpHandler_createEntry(pubsub_tcpHandler_t *handle, int fd, char *url, ch
entry->syncSize = size;
entry->bufferSize = handle->bufferSize;
entry->connected = false;
+ entry->maxMsgSize = (handle->maxMsgSize) ? handle->maxMsgSize : UINT32_MAX;
entry->msg.msg_iov = calloc(sizeof(struct iovec), IOV_MAX);
if (entry->headerBufferSize) {
entry->headerBuffer = calloc(sizeof(char), entry->headerSize);
@@ -420,7 +422,10 @@ int pubsub_tcpHandler_connect(pubsub_tcpHandler_t *handle, char *url) {
if ((rc >= 0) && addr) {
rc = connect(fd, (struct sockaddr *) addr, sizeof(struct sockaddr));
if (rc < 0 && errno != EINPROGRESS) {
- L_ERROR("[TCP Socket] Cannot connect to %s:%d: using; %s err: %s\n", url_info->hostname, url_info->portnr, interface_url,
+ L_ERROR("[TCP Socket] Cannot connect to %s:%d: using; %s err: %s\n",
+ url_info->hostname,
+ url_info->portnr,
+ interface_url,
strerror(errno));
close(fd);
} else {
@@ -434,7 +439,7 @@ int pubsub_tcpHandler_connect(pubsub_tcpHandler_t *handle, char *url) {
#if defined(__APPLE__)
struct kevent ev;
EV_SET (&ev, entry->fd, EVFILT_READ | EVFILT_WRITE, EV_ADD | EV_ENABLE, 0, 0, 0);
- rc = kevent (handle->efd, &ev, 1, NULL, 0, NULL);
+ rc = kevent(handle->efd, &ev, 1, NULL, 0, NULL);
#else
struct epoll_event event;
bzero(&event, sizeof(struct epoll_event)); // zero the struct
@@ -488,9 +493,9 @@ static inline int pubsub_tcpHandler_closeConnectionEntry(
hashMap_remove(handle->connection_fd_map, (void *) (intptr_t) entry->fd);
if ((handle->efd >= 0)) {
#if defined(__APPLE__)
- struct kevent ev;
- EV_SET (&ev, entry->fd, EVFILT_READ, EV_DELETE , 0, 0, 0);
- rc = kevent (handle->efd, &ev, 1, NULL, 0, NULL);
+ struct kevent ev;
+ EV_SET (&ev, entry->fd, EVFILT_READ, EV_DELETE, 0, 0, 0);
+ rc = kevent(handle->efd, &ev, 1, NULL, 0, NULL);
#else
struct epoll_event event;
bzero(&event, sizeof(struct epoll_event)); // zero the struct
@@ -525,8 +530,8 @@ pubsub_tcpHandler_closeInterfaceEntry(pubsub_tcpHandler_t *handle,
if ((handle->efd >= 0)) {
#if defined(__APPLE__)
struct kevent ev;
- EV_SET (&ev, entry->fd, EVFILT_READ, EV_DELETE , 0, 0, 0);
- rc = kevent (handle->efd, &ev, 1, NULL, 0, NULL);
+ EV_SET (&ev, entry->fd, EVFILT_READ, EV_DELETE, 0, 0, 0);
+ rc = kevent(handle->efd, &ev, 1, NULL, 0, NULL);
#else
struct epoll_event event;
bzero(&event, sizeof(struct epoll_event)); // zero the struct
@@ -598,8 +603,8 @@ int pubsub_tcpHandler_listen(pubsub_tcpHandler_t *handle, char *url) {
if ((rc >= 0) && (handle->efd >= 0)) {
#if defined(__APPLE__)
struct kevent ev;
- EV_SET (&ev, fd, EVFILT_READ, EV_ADD | EV_ENABLE , 0, 0, 0);
- rc = kevent (handle->efd, &ev, 1, NULL, 0, NULL);
+ EV_SET (&ev, fd, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, 0);
+ rc = kevent(handle->efd, &ev, 1, NULL, 0, NULL);
#else
struct epoll_event event;
bzero(&event, sizeof(event)); // zero the struct
@@ -691,10 +696,10 @@ void pubsub_tcpHandler_setThreadPriority(pubsub_tcpHandler_t *handle, long prio,
if (strncmp("SCHED_OTHER", sched, 16) == 0) {
policy = SCHED_OTHER;
#if !defined(__APPLE__)
- } else if (strncmp("SCHED_BATCH", sched, 16) == 0) {
- policy = SCHED_BATCH;
- } else if (strncmp("SCHED_IDLE", sched, 16) == 0) {
- policy = SCHED_IDLE;
+ } else if (strncmp("SCHED_BATCH", sched, 16) == 0) {
+ policy = SCHED_BATCH;
+ } else if (strncmp("SCHED_IDLE", sched, 16) == 0) {
+ policy = SCHED_IDLE;
#endif
} else if (strncmp("SCHED_FIFO", sched, 16) == 0) {
policy = SCHED_FIFO;
@@ -830,9 +835,10 @@ int pubsub_tcpHandler_dataAvailable(pubsub_tcpHandler_t *handle, int fd, unsigne
if (entry->headerBufferSize)
entry->msg.msg_iovlen++;
- if (entry->header.header.payloadSize) {
- entry->msg.msg_iov[entry->msg.msg_iovlen].iov_base = entry->buffer;
- entry->msg.msg_iov[entry->msg.msg_iovlen].iov_len = entry->header.header.payloadSize;
+ if (entry->header.header.payloadPartSize) {
+ char *buffer = entry->buffer;
+ entry->msg.msg_iov[entry->msg.msg_iovlen].iov_base = &buffer[entry->header.header.payloadOffset];
+ entry->msg.msg_iov[entry->msg.msg_iovlen].iov_len = entry->header.header.payloadPartSize;
entry->msg.msg_iovlen++;
}
if (entry->header.header.metadataSize) {
@@ -861,9 +867,9 @@ int pubsub_tcpHandler_dataAvailable(pubsub_tcpHandler_t *handle, int fd, unsigne
for (int i = 0; i < entry->msg.msg_iovlen; i++) {
msgSize += entry->msg.msg_iov[i].iov_len;
}
- if (nbytes == msgSize) {
+ if ((nbytes == msgSize) && (entry->header.header.isLastSegment)) {
*readMsg = true;
- } else {
+ } else if (nbytes != msgSize) {
L_ERROR("[TCP Socket] Failed to receive complete message (fd: %d) nbytes : %d = msgSize %d", entry->fd,
nbytes, msgSize);
}
@@ -956,6 +962,7 @@ int pubsub_tcpHandler_write(pubsub_tcpHandler_t *handle, pubsub_protocol_message
int nofConnToClose = 0;
if (handle) {
hash_map_iterator_t iter = hashMapIterator_construct(handle->connection_fd_map);
+ size_t max_msg_iov_len = IOV_MAX - 2;
while (hashMapIterator_hasNext(&iter)) {
psa_tcp_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
void *payloadData = NULL;
@@ -982,6 +989,7 @@ int pubsub_tcpHandler_write(pubsub_tcpHandler_t *handle, pubsub_protocol_message
message->header.metadataSize = metadataSize;
size_t totalMessageSize = payloadSize + metadataSize;
+
bool isMessageSegmentationSupported = false;
handle->protocol->isMessageSegmentationSupported(handle->protocol->handle, &isMessageSegmentationSupported);
if (((!isMessageSegmentationSupported) && (msg_iov_len > max_msg_iov_len)) ||
@@ -1051,319 +1059,331 @@ int pubsub_tcpHandler_write(pubsub_tcpHandler_t *handle, pubsub_protocol_message
message->header.isLastSegment = 0x1;
}
- void *headerData = NULL;
- size_t headerSize = 0;
- // check if header is not part of the payload (=> headerBufferSize = 0)s
- if (entry->headerBufferSize) {
- // Encode the header, with payload size and metadata size
- handle->protocol->encodeHeader(handle->protocol->handle, message,
- &headerData,
- &headerSize);
- if (!entry->headerBufferSize) {
- // Skip header buffer, when header is part of payload;
- msg.msg_iov = &msg_iov[1];
- } else if (headerSize && headerData) {
- // Write header in 1st vector buffer item
- msg.msg_iov[0].iov_base = headerData;
- msg.msg_iov[0].iov_len = headerSize;
- msgSize += msg.msg_iov[0].iov_len;
- msg.msg_iovlen++;
- } else {
- L_ERROR("[TCP Socket] No header buffer is generated");
- msg.msg_iovlen = 0;
- }
- nbytes = 0;
- if (entry->fd >= 0 && msgSize && headerData)
- nbytes = sendmsg(entry->fd, &msg, flags | MSG_NOSIGNAL);
- // When a specific socket keeps reporting errors can indicate a subscriber
- // which is not active anymore, the connection will remain until the retry
- // counter exceeds the maximum retry count.
- // Btw, also, SIGSTOP issued by a debugging tool can result in EINTR error.
- if (nbytes == -1) {
- if (entry->retryCount < handle->maxSendRetryCount) {
- entry->retryCount++;
- L_ERROR(
- "[TCP Socket] Failed to send message (fd: %d), error: %s. try again. Retry count %u of %u, ",
- entry->fd, strerror(errno), entry->retryCount, handle->maxSendRetryCount);
+ void *headerData = NULL;
+ size_t headerSize = 0;
+ // check if header is not part of the payload (=> headerBufferSize = 0)s
+ if (entry->headerBufferSize) {
+ // Encode the header, with payload size and metadata size
+ handle->protocol->encodeHeader(handle->protocol->handle, message,
+ &headerData,
+ &headerSize);
+ }
+ if (!entry->headerBufferSize) {
+ // Skip header buffer, when header is part of payload;
+ msg.msg_iov = &msg_iov[1];
+ } else if (headerSize && headerData) {
+ // Write header in 1st vector buffer item
+ msg.msg_iov[0].iov_base = headerData;
+ msg.msg_iov[0].iov_len = headerSize;
+ msgPartSize += msg.msg_iov[0].iov_len;
+ msg.msg_iovlen++;
} else {
- L_ERROR(
- "[TCP Socket] Failed to send message (fd: %d) after %u retries! Closing connection... Error: %s",
- entry->fd, handle->maxSendRetryCount, strerror(errno));
- connFdCloseQueue[nofConnToClose++] = entry->fd;
+ L_ERROR("[TCP Socket] No header buffer is generated");
+ msg.msg_iovlen = 0;
}
- result = -1; //At least one connection failed sending
- } else if (msgSize) {
- entry->retryCount = 0;
- if (nbytes != msgSize) {
- L_ERROR("[TCP Socket] MsgSize not correct: %d != %d\n", msgSize, nbytes);
+ nbytes = 0;
+ if (entry->fd >= 0 && msgSize && headerData)
+ nbytes = sendmsg(entry->fd, &msg, flags | MSG_NOSIGNAL);
+ // When a specific socket keeps reporting errors can indicate a subscriber
+ // which is not active anymore, the connection will remain until the retry
+ // counter exceeds the maximum retry count.
+ // Btw, also, SIGSTOP issued by a debugging tool can result in EINTR error.
+ if (nbytes == -1) {
+ if (entry->retryCount < handle->maxSendRetryCount) {
+ entry->retryCount++;
+ L_ERROR(
+ "[TCP Socket] Failed to send message (fd: %d), error: %s. try again. Retry count %u of %u, ",
+ entry->fd,
+ strerror(errno),
+ entry->retryCount,
+ handle->maxSendRetryCount);
+ } else {
+ L_ERROR(
+ "[TCP Socket] Failed to send message (fd: %d) after %u retries! Closing connection... Error: %s",
+ entry->fd,
+ handle->maxSendRetryCount,
+ strerror(errno));
+ connFdCloseQueue[nofConnToClose++] = entry->fd;
+ }
+ result = -1; //At least one connection failed sending
+ } else if (msgPartSize) {
+ entry->retryCount = 0;
+ if (nbytes != msgPartSize) {
+ L_ERROR("[TCP Socket] MsgSize not correct: %d != %d\n", msgPartSize, nbytes);
+ }
+ }
+ // Release data
+ if (headerData)
+ free(headerData);
+ // Note: serialized Payload is deleted by serializer
+ if (payloadData && (payloadData != message->payload.payload)) {
+ free(payloadData);
+ }
+ if (metadataData)
+ free(metadataData);
}
}
- // Release data
- if (headerData)
- free(headerData);
- // Note: serialized Payload is deleted by serializer
- if (payloadData && (payloadData != message->payload.payload)) {
- free(payloadData);
- }
- if (metadataData)
- free(metadataData);
}
+ celixThreadRwlock_unlock(&handle->dbLock);
+ //Force close all connections that are queued in a list, done outside of locking handle->dbLock to prevent deadlock
+ for (int i = 0; i < nofConnToClose; i++) {
+ pubsub_tcpHandler_close(handle, connFdCloseQueue[i]);
+ }
+ return result;
}
- celixThreadRwlock_unlock(&handle->dbLock);
- //Force close all connections that are queued in a list, done outside of locking handle->dbLock to prevent deadlock
- for (int i = 0; i < nofConnToClose; i++) {
- pubsub_tcpHandler_close(handle, connFdCloseQueue[i]);
- }
- return result;
-}
//
// get interface URL
//
-char *pubsub_tcpHandler_get_interface_url(pubsub_tcpHandler_t *handle) {
- hash_map_iterator_t interface_iter =
- hashMapIterator_construct(handle->interface_url_map);
- char *url = NULL;
- while (hashMapIterator_hasNext(&interface_iter)) {
- psa_tcp_connection_entry_t *entry =
- hashMapIterator_nextValue(&interface_iter);
- if (entry && entry->url) {
- if (!url) {
- url = celix_utils_strdup(entry->url);
- } else {
- char *tmp = url;
- asprintf(&url, "%s %s", tmp, entry->url);
- free(tmp);
+ char *pubsub_tcpHandler_get_interface_url(pubsub_tcpHandler_t *handle) {
+ hash_map_iterator_t interface_iter =
+ hashMapIterator_construct(handle->interface_url_map);
+ char *url = NULL;
+ while (hashMapIterator_hasNext(&interface_iter)) {
+ psa_tcp_connection_entry_t *entry =
+ hashMapIterator_nextValue(&interface_iter);
+ if (entry && entry->url) {
+ if (!url) {
+ url = celix_utils_strdup(entry->url);
+ } else {
+ char *tmp = url;
+ asprintf(&url, "%s %s", tmp, entry->url);
+ free(tmp);
+ }
}
}
+ return url;
}
- return url;
-}
//
// Handle non-blocking accept (sender)
//
-static inline
-int pubsub_tcpHandler_acceptHandler(pubsub_tcpHandler_t *handle, psa_tcp_connection_entry_t *pendingConnectionEntry) {
- celixThreadRwlock_writeLock(&handle->dbLock);
- // new connection available
- struct sockaddr_in their_addr;
- socklen_t len = sizeof(struct sockaddr_in);
- int fd = accept(pendingConnectionEntry->fd, (struct sockaddr*)&their_addr, &len);
- int rc = fd;
- if (rc == -1) {
- L_ERROR("[TCP Socket] accept failed: %s\n", strerror(errno));
- }
- if (rc >= 0) {
- // handle new connection:
- struct sockaddr_in sin;
- getsockname(pendingConnectionEntry->fd, (struct sockaddr *) &sin, &len);
- char *interface_url = pubsub_utils_url_get_url(&sin, NULL);
- char *url = pubsub_utils_url_get_url(&their_addr, NULL);
- psa_tcp_connection_entry_t *entry = pubsub_tcpHandler_createEntry(handle, fd, url, interface_url, &their_addr);
+ static inline
+ int pubsub_tcpHandler_acceptHandler(pubsub_tcpHandler_t *handle,
+ psa_tcp_connection_entry_t *pendingConnectionEntry) {
+ celixThreadRwlock_writeLock(&handle->dbLock);
+ // new connection available
+ struct sockaddr_in their_addr;
+ socklen_t len = sizeof(struct sockaddr_in);
+ int fd = accept(pendingConnectionEntry->fd, (struct sockaddr *) &their_addr, &len);
+ int rc = fd;
+ if (rc == -1) {
+ L_ERROR("[TCP Socket] accept failed: %s\n", strerror(errno));
+ }
+ if (rc >= 0) {
+ // handle new connection:
+ struct sockaddr_in sin;
+ getsockname(pendingConnectionEntry->fd, (struct sockaddr *) &sin, &len);
+ char *interface_url = pubsub_utils_url_get_url(&sin, NULL);
+ char *url = pubsub_utils_url_get_url(&their_addr, NULL);
+ psa_tcp_connection_entry_t
+ *entry = pubsub_tcpHandler_createEntry(handle, fd, url, interface_url, &their_addr);
#if defined(__APPLE__)
- struct kevent ev;
- EV_SET (&ev, entry->fd, EVFILT_READ, EV_ADD | EV_ENABLE , 0, 0, 0);
- rc = kevent (handle->efd, &ev, 1, NULL, 0, NULL);
+ struct kevent ev;
+ EV_SET (&ev, entry->fd, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, 0);
+ rc = kevent(handle->efd, &ev, 1, NULL, 0, NULL);
#else
- struct epoll_event event;
- bzero(&event, sizeof(event)); // zero the struct
- event.events = EPOLLIN | EPOLLRDHUP | EPOLLERR | EPOLLOUT;
- event.data.fd = entry->fd;
- // Register Read to epoll
- rc = epoll_ctl(handle->efd, EPOLL_CTL_ADD, entry->fd, &event);
+ struct epoll_event event;
+ bzero(&event, sizeof(event)); // zero the struct
+ event.events = EPOLLIN | EPOLLRDHUP | EPOLLERR | EPOLLOUT;
+ event.data.fd = entry->fd;
+ // Register Read to epoll
+ rc = epoll_ctl(handle->efd, EPOLL_CTL_ADD, entry->fd, &event);
#endif
- if (rc < 0) {
- pubsub_tcpHandler_freeEntry(entry);
- free(entry);
- L_ERROR("[TCP Socket] Cannot create epoll\n");
- } else {
- // Call Accept Connection callback
- if (handle->acceptConnectMessageCallback)
- handle->acceptConnectMessageCallback(handle->acceptConnectPayload, url);
- hashMap_put(handle->connection_fd_map, (void *) (intptr_t) entry->fd, entry);
- hashMap_put(handle->connection_url_map, entry->url, entry);
- L_INFO("[TCP Socket] New connection to url: %s: \n", url);
+ if (rc < 0) {
+ pubsub_tcpHandler_freeEntry(entry);
+ free(entry);
+ L_ERROR("[TCP Socket] Cannot create epoll\n");
+ } else {
+ // Call Accept Connection callback
+ if (handle->acceptConnectMessageCallback)
+ handle->acceptConnectMessageCallback(handle->acceptConnectPayload, url);
+ hashMap_put(handle->connection_fd_map, (void *) (intptr_t) entry->fd, entry);
+ hashMap_put(handle->connection_url_map, entry->url, entry);
+ L_INFO("[TCP Socket] New connection to url: %s: \n", url);
+ }
+ free(url);
+ free(interface_url);
}
- free(url);
- free(interface_url);
+ celixThreadRwlock_unlock(&handle->dbLock);
+ return fd;
}
- celixThreadRwlock_unlock(&handle->dbLock);
- return fd;
-}
//
// Handle sockets reads (blocking)
//
-static inline
-void pubsub_tcpHandler_readHandler(pubsub_tcpHandler_t *handle, int fd) {
- unsigned int index = 0;
- bool readMsg = false;
- int rc = pubsub_tcpHandler_dataAvailable(handle, fd, &index, &readMsg);
- if (rc <= 0) {
- // close connection.
- if (rc == 0)
- pubsub_tcpHandler_close(handle, fd);
- return;
- }
- if (readMsg) {
- // Handle data
- pubsub_protocol_message_t *header = NULL;
- rc = pubsub_tcpHandler_read(handle, fd, index, &header);
- if (rc < 0)
+ static inline
+ void pubsub_tcpHandler_readHandler(pubsub_tcpHandler_t *handle, int fd) {
+ unsigned int index = 0;
+ bool readMsg = false;
+ int rc = pubsub_tcpHandler_dataAvailable(handle, fd, &index, &readMsg);
+ if (rc <= 0) {
+ // close connection.
+ if (rc == 0)
+ pubsub_tcpHandler_close(handle, fd);
return;
- celixThreadRwlock_readLock(&handle->dbLock);
- if (handle->processMessageCallback && header != NULL && header->payload.payload != NULL &&
- header->payload.length) {
- struct timespec receiveTime;
- clock_gettime(CLOCK_REALTIME, &receiveTime);
- bool releaseEntryBuffer = false;
- handle->processMessageCallback(handle->processMessagePayload, header, &releaseEntryBuffer, &receiveTime);
- if (releaseEntryBuffer)
- pubsub_tcpHandler_releaseEntryBuffer(handle, fd, index);
}
- celixThreadRwlock_unlock(&handle->dbLock);
+ if (readMsg) {
+ // Handle data
+ pubsub_protocol_message_t *header = NULL;
+ rc = pubsub_tcpHandler_read(handle, fd, index, &header);
+ if (rc < 0)
+ return;
+ celixThreadRwlock_readLock(&handle->dbLock);
+ if (handle->processMessageCallback && header != NULL && header->payload.payload != NULL &&
+ header->payload.length) {
+ struct timespec receiveTime;
+ clock_gettime(CLOCK_REALTIME, &receiveTime);
+ bool releaseEntryBuffer = false;
+ handle->processMessageCallback(handle->processMessagePayload,
+ header,
+ &releaseEntryBuffer,
+ &receiveTime);
+ if (releaseEntryBuffer)
+ pubsub_tcpHandler_releaseEntryBuffer(handle, fd, index);
+ }
+ celixThreadRwlock_unlock(&handle->dbLock);
+ }
}
-}
//
// Handle sockets connection (sender)
//
-static inline
-void pubsub_tcpHandler_connectionHandler(pubsub_tcpHandler_t *handle, int fd) {
- int err = 0;
- socklen_t len = sizeof(int);
- int rc = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len);
- if (rc != 0) {
- L_ERROR("[TCP Socket]:EPOLLOUT ERROR read from socket %s\n", strerror(errno));
- return;
- }
- celixThreadRwlock_readLock(&handle->dbLock);
- psa_tcp_connection_entry_t *entry = hashMap_get(handle->connection_fd_map, (void *) (intptr_t) fd);
- if (entry)
- if ((!entry->connected)) {
- // tell sender that an receiver is connected
- if (handle->receiverConnectMessageCallback)
- handle->receiverConnectMessageCallback(handle->receiverConnectPayload, entry->url, false);
- entry->connected = true;
+ static inline
+ void pubsub_tcpHandler_connectionHandler(pubsub_tcpHandler_t *handle, int fd) {
+ int err = 0;
+ socklen_t len = sizeof(int);
+ int rc = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len);
+ if (rc != 0) {
+ L_ERROR("[TCP Socket]:EPOLLOUT ERROR read from socket %s\n", strerror(errno));
+ return;
}
- celixThreadRwlock_unlock(&handle->dbLock);
-}
+ celixThreadRwlock_readLock(&handle->dbLock);
+ psa_tcp_connection_entry_t *entry = hashMap_get(handle->connection_fd_map, (void *) (intptr_t) fd);
+ if (entry)
+ if ((!entry->connected)) {
+ // tell sender that an receiver is connected
+ if (handle->receiverConnectMessageCallback)
+ handle->receiverConnectMessageCallback(handle->receiverConnectPayload, entry->url, false);
+ entry->connected = true;
+ }
+ celixThreadRwlock_unlock(&handle->dbLock);
+ }
#if defined(__APPLE__)
//
// The main socket event loop
//
-static inline
-void pubsub_tcpHandler_handler(pubsub_tcpHandler_t *handle) {
- int rc = 0;
- if (handle->efd >= 0) {
- int nof_events = 0;
- // Wait for events.
- struct kevent events[MAX_EVENTS];
- struct timespec ts = {handle->timeout / 1000, (handle->timeout % 1000) * 1000000};
- nof_events = kevent (handle->efd, NULL, 0, &events[0], MAX_EVENTS, handle->timeout ? &ts : NULL);
- if (nof_events < 0) {
- if ((errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) {
- } else
- L_ERROR("[TCP Socket] Cannot create poll wait (%d) %s\n", nof_events, strerror(errno));
- }
- for (int i = 0; i < nof_events; i++) {
- hash_map_iterator_t iter = hashMapIterator_construct(handle->interface_fd_map);
- psa_tcp_connection_entry_t *pendingConnectionEntry = NULL;
- while (hashMapIterator_hasNext(&iter)) {
- psa_tcp_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (events[i].ident == entry->fd)
- pendingConnectionEntry = entry;
- }
- if (pendingConnectionEntry) {
- int fd = pubsub_tcpHandler_acceptHandler(handle, pendingConnectionEntry);
- pubsub_tcpHandler_connectionHandler(handle, fd);
- } else if (events[i].filter & EVFILT_READ) {
- pubsub_tcpHandler_readHandler(handle, events[i].ident);
- } else if (events[i].flags & EV_EOF) {
- int err = 0;
- socklen_t len = sizeof(int);
- rc = getsockopt(events[i].ident, SOL_SOCKET, SO_ERROR, &err, &len);
- if (rc != 0) {
- L_ERROR("[TCP Socket]:EPOLLRDHUP ERROR read from socket %s\n", strerror(errno));
- continue;
+ static inline
+ void pubsub_tcpHandler_handler(pubsub_tcpHandler_t *handle) {
+ int rc = 0;
+ if (handle->efd >= 0) {
+ int nof_events = 0;
+ // Wait for events.
+ struct kevent events[MAX_EVENTS];
+ struct timespec ts = {handle->timeout / 1000, (handle->timeout % 1000) * 1000000};
+ nof_events = kevent(handle->efd, NULL, 0, &events[0], MAX_EVENTS, handle->timeout ? &ts : NULL);
+ if (nof_events < 0) {
+ if ((errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) {
+ } else
+ L_ERROR("[TCP Socket] Cannot create poll wait (%d) %s\n", nof_events, strerror(errno));
+ }
+ for (int i = 0; i < nof_events; i++) {
+ hash_map_iterator_t iter = hashMapIterator_construct(handle->interface_fd_map);
+ psa_tcp_connection_entry_t *pendingConnectionEntry = NULL;
+ while (hashMapIterator_hasNext(&iter)) {
+ psa_tcp_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
+ if (events[i].ident == entry->fd)
+ pendingConnectionEntry = entry;
+ }
+ if (pendingConnectionEntry) {
+ int fd = pubsub_tcpHandler_acceptHandler(handle, pendingConnectionEntry);
+ pubsub_tcpHandler_connectionHandler(handle, fd);
+ } else if (events[i].filter & EVFILT_READ) {
+ pubsub_tcpHandler_readHandler(handle, events[i].ident);
+ } else if (events[i].flags & EV_EOF) {
+ int err = 0;
+ socklen_t len = sizeof(int);
+ rc = getsockopt(events[i].ident, SOL_SOCKET, SO_ERROR, &err, &len);
+ if (rc != 0) {
+ L_ERROR("[TCP Socket]:EPOLLRDHUP ERROR read from socket %s\n", strerror(errno));
+ continue;
+ }
+ pubsub_tcpHandler_close(handle, events[i].ident);
+ } else if (events[i].flags & EV_ERROR) {
+ L_ERROR("[TCP Socket]:EPOLLERR ERROR read from socket %s\n", strerror(errno));
+ pubsub_tcpHandler_close(handle, events[i].ident);
+ continue;
+ }
+ }
}
- pubsub_tcpHandler_close(handle, events[i].ident);
- } else if (events[i].flags & EV_ERROR) {
- L_ERROR("[TCP Socket]:EPOLLERR ERROR read from socket %s\n", strerror(errno));
- pubsub_tcpHandler_close(handle, events[i].ident);
- continue;
- }
+ return;
}
- }
- return;
-}
#else
-//
-// The main socket event loop
-//
-static inline
-void pubsub_tcpHandler_handler(pubsub_tcpHandler_t *handle) {
- int rc = 0;
- if (handle->efd >= 0) {
- int nof_events = 0;
- struct epoll_event events[MAX_EVENTS];
- nof_events = epoll_wait(handle->efd, events, MAX_EVENTS, handle->timeout);
- if (nof_events < 0) {
- if ((errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) {
- } else
- L_ERROR("[TCP Socket] Cannot create epoll wait (%d) %s\n", nof_events, strerror(errno));
- }
- for (int i = 0; i < nof_events; i++) {
- hash_map_iterator_t iter = hashMapIterator_construct(handle->interface_fd_map);
- psa_tcp_connection_entry_t *pendingConnectionEntry = NULL;
- while (hashMapIterator_hasNext(&iter)) {
- psa_tcp_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (events[i].data.fd == entry->fd)
- pendingConnectionEntry = entry;
+ //
+ // The main socket event loop
+ //
+ static inline
+ void pubsub_tcpHandler_handler(pubsub_tcpHandler_t *handle) {
+ int rc = 0;
+ if (handle->efd >= 0) {
+ int nof_events = 0;
+ struct epoll_event events[MAX_EVENTS];
+ nof_events = epoll_wait(handle->efd, events, MAX_EVENTS, handle->timeout);
+ if (nof_events < 0) {
+ if ((errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) {
+ } else
+ L_ERROR("[TCP Socket] Cannot create epoll wait (%d) %s\n", nof_events, strerror(errno));
}
- if (pendingConnectionEntry) {
- int fd = pubsub_tcpHandler_acceptHandler(handle, pendingConnectionEntry);
- pubsub_tcpHandler_connectionHandler(handle, fd);
- } else if (events[i].events & EPOLLIN) {
- pubsub_tcpHandler_readHandler(handle, events[i].data.fd);
- } else if (events[i].events & EPOLLRDHUP) {
- int err = 0;
- socklen_t len = sizeof(int);
- rc = getsockopt(events[i].data.fd, SOL_SOCKET, SO_ERROR, &err, &len);
- if (rc != 0) {
- L_ERROR("[TCP Socket]:EPOLLRDHUP ERROR read from socket %s\n", strerror(errno));
+ for (int i = 0; i < nof_events; i++) {
+ hash_map_iterator_t iter = hashMapIterator_construct(handle->interface_fd_map);
+ psa_tcp_connection_entry_t *pendingConnectionEntry = NULL;
+ while (hashMapIterator_hasNext(&iter)) {
+ psa_tcp_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
+ if (events[i].data.fd == entry->fd)
+ pendingConnectionEntry = entry;
+ }
+ if (pendingConnectionEntry) {
+ int fd = pubsub_tcpHandler_acceptHandler(handle, pendingConnectionEntry);
+ pubsub_tcpHandler_connectionHandler(handle, fd);
+ } else if (events[i].events & EPOLLIN) {
+ pubsub_tcpHandler_readHandler(handle, events[i].data.fd);
+ } else if (events[i].events & EPOLLRDHUP) {
+ int err = 0;
+ socklen_t len = sizeof(int);
+ rc = getsockopt(events[i].data.fd, SOL_SOCKET, SO_ERROR, &err, &len);
+ if (rc != 0) {
+ L_ERROR("[TCP Socket]:EPOLLRDHUP ERROR read from socket %s\n", strerror(errno));
+ continue;
+ }
+ pubsub_tcpHandler_close(handle, events[i].data.fd);
+ } else if (events[i].events & EPOLLERR) {
+ L_ERROR("[TCP Socket]:EPOLLERR ERROR read from socket %s\n", strerror(errno));
+ pubsub_tcpHandler_close(handle, events[i].data.fd);
continue;
}
- pubsub_tcpHandler_close(handle, events[i].data.fd);
- } else if (events[i].events & EPOLLERR) {
- L_ERROR("[TCP Socket]:EPOLLERR ERROR read from socket %s\n", strerror(errno));
- pubsub_tcpHandler_close(handle, events[i].data.fd);
- continue;
}
}
+ return;
}
- return;
-}
#endif
//
// The socket thread
//
-static void *pubsub_tcpHandler_thread(void *data) {
- pubsub_tcpHandler_t *handle = data;
- celixThreadRwlock_readLock(&handle->dbLock);
- bool running = handle->running;
- celixThreadRwlock_unlock(&handle->dbLock);
-
- while (running) {
- pubsub_tcpHandler_handler(handle);
+ static void *pubsub_tcpHandler_thread(void *data) {
+ pubsub_tcpHandler_t *handle = data;
celixThreadRwlock_readLock(&handle->dbLock);
- running = handle->running;
+ bool running = handle->running;
celixThreadRwlock_unlock(&handle->dbLock);
- } // while
- return NULL;
-}
\ No newline at end of file
+
+ while (running) {
+ pubsub_tcpHandler_handler(handle);
+ celixThreadRwlock_readLock(&handle->dbLock);
+ running = handle->running;
+ celixThreadRwlock_unlock(&handle->dbLock);
+ } // while
+ return NULL;
+ }
\ No newline at end of file
diff --git a/bundles/pubsub/pubsub_protocol_wire_v1/CMakeLists.txt b/bundles/pubsub/pubsub_protocol_wire_v1/CMakeLists.txt
index 6df3805..793ade8 100644
--- a/bundles/pubsub/pubsub_protocol_wire_v1/CMakeLists.txt
+++ b/bundles/pubsub/pubsub_protocol_wire_v1/CMakeLists.txt
@@ -15,12 +15,12 @@
# specific language governing permissions and limitations
# under the License.
-add_library(celix_protocol_wire_v1_impl STATIC
+add_library(celix_wire_protocol_v1_impl STATIC
src/pubsub_wire_protocol_impl.c
src/pubsub_wire_protocol_common.c
)
-target_include_directories(celix_protocol_wire_v1_impl PRIVATE src)
-target_link_libraries(celix_protocol_wire_v1_impl PRIVATE Celix::pubsub_spi Celix::pubsub_utils)
+target_include_directories(celix_wire_protocol_v1_impl PRIVATE src)
+target_link_libraries(celix_wire_protocol_v1_impl PRIVATE Celix::pubsub_spi )
add_celix_bundle(celix_pubsub_protocol_wire_v1
@@ -32,7 +32,8 @@ add_celix_bundle(celix_pubsub_protocol_wire_v1
)
target_include_directories(celix_pubsub_protocol_wire_v1 PRIVATE src)
target_link_libraries(celix_pubsub_protocol_wire_v1 PRIVATE Celix::pubsub_spi Celix::pubsub_utils)
-target_link_libraries(celix_pubsub_protocol_wire_v1 PRIVATE celix_protocol_wire_v1_impl)
+target_link_libraries(celix_pubsub_protocol_wire_v1 PRIVATE celix_wire_protocol_v1_impl)
+
install_celix_bundle(celix_pubsub_protocol_wire_v1 EXPORT celix COMPONENT pubsub)
add_library(Celix::pubsub_protocol_wire_v1 ALIAS celix_pubsub_protocol_wire_v1)
diff --git a/bundles/pubsub/pubsub_protocol_wire_v2/CMakeLists.txt b/bundles/pubsub/pubsub_protocol_wire_v2/CMakeLists.txt
index 637929d..3db78d9 100644
--- a/bundles/pubsub/pubsub_protocol_wire_v2/CMakeLists.txt
+++ b/bundles/pubsub/pubsub_protocol_wire_v2/CMakeLists.txt
@@ -15,12 +15,12 @@
# specific language governing permissions and limitations
# under the License.
-add_library(celix_protocol_wire_v2_impl STATIC
+add_library(celix_wire_protocol_v2_impl STATIC
src/pubsub_wire_v2_protocol_impl.c
src/pubsub_wire_v2_protocol_common.c)
-target_include_directories(celix_protocol_wire_v2_impl PRIVATE src)
-target_link_libraries(celix_protocol_wire_v2_impl PRIVATE Celix::pubsub_spi Celix::pubsub_utils)
-target_include_directories(celix_protocol_wire_v2_impl PRIVATE src ${CMAKE_SOURCE_DIR}/bundles/pubsub/pubsub_protocol_wire_v1/src )
+target_include_directories(celix_wire_protocol_v2_impl PRIVATE src)
+target_link_libraries(celix_wire_protocol_v2_impl PRIVATE Celix::pubsub_spi Celix::pubsub_utils)
+target_include_directories(celix_wire_protocol_v2_impl PRIVATE src ${CMAKE_SOURCE_DIR}/bundles/pubsub/pubsub_protocol_wire_v1/src )
add_celix_bundle(celix_pubsub_protocol_wire_v2
BUNDLE_SYMBOLICNAME "apache_celix_pubsub_protocol_wire_v2"
@@ -32,8 +32,8 @@ add_celix_bundle(celix_pubsub_protocol_wire_v2
target_include_directories(celix_pubsub_protocol_wire_v2 PRIVATE src ${CMAKE_SOURCE_DIR}/bundles/pubsub/pubsub_protocol_wire_v1/src)
target_link_libraries(celix_pubsub_protocol_wire_v2 PRIVATE Celix::pubsub_spi Celix::pubsub_utils)
-target_link_libraries(celix_pubsub_protocol_wire_v2 PRIVATE celix_protocol_wire_v2_impl)
-target_link_libraries(celix_pubsub_protocol_wire_v2 PRIVATE celix_protocol_wire_v1_impl)
+target_link_libraries(celix_pubsub_protocol_wire_v2 PRIVATE celix_wire_protocol_v2_impl)
+target_link_libraries(celix_pubsub_protocol_wire_v2 PRIVATE celix_wire_protocol_v1_impl)
install_celix_bundle(celix_pubsub_protocol_wire_v2 EXPORT celix COMPONENT pubsub)
diff --git a/bundles/pubsub/pubsub_protocol_wire_v2/gtest/CMakeLists.txt b/bundles/pubsub/pubsub_protocol_wire_v2/gtest/CMakeLists.txt
index 682891a..afc165a 100644
--- a/bundles/pubsub/pubsub_protocol_wire_v2/gtest/CMakeLists.txt
+++ b/bundles/pubsub/pubsub_protocol_wire_v2/gtest/CMakeLists.txt
@@ -22,7 +22,7 @@ set(SOURCES
add_executable(celix_pswp_v2_tests ${SOURCES})
#target_include_directories(celix_cxx_pswp_tests SYSTEM PRIVATE gtest)
target_include_directories(celix_pswp_v2_tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../src)
-target_link_libraries(celix_pswp_v2_tests PRIVATE celix_protocol_wire_v2_impl GTest::GTest Celix::pubsub_spi)
+target_link_libraries(celix_pswp_v2_tests PRIVATE celix_wire_protocol_v2_impl GTest::GTest Celix::pubsub_spi)
add_test(NAME celix_pswp_v2_tests COMMAND celix_pswp_v2_tests)
SETUP_TARGET_FOR_COVERAGE(celix_pswp_v2_tests_cov celix_cxx_pswp_v2_tests ${CMAKE_BINARY_DIR}/coverage/bundles/pubsub/protocol_wire_v2 ..)
diff --git a/bundles/pubsub/test/CMakeLists.txt b/bundles/pubsub/test/CMakeLists.txt
index 0763d4c..8361ea0 100644
--- a/bundles/pubsub/test/CMakeLists.txt
+++ b/bundles/pubsub/test/CMakeLists.txt
@@ -141,7 +141,7 @@ if (BUILD_PUBSUB_PSA_TCP)
Celix::pubsub_serializer_json
Celix::pubsub_topology_manager
Celix::pubsub_admin_tcp
- Celix::pubsub_protocol_wire_v2
+ Celix::pubsub_protocol_wire_v1
pubsub_sut
pubsub_tst
)
[celix] 01/05: Merge branch 'feature/fix_tcp_endpoint_handling'
into feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
Posted by rb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rbulter pushed a commit to branch feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
in repository https://gitbox.apache.org/repos/asf/celix.git
commit 543a12f0bd14f3e90a80379e88bb932a52a88be0
Merge: 6fb06e1 57c3f5a
Author: Roy Bulter <ro...@gmail.com>
AuthorDate: Wed Apr 29 13:14:30 2020 +0200
Merge branch 'feature/fix_tcp_endpoint_handling' into feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
# Conflicts:
# bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
.../pubsub_admin_tcp/src/pubsub_tcp_handler.c | 119 +++++++++++----------
.../src/pubsub_tcp_topic_receiver.c | 15 +--
libs/framework/src/framework.c | 5 +
libs/utils/private/test/celix_threads_test.cpp | 1 +
libs/utils/src/utils.c | 4 +-
5 files changed, 78 insertions(+), 66 deletions(-)
diff --cc bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
index 24e1202,0e31222..99a5653
--- a/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
+++ b/bundles/pubsub/pubsub_admin_tcp/src/pubsub_tcp_handler.c
@@@ -979,131 -980,100 +980,133 @@@ int pubsub_tcpHandler_write(pubsub_tcpH
&metadataSize);
}
message->header.metadataSize = metadataSize;
+ size_t totalMessageSize = payloadSize + metadataSize;
+
+ bool isMessageSegmentationSupported = false;
+ handle->protocol->isMessageSegmentationSupported(handle->protocol->handle, &isMessageSegmentationSupported);
+ if (((!isMessageSegmentationSupported) && (msg_iov_len > max_msg_iov_len)) ||
+ ((!isMessageSegmentationSupported) && (totalMessageSize > entry->maxMsgSize))) {
+ L_WARN("[TCP Socket] Failed to send message (fd: %d), Message segmentation is not supported\n",
+ entry->fd);
+ continue;
+ }
size_t msgSize = 0;
- long int nbytes = 0;
- struct msghdr msg;
- struct iovec msg_iov[IOV_MAX];
- msg.msg_name = &entry->addr;
- msg.msg_namelen = entry->len;
- msg.msg_flags = flags;
- msg.msg_control = NULL;
- msg.msg_controllen = 0;
- msg.msg_iovlen = 0;
- msg.msg_iov = msg_iov;
+ size_t msgIovLen = 0;
+ long int nbytes = UINT32_MAX;
+ while (msgSize < totalMessageSize && nbytes > 0) {
+ struct msghdr msg;
+ struct iovec msg_iov[IOV_MAX];
+ msg.msg_name = &entry->addr;
+ msg.msg_namelen = entry->len;
+ msg.msg_flags = flags;
+ msg.msg_control = NULL;
+ msg.msg_controllen = 0;
+ msg.msg_iovlen = 0;
+ msg.msg_iov = msg_iov;
+ size_t msgPartSize = 0;
+ message->header.payloadPartSize = 0;
+ message->header.payloadOffset = 0;
+ message->header.metadataSize = 0;
+ message->header.isLastSegment = 0;
- // Write generic seralized payload in vector buffer
- if (payloadSize && payloadData) {
- msg.msg_iovlen++;
- msg.msg_iov[msg.msg_iovlen].iov_base = payloadData;
- msg.msg_iov[msg.msg_iovlen].iov_len = payloadSize;
- msgSize += msg.msg_iov[msg.msg_iovlen].iov_len;
- } else {
- // copy serialized vector into vector buffer
- for (size_t i = 0; i < MIN(msg_iov_len, IOV_MAX - 2); i++) {
+ // Write generic seralized payload in vector buffer
+ if (payloadSize && payloadData) {
+ char *payloadDataBuffer = payloadData;
msg.msg_iovlen++;
- msg.msg_iov[msg.msg_iovlen].iov_base = msgIoVec[i].iov_base;
- msg.msg_iov[msg.msg_iovlen].iov_len = msgIoVec[i].iov_len;
+ msg.msg_iov[msg.msg_iovlen].iov_base = &payloadDataBuffer[msgSize];
+ msg.msg_iov[msg.msg_iovlen].iov_len = MIN((payloadSize - msgSize), entry->maxMsgSize);
+ msgPartSize += msg.msg_iov[msg.msg_iovlen].iov_len;
+ message->header.payloadPartSize = msgPartSize;
+ message->header.payloadOffset = msgSize;
msgSize += msg.msg_iov[msg.msg_iovlen].iov_len;
+ } else {
+ // copy serialized vector into vector buffer
+ for (size_t i = 0; i < MIN(msg_iov_len, max_msg_iov_len); i++) {
+ msg.msg_iovlen++;
+ msg.msg_iov[msg.msg_iovlen].iov_base = msgIoVec[msgIovLen + i].iov_base;
+ msg.msg_iov[msg.msg_iovlen].iov_len = msgIoVec[msgIovLen + i].iov_len;
+ if ((msgPartSize + msg.msg_iov[msg.msg_iovlen].iov_len) > entry->maxMsgSize)
+ break;
+ msgSize += msg.msg_iov[msg.msg_iovlen].iov_len;
+ }
+ message->header.payloadPartSize = msgPartSize;
+ message->header.payloadOffset = msgSize;
+ msgSize += msgPartSize;
+ msgIovLen += (msg.msg_iovlen - 1);
}
- }
- // Write optional metadata in vector buffer
- if (metadataSize && metadataData) {
- msg.msg_iovlen++;
- msg.msg_iov[msg.msg_iovlen].iov_base = metadataData;
- msg.msg_iov[msg.msg_iovlen].iov_len = metadataSize;
- msgSize += msg.msg_iov[msg.msg_iovlen].iov_len;
- }
+ // Write optional metadata in vector buffer
+ if ((msgSize < (payloadSize + metadataSize)) &&
+ (msgPartSize < entry->maxMsgSize) &&
+ (metadataSize && metadataData)) {
+ msg.msg_iovlen++;
+ msg.msg_iov[msg.msg_iovlen].iov_base = metadataData;
+ msg.msg_iov[msg.msg_iovlen].iov_len = metadataSize;
+ msgPartSize += msg.msg_iov[msg.msg_iovlen].iov_len;
+ message->header.metadataSize = metadataSize;
+ msgSize += metadataSize;
+ }
+ if (msgSize >= totalMessageSize) {
+ message->header.isLastSegment = 0x1;
+ }
- void *headerData = NULL;
- size_t headerSize = 0;
- // Encode the header, with payload size and metadata size
- handle->protocol->encodeHeader(handle->protocol->handle, message,
- &headerData,
- &headerSize);
+ void *headerData = NULL;
+ size_t headerSize = 0;
+ // check if header is not part of the payload (=> headerBufferSize = 0)s
+ if (entry->headerBufferSize) {
+ // Encode the header, with payload size and metadata size
+ handle->protocol->encodeHeader(handle->protocol->handle, message,
- &headerData,
- &headerSize);
- }
- if (!entry->headerBufferSize) {
- // Skip header buffer, when header is part of payload;
- msg.msg_iov = &msg_iov[1];
- } else if (headerSize && headerData) {
- // Write header in 1st vector buffer item
++ &headerData,
++ &headerSize);
++ if (!entry->headerBufferSize) {
++ // Skip header buffer, when header is part of payload;
++ msg.msg_iov = &msg_iov[1];
++ } else if (headerSize && headerData) {
+ // Write header in 1st vector buffer item
- if (headerSize && headerData) {
- msg.msg_iov[0].iov_base = headerData;
- msg.msg_iov[0].iov_len = headerSize;
- msgPartSize += msg.msg_iov[0].iov_len;
- msg.msg_iovlen++;
- }
- nbytes = 0;
- if (entry->fd >= 0 && msgSize && headerData)
- nbytes = sendmsg(entry->fd, &msg, MSG_NOSIGNAL | flags);
- // When a specific socket keeps reporting errors can indicate a subscriber
- // which is not active anymore, the connection will remain until the retry
- // counter exceeds the maximum retry count.
- // Btw, also, SIGSTOP issued by a debugging tool can result in EINTR error.
- if (nbytes == -1) {
- if (entry->retryCount < handle->maxSendRetryCount) {
- entry->retryCount++;
- L_ERROR(
- "[TCP Socket] Failed to send message (fd: %d), error: %s. try again. Retry count %u of %u, ",
- entry->fd,
- strerror(errno),
- entry->retryCount,
- handle->maxSendRetryCount);
- } else {
- L_ERROR(
- "[TCP Socket] Failed to send message (fd: %d) after %u retries! Closing connection... Error: %s",
- entry->fd,
- handle->maxSendRetryCount,
- strerror(errno));
- connFdCloseQueue[nofConnToClose++] = entry->fd;
- }
- result = -1; //At least one connection failed sending
- } else if (msgPartSize) {
- entry->retryCount = 0;
- if (nbytes != msgPartSize) {
- L_ERROR("[TCP Socket] MsgSize not correct: %d != %d\n", msgPartSize, nbytes);
- }
+ msg.msg_iov[0].iov_base = headerData;
+ msg.msg_iov[0].iov_len = headerSize;
+ msgSize += msg.msg_iov[0].iov_len;
+ msg.msg_iovlen++;
- } else {
- L_ERROR("[TCP Socket] No header buffer is generated");
- msg.msg_iovlen = 0;
- }
++ } else {
++ L_ERROR("[TCP Socket] No header buffer is generated");
++ msg.msg_iovlen = 0;
++ }
+ nbytes = 0;
- if (entry->fd >= 0 && msgSize && msg.msg_iovlen)
++ if (entry->fd >= 0 && msgSize && headerData)
+ nbytes = sendmsg(entry->fd, &msg, flags | MSG_NOSIGNAL);
+ // When a specific socket keeps reporting errors can indicate a subscriber
+ // which is not active anymore, the connection will remain until the retry
+ // counter exceeds the maximum retry count.
+ // Btw, also, SIGSTOP issued by a debugging tool can result in EINTR error.
+ if (nbytes == -1) {
+ if (entry->retryCount < handle->maxSendRetryCount) {
+ entry->retryCount++;
+ L_ERROR(
+ "[TCP Socket] Failed to send message (fd: %d), error: %s. try again. Retry count %u of %u, ",
+ entry->fd, strerror(errno), entry->retryCount, handle->maxSendRetryCount);
+ } else {
+ L_ERROR(
+ "[TCP Socket] Failed to send message (fd: %d) after %u retries! Closing connection... Error: %s",
+ entry->fd, handle->maxSendRetryCount, strerror(errno));
+ connFdCloseQueue[nofConnToClose++] = entry->fd;
}
- // Release data
- if (headerData)
- free(headerData);
- // Note: serialized Payload is deleted by serializer
- if (payloadData && (payloadData != message->payload.payload)) {
- free(payloadData);
+ result = -1; //At least one connection failed sending
+ } else if (msgSize) {
+ entry->retryCount = 0;
+ if (nbytes != msgSize) {
+ L_ERROR("[TCP Socket] MsgSize not correct: %d != %d\n", msgSize, nbytes);
}
- if (metadataData)
- free(metadataData);
}
+ // Release data
+ if (headerData)
+ free(headerData);
+ // Note: serialized Payload is deleted by serializer
+ if (payloadData && (payloadData != message->payload.payload)) {
+ free(payloadData);
+ }
+ if (metadataData)
+ free(metadataData);
}
}
celixThreadRwlock_unlock(&handle->dbLock);
[celix] 05/05: Fix wire 2
Posted by rb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rbulter pushed a commit to branch feature/add_msg_segemenation_to_tcp_admin_with_wire_v2
in repository https://gitbox.apache.org/repos/asf/celix.git
commit 209a081c7921d8465f1f5ffdcc3c6aaf258e18bb
Author: Roy Bulter <ro...@gmail.com>
AuthorDate: Wed Apr 29 21:58:22 2020 +0200
Fix wire 2
---
.../pubsub/pubsub_protocol_wire_v2/src/pubsub_wire_v2_protocol_impl.h | 2 +-
bundles/pubsub/test/CMakeLists.txt | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/bundles/pubsub/pubsub_protocol_wire_v2/src/pubsub_wire_v2_protocol_impl.h b/bundles/pubsub/pubsub_protocol_wire_v2/src/pubsub_wire_v2_protocol_impl.h
index 6a1b748..93724fc 100644
--- a/bundles/pubsub/pubsub_protocol_wire_v2/src/pubsub_wire_v2_protocol_impl.h
+++ b/bundles/pubsub/pubsub_protocol_wire_v2/src/pubsub_wire_v2_protocol_impl.h
@@ -26,7 +26,7 @@
extern "C" {
#endif
-#define PUBSUB_WIRE_V2_PROTOCOL_TYPE "wire_v2"
+#define PUBSUB_WIRE_V2_PROTOCOL_TYPE "envelope-v2"
typedef struct pubsub_protocol_wire_v2 pubsub_protocol_wire_v2_t;
diff --git a/bundles/pubsub/test/CMakeLists.txt b/bundles/pubsub/test/CMakeLists.txt
index 8361ea0..0763d4c 100644
--- a/bundles/pubsub/test/CMakeLists.txt
+++ b/bundles/pubsub/test/CMakeLists.txt
@@ -141,7 +141,7 @@ if (BUILD_PUBSUB_PSA_TCP)
Celix::pubsub_serializer_json
Celix::pubsub_topology_manager
Celix::pubsub_admin_tcp
- Celix::pubsub_protocol_wire_v1
+ Celix::pubsub_protocol_wire_v2
pubsub_sut
pubsub_tst
)