You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celix.apache.org by pn...@apache.org on 2020/07/02 08:27:24 UTC
[celix] branch master updated: Bugfix/pstm deadlock work outside of
locks (#265)
This is an automated email from the ASF dual-hosted git repository.
pnoltes pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/celix.git
The following commit(s) were added to refs/heads/master by this push:
new 69209ca Bugfix/pstm deadlock work outside of locks (#265)
69209ca is described below
commit 69209ca164748b80827cf29c0a7df5c74aa6a699
Author: Pepijn Noltes <pe...@gmail.com>
AuthorDate: Thu Jul 2 10:27:16 2020 +0200
Bugfix/pstm deadlock work outside of locks (#265)
* Adds deadlock tests to other pubsub admins.
* Refactors usage locking in pstm so that recursive lock are not needed.
Co-authored-by: Michael de Lang <ki...@gmail.com>
---
.../src/pubsub_topology_manager.c | 447 +++++++++++++--------
.../src/pubsub_topology_manager.h | 14 +-
bundles/pubsub/test/CMakeLists.txt | 148 ++++++-
.../test/meta_data/deadlock.scope.properties | 28 ++
.../test/meta_data/deadlock.scope2.properties | 28 ++
.../pubsub/test/pstm_deadlock_test/test_runner.cc | 142 +++++++
bundles/pubsub/test/test/test_endpoint_runner.cc | 4 +-
bundles/pubsub/test/test/test_runner.cc | 4 +-
libs/framework/src/dm_dependency_manager_impl.c | 4 +-
9 files changed, 647 insertions(+), 172 deletions(-)
diff --git a/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c b/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c
index 2b063ef..8397a67 100644
--- a/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c
+++ b/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c
@@ -60,12 +60,7 @@ celix_status_t pubsub_topologyManager_create(celix_bundle_context_t *context, ce
manager->context = context;
- celix_thread_mutexattr_t psaAttr;
- celixThreadMutexAttr_create(&psaAttr);
- celixThreadMutexAttr_settype(&psaAttr, CELIX_THREAD_MUTEX_RECURSIVE);
- status |= celixThreadMutex_create(&manager->pubsubadmins.mutex, &psaAttr);
- celixThreadMutexAttr_destroy(&psaAttr);
-
+ status |= celixThreadMutex_create(&manager->pubsubadmins.mutex, NULL);
status |= celixThreadMutex_create(&manager->discoveredEndpoints.mutex, NULL);
status |= celixThreadMutex_create(&manager->announceEndpointListeners.mutex, NULL);
status |= celixThreadMutex_create(&manager->topicReceivers.mutex, NULL);
@@ -184,7 +179,8 @@ void pubsub_topologyManager_psaAdded(void *handle, void *svc, const celix_proper
long svcId = celix_properties_getAsLong(props, OSGI_FRAMEWORK_SERVICE_ID, -1L);
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_DEBUG, "PSTM: Added PSA");
+ const char* psaType = celix_properties_get(props, PUBSUB_ADMIN_SERVICE_TYPE, "!Error!");
+ celix_logHelper_debug(manager->loghelper, "Added %s PSA", psaType);
if (svcId >= 0) {
celixThreadMutex_lock(&manager->pubsubadmins.mutex);
@@ -194,14 +190,15 @@ void pubsub_topologyManager_psaAdded(void *handle, void *svc, const celix_proper
//NOTE new psa, so no endpoints announce yet
- //new PSA -> every topic sender/receiver entry needs a rematch
+ //Because this is a new PSA every topic sender/receiver entry needs a rematch.
+ //This is needed because the new PSA can be a better match than currently used.
int needsRematchCount = 0;
celixThreadMutex_lock(&manager->topicSenders.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(manager->topicSenders.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- entry->needsMatch = true;
+ entry->matching.needsMatch = true;
++needsRematchCount;
}
celixThreadMutex_unlock(&manager->topicSenders.mutex);
@@ -209,14 +206,14 @@ void pubsub_topologyManager_psaAdded(void *handle, void *svc, const celix_proper
iter = hashMapIterator_construct(manager->topicReceivers.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- entry->needsMatch = true;
+ entry->matching.needsMatch = true;
++needsRematchCount;
}
celixThreadMutex_unlock(&manager->topicReceivers.mutex);
if (needsRematchCount > 0) {
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_INFO,
- "A PSA is added after at least one active publisher/provided. \
+ celix_logHelper_info(manager->loghelper,
+ "A new PSA is added after at least one active publisher/provided. \
It is preferred that all PSA are started before publiser/subscriber are started!\n\
Current topic/sender count is %i", needsRematchCount);
}
@@ -227,6 +224,8 @@ void pubsub_topologyManager_psaRemoved(void *handle, void *svc __attribute__((un
pubsub_topology_manager_t *manager = handle;
//pubsub_admin_service_t *psa = (pubsub_admin_service_t*) svc;
long svcId = celix_properties_getAsLong(props, OSGI_FRAMEWORK_SERVICE_ID, -1L);
+ const char* psaType = celix_properties_get(props, PUBSUB_ADMIN_SERVICE_TYPE, "!Error!");
+ celix_logHelper_debug(manager->loghelper, "Removing %s PSA", psaType);
// Remove the svcId from the hashmap, because the service is not available
celixThreadMutex_lock(&manager->pubsubadmins.mutex);
@@ -248,30 +247,21 @@ void pubsub_topologyManager_psaRemoved(void *handle, void *svc __attribute__((un
//de-setup all topic receivers/senders for the removed psa.
//the next psaHandling run will try to find new psa.
+ celix_array_list_t* revokedEndpoints = celix_arrayList_create();
celixThreadMutex_lock(&manager->topicSenders.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(manager->topicSenders.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (entry->selectedPsaSvcId == svcId) {
- /* de-announce all senders */
+ if (entry->matching.selectedPsaSvcId == svcId) {
if (entry->endpoint != NULL) {
- celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
- for (int j = 0; j < celix_arrayList_size(manager->announceEndpointListeners.list); ++j) {
- pubsub_announce_endpoint_listener_t *listener;
- listener = celix_arrayList_get(manager->announceEndpointListeners.list, j);
- listener->revokeEndpoint(listener->handle, entry->endpoint);
- }
- celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+ celix_arrayList_add(revokedEndpoints, entry->endpoint);
}
- entry->needsMatch = true;
- entry->selectedSerializerSvcId = -1L;
- entry->selectedProtocolSvcId = -1L;
- entry->selectedPsaSvcId = -1L;
- if (entry->endpoint != NULL) {
- celix_properties_destroy(entry->endpoint);
- entry->endpoint = NULL;
- }
+ entry->matching.needsMatch = true;
+ entry->matching.selectedSerializerSvcId = -1L;
+ entry->matching.selectedProtocolSvcId = -1L;
+ entry->matching.selectedPsaSvcId = -1L;
+ entry->endpoint = NULL;
}
}
celixThreadMutex_unlock(&manager->topicSenders.mutex);
@@ -280,32 +270,33 @@ void pubsub_topologyManager_psaRemoved(void *handle, void *svc __attribute__((un
iter = hashMapIterator_construct(manager->topicReceivers.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (entry->selectedPsaSvcId == svcId) {
- /* de-announce all receivers */
+ if (entry->matching.selectedPsaSvcId == svcId) {
if (entry->endpoint != NULL) {
- celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
- for (int j = 0; j < celix_arrayList_size(manager->announceEndpointListeners.list); ++j) {
- pubsub_announce_endpoint_listener_t *listener;
- listener = celix_arrayList_get(manager->announceEndpointListeners.list, j);
- listener->revokeEndpoint(listener->handle, entry->endpoint);
- }
- celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+ celix_arrayList_add(revokedEndpoints, entry->endpoint);
}
- entry->needsMatch = true;
- entry->selectedSerializerSvcId = -1L;
- entry->selectedProtocolSvcId = -1L;
- entry->selectedPsaSvcId = -1L;
- if (entry->endpoint != NULL) {
- celix_properties_destroy(entry->endpoint);
- entry->endpoint = NULL;
- }
+ entry->matching.needsMatch = true;
+ entry->matching.selectedSerializerSvcId = -1L;
+ entry->matching.selectedProtocolSvcId = -1L;
+ entry->matching.selectedPsaSvcId = -1L;
+ entry->endpoint = NULL;
}
}
celixThreadMutex_unlock(&manager->topicReceivers.mutex);
-
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_DEBUG, "PSTM: Removed PSA");
+ /* de-announce all senders & receiver endpoints */
+ for (int i = 0; i < celix_arrayList_size(revokedEndpoints); ++i) {
+ celix_properties_t* endpoint = celix_arrayList_get(revokedEndpoints, i);
+ celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
+ for (int j = 0; j < celix_arrayList_size(manager->announceEndpointListeners.list); ++j) {
+ pubsub_announce_endpoint_listener_t *listener;
+ listener = celix_arrayList_get(manager->announceEndpointListeners.list, j);
+ listener->revokeEndpoint(listener->handle, endpoint);
+ }
+ celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+ celix_properties_destroy(endpoint);
+ }
+ celix_arrayList_destroy(revokedEndpoints);
}
void pubsub_topologyManager_subscriberAdded(void *handle, void *svc __attribute__((unused)), const celix_properties_t *props, const celix_bundle_t *bnd) {
@@ -319,11 +310,12 @@ void pubsub_topologyManager_subscriberAdded(void *handle, void *svc __attribute_
const char *topic = celix_properties_get(props, PUBSUB_SUBSCRIBER_TOPIC, NULL);
const char *scope = celix_properties_get(props, PUBSUB_SUBSCRIBER_SCOPE, NULL);
if (topic == NULL) {
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_WARNING,
- "[PSTM] Warning found subscriber service without mandatory '%s' property.",
+ celix_logHelper_warning(manager->loghelper,
+ "Warning found subscriber service without mandatory '%s' property.",
PUBSUB_SUBSCRIBER_TOPIC);
return;
}
+ celix_logHelper_trace(manager->loghelper, "Adding subscriber %s/%s", scope, topic);
long bndId = celix_bundle_getId(bnd);
char *scopeAndTopicKey = NULL;
@@ -340,12 +332,13 @@ void pubsub_topologyManager_subscriberAdded(void *handle, void *svc __attribute_
entry->scope = scope == NULL ? NULL : strndup(scope, 1024 * 1024);
entry->topic = strndup(topic, 1024 * 1024);
entry->usageCount = 1;
- entry->selectedPsaSvcId = -1L;
- entry->selectedSerializerSvcId = -1L;
- entry->needsMatch = true;
+ entry->matching.selectedPsaSvcId = -1L;
+ entry->matching.selectedSerializerSvcId = -1L;
+ entry->matching.needsMatch = true;
entry->bndId = bndId;
entry->subscriberProperties = celix_properties_copy(props);
hashMap_put(manager->topicReceivers.map, entry->scopeAndTopicKey, entry);
+ celix_logHelper_trace(manager->loghelper, "Created new topic receiver entry %s", entry->scopeAndTopicKey);
}
//signal psa handling thread
bool triggerCondition = (entry->usageCount == 1);
@@ -370,6 +363,7 @@ void pubsub_topologyManager_subscriberRemoved(void *handle, void *svc __attribut
if (topic == NULL) {
return;
}
+ celix_logHelper_trace(manager->loghelper, "Removing subscriber %s/%s", scope, topic);
char *scopeAndTopicKey = pubsubEndpoint_createScopeTopicKey(scope, topic);
celixThreadMutex_lock(&manager->topicReceivers.mutex);
@@ -387,15 +381,22 @@ void pubsub_topologyManager_pubsubAnnounceEndpointListenerAdded(void *handle, vo
pubsub_topology_manager_t *manager = (pubsub_topology_manager_t *) handle;
pubsub_announce_endpoint_listener_t *listener = svc;
+ celix_logHelper_trace(manager->loghelper, "Added EndpointListener");
+
+
//1) retroactively call announceEndpoint for already existing endpoints (manager->announcedEndpoints)
//2) Add listener to manager->announceEndpointListeners
+ celix_array_list_t* announceEndpoints = celix_arrayList_create();
+
+ celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
+
celixThreadMutex_lock(&manager->topicSenders.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(manager->topicSenders.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
if (entry != NULL && entry->endpoint != NULL) {
- listener->announceEndpoint(listener->handle, entry->endpoint);
+ celix_arrayList_add(announceEndpoints, celix_properties_copy(entry->endpoint));
}
}
celixThreadMutex_unlock(&manager->topicSenders.mutex);
@@ -405,14 +406,20 @@ void pubsub_topologyManager_pubsubAnnounceEndpointListenerAdded(void *handle, vo
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
if (entry != NULL && entry->endpoint != NULL) {
- listener->announceEndpoint(listener->handle, entry->endpoint);
+ celix_arrayList_add(announceEndpoints, celix_properties_copy(entry->endpoint));
}
}
celixThreadMutex_unlock(&manager->topicReceivers.mutex);
- celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
celix_arrayList_add(manager->announceEndpointListeners.list, listener);
celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+
+ for (int i = 0; i < celix_arrayList_size(announceEndpoints); ++i) {
+ celix_properties_t* endpoint = celix_arrayList_get(announceEndpoints, i);
+ listener->announceEndpoint(listener->handle, endpoint);
+ celix_properties_destroy(endpoint);
+ }
+ celix_arrayList_destroy(announceEndpoints);
}
@@ -420,8 +427,7 @@ void pubsub_topologyManager_pubsubAnnounceEndpointListenerRemoved(void *handle,
pubsub_topology_manager_t *manager = (pubsub_topology_manager_t *) handle;
pubsub_announce_endpoint_listener_t *listener = svc;
- //1) Remove listener from manager->announceEndpointListeners
- //2) call removeEndpoint for already existing endpoints (manager->announcedEndpoints)
+ celix_logHelper_trace(manager->loghelper, "Removing EndpointListener");
celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
celix_arrayList_remove(manager->announceEndpointListeners.list, listener);
@@ -450,6 +456,8 @@ void pubsub_topologyManager_publisherTrackerAdded(void *handle, const celix_serv
PUBSUB_SUBSCRIBER_TOPIC);
return;
}
+ celix_logHelper_trace(manager->loghelper, "Adding new request for publisher %s/%s", scope, topic);
+
scopeAndTopicKey = pubsubEndpoint_createScopeTopicKey(scope, topic);
celixThreadMutex_lock(&manager->topicSenders.mutex);
@@ -464,16 +472,17 @@ void pubsub_topologyManager_publisherTrackerAdded(void *handle, const celix_serv
} else {
entry = calloc(1, sizeof(*entry));
entry->usageCount = 1;
- entry->selectedSerializerSvcId = -1L;
- entry->selectedProtocolSvcId = -1L;
- entry->selectedPsaSvcId = -1L;
+ entry->matching.needsMatch = true;
+ entry->matching.selectedSerializerSvcId = -1L;
+ entry->matching.selectedProtocolSvcId = -1L;
+ entry->matching.selectedPsaSvcId = -1L;
entry->scope = scope; //taking ownership
entry->topic = topic; //taking ownership
entry->scopeAndTopicKey = scopeAndTopicKey; //taking ownership
- entry->needsMatch = true;
entry->publisherFilter = celix_filter_create(info->filter->filterStr);
entry->bndId = info->bundleId;
hashMap_put(manager->topicSenders.map, entry->scopeAndTopicKey, entry);
+ celix_logHelper_trace(manager->loghelper, "Created new topic sender entry %s", entry->scopeAndTopicKey);
}
//new entry -> wakeup psaHandling thread
bool triggerCondition = (entry->usageCount == 1);
@@ -504,6 +513,7 @@ void pubsub_topologyManager_publisherTrackerRemoved(void *handle, const celix_se
}
return;
}
+ celix_logHelper_trace(manager->loghelper, "Removing request for publisher %s/%s", scope, topic);
char *scopeAndTopicKey = pubsubEndpoint_createScopeTopicKey(scope, topic);
@@ -534,8 +544,8 @@ celix_status_t pubsub_topologyManager_addDiscoveredEndpoint(void *handle, const
bool triggerCondition = false;
if (manager->verbose) {
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_DEBUG,
- "PSTM: Discovered endpoint added for topic %s with scope %s [fwUUID=%s, epUUID=%s]\n",
+ celix_logHelper_trace(manager->loghelper,
+ "Adding discovered endpoint for topic %s with scope %s [fwUUID=%s, epUUID=%s]\n",
celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_NAME, NULL),
celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_SCOPE, "(null)"),
celix_properties_get(endpoint, PUBSUB_ENDPOINT_FRAMEWORK_UUID, NULL),
@@ -556,6 +566,7 @@ celix_status_t pubsub_topologyManager_addDiscoveredEndpoint(void *handle, const
entry->uuid = celix_properties_get(entry->endpoint, PUBSUB_ENDPOINT_UUID, NULL);
entry->selectedPsaSvcId = -1L; //NOTE not selected a psa yet
hashMap_put(manager->discoveredEndpoints.map, (void *) entry->uuid, entry);
+ celix_logHelper_trace(manager->loghelper, "Created new discovered endpoint entry %s", uuid);
//waking up psa handling thread to select psa
triggerCondition = true;
@@ -589,8 +600,8 @@ celix_status_t pubsub_topologyManager_removeDiscoveredEndpoint(void *handle, con
assert(uuid != NULL); //discovery should check if endpoint is valid -> pubsubEndoint_isValid.
if (manager->verbose) {
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_DEBUG,
- "PSTM: Discovered endpoint removed for topic %s with scope %s [fwUUID=%s, epUUID=%s]\n",
+ celix_logHelper_trace(manager->loghelper,
+ "Removing discovered endpoint for topic %s with scope %s [fwUUID=%s, epUUID=%s]\n",
celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_NAME, "(null)"),
celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_SCOPE, "(null)"),
celix_properties_get(endpoint, PUBSUB_ENDPOINT_FRAMEWORK_UUID, "(null)"),
@@ -619,6 +630,7 @@ celix_status_t pubsub_topologyManager_removeDiscoveredEndpoint(void *handle, con
} else {
celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_DEBUG, "No selected psa for endpoint %s\n", entry->uuid);
}
+ celix_logHelper_trace(manager->loghelper, "Destroying discovered endpoint entry %s", uuid);
celix_properties_destroy(entry->endpoint);
free(entry);
}
@@ -627,39 +639,42 @@ celix_status_t pubsub_topologyManager_removeDiscoveredEndpoint(void *handle, con
return CELIX_SUCCESS;
}
+struct pstm_teardown_entry {
+ char* scope;
+ char* topic;
+ long psaSvcId;
+};
static void pstm_teardownTopicSenderCallback(void *handle, void *svc) {
- pstm_topic_receiver_or_sender_entry_t *entry = handle;
+ struct pstm_teardown_entry* entry = handle;
pubsub_admin_service_t *psa = svc;
psa->teardownTopicSender(psa->handle, entry->scope, entry->topic);
}
+//Note called on pstm update thread
static void pstm_teardownTopicSenders(pubsub_topology_manager_t *manager) {
+ celix_array_list_t* revokeEndpoints = celix_arrayList_create();
+ celix_array_list_t* teardownEntries = celix_arrayList_create();
+
celixThreadMutex_lock(&manager->topicSenders.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(manager->topicSenders.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (entry != NULL && (entry->usageCount <= 0 || entry->needsMatch)) {
+ if (entry != NULL && (entry->usageCount <= 0 || entry->matching.needsMatch)) {
if (manager->verbose && entry->endpoint != NULL) {
celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_DEBUG,
- "[PSTM] Tearing down TopicSender for scope/topic %s/%s\n", entry->scope == NULL ? "(null)" : entry->scope, entry->topic);
+ "Tearing down TopicSender for scope/topic %s/%s\n", entry->scope == NULL ? "(null)" : entry->scope, entry->topic);
}
-
if (entry->endpoint != NULL) {
- celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
- for (int i = 0; i < celix_arrayList_size(manager->announceEndpointListeners.list); ++i) {
- pubsub_announce_endpoint_listener_t *listener;
- listener = celix_arrayList_get(manager->announceEndpointListeners.list, i);
- listener->revokeEndpoint(listener->handle, entry->endpoint);
- }
- celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
- celix_bundleContext_useServiceWithId(manager->context, entry->selectedPsaSvcId,
- PUBSUB_ADMIN_SERVICE_NAME,
- entry, pstm_teardownTopicSenderCallback);
+ celix_arrayList_add(revokeEndpoints, celix_properties_copy(entry->endpoint));
+ struct pstm_teardown_entry* teardownEntry = malloc(sizeof(*teardownEntry));
+ teardownEntry->scope = celix_utils_strdup(entry->scope);
+ teardownEntry->topic = celix_utils_strdup(entry->topic);
+ teardownEntry->psaSvcId = entry->matching.selectedPsaSvcId;
+ celix_arrayList_add(teardownEntries, teardownEntry);
}
-
//cleanup entry
if (entry->usageCount <= 0) {
//no usage -> remove
@@ -685,46 +700,71 @@ static void pstm_teardownTopicSenders(pubsub_topology_manager_t *manager) {
celix_properties_destroy(entry->endpoint);
}
entry->endpoint = NULL;
- entry->selectedPsaSvcId = -1L;
- entry->selectedSerializerSvcId = -1L;
- entry->selectedProtocolSvcId = -1L;
+ entry->matching.selectedPsaSvcId = -1L;
+ entry->matching.selectedSerializerSvcId = -1L;
+ entry->matching.selectedProtocolSvcId = -1L;
}
}
}
celixThreadMutex_unlock(&manager->topicSenders.mutex);
+
+
+
+ celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
+ for (int i = 0; i < celix_arrayList_size(manager->announceEndpointListeners.list); ++i) {
+ pubsub_announce_endpoint_listener_t *listener;
+ listener = celix_arrayList_get(manager->announceEndpointListeners.list, i);
+ for (int k = 0; k < celix_arrayList_size(revokeEndpoints); ++k) {
+ celix_properties_t* endpoint = celix_arrayList_get(revokeEndpoints, k);
+ listener->revokeEndpoint(listener->handle, endpoint);
+ celix_properties_destroy(endpoint);
+ }
+ }
+ celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+ celix_arrayList_destroy(revokeEndpoints);
+
+ for (int i = 0; i < celix_arrayList_size(teardownEntries); ++i) {
+ struct pstm_teardown_entry* entry = celix_arrayList_get(teardownEntries, i);
+ celix_bundleContext_useServiceWithId(manager->context, entry->psaSvcId,
+ PUBSUB_ADMIN_SERVICE_NAME,
+ entry, pstm_teardownTopicSenderCallback);
+ free(entry->scope);
+ free(entry->topic);
+ free(entry);
+ }
+ celix_arrayList_destroy(teardownEntries);
}
static void pstm_teardownTopicReceiverCallback(void *handle, void *svc) {
- pstm_topic_receiver_or_sender_entry_t *entry = handle;
+ struct pstm_teardown_entry* entry = handle;
pubsub_admin_service_t *psa = svc;
psa->teardownTopicReceiver(psa->handle, entry->scope, entry->topic);
}
static void pstm_teardownTopicReceivers(pubsub_topology_manager_t *manager) {
+ celix_array_list_t* revokeEndpoints = celix_arrayList_create();
+ celix_array_list_t* teardownEntries = celix_arrayList_create();
+
celixThreadMutex_lock(&manager->topicReceivers.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(manager->topicReceivers.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (entry != NULL && (entry->usageCount <= 0 || entry->needsMatch)) {
+ if (entry != NULL && (entry->usageCount <= 0 || entry->matching.needsMatch)) {
if (manager->verbose && entry->endpoint != NULL) {
const char *adminType = celix_properties_get(entry->endpoint, PUBSUB_ENDPOINT_ADMIN_TYPE, "!Error!");
const char *serType = celix_properties_get(entry->endpoint, PUBSUB_ENDPOINT_SERIALIZER, "!Error!");
celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_DEBUG,
- "[PSTM] Tearing down TopicReceiver for scope/topic %s/%s with psa admin type %s and serializer %s\n",
+ "Tearing down TopicReceiver for scope/topic %s/%s with psa admin type %s and serializer %s\n",
entry->scope == NULL ? "(null)" : entry->scope, entry->topic, adminType, serType);
}
if (entry->endpoint != NULL) {
- celix_bundleContext_useServiceWithId(manager->context, entry->selectedPsaSvcId,
- PUBSUB_ADMIN_SERVICE_NAME, entry,
- pstm_teardownTopicReceiverCallback);
- celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
- for (int i = 0; i < celix_arrayList_size(manager->announceEndpointListeners.list); ++i) {
- pubsub_announce_endpoint_listener_t *listener = celix_arrayList_get(
- manager->announceEndpointListeners.list, i);
- listener->revokeEndpoint(listener->handle, entry->endpoint);
- }
- celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+ celix_arrayList_add(revokeEndpoints, celix_properties_copy(entry->endpoint));
+ struct pstm_teardown_entry* teardownEntry = malloc(sizeof(*teardownEntry));
+ teardownEntry->scope = celix_utils_strdup(entry->scope);
+ teardownEntry->topic = celix_utils_strdup(entry->topic);
+ teardownEntry->psaSvcId = entry->matching.selectedPsaSvcId;
+ celix_arrayList_add(teardownEntries, teardownEntry);
}
if (entry->usageCount <= 0) {
@@ -752,13 +792,37 @@ static void pstm_teardownTopicReceivers(pubsub_topology_manager_t *manager) {
celix_properties_destroy(entry->endpoint);
}
entry->endpoint = NULL;
- entry->selectedPsaSvcId = -1L;
- entry->selectedSerializerSvcId = -1L;
- entry->selectedProtocolSvcId = -1L;
+ entry->matching.selectedPsaSvcId = -1L;
+ entry->matching.selectedSerializerSvcId = -1L;
+ entry->matching.selectedProtocolSvcId = -1L;
}
}
}
celixThreadMutex_unlock(&manager->topicReceivers.mutex);
+
+ celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
+ for (int i = 0; i < celix_arrayList_size(manager->announceEndpointListeners.list); ++i) {
+ pubsub_announce_endpoint_listener_t *listener;
+ listener = celix_arrayList_get(manager->announceEndpointListeners.list, i);
+ for (int k = 0; k < celix_arrayList_size(revokeEndpoints); ++k) {
+ celix_properties_t* endpoint = celix_arrayList_get(revokeEndpoints, k);
+ listener->revokeEndpoint(listener->handle, endpoint);
+ celix_properties_destroy(endpoint);
+ }
+ }
+ celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+ celix_arrayList_destroy(revokeEndpoints);
+
+ for (int i = 0; i < celix_arrayList_size(teardownEntries); ++i) {
+ struct pstm_teardown_entry* entry = celix_arrayList_get(teardownEntries, i);
+ celix_bundleContext_useServiceWithId(manager->context, entry->psaSvcId,
+ PUBSUB_ADMIN_SERVICE_NAME,
+ entry, pstm_teardownTopicReceiverCallback);
+ free(entry->scope);
+ free(entry->topic);
+ free(entry);
+ }
+ celix_arrayList_destroy(teardownEntries);
}
static void pstm_addEndpointCallback(void *handle, void *svc) {
@@ -782,6 +846,7 @@ static void pstm_findPsaForEndpoints(pubsub_topology_manager_t *manager) {
pubsub_admin_service_t *psa = hashMapEntry_getValue(mapEntry);
long svcId = (long) hashMapEntry_getKey(mapEntry);
bool match = false;
+ //NOTE assuming match is safe to call within a lock
psa->matchDiscoveredEndpoint(psa->handle, entry->endpoint, &match);
if (match) {
psaSvcId = svcId;
@@ -791,6 +856,7 @@ static void pstm_findPsaForEndpoints(pubsub_topology_manager_t *manager) {
celixThreadMutex_unlock(&manager->pubsubadmins.mutex);
if (psaSvcId >= 0) {
+ //NOTE assuming adding discovered endpoint is safe to call within a lock
celix_bundleContext_useServiceWithId(manager->context, psaSvcId, PUBSUB_ADMIN_SERVICE_NAME,
(void *) entry->endpoint, pstm_addEndpointCallback);
} else {
@@ -803,19 +869,31 @@ static void pstm_findPsaForEndpoints(pubsub_topology_manager_t *manager) {
celixThreadMutex_unlock(&manager->discoveredEndpoints.mutex);
}
+struct pstm_setup_entry {
+ char* scope;
+ char* topic;
+ char* key;
+ celix_properties_t* topicProperties;
+ long selectedSerializerSvcId;
+ long selectedProtocolSvcId;
+ long psaSvcId;
+ celix_properties_t* endpointResult;
+};
static void pstm_setupTopicSenderCallback(void *handle, void *svc) {
- pstm_topic_receiver_or_sender_entry_t *entry = handle;
+ struct pstm_setup_entry *entry = handle;
pubsub_admin_service_t *psa = svc;
- psa->setupTopicSender(psa->handle, entry->scope, entry->topic, entry->topicProperties, entry->selectedSerializerSvcId, entry->selectedProtocolSvcId, &entry->endpoint);
+ psa->setupTopicSender(psa->handle, entry->scope, entry->topic, entry->topicProperties, entry->selectedSerializerSvcId, entry->selectedProtocolSvcId, &entry->endpointResult);
}
static void pstm_setupTopicSenders(pubsub_topology_manager_t *manager) {
+ celix_array_list_t* setupEntries = celix_arrayList_create();
+
celixThreadMutex_lock(&manager->topicSenders.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(manager->topicSenders.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (entry != NULL && entry->needsMatch && entry->usageCount > 0) {
+ if (entry != NULL && entry->matching.needsMatch && entry->usageCount > 0) {
//new topic sender needed, requesting match with current psa
double highestScore = PUBSUB_ADMIN_NO_MATCH_SCORE;
long serializerSvcId = -1L;
@@ -833,7 +911,9 @@ static void pstm_setupTopicSenders(pubsub_topology_manager_t *manager) {
long serSvcId = -1L;
long protSvcId = -1L;
celix_properties_t *topicProps = NULL;
- psa->matchPublisher(psa->handle, entry->bndId, entry->publisherFilter, &topicProps, &score, &serSvcId, &protSvcId);
+ //NOTE assuming matchPublisher is safe to call within lock
+ psa->matchPublisher(psa->handle, entry->bndId, entry->publisherFilter, &topicProps, &score, &serSvcId,
+ &protSvcId);
if (score > highestScore) {
if (topicPropertiesForHighestMatch != NULL) {
celix_properties_destroy(topicPropertiesForHighestMatch);
@@ -850,45 +930,73 @@ static void pstm_setupTopicSenders(pubsub_topology_manager_t *manager) {
celixThreadMutex_unlock(&manager->pubsubadmins.mutex);
if (highestScore > PUBSUB_ADMIN_NO_MATCH_SCORE) {
- entry->selectedPsaSvcId = selectedPsaSvcId;
- entry->selectedSerializerSvcId = serializerSvcId;
- entry->selectedProtocolSvcId = protocolSvcId;
- entry->topicProperties = topicPropertiesForHighestMatch;
- bool called = celix_bundleContext_useServiceWithId(manager->context, selectedPsaSvcId, PUBSUB_ADMIN_SERVICE_NAME, entry, pstm_setupTopicSenderCallback);
-
- if (called && entry->endpoint != NULL) {
- entry->needsMatch = false;
-
- //announce new endpoint through the network
- celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
- for (int i = 0; i < celix_arrayList_size(manager->announceEndpointListeners.list); ++i) {
- pubsub_announce_endpoint_listener_t *listener = celix_arrayList_get(manager->announceEndpointListeners.list, i);
- listener->announceEndpoint(listener->handle, entry->endpoint);
- }
- celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
- }
+ entry->matching.needsMatch = false;
+ //NOTE the needsMatch can be updated as soon as the lock manager->topicSenders.mutex is released.
+ //This is ok, because the next pstm update loop will then do a new teardown/setup
+
+ struct pstm_setup_entry *setupEntry = malloc(sizeof(*setupEntry));
+ setupEntry->endpointResult = NULL;
+ setupEntry->scope = celix_utils_strdup(entry->scope);
+ setupEntry->topic = celix_utils_strdup(entry->topic);
+ setupEntry->key = pubsubEndpoint_createScopeTopicKey(entry->scope, entry->topic);
+ setupEntry->topicProperties = topicPropertiesForHighestMatch;
+ setupEntry->psaSvcId = selectedPsaSvcId;
+ setupEntry->selectedSerializerSvcId = serializerSvcId;
+ setupEntry->selectedProtocolSvcId = protocolSvcId;
+ celix_arrayList_add(setupEntries, setupEntry);
}
+ }
+ }
+ celixThreadMutex_unlock(&manager->topicSenders.mutex);
+
+ for (int i = 0; i < celix_arrayList_size(setupEntries); ++i) {
+ struct pstm_setup_entry* setupEntry = celix_arrayList_get(setupEntries, i);
+ bool called = celix_bundleContext_useServiceWithId(manager->context, setupEntry->psaSvcId, PUBSUB_ADMIN_SERVICE_NAME, setupEntry, pstm_setupTopicSenderCallback);
+ if (called && setupEntry->endpointResult != NULL) {
+ celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
+ for (int k = 0; k < celix_arrayList_size(manager->announceEndpointListeners.list); ++k) {
+ pubsub_announce_endpoint_listener_t *listener = celix_arrayList_get(manager->announceEndpointListeners.list, k);
+ listener->announceEndpoint(listener->handle, setupEntry->endpointResult);
+ }
+ celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
- if (entry->needsMatch) {
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_WARNING, "Cannot setup TopicSender for %s/%s\n", entry->scope == NULL ? "(null)" : entry->scope, entry->topic);
+ celixThreadMutex_lock(&manager->topicSenders.mutex);
+ pstm_topic_receiver_or_sender_entry_t* entry = hashMap_get(manager->topicSenders.map, setupEntry->key);
+ if (entry->endpoint != NULL) {
+ celix_properties_destroy(entry->endpoint);
}
+ entry->endpoint = setupEntry->endpointResult;
+ entry->topicProperties = setupEntry->topicProperties;
+ entry->matching.selectedPsaSvcId = setupEntry->psaSvcId;
+ entry->matching.selectedSerializerSvcId = setupEntry->selectedSerializerSvcId;
+ entry->matching.selectedProtocolSvcId = setupEntry->selectedProtocolSvcId;
+ celixThreadMutex_unlock(&manager->topicSenders.mutex);
+ } else {
+ celix_logHelper_warning(manager->loghelper, "Cannot setup TopicSender for %s/%s\n", setupEntry->scope == NULL ? "(null)" : setupEntry->scope, setupEntry->topic);
}
+ free(setupEntry->scope);
+ free(setupEntry->topic);
+ free(setupEntry->key);
+ free(setupEntry);
}
- celixThreadMutex_unlock(&manager->topicSenders.mutex);
+
+ celix_arrayList_destroy(setupEntries);
}
static void pstm_setupTopicReceiverCallback(void *handle, void *svc) {
- pstm_topic_receiver_or_sender_entry_t *entry = handle;
+ struct pstm_setup_entry *entry = handle;
pubsub_admin_service_t *psa = svc;
- psa->setupTopicReceiver(psa->handle, entry->scope, entry->topic, entry->topicProperties, entry->selectedSerializerSvcId, entry->selectedProtocolSvcId, &entry->endpoint);
+ psa->setupTopicReceiver(psa->handle, entry->scope, entry->topic, entry->topicProperties, entry->selectedSerializerSvcId, entry->selectedProtocolSvcId, &entry->endpointResult);
}
static void pstm_setupTopicReceivers(pubsub_topology_manager_t *manager) {
+ celix_array_list_t* setupEntries = celix_arrayList_create();
+
celixThreadMutex_lock(&manager->topicReceivers.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(manager->topicReceivers.map);
while (hashMapIterator_hasNext(&iter)) {
pstm_topic_receiver_or_sender_entry_t *entry = hashMapIterator_nextValue(&iter);
- if (entry != NULL && entry->needsMatch && entry->usageCount > 0) {
+ if (entry != NULL && entry->matching.needsMatch && entry->usageCount > 0) {
double highestScore = PUBSUB_ADMIN_NO_MATCH_SCORE;
long serializerSvcId = -1L;
@@ -907,7 +1015,8 @@ static void pstm_setupTopicReceivers(pubsub_topology_manager_t *manager) {
long protSvcId = -1L;
celix_properties_t *topicProps = NULL;
- psa->matchSubscriber(psa->handle, entry->bndId, entry->subscriberProperties, &topicProps, &score, &serSvcId, &protSvcId);
+ psa->matchSubscriber(psa->handle, entry->bndId, entry->subscriberProperties, &topicProps, &score,
+ &serSvcId, &protSvcId);
if (score > highestScore) {
if (highestMatchTopicProperties != NULL) {
celix_properties_destroy(highestMatchTopicProperties);
@@ -924,35 +1033,59 @@ static void pstm_setupTopicReceivers(pubsub_topology_manager_t *manager) {
celixThreadMutex_unlock(&manager->pubsubadmins.mutex);
if (highestScore > PUBSUB_ADMIN_NO_MATCH_SCORE) {
- entry->selectedPsaSvcId = selectedPsaSvcId;
- entry->selectedSerializerSvcId = serializerSvcId;
- entry->selectedProtocolSvcId = protocolSvcId;
- entry->topicProperties = highestMatchTopicProperties;
-
- bool called = celix_bundleContext_useServiceWithId(manager->context, selectedPsaSvcId, PUBSUB_ADMIN_SERVICE_NAME,
- entry,
- pstm_setupTopicReceiverCallback);
-
- if (called && entry->endpoint != NULL) {
- entry->needsMatch = false;
- //announce new endpoint through the network
- celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
- for (int i = 0; i < celix_arrayList_size(manager->announceEndpointListeners.list); ++i) {
- pubsub_announce_endpoint_listener_t *listener = celix_arrayList_get(
- manager->announceEndpointListeners.list, i);
- listener->announceEndpoint(listener->handle, entry->endpoint);
- }
- celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
- }
+ entry->matching.needsMatch = false;
+ //NOTE the needsMatch can be updated as soon as the lock manager->topicSenders.mutex is released.
+ //This is ok, because the next pstm update loop will then do a new teardown/setup
+
+ struct pstm_setup_entry *setupEntry = malloc(sizeof(*setupEntry));
+ setupEntry->endpointResult = NULL;
+ setupEntry->scope = celix_utils_strdup(entry->scope);
+ setupEntry->topic = celix_utils_strdup(entry->topic);
+ setupEntry->key = pubsubEndpoint_createScopeTopicKey(entry->scope, entry->topic);
+ setupEntry->topicProperties = highestMatchTopicProperties;
+ setupEntry->psaSvcId = selectedPsaSvcId;
+ setupEntry->selectedSerializerSvcId = serializerSvcId;
+ setupEntry->selectedProtocolSvcId = protocolSvcId;
+ celix_arrayList_add(setupEntries, setupEntry);
}
+ }
+ }
+ celixThreadMutex_unlock(&manager->topicReceivers.mutex);
+
- if (entry->needsMatch) {
- celix_logHelper_log(manager->loghelper, CELIX_LOG_LEVEL_WARNING, "Cannot setup TopicReceiver for %s/%s\n", entry->scope == NULL ? "(null)" : entry->scope, entry->topic);
+ for (int i = 0; i < celix_arrayList_size(setupEntries); ++i) {
+ struct pstm_setup_entry* setupEntry = celix_arrayList_get(setupEntries, i);
+ bool called = celix_bundleContext_useServiceWithId(manager->context, setupEntry->psaSvcId, PUBSUB_ADMIN_SERVICE_NAME, setupEntry, pstm_setupTopicReceiverCallback);
+ if (called && setupEntry->endpointResult != NULL) {
+ celixThreadMutex_lock(&manager->announceEndpointListeners.mutex);
+ for (int k = 0; k < celix_arrayList_size(manager->announceEndpointListeners.list); ++k) {
+ pubsub_announce_endpoint_listener_t *listener = celix_arrayList_get(manager->announceEndpointListeners.list, k);
+ listener->announceEndpoint(listener->handle, setupEntry->endpointResult);
}
+ celixThreadMutex_unlock(&manager->announceEndpointListeners.mutex);
+
+ celixThreadMutex_lock(&manager->topicReceivers.mutex);
+ pstm_topic_receiver_or_sender_entry_t* entry = hashMap_get(manager->topicReceivers.map, setupEntry->key);
+ if (entry->endpoint != NULL) {
+ celix_properties_destroy(entry->endpoint);
+ }
+ entry->endpoint = setupEntry->endpointResult;
+ entry->topicProperties = setupEntry->topicProperties;
+ entry->matching.selectedPsaSvcId = setupEntry->psaSvcId;
+ entry->matching.selectedSerializerSvcId = setupEntry->selectedSerializerSvcId;
+ entry->matching.selectedProtocolSvcId = setupEntry->selectedProtocolSvcId;
+ celixThreadMutex_unlock(&manager->topicReceivers.mutex);
+ } else {
+ celix_logHelper_warning(manager->loghelper, "Cannot setup TopicReceiver for %s/%s\n", setupEntry->scope == NULL ? "(null)" : setupEntry->scope, setupEntry->topic);
}
+ free(setupEntry->scope);
+ free(setupEntry->topic);
+ free(setupEntry->key);
+ free(setupEntry);
}
- celixThreadMutex_unlock(&manager->topicReceivers.mutex);
+
+ celix_arrayList_destroy(setupEntries);
}
static void *pstm_psaHandlingThread(void *data) {
@@ -1053,9 +1186,9 @@ static celix_status_t pubsub_topologyManager_topology(pubsub_topology_manager_t
fprintf(os, " |- serializer = %s\n", serType);
fprintf(os, " |- protocol = %s\n", protType);
if (manager->verbose) {
- fprintf(os, " |- psa svc id = %li\n", entry->selectedPsaSvcId);
- fprintf(os, " |- ser svc id = %li\n", entry->selectedSerializerSvcId);
- fprintf(os, " |- prot svc id = %li\n", entry->selectedProtocolSvcId);
+ fprintf(os, " |- psa svc id = %li\n", entry->matching.selectedPsaSvcId);
+ fprintf(os, " |- ser svc id = %li\n", entry->matching.selectedSerializerSvcId);
+ fprintf(os, " |- prot svc id = %li\n", entry->matching.selectedProtocolSvcId);
fprintf(os, " |- usage count = %i\n", entry->usageCount);
}
}
@@ -1083,9 +1216,9 @@ static celix_status_t pubsub_topologyManager_topology(pubsub_topology_manager_t
fprintf(os, " |- serializer = %s\n", serType);
fprintf(os, " |- protocol = %s\n", protType);
if (manager->verbose) {
- fprintf(os, " |- psa svc id = %li\n", entry->selectedPsaSvcId);
- fprintf(os, " |- ser svc id = %li\n", entry->selectedSerializerSvcId);
- fprintf(os, " |- prot svc id = %li\n", entry->selectedProtocolSvcId);
+ fprintf(os, " |- psa svc id = %li\n", entry->matching.selectedPsaSvcId);
+ fprintf(os, " |- ser svc id = %li\n", entry->matching.selectedSerializerSvcId);
+ fprintf(os, " |- prot svc id = %li\n", entry->matching.selectedProtocolSvcId);
fprintf(os, " |- usage count = %i\n", entry->usageCount);
}
}
diff --git a/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.h b/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.h
index 853811b..9f5ef49 100644
--- a/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.h
+++ b/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.h
@@ -87,16 +87,11 @@ typedef struct pstm_discovered_endpoint_entry {
} pstm_discovered_endpoint_entry_t;
typedef struct pstm_topic_receiver_or_sender_entry {
- bool needsMatch; //true if a psa needs to be selected or if a new psa has to be considered.
-
char *scopeAndTopicKey; //key of the combined value of the scope and topic
celix_properties_t *endpoint;
char *topic;
char *scope;
- int usageCount; //nr of subscriber service for the topic receiver (matching scope & topic)
- long selectedPsaSvcId;
- long selectedSerializerSvcId;
- long selectedProtocolSvcId;
+ int usageCount; //nr of provided subscriber services for the topic receiver (matching scope & topic) or nr of publisher requested for matching scope & topic.
long bndId;
celix_properties_t *topicProperties; //found in META-INF/(pub|sub)/(topic).properties
@@ -105,6 +100,13 @@ typedef struct pstm_topic_receiver_or_sender_entry {
//for receiver entry
celix_properties_t *subscriberProperties;
+
+ struct {
+ bool needsMatch; //true if a psa needs to be selected or if a new psa has to be considered.
+ long selectedPsaSvcId;
+ long selectedSerializerSvcId;
+ long selectedProtocolSvcId;
+ } matching;
} pstm_topic_receiver_or_sender_entry_t;
celix_status_t pubsub_topologyManager_create(celix_bundle_context_t *context, celix_log_helper_t *logHelper, pubsub_topology_manager_t **manager);
diff --git a/bundles/pubsub/test/CMakeLists.txt b/bundles/pubsub/test/CMakeLists.txt
index 80859fa..8b7e061 100644
--- a/bundles/pubsub/test/CMakeLists.txt
+++ b/bundles/pubsub/test/CMakeLists.txt
@@ -106,6 +106,28 @@ celix_bundle_files(pubsub_tst
DESTINATION "META-INF/topics/sub"
)
+add_celix_bundle(pubsub_deadlock_sut
+ #"Vanilla" bundle which is used to trigger a publisher added call
+ SOURCES
+ test/sut_activator.c
+ VERSION 1.0.0
+)
+celix_bundle_files(pubsub_deadlock_sut
+ meta_data/msg.descriptor
+ DESTINATION "META-INF/descriptors"
+)
+celix_bundle_files(pubsub_deadlock_sut
+ meta_data/deadlock.scope.properties
+ DESTINATION "META-INF/topics/pub"
+)
+celix_bundle_files(pubsub_deadlock_sut
+ meta_data/deadlock.scope2.properties
+ DESTINATION "META-INF/topics/pub"
+)
+target_link_libraries(pubsub_deadlock_sut PRIVATE Celix::pubsub_api)
+
+celix_get_bundle_file(pubsub_deadlock_sut DEADLOCK_SUT_BUNDLE_FILE)
+
if (BUILD_PUBSUB_PSA_UDP_MC)
add_celix_container(pubsub_udpmc_tests
USE_CONFIG #ensures that a config.properties will be created with the launch bundles.
@@ -123,8 +145,43 @@ if (BUILD_PUBSUB_PSA_UDP_MC)
target_link_libraries(pubsub_udpmc_tests PRIVATE Celix::pubsub_api ${CppUTest_LIBRARIES} Jansson Celix::dfi)
target_include_directories(pubsub_udpmc_tests SYSTEM PRIVATE ${CppUTest_INCLUDE_DIR} test)
- add_test(NAME pubsub_udpmc_tests COMMAND pubsub_udpmc_tests WORKING_DIRECTORY $<TARGET_PROPERTY:pubsub_udpmc_tests,CONTAINER_LOC>)
- setup_target_for_coverage(pubsub_udpmc_tests_cov SCAN_DIR ..)
+ add_celix_container(pstm_deadlock_udpmc_test
+ USE_CONFIG #ensures that a config.properties will be created with the launch bundles.
+ LAUNCHER_SRC ${CMAKE_CURRENT_LIST_DIR}/pstm_deadlock_test/test_runner.cc
+ DIR ${CMAKE_CURRENT_BINARY_DIR}
+ PROPERTIES
+ LOGHELPER_STDOUT_FALLBACK_INCLUDE_DEBUG=true
+ BUNDLES
+ Celix::pubsub_serializer_json
+ Celix::pubsub_topology_manager
+ Celix::pubsub_admin_zmq
+ Celix::pubsub_protocol_wire_v2
+ Celix::shell
+ Celix::shell_tui
+ )
+ target_compile_definitions(pstm_deadlock_udpmc_test PRIVATE -DDEADLOCK_SUT_BUNDLE_FILE=\"${DEADLOCK_SUT_BUNDLE_FILE}\")
+ target_link_libraries(pstm_deadlock_udpmc_test PRIVATE Celix::pubsub_api GTest::gtest GTest::gtest_main Jansson Celix::dfi ZMQ::lib CZMQ::lib)
+ target_include_directories(pstm_deadlock_udpmc_test SYSTEM PRIVATE pstm_deadlock_udpmc_test)
+
+ #Note we do not link to bundles, as result (to ensure a bundle zip file is created) an dependency on the bundle is needed.
+ add_dependencies(pstm_deadlock_udpmc_test pubsub_deadlock_sut_bundle)
+
+ #Framework "bundle" has no cache dir. Default as "cache dir" the cwd is used.
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/msg.descriptor ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_udpmc_test/META-INF/descriptors/msg.descriptor COPYONLY)
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/deadlock.scope.properties ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_udpmc_test/META-INF/topics/pub/deadlock.properties COPYONLY)
+
+ add_test(NAME pstm_deadlock_udpmc_test COMMAND pstm_deadlock_udpmc_test WORKING_DIRECTORY $<TARGET_PROPERTY:pstm_deadlock_udpmc_test,CONTAINER_LOC>)
+ setup_target_for_coverage(pstm_deadlock_udpmc_test SCAN_DIR ..)
+
+ #TCP Endpoint test is disabled because the test is not stable when running on Travis
+ if (ENABLE_PUBSUB_PSA_TCP_ENDPOINT_TEST)
+ add_test(NAME pubsub_tcp_endpoint_tests COMMAND pubsub_tcp_endpoint_tests WORKING_DIRECTORY $<TARGET_PROPERTY:pubsub_tcp_endpoint_tests,CONTAINER_LOC>)
+ setup_target_for_coverage(pubsub_tcp_endpoint_tests SCAN_DIR ..)
+ endif()
+
+ #TODO fix issues with UDPMC and reanble test again
+ #add_test(NAME pubsub_udpmc_tests COMMAND pubsub_udpmc_tests WORKING_DIRECTORY $<TARGET_PROPERTY:pubsub_udpmc_tests,CONTAINER_LOC>)
+ #setup_target_for_coverage(pubsub_udpmc_tests SCAN_DIR ..)
endif()
if (BUILD_PUBSUB_PSA_TCP)
@@ -170,6 +227,34 @@ if (BUILD_PUBSUB_PSA_TCP)
target_link_libraries(pubsub_tcp_endpoint_tests PRIVATE Celix::pubsub_api ${CppUTest_LIBRARIES} Jansson Celix::dfi)
target_include_directories(pubsub_tcp_endpoint_tests SYSTEM PRIVATE ${CppUTest_INCLUDE_DIR} test)
+ add_celix_container(pstm_deadlock_tcp_test
+ USE_CONFIG #ensures that a config.properties will be created with the launch bundles.
+ LAUNCHER_SRC ${CMAKE_CURRENT_LIST_DIR}/pstm_deadlock_test/test_runner.cc
+ DIR ${CMAKE_CURRENT_BINARY_DIR}
+ PROPERTIES
+ LOGHELPER_STDOUT_FALLBACK_INCLUDE_DEBUG=true
+ BUNDLES
+ Celix::pubsub_serializer_json
+ Celix::pubsub_topology_manager
+ Celix::pubsub_admin_zmq
+ Celix::pubsub_protocol_wire_v2
+ Celix::shell
+ Celix::shell_tui
+ )
+ target_compile_definitions(pstm_deadlock_tcp_test PRIVATE -DDEADLOCK_SUT_BUNDLE_FILE=\"${DEADLOCK_SUT_BUNDLE_FILE}\")
+ target_link_libraries(pstm_deadlock_tcp_test PRIVATE Celix::pubsub_api GTest::gtest GTest::gtest_main Jansson Celix::dfi ZMQ::lib CZMQ::lib)
+ target_include_directories(pstm_deadlock_tcp_test SYSTEM PRIVATE pstm_deadlock_tcp_test)
+
+ #Note we do not link to bundles, as result (to ensure a bundle zip file is created) an dependency on the bundle is needed.
+ add_dependencies(pstm_deadlock_tcp_test pubsub_deadlock_sut_bundle)
+
+ #Framework "bundle" has no cache dir. Default as "cache dir" the cwd is used.
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/msg.descriptor ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_tcp_test/META-INF/descriptors/msg.descriptor COPYONLY)
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/deadlock.scope.properties ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_tcp_test/META-INF/topics/pub/deadlock.properties COPYONLY)
+
+ add_test(NAME pstm_deadlock_tcp_test COMMAND pstm_deadlock_tcp_test WORKING_DIRECTORY $<TARGET_PROPERTY:pstm_deadlock_tcp_test,CONTAINER_LOC>)
+ setup_target_for_coverage(pstm_deadlock_tcp_test SCAN_DIR ..)
+
#TCP Endpoint test is disabled because the test is not stable when running on Travis
if (ENABLE_PUBSUB_PSA_TCP_ENDPOINT_TEST)
add_test(NAME pubsub_tcp_endpoint_tests COMMAND pubsub_tcp_endpoint_tests WORKING_DIRECTORY $<TARGET_PROPERTY:pubsub_tcp_endpoint_tests,CONTAINER_LOC>)
@@ -198,7 +283,35 @@ if (BUILD_PUBSUB_PSA_WS)
target_include_directories(pubsub_websocket_tests SYSTEM PRIVATE ${CppUTest_INCLUDE_DIR} test)
add_test(NAME pubsub_websocket_tests COMMAND pubsub_websocket_tests WORKING_DIRECTORY $<TARGET_PROPERTY:pubsub_websocket_tests,CONTAINER_LOC>)
setup_target_for_coverage(pubsub_websocket_tests SCAN_DIR ..)
- endif()
+
+ add_celix_container(pstm_deadlock_websocket_test
+ USE_CONFIG #ensures that a config.properties will be created with the launch bundles.
+ LAUNCHER_SRC ${CMAKE_CURRENT_LIST_DIR}/pstm_deadlock_test/test_runner.cc
+ DIR ${CMAKE_CURRENT_BINARY_DIR}
+ PROPERTIES
+ LOGHELPER_STDOUT_FALLBACK_INCLUDE_DEBUG=true
+ BUNDLES
+ Celix::pubsub_serializer_json
+ Celix::pubsub_topology_manager
+ Celix::pubsub_admin_zmq
+ Celix::pubsub_protocol_wire_v2
+ Celix::shell
+ Celix::shell_tui
+ )
+ target_compile_definitions(pstm_deadlock_websocket_test PRIVATE -DDEADLOCK_SUT_BUNDLE_FILE=\"${DEADLOCK_SUT_BUNDLE_FILE}\")
+ target_link_libraries(pstm_deadlock_websocket_test PRIVATE Celix::pubsub_api GTest::gtest GTest::gtest_main Jansson Celix::dfi ZMQ::lib CZMQ::lib)
+ target_include_directories(pstm_deadlock_websocket_test SYSTEM PRIVATE pstm_deadlock_websocket_test)
+
+ #Note we do not link to bundles, as result (to ensure a bundle zip file is created) an dependency on the bundle is needed.
+ add_dependencies(pstm_deadlock_websocket_test pubsub_deadlock_sut_bundle)
+
+ #Framework "bundle" has no cache dir. Default as "cache dir" the cwd is used.
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/msg.descriptor ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_websocket_test/META-INF/descriptors/msg.descriptor COPYONLY)
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/deadlock.scope.properties ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_websocket_test/META-INF/topics/pub/deadlock.properties COPYONLY)
+
+ add_test(NAME pstm_deadlock_websocket_test COMMAND pstm_deadlock_websocket_test WORKING_DIRECTORY $<TARGET_PROPERTY:pstm_deadlock_websocket_test,CONTAINER_LOC>)
+ setup_target_for_coverage(pstm_deadlock_websocket_test SCAN_DIR ..)
+endif()
if (BUILD_PUBSUB_PSA_ZMQ)
find_package(ZMQ REQUIRED)
@@ -288,7 +401,34 @@ if (BUILD_PUBSUB_PSA_ZMQ)
target_link_libraries(pubsub_zmq_zerocopy_wire_v2_tests PRIVATE Celix::pubsub_api ${CppUTest_LIBRARIES} Jansson Celix::dfi ZMQ::lib CZMQ::lib)
target_include_directories(pubsub_zmq_zerocopy_wire_v2_tests SYSTEM PRIVATE ${CppUTest_INCLUDE_DIR} test)
- #TODO fix issues with ZeroCopy and reanble test again
add_test(NAME pubsub_zmq_zerocopy_wire_v2_tests COMMAND pubsub_zmq_zerocopy_wire_v2_tests WORKING_DIRECTORY $<TARGET_PROPERTY:pubsub_zmq_zerocopy_wire_v2_tests,CONTAINER_LOC>)
setup_target_for_coverage(pubsub_zmq_zerocopy_wire_v2_tests SCAN_DIR ..)
+
+ add_celix_container(pstm_deadlock_zmq_test
+ USE_CONFIG #ensures that a config.properties will be created with the launch bundles.
+ LAUNCHER_SRC ${CMAKE_CURRENT_LIST_DIR}/pstm_deadlock_test/test_runner.cc
+ DIR ${CMAKE_CURRENT_BINARY_DIR}
+ PROPERTIES
+ LOGHELPER_STDOUT_FALLBACK_INCLUDE_DEBUG=true
+ BUNDLES
+ Celix::pubsub_serializer_json
+ Celix::pubsub_topology_manager
+ Celix::pubsub_admin_zmq
+ Celix::pubsub_protocol_wire_v2
+ Celix::shell
+ Celix::shell_tui
+ )
+ target_compile_definitions(pstm_deadlock_zmq_test PRIVATE -DDEADLOCK_SUT_BUNDLE_FILE=\"${DEADLOCK_SUT_BUNDLE_FILE}\")
+ target_link_libraries(pstm_deadlock_zmq_test PRIVATE Celix::pubsub_api GTest::gtest GTest::gtest_main Jansson Celix::dfi ZMQ::lib CZMQ::lib)
+ target_include_directories(pstm_deadlock_zmq_test SYSTEM PRIVATE pstm_deadlock_zmq_test)
+
+ #Note we do not link to bundles, as result (to ensure a bundle zip file is created) an dependency on the bundle is needed.
+ add_dependencies(pstm_deadlock_zmq_test pubsub_deadlock_sut_bundle)
+
+ #Framework "bundle" has no cache dir. Default as "cache dir" the cwd is used.
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/msg.descriptor ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_zmq_test/META-INF/descriptors/msg.descriptor COPYONLY)
+ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/meta_data/deadlock.scope.properties ${CMAKE_CURRENT_BINARY_DIR}/pstm_deadlock_zmq_test/META-INF/topics/pub/deadlock.properties COPYONLY)
+
+ add_test(NAME pstm_deadlock_zmq_test COMMAND pstm_deadlock_zmq_test WORKING_DIRECTORY $<TARGET_PROPERTY:pstm_deadlock_zmq_test,CONTAINER_LOC>)
+ setup_target_for_coverage(pstm_deadlock_zmq_test SCAN_DIR ..)
endif ()
diff --git a/bundles/pubsub/test/meta_data/deadlock.scope.properties b/bundles/pubsub/test/meta_data/deadlock.scope.properties
new file mode 100644
index 0000000..2028e72
--- /dev/null
+++ b/bundles/pubsub/test/meta_data/deadlock.scope.properties
@@ -0,0 +1,28 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+zmq.static.bind.url=ipc:///tmp/pubsub-deadlock
+zmq.static.connect.urls=ipc:///tmp/pubsub-deadlock
+tcp.static.bind.url=tcp://localhost:9000
+tcp.static.connect.urls=tcp://localhost:9000
+udpmc.static.bind.port=50678
+udpmc.static.connect.socket_addresses=224.100.0.1:50678
+websocket.static.connect.socket_addresses=127.0.0.1:8080
+
+#note only effective if run as root
+thread.realtime.shed=SCHED_FIFO
+thread.realtime.prio=50
+
diff --git a/bundles/pubsub/test/meta_data/deadlock.scope2.properties b/bundles/pubsub/test/meta_data/deadlock.scope2.properties
new file mode 100644
index 0000000..2028e72
--- /dev/null
+++ b/bundles/pubsub/test/meta_data/deadlock.scope2.properties
@@ -0,0 +1,28 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+zmq.static.bind.url=ipc:///tmp/pubsub-deadlock
+zmq.static.connect.urls=ipc:///tmp/pubsub-deadlock
+tcp.static.bind.url=tcp://localhost:9000
+tcp.static.connect.urls=tcp://localhost:9000
+udpmc.static.bind.port=50678
+udpmc.static.connect.socket_addresses=224.100.0.1:50678
+websocket.static.connect.socket_addresses=127.0.0.1:8080
+
+#note only effective if run as root
+thread.realtime.shed=SCHED_FIFO
+thread.realtime.prio=50
+
diff --git a/bundles/pubsub/test/pstm_deadlock_test/test_runner.cc b/bundles/pubsub/test/pstm_deadlock_test/test_runner.cc
new file mode 100644
index 0000000..1e82330
--- /dev/null
+++ b/bundles/pubsub/test/pstm_deadlock_test/test_runner.cc
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "celix_api.h"
+#include <celix/dm/Component.h>
+#include <celix/dm/DependencyManager.h>
+#include "pubsub/api.h"
+#include <unistd.h>
+#include <memory>
+
+#include <gtest/gtest.h>
+
+constexpr const char *deadlockSutBundleFile = DEADLOCK_SUT_BUNDLE_FILE;
+
+struct DeadlockTestSuite : public ::testing::Test {
+ celix_framework_t *fw = NULL;
+ celix_bundle_context_t *ctx = NULL;
+ std::unique_ptr<DependencyManager> mng = NULL;
+ long sutBundleId = 0;
+
+ DeadlockTestSuite() {
+ celixLauncher_launch("config.properties", &fw);
+ ctx = celix_framework_getFrameworkContext(fw);
+ mng = std::unique_ptr<DependencyManager>(new DependencyManager(ctx));
+ sutBundleId = celix_bundleContext_installBundle(ctx, deadlockSutBundleFile, false);
+ }
+
+ ~DeadlockTestSuite() override {
+ celix_bundleContext_uninstallBundle(ctx, sutBundleId);
+ celixLauncher_stop(fw);
+ celixLauncher_waitForShutdown(fw);
+ celixLauncher_destroy(fw);
+ ctx = NULL;
+ fw = NULL;
+ }
+
+ DeadlockTestSuite(const DeadlockTestSuite&) = delete;
+ DeadlockTestSuite& operator=(const DeadlockTestSuite&) = delete;
+};
+
+class DependencyCmp;
+
+struct activator {
+ Component<DependencyCmp> *cmp;
+ celix_bundle_context_t *ctx;
+};
+
+class IDependency {
+protected:
+ IDependency() = default;
+ ~IDependency() = default;
+public:
+ virtual double getData() = 0;
+};
+
+class DependencyCmp : public IDependency {
+public:
+ DependencyCmp() {
+ std::cout << "Creating DependencyCmp\n";
+ }
+ virtual ~DependencyCmp() { std::cout << "Destroying DependencyCmp\n"; };
+
+ DependencyCmp(DependencyCmp&& other) noexcept = default;
+ DependencyCmp& operator=(DependencyCmp&&) = default;
+
+ DependencyCmp(const DependencyCmp& other) = delete;
+ DependencyCmp operator=(const DependencyCmp&) = delete;
+
+ double getData() override {
+ return 1.0;
+ }
+
+ void setPublisher(const pubsub_publisher_t *pub __attribute__((unused))) {
+ act->cmp->createCServiceDependency<pubsub_publisher_t>(PUBSUB_PUBLISHER_SERVICE_NAME)
+ .setVersionRange("[3.0.0,4)")
+ .setFilter("(topic=deadlock)(scope=scope2)")
+ .setStrategy(celix::dm::DependencyUpdateStrategy::suspend)
+ .setCallbacks([](const pubsub_publisher_t *, Properties&&) { std::cout << "success\n"; })
+ .setRequired(true);
+ }
+
+ int init() {
+ return CELIX_SUCCESS;
+ }
+
+ int start() {
+ return CELIX_SUCCESS;
+ }
+
+ int stop() {
+ return CELIX_SUCCESS;
+ }
+
+ int deinit() {
+ return CELIX_SUCCESS;
+ }
+
+ activator *act{nullptr};
+};
+
+TEST_F(DeadlockTestSuite, test) {
+
+ Component<DependencyCmp>& cmp = mng->createComponent<DependencyCmp>()
+ .addInterface<IDependency>("1.0.0", {});
+
+ cmp.setCallbacks(&DependencyCmp::init, &DependencyCmp::start, &DependencyCmp::stop, &DependencyCmp::deinit);
+
+ activator act{};
+ act.ctx = ctx;
+ act.cmp = &cmp;
+ cmp.getInstance().act = &act;
+
+ cmp.createCServiceDependency<pubsub_publisher_t>(PUBSUB_PUBLISHER_SERVICE_NAME)
+ .setVersionRange("[3.0.0,4)")
+ .setFilter("(topic=deadlock)(scope=scope)")
+ .setStrategy(celix::dm::DependencyUpdateStrategy::suspend)
+ .setCallbacks([&cmp](const pubsub_publisher_t *pub, Properties&&) { cmp.getInstance().setPublisher(pub); })
+ .setRequired(true);
+
+ ASSERT_TRUE(cmp.isValid());
+ mng->start();
+
+ celix_bundleContext_startBundle(ctx, sutBundleId);
+
+ mng->stop();
+}
diff --git a/bundles/pubsub/test/test/test_endpoint_runner.cc b/bundles/pubsub/test/test/test_endpoint_runner.cc
index e1daa6e..ae65175 100644
--- a/bundles/pubsub/test/test/test_endpoint_runner.cc
+++ b/bundles/pubsub/test/test/test_endpoint_runner.cc
@@ -48,7 +48,7 @@ TEST_GROUP(PUBSUB_INT_GROUP) {
};
TEST(PUBSUB_INT_GROUP, recvTest) {
- constexpr int TRIES = 25;
+ constexpr int TRIES = 50;
constexpr int TIMEOUT = 250000;
constexpr int MSG_COUNT = 100;
@@ -68,4 +68,4 @@ TEST(PUBSUB_INT_GROUP, recvTest) {
usleep(TIMEOUT);
}
CHECK(count >= MSG_COUNT);
-}
\ No newline at end of file
+}
diff --git a/bundles/pubsub/test/test/test_runner.cc b/bundles/pubsub/test/test/test_runner.cc
index 3ef3faa..28975e0 100644
--- a/bundles/pubsub/test/test/test_runner.cc
+++ b/bundles/pubsub/test/test/test_runner.cc
@@ -82,7 +82,7 @@ TEST_GROUP(PUBSUB_INT_ENV_GROUP) {
};
void receiveTest(celix_bundle_context_t *ctx) {
- constexpr int TRIES = 40;
+ constexpr int TRIES = 50;
constexpr int TIMEOUT = 250000;
constexpr int MSG_COUNT = 100;
@@ -110,4 +110,4 @@ TEST(PUBSUB_INT_GROUP, recvTest) {
TEST(PUBSUB_INT_ENV_GROUP, recvTest) {
receiveTest(ctx);
-}
\ No newline at end of file
+}
diff --git a/libs/framework/src/dm_dependency_manager_impl.c b/libs/framework/src/dm_dependency_manager_impl.c
index 2576ca5..80438e9 100644
--- a/libs/framework/src/dm_dependency_manager_impl.c
+++ b/libs/framework/src/dm_dependency_manager_impl.c
@@ -27,6 +27,7 @@
#include "dm_dependency_manager_impl.h"
#include "celix_dependency_manager.h"
#include "celix_bundle.h"
+#include "celix_framework.h"
celix_dependency_manager_t* celix_private_dependencyManager_create(celix_bundle_context_t *context) {
@@ -173,7 +174,8 @@ static void celix_dm_getInfosCallback(void *handle, const celix_bundle_t *bnd) {
celix_array_list_t * celix_dependencyManager_createInfos(celix_dependency_manager_t *manager) {
celix_array_list_t *infos = celix_arrayList_create();
- celix_bundleContext_useBundles(manager->ctx, infos, celix_dm_getInfosCallback);
+ celix_framework_t* fw = celix_bundleContext_getFramework(manager->ctx);
+ celix_framework_useBundles(fw, true, infos, celix_dm_getInfosCallback);
return infos;
}