loopback: support locality and local delivery for same-session messages

- Deliver PUT/DELETE, query, reply, and reply-final messages for a key
expression that is defined in the same local session locally, without
causing network activity.
- Extend z_api_source_info_test to validate deliveries both to remote and
local subscribers.
- Support locality of samples to be received by subscribers or targeted by
publishers.
- Enable the locality-specific CMake options in CI.

User can now set allowed_origin / allowed_destination in
z_put_options_t, z_publisher_put_options_t, z_subscriber_options_t,
z_queryable_options_t, etc. with the following values:
- ZP_LOCALITY_SESSION_LOCAL - stay inside the session (don't send
  anything to network)
- ZP_LOCALITY_REMOTE - send/accept only over the transport (no
  same-session samples)
- ZP_LOCALITY_ANY - allow both,
exactly as with zenoh-c's zc_locality_t.

Co-developed-by: Denis Biryukov <[email protected]>

Author: steils

.github/workflows/ci.yml

@@ -32,7 +32,7 @@ jobs:
       - name: Build & run tests
         run: |
           sudo apt install -y ninja-build libmbedtls-dev
-          Z_FEATURE_LINK_TLS=1 Z_FEATURE_UNSTABLE_API=1 Z_FEATURE_PERIODIC_TASKS=1 CMAKE_GENERATOR=Ninja ASAN=ON make BUILD_TYPE=Debug test
+          Z_FEATURE_LINK_TLS=1 Z_FEATURE_LOCAL_QUERYABLE=1 Z_FEATURE_LOCAL_SUBSCRIBER=1 Z_FEATURE_UNSTABLE_API=1 Z_FEATURE_PERIODIC_TASKS=1 CMAKE_GENERATOR=Ninja ASAN=ON make BUILD_TYPE=Debug test
 
   run_windows_test:
     name: Run peer unicast test on windows-latest
@@ -62,7 +62,7 @@ jobs:
       - name: Build tests
         run: |
           sudo apt install -y ninja-build libmbedtls-dev
-          Z_FEATURE_LINK_TLS=1 Z_FEATURE_UNSTABLE_API=1 Z_FEATURE_PERIODIC_TASKS=1 CMAKE_GENERATOR=Ninja make BUILD_TYPE=Debug
+          Z_FEATURE_LINK_TLS=1 Z_FEATURE_LOCAL_QUERYABLE=1 Z_FEATURE_LOCAL_SUBSCRIBER=1 Z_FEATURE_UNSTABLE_API=1 Z_FEATURE_PERIODIC_TASKS=1 CMAKE_GENERATOR=Ninja make BUILD_TYPE=Debug
 
       - name: Install valgrind
         run: |
@@ -423,7 +423,7 @@ jobs:
       - name: Build project
         run: |
           sudo apt install -y ninja-build
-          Z_FEATURE_UNSTABLE_API=1 Z_FEATURE_LIVELINESS=1 CMAKE_GENERATOR=Ninja make
+          Z_FEATURE_LINK_TLS=1 Z_FEATURE_LOCAL_QUERYABLE=1 Z_FEATURE_LOCAL_SUBSCRIBER=1 Z_FEATURE_UNSTABLE_API=1 Z_FEATURE_LIVELINESS=1 CMAKE_GENERATOR=Ninja make
 
       - name: Run test
         run: python3 -u ./build/tests/memory_leak.py

.github/workflows/integration.yaml

@@ -47,6 +47,7 @@ jobs:
           BUILD_INTEGRATION: ON
           Z_FEATURE_LINK_TLS: 1
           Z_FEATURE_UNSTABLE_API: 1
+          Z_FEATURE_LOCAL_QUERYABLE: 1
           Z_FEATURE_LOCAL_SUBSCRIBER: 1
           Z_FEATURE_PERIODIC_TASKS: 1
           Z_FEATURE_ADVANCED_PUBLICATION: 1
@@ -60,8 +61,9 @@ jobs:
           BUILD_TESTING: OFF # Workaround for Windows as it seems the previous step is being ignored
           BUILD_INTEGRATION: ON # Workaround for Windows as it seems the previous step is being ignored
           Z_FEATURE_LINK_TLS: 1
-          Z_FEATURE_UNSTABLE_API: 1
+          Z_FEATURE_LOCAL_QUERYABLE: 1
           Z_FEATURE_LOCAL_SUBSCRIBER: 1
+          Z_FEATURE_UNSTABLE_API: 1
           ZENOH_BRANCH: main
 
   asan-build:

CMakeLists.txt

@@ -278,6 +278,7 @@ set(Z_FEATURE_UNICAST_PEER 1 CACHE STRING "Toggle Unicast peer mode")
 set(Z_FEATURE_AUTO_RECONNECT 1 CACHE STRING "Toggle automatic reconnection")
 set(Z_FEATURE_MULTICAST_DECLARATIONS 0 CACHE STRING "Toggle multicast resource declarations")
 set(Z_FEATURE_PERIODIC_TASKS 0 CACHE STRING "Toggle periodic task support")
+set(Z_FEATURE_LOCAL_QUERYABLE 0 CACHE STRING "Toggle local queriables")
 
 # Add a warning message if someone tries to enable Z_FEATURE_LINK_SERIAL_USB directly
 if(Z_FEATURE_LINK_SERIAL_USB AND NOT Z_FEATURE_UNSTABLE_API)
@@ -695,6 +696,7 @@ if(UNIX OR MSVC)
       add_executable(z_api_queryable_test ${PROJECT_SOURCE_DIR}/tests/z_api_queryable_test.c)
       add_executable(z_api_scheduler_test ${PROJECT_SOURCE_DIR}/tests/z_api_scheduler_test.c)
       add_executable(z_api_cancellation_test ${PROJECT_SOURCE_DIR}/tests/z_api_cancellation_test.c)
+      add_executable(z_local_loopback_test ${PROJECT_SOURCE_DIR}/tests/z_local_loopback_test.c)
 if(UNIX)
       add_executable(z_api_advanced_pubsub_test ${PROJECT_SOURCE_DIR}/tests/z_api_advanced_pubsub_test.c
                                                 ${PROJECT_SOURCE_DIR}/tests/utils/tcp_proxy.c)
@@ -712,6 +714,15 @@ endif()
       target_link_libraries(z_api_source_info_test zenohpico::lib)
       target_link_libraries(z_api_queryable_test zenohpico::lib)
       target_link_libraries(z_api_scheduler_test zenohpico::lib)
+      target_link_libraries(z_local_loopback_test zenohpico::lib)
+      target_include_directories(z_local_loopback_test PRIVATE ${PROJECT_SOURCE_DIR}/include)
+      target_compile_definitions(z_local_loopback_test PRIVATE Z_LOOPBACK_TESTING=1)
+      if(PICO_SHARED)
+        target_compile_definitions(${Libname}_shared PRIVATE Z_LOOPBACK_TESTING=1)
+      endif()
+      if(PICO_STATIC)
+        target_compile_definitions(${Libname}_static PRIVATE Z_LOOPBACK_TESTING=1)
+      endif()
       target_link_libraries(z_api_advanced_pubsub_test zenohpico::lib)
       target_link_libraries(z_api_cancellation_test zenohpico::lib)
 
@@ -735,6 +746,7 @@ endif()
       add_test(z_api_source_info_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_source_info_test)
       add_test(z_api_queryable_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_queryable_test)
       add_test(z_api_scheduler_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_scheduler_test)
+      add_test(z_local_loopback_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_local_loopback_test)
       add_test(z_api_advanced_pubsub_test bash ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/api.sh z_api_advanced_pubsub_test)
     endif()
   endif()

docs/api.rst

@@ -1128,6 +1128,7 @@ Constants
 .. autocenum:: constants.h::z_congestion_control_t
 .. autocenum:: constants.h::z_priority_t
 .. autocenum:: constants.h::z_reliability_t
+.. autocenum:: constants.h::z_locality_t
 
 Functions
 ---------

include/zenoh-pico/api/constants.h

@@ -54,6 +54,22 @@ typedef enum z_whatami_t {
 } z_whatami_t;
 #define Z_WHATAMI_DEFAULT Z_WHATAMI_ROUTER;
 
+/**
+ * The locality of samples to be received by subscribers or targeted by publishers.
+ *
+ * Enumerators:
+ *   Z_LOCALITY_ANY: Allow both session-local and remote traffic.
+ *   Z_LOCALITY_SESSION_LOCAL: Allow session-local traffic only.
+ *   Z_LOCALITY_REMOTE: Allow remote traffic only.
+ */
+typedef enum z_locality_t {
+    Z_LOCALITY_ANY = 0,
+    Z_LOCALITY_SESSION_LOCAL = 1,
+    Z_LOCALITY_REMOTE = 2,
+} z_locality_t;
+
+static inline z_locality_t z_locality_default(void) { return Z_LOCALITY_ANY; }
+
 /**
  * Status values for keyexpr canonization operation.
  * Used as return value of canonization-related functions,

include/zenoh-pico/api/types.h

@@ -165,7 +165,11 @@ typedef _z_matching_status_t z_matching_status_t;
  * Represents the configuration used to configure a subscriber upon declaration :c:func:`z_declare_subscriber`.
  */
 typedef struct {
-    uint8_t __dummy;  // Just to avoid empty structures that might cause undefined behavior
+#if Z_FEATURE_LOCAL_SUBSCRIBER == 1
+    z_locality_t allowed_origin;
+#else
+    uint8_t __dummy;  // keep struct non-empty when locality is disabled
+#endif
 } z_subscriber_options_t;
 
 /**
@@ -218,6 +222,9 @@ typedef struct {
 #ifdef Z_FEATURE_UNSTABLE_API
     z_reliability_t reliability;
 #endif
+#if Z_FEATURE_LOCAL_SUBSCRIBER == 1
+    z_locality_t allowed_destination;
+#endif
 } z_publisher_options_t;
 
 /**
@@ -241,6 +248,9 @@ typedef struct z_querier_options_t {
     z_query_consolidation_t consolidation;
     z_congestion_control_t congestion_control;
     bool is_express;
+#if Z_FEATURE_LOCAL_QUERYABLE == 1
+    z_locality_t allowed_destination;
+#endif
     z_priority_t priority;
     uint64_t timeout_ms;
 } z_querier_options_t;
@@ -271,6 +281,9 @@ typedef struct z_querier_get_options_t {
  */
 typedef struct {
     bool complete;
+#if Z_FEATURE_LOCAL_QUERYABLE == 1
+    z_locality_t allowed_origin;
+#endif
 } z_queryable_options_t;
 
 /**
@@ -349,6 +362,9 @@ typedef struct {
     z_timestamp_t *timestamp;
     bool is_express;
     z_moved_bytes_t *attachment;
+#if Z_FEATURE_LOCAL_SUBSCRIBER == 1
+    z_locality_t allowed_destination;
+#endif
 #ifdef Z_FEATURE_UNSTABLE_API
     z_reliability_t reliability;
     z_moved_source_info_t *source_info;
@@ -371,6 +387,9 @@ typedef struct {
     z_priority_t priority;
     bool is_express;
     z_timestamp_t *timestamp;
+#if Z_FEATURE_LOCAL_SUBSCRIBER == 1
+    z_locality_t allowed_destination;
+#endif
 #ifdef Z_FEATURE_UNSTABLE_API
     z_reliability_t reliability;
     z_moved_source_info_t *source_info;
@@ -433,6 +452,9 @@ typedef struct {
     z_congestion_control_t congestion_control;
     z_priority_t priority;
     bool is_express;
+#if Z_FEATURE_LOCAL_QUERYABLE == 1
+    z_locality_t allowed_destination;
+#endif
     z_query_target_t target;
     uint64_t timeout_ms;
     z_moved_bytes_t *attachment;

include/zenoh-pico/config.h

@@ -54,6 +54,7 @@
 #define Z_FEATURE_ENCODING_VALUES 1
 #define Z_FEATURE_TCP_NODELAY 1
 #define Z_FEATURE_LOCAL_SUBSCRIBER 0
+#define Z_FEATURE_LOCAL_QUERYABLE 0
 #define Z_FEATURE_SESSION_CHECK 1
 #define Z_FEATURE_BATCHING 1
 #define Z_FEATURE_BATCH_TX_MUTEX 0

include/zenoh-pico/config.h.in

@@ -54,6 +54,7 @@
 #define Z_FEATURE_ENCODING_VALUES @Z_FEATURE_ENCODING_VALUES@
 #define Z_FEATURE_TCP_NODELAY @Z_FEATURE_TCP_NODELAY@
 #define Z_FEATURE_LOCAL_SUBSCRIBER @Z_FEATURE_LOCAL_SUBSCRIBER@
+#define Z_FEATURE_LOCAL_QUERYABLE @Z_FEATURE_LOCAL_QUERYABLE@
 #define Z_FEATURE_SESSION_CHECK @Z_FEATURE_SESSION_CHECK@
 #define Z_FEATURE_BATCHING @Z_FEATURE_BATCHING@
 #define Z_FEATURE_BATCH_TX_MUTEX @Z_FEATURE_BATCH_TX_MUTEX@

include/zenoh-pico/net/filtering.h

@@ -19,6 +19,7 @@
 #include "zenoh-pico/api/constants.h"
 #include "zenoh-pico/net/session.h"
 #include "zenoh-pico/protocol/core.h"
+#include "zenoh-pico/utils/locality.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -39,6 +40,13 @@ typedef enum {
     WRITE_FILTER_OFF = 1,
 } _z_write_filter_state_t;
 
+struct _z_write_filter_registration_t;
+
+typedef enum {
+    _Z_WRITE_FILTER_SUBSCRIBER = 0,
+    _Z_WRITE_FILTER_QUERYABLE = 1,
+} _z_write_filter_target_type_t;
+
 typedef struct {
     _z_session_weak_t zn;
 #if Z_FEATURE_MULTI_THREAD == 1
@@ -52,6 +60,11 @@ typedef struct {
     uint8_t state;
     bool is_complete;
     bool is_aggregate;
+    bool allow_local;
+    bool allow_remote;
+    _z_write_filter_target_type_t target_type;
+    size_t local_targets;
+    struct _z_write_filter_registration_t *registration;
 } _z_write_filter_ctx_t;
 
 z_result_t _z_write_filter_ctx_clear(_z_write_filter_ctx_t *filter);
@@ -67,8 +80,12 @@ typedef struct _z_write_filter_t {
 } _z_write_filter_t;
 
 z_result_t _z_write_filter_create(const _z_session_rc_t *zn, _z_write_filter_t *filter, _z_keyexpr_t keyexpr,
-                                  uint8_t interest_flag, bool complete);
+                                  uint8_t interest_flag, bool complete, z_locality_t locality);
 z_result_t _z_write_filter_clear(_z_write_filter_t *filter);
+void _z_write_filter_notify_subscriber(struct _z_session_t *session, const _z_keyexpr_t *key,
+                                       z_locality_t allowed_origin, bool add);
+void _z_write_filter_notify_queryable(struct _z_session_t *session, const _z_keyexpr_t *key,
+                                      z_locality_t allowed_origin, bool is_complete, bool add);
 
 #if Z_FEATURE_MATCHING
 z_result_t _z_write_filter_ctx_add_callback(_z_write_filter_ctx_t *filter, size_t id, _z_closure_matching_status_t *v);

include/zenoh-pico/net/primitives.h

@@ -110,7 +110,7 @@ _z_keyexpr_t _z_update_keyexpr_to_declared(_z_session_t *zs, _z_keyexpr_t keyexp
  */
 _z_publisher_t _z_declare_publisher(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr, _z_encoding_t *encoding,
                                     z_congestion_control_t congestion_control, z_priority_t priority, bool is_express,
-                                    z_reliability_t reliability);
+                                    z_reliability_t reliability, z_locality_t allowed_destination);
 
 /**
  * Undeclare a :c:type:`_z_publisher_t`.
@@ -141,13 +141,15 @@ z_result_t _z_undeclare_publisher(_z_publisher_t *pub);
  *     attachment: An optional attachment to this write.
  *     reliability: The message reliability.
  *     source_info: The message source info.
+ *     allowed_destination: The allowed destination locality.
  * Returns:
  *     ``0`` in case of success, ``-1`` in case of failure.
  */
 z_result_t _z_write(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const _z_bytes_t *payload,
                     const _z_encoding_t *encoding, const z_sample_kind_t kind, const z_congestion_control_t cong_ctrl,
                     z_priority_t priority, bool is_express, const _z_timestamp_t *timestamp,
-                    const _z_bytes_t *attachment, z_reliability_t reliability, const _z_source_info_t *source_info);
+                    const _z_bytes_t *attachment, z_reliability_t reliability, const _z_source_info_t *source_info,
+                    z_locality_t allowed_destination);
 #endif
 
 #if Z_FEATURE_SUBSCRIPTION == 1
@@ -165,7 +167,8 @@ z_result_t _z_write(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const _z_byte
  *    The created :c:type:`_z_subscriber_t` (in null state if the declaration failed).
  */
 _z_subscriber_t _z_declare_subscriber(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr,
-                                      _z_closure_sample_callback_t callback, _z_drop_handler_t dropper, void *arg);
+                                      _z_closure_sample_callback_t callback, _z_drop_handler_t dropper, void *arg,
+                                      z_locality_t allowed_origin);
 
 /**
  * Undeclare a :c:type:`_z_subscriber_t`.
@@ -195,7 +198,8 @@ z_result_t _z_undeclare_subscriber(_z_subscriber_t *sub);
  *    The created :c:type:`_z_queryable_t` (in null state if the declaration failed)..
  */
 _z_queryable_t _z_declare_queryable(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr, bool complete,
-                                    _z_closure_query_callback_t callback, _z_drop_handler_t dropper, void *arg);
+                                    _z_closure_query_callback_t callback, _z_drop_handler_t dropper, void *arg,
+                                    z_locality_t allowed_origin);
 
 /**
  * Undeclare a :c:type:`_z_queryable_t`.
@@ -270,7 +274,7 @@ z_result_t _z_send_reply_err(const _z_query_t *query, const _z_session_rc_t *zsr
 _z_querier_t _z_declare_querier(const _z_session_rc_t *zn, _z_keyexpr_t keyexpr,
                                 z_consolidation_mode_t consolidation_mode, z_congestion_control_t congestion_control,
                                 z_query_target_t target, z_priority_t priority, bool is_express, uint64_t timeout_ms,
-                                _z_encoding_t *encoding, z_reliability_t reliability);
+                                _z_encoding_t *encoding, z_reliability_t reliability, z_locality_t allowed_destination);
 
 /**
  * Undeclare a :c:type:`_z_querier_t`.
@@ -301,14 +305,15 @@ z_result_t _z_undeclare_querier(_z_querier_t *querier);
  *     timeout_ms: The timeout value of this query.
  *     attachment: An optional attachment to this query.
  *     qos: QoS to apply when routing this query.
+ *     allowed_destination: Locality restrictions for delivery.
  *     out_id: In case of success the query id will be written to it.
  *
  */
 z_result_t _z_query(_z_session_t *zn, const _z_keyexpr_t *keyexpr, const char *parameters, size_t parameters_len,
                     z_query_target_t target, z_consolidation_mode_t consolidation, const _z_bytes_t *payload,
                     const _z_encoding_t *encoding, _z_closure_reply_callback_t callback, _z_drop_handler_t dropper,
-                    void *arg, uint64_t timeout_ms, const _z_bytes_t *attachment, _z_n_qos_t qos, _z_zint_t *out_id);
-
+                    void *arg, uint64_t timeout_ms, const _z_bytes_t *attachment, _z_n_qos_t qos,
+                    z_locality_t allowed_destination, _z_zint_t *out_id);
 #endif
 
 #if Z_FEATURE_INTEREST == 1