pnoltes commented on code in PR #773:
URL: https://github.com/apache/celix/pull/773#discussion_r1855517662
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -415,6 +454,131 @@ int
celix_eventAdmin_removeEventHandlerWithProperties(void* handle, void* svc, c
return CELIX_SUCCESS;
}
+int celix_eventAdmin_addRemoteProviderService(void* handle, void* svc, const
celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ celix_event_admin_t* ea = handle;
+ long serviceId = celix_properties_getAsLong(props,
CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Remote provider service id is
invalid.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ celix_longHashMap_put(ea->remoteProviderServices, serviceId, svc);
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_removeRemoteProviderService(void* handle, void* svc,
const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ celix_event_admin_t* ea = handle;
+ long serviceId = celix_properties_getAsLong(props,
CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Remote provider service id is
invalid.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ celix_longHashMap_remove(ea->remoteProviderServices, serviceId);
+ return CELIX_SUCCESS;
+}
+
+static void
celix_eventAdmin_retrieveLongTimeUnusedEventSeqIdCache(celix_event_admin_t* ea)
{
+ if (celix_stringHashMap_size(ea->eventSeqIdCache) > 16) {
+ celix_string_hash_map_iterator_t iter =
celix_stringHashMap_begin(ea->eventSeqIdCache);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_seq_id_cache_t* cache = iter.value.ptrValue;
+ if (celix_elapsedtime(CLOCK_MONOTONIC, cache->lastModified) >
60*60/*1h*/) {
+ celix_stringHashMapIterator_remove(&iter);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+ }
+ return;
+}
+
+static bool celix_eventAdmin_isDuplicateEvent(celix_event_admin_t* ea, const
char* topic CELIX_UNUSED, const celix_properties_t* properties) {
+ const char* remoteFwUUID = celix_properties_get(properties,
CELIX_EVENT_REMOTE_FRAMEWORK_UUID, NULL);
+ if (remoteFwUUID == NULL) {
+ return false;
+ }
+ long seqId = celix_properties_getAsLong(properties,
CELIX_EVENT_REMOTE_SEQ_ID, -1L);
+ if (seqId <= 0) {
+ return false;
+ }
+ long seqIdMod = seqId % CELIX_EVENT_ADMIN_MAX_EVENT_SEQ_ID_CACHE_SIZE;
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ celix_event_seq_id_cache_t* seqIdCache =
celix_stringHashMap_get(ea->eventSeqIdCache, remoteFwUUID);
+ if (seqIdCache == NULL) {
+ celix_autofree celix_event_seq_id_cache_t* cache = calloc(1,
sizeof(*cache));
+ if (cache == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event seq
id cache for %s.", remoteFwUUID);
+ return false;
+ }
+ celix_status_t status = celix_stringHashMap_put(ea->eventSeqIdCache,
remoteFwUUID, cache);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event seq id
cache for %s.", remoteFwUUID);
+ return false;
+ }
+ seqIdCache = celix_steal_ptr(cache);
+ }
+ seqIdCache->lastModified = celix_gettime(CLOCK_MONOTONIC);
+ if (seqIdCache->seqIdBuffer[seqIdMod] == seqId) {
+ return true;
+ }
+ seqIdCache->seqIdBuffer[seqIdMod] = seqId;
+
+ celix_eventAdmin_retrieveLongTimeUnusedEventSeqIdCache(ea);
+
+ return false;
+}
+
+static long celix_eventAdmin_getEventSeqId(celix_event_admin_t* ea) {
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ long seqId = ea->nextSeqId++;
+ if (seqId <= 0) {
+ seqId = 1;
+ ea->nextSeqId = seqId + 1;
+ }
+ return seqId;
+}
+
+static int celix_eventAdmin_deliverEventToRemote(celix_event_admin_t* ea,
const char* topic, const celix_properties_t* props, bool async) {
+ celix_autoptr(celix_properties_t) remoteProps =
celix_properties_copy(props);
+ if (remoteProps == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to copy remote properties
for event %s.", topic);
+ return ENOMEM;
+ }
+ celix_properties_unset(remoteProps, CELIX_EVENT_REMOTE_ENABLE);
+ celix_status_t status = celix_properties_set(remoteProps,
CELIX_EVENT_REMOTE_FRAMEWORK_UUID, ea->fwUUID);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to set remote framework
uuid for event %s.", topic);
+ return status;
+ }
+ long seqId = celix_eventAdmin_getEventSeqId(ea);
+ status = celix_properties_setLong(remoteProps, CELIX_EVENT_REMOTE_SEQ_ID,
seqId);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to set remote seq id for
event %s.", topic);
+ return status;
+ }
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard =
celixRwlockRlockGuard_init(&ea->lock);
+ CELIX_LONG_HASH_MAP_ITERATE(ea->remoteProviderServices, iter) {
+ celix_event_remote_provider_service_t* remoteProvider =
iter.value.ptrValue;
+ if (async) {
+ celix_logHelper_trace(ea->logHelper, "Post event %s to remote
provider.", topic);
+ status = remoteProvider->postEvent(remoteProvider->handle, topic,
remoteProps);
Review Comment:
Yes, theoretically, a deadlock is still possible because a write lock cannot
be taken while a read lock is active. If the sendEvent on the remote provider
returns to the process and tries to register a new remote provider
synchronously, it could result in a deadlock. Such a situation has been
observed in remote services, but there the likelihood of a (remote) call loop
is much higher.
That being said, these king of deadlock scenarios is the primary reason why
async service registrations in ASF Celix exists and why services are registered
asynchronously during component activations. I was too hasty in my judgment,
and in my opinion, the current code is good as long as we continue to promote
the use of asynchronous service registrations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
