add synchronization for callback drop for subscribers, queryables, queriers and sessions

Author: DenisBiryukov91

include/zenoh-pico/collections/sync_group.h

@@ -51,6 +51,12 @@ static inline _z_sync_group_notifier_t _z_sync_group_notifier_null(void) {
     return n;
 }
 
+static inline _z_sync_group_notifier_t _z_sync_group_notifier_steal(_z_sync_group_notifier_t* notifier) {
+    _z_sync_group_notifier_t n = *notifier;
+    *notifier = _z_sync_group_notifier_null();
+    return n;
+}
+
 z_result_t _z_sync_group_create(_z_sync_group_t* sync_group);
 z_result_t _z_sync_group_wait(_z_sync_group_t* sync_group);
 void _z_sync_group_close(_z_sync_group_t* sync_group);
@@ -60,7 +66,7 @@ static inline bool _z_sync_group_check(const _z_sync_group_t* sync_group) {
 }
 bool _z_sync_group_is_closed(const _z_sync_group_t* sync_group);
 void _z_sync_group_drop(_z_sync_group_t* sync_group);
-z_result_t _z_sync_group_create_notifier(_z_sync_group_t* sync_group, _z_sync_group_notifier_t* notifier);
+z_result_t _z_sync_group_create_notifier(const _z_sync_group_t* sync_group, _z_sync_group_notifier_t* notifier);
 void _z_sync_group_notifier_drop(_z_sync_group_notifier_t* notifier);
 static inline bool _z_sync_group_notifier_check(const _z_sync_group_notifier_t* notifier) {
     return !_Z_RC_IS_NULL(&notifier->_state);

include/zenoh-pico/net/liveliness.h

@@ -48,6 +48,12 @@ z_result_t _z_declare_liveliness_subscriber(_z_subscriber_t *subscriber, const _
                                             const _z_declared_keyexpr_t *keyexpr, _z_closure_sample_callback_t callback,
                                             _z_drop_handler_t dropper, bool history, void *arg);
 
+z_result_t _z_register_liveliness_subscriber(uint32_t *out_sub_id, const _z_session_rc_t *zn,
+                                             const _z_declared_keyexpr_t *keyexpr,
+                                             _z_closure_sample_callback_t callback, _z_drop_handler_t dropper,
+                                             bool history, void *arg,
+                                             const _z_sync_group_t *opt_callback_drop_sync_group);
+
 /**
  * Undeclare a liveliness :c:type:`_z_subscriber_t`.
  *

include/zenoh-pico/net/matching.h

@@ -27,8 +27,10 @@ typedef struct _z_matching_listener_t {
 } _z_matching_listener_t;
 
 #if Z_FEATURE_MATCHING == 1
-_z_matching_listener_t _z_matching_listener_declare(const _z_write_filter_ctx_rc_t *ctx, uint32_t id,
-                                                    _z_closure_matching_status_t *callback);
+z_result_t _z_matching_listener_declare(_z_matching_listener_t *listener, _z_session_t *s,
+                                        const _z_write_filter_ctx_rc_t *ctx, _z_closure_matching_status_t *callback);
+z_result_t _z_matching_listener_register(uint32_t *listener_id, _z_session_t *s, const _z_write_filter_ctx_rc_t *ctx,
+                                         _z_closure_matching_status_t *callback);
 z_result_t _z_matching_listener_undeclare(_z_matching_listener_t *listener);
 // Warning: None of the sub-types require a non-0 initialization. Add a init function if it changes.
 static inline _z_matching_listener_t _z_matching_listener_null(void) { return (_z_matching_listener_t){0}; }

include/zenoh-pico/net/primitives.h

@@ -161,7 +161,9 @@ z_result_t _z_write(_z_session_t *zn, const _z_declared_keyexpr_t *keyexpr, _z_b
  *     zn: The zenoh-net session. The caller keeps its ownership.
  *     keyexpr: The resource key to subscribe.
  *     callback: The callback function that will be called each time a data matching the subscribed resource is
- * received. arg: A pointer that will be passed to the **callback** on each call.
+ * received.
+ *     dropper: A function that will be called once subscriber is undeclared.
+ *     arg: A pointer that will be passed to the **callback** on each call.
  *
  * Returns:
  *    0 in case of success, negative error code otherwise.
@@ -170,6 +172,10 @@ z_result_t _z_declare_subscriber(_z_subscriber_t *subscriber, const _z_session_r
                                  const _z_declared_keyexpr_t *keyexpr, _z_closure_sample_callback_t callback,
                                  _z_drop_handler_t dropper, void *arg, z_locality_t allowed_origin);
 
+z_result_t _z_register_subscriber(uint32_t *out_sub_id, const _z_session_rc_t *zn, const _z_declared_keyexpr_t *keyexpr,
+                                  _z_closure_sample_callback_t callback, _z_drop_handler_t dropper, void *arg,
+                                  z_locality_t allowed_origin, const _z_sync_group_t *opt_callback_drop_sync_group);
+
 /**
  * Undeclare a :c:type:`_z_subscriber_t`.
  *
@@ -202,6 +208,11 @@ z_result_t _z_declare_queryable(_z_queryable_t *queryable, const _z_session_rc_t
                                 _z_closure_query_callback_t callback, _z_drop_handler_t dropper, void *arg,
                                 z_locality_t allowed_origin);
 
+z_result_t _z_register_queryable(uint32_t *queryable_id, const _z_session_rc_t *zn,
+                                 const _z_declared_keyexpr_t *keyexpr, bool complete,
+                                 _z_closure_query_callback_t callback, _z_drop_handler_t dropper, void *arg,
+                                 z_locality_t allowed_origin, const _z_sync_group_t *sync_group);
+
 /**
  * Undeclare a :c:type:`_z_queryable_t`.
  *
@@ -292,6 +303,7 @@ z_result_t _z_undeclare_querier(_z_querier_t *querier);
  *
  * Parameters:
  *     session: The zenoh-net session.
+ *     querier_id: Optional id of querier.
  *     keyexpr: The resource key to query.
  *     parameters: An indication to matching queryables about the queried data.
  *     parameters_len: Length of the parameters string.
@@ -309,12 +321,12 @@ z_result_t _z_undeclare_querier(_z_querier_t *querier);
  *     opt_cancellation_token: Optional cancellation token to cancel the query, can be null.
  *
  */
-z_result_t _z_query(const _z_session_rc_t *session, const _z_declared_keyexpr_t *keyexpr, const char *parameters,
-                    size_t parameters_len, z_query_target_t target, z_consolidation_mode_t consolidation,
-                    _z_bytes_t *payload, _z_encoding_t *encoding, _z_closure_reply_callback_t callback,
-                    _z_drop_handler_t dropper, void *arg, uint64_t timeout_ms, _z_bytes_t *attachment, _z_n_qos_t qos,
-                    _z_source_info_t *source_info, z_locality_t allowed_destination,
-                    _z_cancellation_token_rc_t *opt_cancellation_token);
+z_result_t _z_query(const _z_session_rc_t *session, _z_optional_id_t querier_id, const _z_declared_keyexpr_t *keyexpr,
+                    const char *parameters, size_t parameters_len, z_query_target_t target,
+                    z_consolidation_mode_t consolidation, _z_bytes_t *payload, _z_encoding_t *encoding,
+                    _z_closure_reply_callback_t callback, _z_drop_handler_t dropper, void *arg, uint64_t timeout_ms,
+                    _z_bytes_t *attachment, _z_n_qos_t qos, _z_source_info_t *source_info,
+                    z_locality_t allowed_destination, _z_cancellation_token_rc_t *opt_cancellation_token);
 #endif
 
 #if Z_FEATURE_INTEREST == 1

include/zenoh-pico/net/query.h

@@ -60,7 +60,7 @@ _Z_REFCOUNT_DEFINE(_z_query, _z_query)
  */
 typedef struct _z_querier_t {
     _z_declared_keyexpr_t _key;
-    _z_zint_t _id;
+    uint32_t _id;
     _z_session_weak_t _zn;
     _z_encoding_t _encoding;
     z_consolidation_mode_t _consolidation_mode;
@@ -86,6 +86,7 @@ static inline bool _z_querier_check(const _z_querier_t *querier) { return !_Z_RC
 typedef struct {
     uint32_t _entity_id;
     _z_session_weak_t _zn;
+    _z_sync_group_t _callback_drop_sync_group;
 } _z_queryable_t;
 
 #if Z_FEATURE_QUERYABLE == 1

include/zenoh-pico/net/session.h

@@ -126,6 +126,7 @@ typedef struct _z_session_t {
     uint32_t _admin_space_queryable_id;
 #endif
 #endif
+    _z_sync_group_t _callback_drop_sync_group;
 } _z_session_t;
 
 /**

include/zenoh-pico/net/subscribe.h

@@ -30,6 +30,7 @@ extern "C" {
 typedef struct {
     uint32_t _entity_id;
     _z_session_weak_t _zn;
+    _z_sync_group_t _callback_drop_sync_group;
 } _z_subscriber_t;
 
 #if Z_FEATURE_SUBSCRIPTION == 1

include/zenoh-pico/session/query.h

@@ -34,6 +34,7 @@ z_result_t _z_trigger_query_reply_err(_z_session_t *zn, _z_zint_t id, _z_msg_err
                                       _z_entity_global_id_t *replier_id);
 z_result_t _z_trigger_query_reply_final(_z_session_t *zn, _z_zint_t id);
 void _z_unregister_pending_query(_z_session_t *zn, _z_zint_t query_id);
+void _z_unregister_pending_queries_from_querier(_z_session_t *zn, uint32_t querier_id);
 void _z_flush_pending_queries(_z_session_t *zn);
 #endif
 

include/zenoh-pico/session/session.h

@@ -82,6 +82,8 @@ typedef struct {
     _z_closure_sample_callback_t _callback;
     _z_drop_handler_t _dropper;
     void *_arg;
+    _z_sync_group_notifier_t _session_callback_drop_notifier;
+    _z_sync_group_notifier_t _subscriber_callback_drop_notifier;
 } _z_subscription_t;
 
 bool _z_subscription_eq(const _z_subscription_t *one, const _z_subscription_t *two);
@@ -115,6 +117,8 @@ typedef struct {
     void *_arg;
     bool _complete;
     z_locality_t _allowed_origin;
+    _z_sync_group_notifier_t _session_callback_drop_notifier;
+    _z_sync_group_notifier_t _queryable_callback_drop_notifier;
 } _z_session_queryable_t;
 
 bool _z_session_queryable_eq(const _z_session_queryable_t *one, const _z_session_queryable_t *two);
@@ -166,6 +170,7 @@ void _z_pending_query_cancellation_data_clear(_z_pending_query_cancellation_data
 #endif
 struct _z_pending_query_t {
     _z_keyexpr_t _key;
+    _z_optional_id_t _querier_id;
     _z_zint_t _id;
     _z_closure_reply_callback_t _callback;
     _z_drop_handler_t _dropper;

src/api/api.c

@@ -737,6 +737,7 @@ static z_result_t _z_session_rc_init(z_owned_session_t *zs, _z_id_t *zid) {
 }
 
 z_result_t z_open(z_owned_session_t *zs, z_moved_config_t *config, const z_open_options_t *options) {
+    z_internal_session_null(zs);
 #if Z_FEATURE_MULTI_THREAD == 1
     z_open_options_t opts;
     if (options == NULL) {
@@ -1337,26 +1338,32 @@ z_entity_global_id_t z_publisher_id(const z_loaned_publisher_t *publisher) {
 #if Z_FEATURE_MATCHING == 1
 z_result_t z_publisher_declare_background_matching_listener(const z_loaned_publisher_t *publisher,
                                                             z_moved_closure_matching_status_t *callback) {
-    z_owned_matching_listener_t listener;
-    _Z_RETURN_IF_ERR(z_publisher_declare_matching_listener(publisher, &listener, callback));
-    _z_matching_listener_clear(&listener._val);
-    return _Z_RES_OK;
+    return z_publisher_declare_matching_listener(publisher, NULL, callback);
 }
 
 z_result_t z_publisher_declare_matching_listener(const z_loaned_publisher_t *publisher,
                                                  z_owned_matching_listener_t *matching_listener,
                                                  z_moved_closure_matching_status_t *callback) {
-    matching_listener->_val = _z_matching_listener_null();
     _z_session_rc_t sess_rc = _z_session_weak_upgrade_if_open(&publisher->_zn);
+    z_result_t ret = _Z_RES_OK;
     if (!_Z_RC_IS_NULL(&sess_rc)) {
-        matching_listener->_val = _z_matching_listener_declare(
-            &publisher->_filter.ctx, _z_get_entity_id(_Z_RC_IN_VAL(&sess_rc)), &callback->_this._val);
+        if (matching_listener != NULL) {
+            matching_listener->_val = _z_matching_listener_null();
+            ret = _z_matching_listener_declare(&matching_listener->_val, _Z_RC_IN_VAL(&sess_rc),
+                                               &publisher->_filter.ctx, &callback->_this._val);
+        } else {
+            uint32_t _listener_id;
+            ret = _z_matching_listener_register(&_listener_id, _Z_RC_IN_VAL(&sess_rc), &publisher->_filter.ctx,
+                                                &callback->_this._val);
+        }
         _z_session_rc_drop(&sess_rc);
     } else {
-        matching_listener->_val = _z_matching_listener_null();
-        z_internal_closure_matching_status_null(&callback->_this);
+        if (matching_listener != NULL) {
+            matching_listener->_val = _z_matching_listener_null();
+        }
+        ret = _Z_ERR_GENERIC;
     }
-    return _z_matching_listener_check(&matching_listener->_val) ? _Z_RES_OK : _Z_ERR_GENERIC;
+    return ret;
 }
 
 z_result_t z_publisher_get_matching_status(const z_loaned_publisher_t *publisher,
@@ -1433,10 +1440,10 @@ z_result_t z_get_with_parameters_substr(const z_loaned_session_t *zs, const z_lo
 #if Z_FEATURE_LOCAL_QUERYABLE == 1
     allowed_destination = opt.allowed_destination;
 #endif
-    ret = _z_query(zs, keyexpr, parameters, parameters_len, opt.target, opt.consolidation.mode,
-                   _z_bytes_from_moved(opt.payload), _z_encoding_from_moved(opt.encoding), closure.call, closure.drop,
-                   closure.context, opt.timeout_ms, _z_bytes_from_moved(opt.attachment), qos, source_info,
-                   allowed_destination, cancellation_token);
+    ret = _z_query(zs, _z_optional_id_make_none(), keyexpr, parameters, parameters_len, opt.target,
+                   opt.consolidation.mode, _z_bytes_from_moved(opt.payload), _z_encoding_from_moved(opt.encoding),
+                   closure.call, closure.drop, closure.context, opt.timeout_ms, _z_bytes_from_moved(opt.attachment),
+                   qos, source_info, allowed_destination, cancellation_token);
     // Clean-up
 #ifdef Z_FEATURE_UNSTABLE_API
     z_cancellation_token_drop(opt.cancellation_token);
@@ -1557,10 +1564,11 @@ z_result_t z_querier_get_with_parameters_substr(const z_loaned_querier_t *querie
     if (should_proceed) {
         _z_n_qos_t qos = _z_n_qos_make(querier->_is_express, querier->_congestion_control == Z_CONGESTION_CONTROL_BLOCK,
                                        querier->_priority);
-        ret = _z_query(&sess_rc, &querier->_key, parameters, parameters_len, querier->_target,
-                       querier->_consolidation_mode, _z_bytes_from_moved(opt.payload), &encoding, closure.call,
-                       closure.drop, closure.context, querier->_timeout_ms, _z_bytes_from_moved(opt.attachment), qos,
-                       source_info, querier->_allowed_destination, cancellation_token);
+        ret = _z_query(&sess_rc, _z_optional_id_make_some(querier->_id), &querier->_key, parameters, parameters_len,
+                       querier->_target, querier->_consolidation_mode, _z_bytes_from_moved(opt.payload), &encoding,
+                       closure.call, closure.drop, closure.context, querier->_timeout_ms,
+                       _z_bytes_from_moved(opt.attachment), qos, source_info, querier->_allowed_destination,
+                       cancellation_token);
     } else if (closure.drop != NULL) {
         closure.drop(closure.context);
     }
@@ -1613,25 +1621,31 @@ z_entity_global_id_t z_querier_id(const z_loaned_querier_t *querier) {
 #if Z_FEATURE_MATCHING == 1
 z_result_t z_querier_declare_background_matching_listener(const z_loaned_querier_t *querier,
                                                           z_moved_closure_matching_status_t *callback) {
-    z_owned_matching_listener_t listener;
-    _Z_RETURN_IF_ERR(z_querier_declare_matching_listener(querier, &listener, callback));
-    _z_matching_listener_clear(&listener._val);
-    return _Z_RES_OK;
+    return z_querier_declare_matching_listener(querier, NULL, callback);
 }
 z_result_t z_querier_declare_matching_listener(const z_loaned_querier_t *querier,
                                                z_owned_matching_listener_t *matching_listener,
                                                z_moved_closure_matching_status_t *callback) {
-    matching_listener->_val = _z_matching_listener_null();
     _z_session_rc_t sess_rc = _z_session_weak_upgrade_if_open(&querier->_zn);
+    z_result_t ret = _Z_RES_OK;
     if (!_Z_RC_IS_NULL(&sess_rc)) {
-        matching_listener->_val = _z_matching_listener_declare(
-            &querier->_filter.ctx, _z_get_entity_id(_Z_RC_IN_VAL(&sess_rc)), &callback->_this._val);
+        if (matching_listener != NULL) {
+            matching_listener->_val = _z_matching_listener_null();
+            ret = _z_matching_listener_declare(&matching_listener->_val, _Z_RC_IN_VAL(&sess_rc), &querier->_filter.ctx,
+                                               &callback->_this._val);
+        } else {
+            uint32_t _listener_id;
+            ret = _z_matching_listener_register(&_listener_id, _Z_RC_IN_VAL(&sess_rc), &querier->_filter.ctx,
+                                                &callback->_this._val);
+        }
         _z_session_rc_drop(&sess_rc);
     } else {
-        matching_listener->_val = _z_matching_listener_null();
-        z_internal_closure_matching_status_null(&callback->_this);
+        if (matching_listener != NULL) {
+            matching_listener->_val = _z_matching_listener_null();
+        }
+        ret = _Z_ERR_GENERIC;
     }
-    return _z_matching_listener_check(&matching_listener->_val) ? _Z_RES_OK : _Z_ERR_GENERIC;
+    return ret;
 }
 
 z_result_t z_querier_get_matching_status(const z_loaned_querier_t *querier, z_matching_status_t *matching_status) {
@@ -1691,10 +1705,7 @@ void z_queryable_options_default(z_queryable_options_t *options) {
 
 z_result_t z_declare_background_queryable(const z_loaned_session_t *zs, const z_loaned_keyexpr_t *keyexpr,
                                           z_moved_closure_query_t *callback, const z_queryable_options_t *options) {
-    z_owned_queryable_t qle;
-    _Z_RETURN_IF_ERR(z_declare_queryable(zs, &qle, keyexpr, callback, options));
-    _z_queryable_clear(&qle._val);
-    return _Z_RES_OK;
+    return z_declare_queryable(zs, NULL, keyexpr, callback, options);
 }
 
 z_result_t z_declare_queryable(const z_loaned_session_t *zs, z_owned_queryable_t *queryable,
@@ -1713,8 +1724,14 @@ z_result_t z_declare_queryable(const z_loaned_session_t *zs, z_owned_queryable_t
 #if Z_FEATURE_LOCAL_QUERYABLE == 1
     allowed_origin = opt.allowed_origin;
 #endif
-    return _z_declare_queryable(&queryable->_val, zs, keyexpr, opt.complete, closure.call, closure.drop,
-                                closure.context, allowed_origin);
+    if (queryable != NULL) {
+        return _z_declare_queryable(&queryable->_val, zs, keyexpr, opt.complete, closure.call, closure.drop,
+                                    closure.context, allowed_origin);
+    } else {
+        uint32_t _queryable_id;
+        return _z_register_queryable(&_queryable_id, zs, keyexpr, opt.complete, closure.call, closure.drop,
+                                     closure.context, allowed_origin, NULL);
+    }
 }
 
 z_result_t z_undeclare_queryable(z_moved_queryable_t *queryable) {
@@ -1979,18 +1996,14 @@ void z_subscriber_options_default(z_subscriber_options_t *options) {
 
 z_result_t z_declare_background_subscriber(const z_loaned_session_t *zs, const z_loaned_keyexpr_t *keyexpr,
                                            z_moved_closure_sample_t *callback, const z_subscriber_options_t *options) {
-    z_owned_subscriber_t sub;
-    _Z_RETURN_IF_ERR(z_declare_subscriber(zs, &sub, keyexpr, callback, options));
-    _z_subscriber_clear(&sub._val);
-    return _Z_RES_OK;
+    return z_declare_subscriber(zs, NULL, keyexpr, callback, options);
 }
 
 z_result_t z_declare_subscriber(const z_loaned_session_t *zs, z_owned_subscriber_t *sub,
                                 const z_loaned_keyexpr_t *keyexpr, z_moved_closure_sample_t *callback,
                                 const z_subscriber_options_t *options) {
     _z_closure_sample_t closure = callback->_this._val;
     z_internal_closure_sample_null(&callback->_this);
-    sub->_val = _z_subscriber_null();
 
     z_subscriber_options_t opt;
     z_subscriber_options_default(&opt);
@@ -2002,7 +2015,15 @@ z_result_t z_declare_subscriber(const z_loaned_session_t *zs, z_owned_subscriber
 #if Z_FEATURE_LOCAL_SUBSCRIBER == 1
     allowed_origin = opt.allowed_origin;
 #endif
-    return _z_declare_subscriber(&sub->_val, zs, keyexpr, closure.call, closure.drop, closure.context, allowed_origin);
+    if (sub != NULL) {
+        sub->_val = _z_subscriber_null();
+        return _z_declare_subscriber(&sub->_val, zs, keyexpr, closure.call, closure.drop, closure.context,
+                                     allowed_origin);
+    } else {
+        uint32_t _sub_id;
+        return _z_register_subscriber(&_sub_id, zs, keyexpr, closure.call, closure.drop, closure.context,
+                                      allowed_origin, NULL);
+    }
 }
 
 z_result_t z_undeclare_subscriber(z_moved_subscriber_t *sub) {