improve trigger subscription performance

Author: DenisBiryukov91

include/zenoh-pico/collections/bytes.h

@@ -58,7 +58,11 @@ z_result_t _z_bytes_append_slice(_z_bytes_t *dst, _z_arc_slice_t *s);
 z_result_t _z_bytes_copy(_z_bytes_t *dst, const _z_bytes_t *src);
 _z_bytes_t _z_bytes_duplicate(const _z_bytes_t *src);
 z_result_t _z_bytes_move(_z_bytes_t *dst, _z_bytes_t *src);
-_z_bytes_t _z_bytes_steal(_z_bytes_t *src);
+static inline _z_bytes_t _z_bytes_steal(_z_bytes_t *src) {
+    _z_bytes_t b = *src;
+    *src = _z_bytes_null();
+    return b;
+}
 void _z_bytes_drop(_z_bytes_t *bytes);
 void _z_bytes_free(_z_bytes_t **bs);
 size_t _z_bytes_num_slices(const _z_bytes_t *bs);

include/zenoh-pico/net/encoding.h

@@ -42,7 +42,11 @@ void _z_encoding_clear(_z_encoding_t *encoding);
 z_result_t _z_encoding_copy(_z_encoding_t *dst, const _z_encoding_t *src);
 _z_encoding_t _z_encoding_alias(_z_encoding_t src);
 z_result_t _z_encoding_move(_z_encoding_t *dst, _z_encoding_t *src);
-_z_encoding_t _z_encoding_steal(_z_encoding_t *val);
+static inline _z_encoding_t _z_encoding_steal(_z_encoding_t *val) {
+    _z_encoding_t ret = *val;
+    *val = _z_encoding_null();
+    return ret;
+}
 
 #ifdef __cplusplus
 }

include/zenoh-pico/protocol/keyexpr.h

@@ -50,7 +50,12 @@ _z_keyexpr_t *_z_keyexpr_clone(const _z_keyexpr_t *src);
 /// or keyexpr defined by its suffix only, with 0 id and no mapping. This is to be used only when forwarding
 /// keyexpr in user api to properly separate declared keyexpr from its suffix.
 _z_keyexpr_t _z_keyexpr_alias_from_user_defined(_z_keyexpr_t src, bool try_declared);
-_z_keyexpr_t _z_keyexpr_steal(_Z_MOVE(_z_keyexpr_t) src);
+static inline _z_keyexpr_t _z_keyexpr_steal(_Z_MOVE(_z_keyexpr_t) src) {
+    _z_keyexpr_t stolen = *src;
+    *src = _z_keyexpr_null();
+    return stolen;
+}
+
 bool _z_keyexpr_equals(const _z_keyexpr_t *left, const _z_keyexpr_t *right);
 z_result_t _z_keyexpr_move(_z_keyexpr_t *dst, _z_keyexpr_t *src);
 void _z_keyexpr_clear(_z_keyexpr_t *rk);

src/collections/bytes.c

@@ -174,12 +174,6 @@ z_result_t _z_bytes_move(_z_bytes_t *dst, _z_bytes_t *src) {
     return _Z_RES_OK;
 }
 
-_z_bytes_t _z_bytes_steal(_z_bytes_t *src) {
-    _z_bytes_t b = *src;
-    *src = _z_bytes_null();
-    return b;
-}
-
 _z_bytes_reader_t _z_bytes_get_reader(const _z_bytes_t *bytes) {
     _z_bytes_reader_t r;
     r.bytes = bytes;

src/net/encoding.c

@@ -75,12 +75,3 @@ z_result_t _z_encoding_move(_z_encoding_t *dst, _z_encoding_t *src) {
     }
     return _Z_RES_OK;
 }
-
-_z_encoding_t _z_encoding_steal(_z_encoding_t *val) {
-    _z_encoding_t ret = {
-        .id = val->id,
-        .schema = _z_string_steal(&val->schema),
-    };
-    val->id = _Z_ENCODING_ID_DEFAULT;
-    return ret;
-}

src/protocol/keyexpr.c

@@ -102,12 +102,6 @@ _z_keyexpr_t *_z_keyexpr_clone(const _z_keyexpr_t *src) {
     return dst;
 }
 
-_z_keyexpr_t _z_keyexpr_steal(_Z_MOVE(_z_keyexpr_t) src) {
-    _z_keyexpr_t stolen = *src;
-    *src = _z_keyexpr_null();
-    return stolen;
-}
-
 z_result_t _z_keyexpr_move(_z_keyexpr_t *dst, _z_keyexpr_t *src) {
     dst->_id = src->_id;
     dst->_mapping = src->_mapping;

src/session/subscription.c

@@ -246,54 +246,45 @@ static z_result_t _z_trigger_subscriptions_inner(_z_session_t *zn, _z_subscriber
     // Retrieve sub infos
     _z_subscription_cache_data_t sub_infos = _z_subscription_cache_data_null();
     sub_infos.ke_in = _z_keyexpr_alias(keyexpr);
-    _Z_RETURN_IF_ERR(_z_subscription_get_infos(zn, sub_kind, &sub_infos));
+    _Z_CLEAN_RETURN_IF_ERR(_z_subscription_get_infos(zn, sub_kind, &sub_infos), _z_keyexpr_clear(keyexpr);
+                           _z_encoding_clear(encoding); _z_bytes_drop(payload); _z_bytes_drop(attachment););
     // Check if there are subs
     _Z_DEBUG("Triggering %ju subs for key %d - %.*s", (uintmax_t)sub_infos.sub_nb, sub_infos.ke_out._id,
              (int)_z_string_len(&sub_infos.ke_out._suffix), _z_string_data(&sub_infos.ke_out._suffix));
-    if (sub_infos.sub_nb == 0) {
-        _z_keyexpr_clear(&sub_infos.ke_out);
-#if Z_FEATURE_RX_CACHE == 0
-        _z_subscription_infos_svec_release(&sub_infos.infos);  // Otherwise it's released with cache
-#endif
-        return _Z_RES_OK;
-    }
-    z_result_t ret = _Z_RES_OK;
     // Create sample
     _z_sample_t sample =
         _z_sample_alias(&sub_infos.ke_out, payload, timestamp, encoding, sample_kind, qos, attachment, reliability);
+
+    z_result_t ret = _Z_RES_OK;
     // Parse subscription infos svec
-    for (size_t i = 1; i < sub_infos.sub_nb; i++) {
-        _z_sample_t sample_copy;
-        ret = _z_sample_copy(&sample_copy, &sample);
-        if (ret != _Z_RES_OK) {
-            break;
-        }
+    for (size_t i = 0; i < sub_infos.sub_nb; i++) {
         _z_subscription_infos_t *sub_info = _z_subscription_infos_svec_get(&sub_infos.infos, i);
-        sub_info->callback(&sample_copy, sub_info->arg);
-        _z_sample_clear(&sample_copy);
+        if (i + 1 == sub_infos.sub_nb) {
+            sub_info->callback(&sample, sub_info->arg);
+        } else {
+            _z_sample_t sample_copy;
+            ret = _z_sample_copy(&sample_copy, &sample);
+            if (ret != _Z_RES_OK) {
+                break;
+            }
+            sub_info->callback(&sample_copy, sub_info->arg);
+            _z_sample_clear(&sample_copy);
+        }
     }
-    _z_subscription_infos_t *sub_info = _z_subscription_infos_svec_get(&sub_infos.infos, 0);
-    sub_info->callback(&sample, sub_info->arg);
     _z_sample_clear(&sample);
 #if Z_FEATURE_RX_CACHE == 0
-    _z_subscription_infos_svec_release(&sub_infos.infos);  // Otherwise it's released with cache
+    _z_subscription_infos_svec_release(&sub_infos.infos);
 #endif
-
+    _z_keyexpr_clear(keyexpr);
     return ret;
 }
 
 z_result_t _z_trigger_subscriptions_impl(_z_session_t *zn, _z_subscriber_kind_t sub_kind, _z_keyexpr_t *keyexpr,
                                          _z_bytes_t *payload, _z_encoding_t *encoding, const _z_zint_t sample_kind,
                                          const _z_timestamp_t *timestamp, const _z_n_qos_t qos, _z_bytes_t *attachment,
                                          z_reliability_t reliability) {
-    z_result_t ret = _z_trigger_subscriptions_inner(zn, sub_kind, keyexpr, payload, encoding, sample_kind, timestamp,
-                                                    qos, attachment, reliability);
-    // Clean up
-    _z_keyexpr_clear(keyexpr);
-    _z_encoding_clear(encoding);
-    _z_bytes_drop(payload);
-    _z_bytes_drop(attachment);
-    return ret;
+    return _z_trigger_subscriptions_inner(zn, sub_kind, keyexpr, payload, encoding, sample_kind, timestamp, qos,
+                                          attachment, reliability);
 }
 
 void _z_unregister_subscription(_z_session_t *zn, _z_subscriber_kind_t kind, _z_subscription_rc_t *sub) {