add synchronization for callback drop for advanced subscribers

Author: DenisBiryukov91

include/zenoh-pico/api/advanced_subscriber.h

@@ -82,7 +82,22 @@ _Z_HASHMAP_DEFINE(_z_entity_global_id, _ze_advanced_subscriber_sequenced_state,
                   _ze_advanced_subscriber_sequenced_state_t)
 _Z_HASHMAP_DEFINE(_z_id, _ze_advanced_subscriber_timestamped_state, z_id_t, _ze_advanced_subscriber_timestamped_state_t)
 
-static void _ze_closure_miss_copy(_ze_closure_miss_t *dst, const _ze_closure_miss_t *src) { *dst = *src; }
+static inline _ze_closure_miss_t _ze_closure_miss_null(void) {
+    _ze_closure_miss_t miss = {0};
+    return miss;
+}
+static inline void _ze_closure_miss_drop(_ze_closure_miss_t *closure) {
+    if (closure->drop != NULL) {
+        closure->drop(closure->context);
+    }
+    *closure = _ze_closure_miss_null();
+}
+
+static inline void _ze_closure_miss_copy(_ze_closure_miss_t *dst, const _ze_closure_miss_t *src) { *dst = *src; }
+static inline void _ze_closure_miss_move(_ze_closure_miss_t *dst, _ze_closure_miss_t *src) {
+    *dst = *src;
+    *src = _ze_closure_miss_null();
+}
 
 _Z_ELEM_DEFINE(_ze_closure_miss, _ze_closure_miss_t, _z_noop_size, _z_noop_clear, _ze_closure_miss_copy, _z_noop_move,
                _z_noop_eq, _z_noop_cmp, _z_noop_hash)
@@ -111,6 +126,8 @@ typedef struct {
     _ze_closure_miss_intmap_t _miss_handlers;
     bool _has_token;
     z_owned_liveliness_token_t _token;
+    z_owned_cancellation_token_t _cancellation_token;
+    bool _is_undeclaring;
 } _ze_advanced_subscriber_state_t;
 
 _ze_advanced_subscriber_state_t _ze_advanced_subscriber_state_null(void);

include/zenoh-pico/net/session.h

@@ -261,7 +261,7 @@ z_result_t _zp_process_periodic_tasks(_z_session_t *z);
  * Returns:
  *     ``0`` in case of success, ``negative`` in case of failure.
  */
-z_result_t _zp_periodic_task_add(_z_session_t *z, const _zp_closure_periodic_task_t *closure, uint64_t period_ms,
+z_result_t _zp_periodic_task_add(_z_session_t *z, _zp_closure_periodic_task_t *closure, uint64_t period_ms,
                                  uint32_t *id);
 
 /*

include/zenoh-pico/utils/result.h

@@ -44,7 +44,6 @@ typedef enum {
     _Z_NO_DATA_PROCESSED = 3,
     Z_NO_DATA_PROCESSED = 3,
     _Z_RESOURCE_POSITIVE_REF_COUNT = 4,
-    Z_CANCELLATION_TOKEN_ALREADY_CANCELLED = 5,
     Z_SYNC_GROUP_CLOSED = 6,
 
     _Z_ERR_MESSAGE_DESERIALIZATION_FAILED = -119,
@@ -94,6 +93,7 @@ typedef enum {
     Z_EDESERIALIZE = -72,
     Z_ETIMEDOUT = -71,
     _Z_ERR_KEYEXPR_DECLARED_ON_ANOTHER_SESSION = -70,
+    Z_ERR_CANCELLED = -69,
 
     _Z_ERR_NULL = -127,
     _Z_ERR_GENERIC = -128

include/zenoh-pico/utils/scheduler.h

@@ -39,6 +39,18 @@ typedef struct {
     z_closure_drop_callback_t drop;
 } _zp_closure_periodic_task_t;
 
+static inline _zp_closure_periodic_task_t _zp_closure_periodic_task_null() {
+    _zp_closure_periodic_task_t c = {0};
+    return c;
+}
+
+static inline void _zp_closure_periodic_task_drop(_zp_closure_periodic_task_t *closure) {
+    if (closure->drop != NULL) {
+        closure->drop(closure->context);
+    };
+    *closure = _zp_closure_periodic_task_null();
+}
+
 typedef struct {
     uint32_t _id;
     uint64_t _period_ms;
@@ -101,7 +113,7 @@ static inline bool _zp_periodic_scheduler_check(const _zp_periodic_scheduler_t *
 }
 z_result_t _zp_periodic_scheduler_init(_zp_periodic_scheduler_t *scheduler);
 void _zp_periodic_scheduler_clear(_zp_periodic_scheduler_t *scheduler);
-z_result_t _zp_periodic_scheduler_add(_zp_periodic_scheduler_t *scheduler, const _zp_closure_periodic_task_t *closure,
+z_result_t _zp_periodic_scheduler_add(_zp_periodic_scheduler_t *scheduler, _zp_closure_periodic_task_t *closure,
                                       uint64_t period_ms, uint32_t *id);
 z_result_t _zp_periodic_scheduler_remove(_zp_periodic_scheduler_t *scheduler, uint32_t id);
 

src/api/advanced_publisher.c

@@ -357,9 +357,7 @@ z_result_t ze_declare_advanced_publisher(const z_loaned_session_t *zs, ze_owned_
             _Z_CLEAN_RETURN_IF_ERR(
                 _zp_periodic_task_add(_Z_RC_IN_VAL(zs), &closure, opt.sample_miss_detection.heartbeat_period_ms,
                                       &state->_state_publisher_task_id),
-
                 z_keyexpr_drop(z_keyexpr_move(&ke));
-                _ze_advanced_publisher_state_weak_drop(ctx); z_free(ctx);
                 _ze_advanced_publisher_state_rc_drop(&pub->_val._state);
                 z_publisher_drop(z_publisher_move(&pub->_val._publisher));
                 z_liveliness_token_drop(z_liveliness_token_move(&pub->_val._liveliness));