Add raweth transport (glue) (#286)

Author: jean-roland

.github/workflows/build-check.yaml

@@ -75,3 +75,23 @@ jobs:
       run: |
         docker stop zenoh_router
         docker rm zenoh_router
+  
+  raweth_build:
+    name: Build and test raweth transport on ubuntu-latest
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        feature_reth: [1, 0]
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Build project
+        run: |
+          sudo apt install -y ninja-build
+          CMAKE_GENERATOR=Ninja make
+          python3 ./build/tests/raweth.py --reth $Z_FEATURE_RAWETH_TRANSPORT
+        timeout-minutes: 5
+        env:
+          Z_FEATURE_RAWETH_TRANSPORT: ${{ matrix.feature_reth }}
+          

CMakeLists.txt

@@ -107,15 +107,18 @@ set(Z_FEATURE_PUBLICATION 1 CACHE STRING "Toggle publication feature")
 set(Z_FEATURE_SUBSCRIPTION 1 CACHE STRING "Toggle subscription feature")
 set(Z_FEATURE_QUERY 1 CACHE STRING "Toggle query feature")
 set(Z_FEATURE_QUERYABLE 1 CACHE STRING "Toggle queryable feature")
+set(Z_FEATURE_RAWETH_TRANSPORT 0 CACHE STRING "Toggle raw ethernet transport feature")
 add_definition(Z_FEATURE_PUBLICATION=${Z_FEATURE_PUBLICATION})
 add_definition(Z_FEATURE_SUBSCRIPTION=${Z_FEATURE_SUBSCRIPTION})
 add_definition(Z_FEATURE_QUERY=${Z_FEATURE_QUERY})
 add_definition(Z_FEATURE_QUERYABLE=${Z_FEATURE_QUERYABLE})
+add_definition(Z_FEATURE_RAWETH_TRANSPORT=${Z_FEATURE_RAWETH_TRANSPORT})
 message(STATUS "Building with feature confing:\n\
 * PUBLICATION: ${Z_FEATURE_PUBLICATION}\n\
 * SUBSCRIPTION: ${Z_FEATURE_SUBSCRIPTION}\n\
 * QUERY: ${Z_FEATURE_QUERY}\n\
-* QUERYABLE: ${Z_FEATURE_QUERYABLE}")
+* QUERYABLE: ${Z_FEATURE_QUERYABLE}\n\
+* RAWETH: ${Z_FEATURE_RAWETH_TRANSPORT}")
 
 # Print summary of CMAKE configurations
 message(STATUS "Building in ${CMAKE_BUILD_TYPE} mode")
@@ -300,6 +303,7 @@ if(UNIX OR MSVC)
     target_link_libraries(z_api_double_drop_test ${Libname})
 
     configure_file(${PROJECT_SOURCE_DIR}/tests/modularity.py ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/modularity.py COPYONLY)
+    configure_file(${PROJECT_SOURCE_DIR}/tests/raweth.py ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/raweth.py COPYONLY)
 
     enable_testing()
     add_test(z_data_struct_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_data_struct_test)

GNUmakefile

@@ -51,6 +51,7 @@ Z_FEATURE_PUBLICATION?=1
 Z_FEATURE_SUBSCRIPTION?=1
 Z_FEATURE_QUERY?=1
 Z_FEATURE_QUERYABLE?=1
+Z_FEATURE_RAWETH_TRANSPORT?=0
 
 # zenoh-pico/ directory
 ROOT_DIR:=$(shell dirname $(realpath $(firstword $(MAKEFILE_LIST))))
@@ -67,7 +68,7 @@ CROSSIMG_PREFIX=zenoh-pico_
 
 CMAKE_OPT=-DZENOH_DEBUG=$(ZENOH_DEBUG) -DBUILD_EXAMPLES=$(BUILD_EXAMPLES) -DCMAKE_BUILD_TYPE=$(BUILD_TYPE) -DBUILD_TESTING=$(BUILD_TESTING) -DBUILD_MULTICAST=$(BUILD_MULTICAST)\
  -DZ_FEATURE_PUBLICATION=$(Z_FEATURE_PUBLICATION) -DZ_FEATURE_SUBSCRIPTION=$(Z_FEATURE_SUBSCRIPTION) -DZ_FEATURE_QUERY=$(Z_FEATURE_QUERY) -DZ_FEATURE_QUERYABLE=$(Z_FEATURE_QUERYABLE)\
- -DBUILD_INTEGRATION=$(BUILD_INTEGRATION) -DBUILD_TOOLS=$(BUILD_TOOLS) -DBUILD_SHARED_LIBS=$(BUILD_SHARED_LIBS) -H.
+ -DZ_FEATURE_RAWETH_TRANSPORT=$(Z_FEATURE_RAWETH_TRANSPORT) -DBUILD_INTEGRATION=$(BUILD_INTEGRATION) -DBUILD_TOOLS=$(BUILD_TOOLS) -DBUILD_SHARED_LIBS=$(BUILD_SHARED_LIBS) -H.
 
 all: make
 

include/zenoh-pico/link/config/raweth.h

@@ -20,13 +20,9 @@
 #include "zenoh-pico/config.h"
 #include "zenoh-pico/link/link.h"
 
-#if Z_FEATURE_RAWETH_TRANSPORT == 1
-
 #define RAWETH_SCHEMA "reth"
 
 int8_t _z_endpoint_raweth_valid(_z_endpoint_t *endpoint);
-
 int8_t _z_new_link_raweth(_z_link_t *zl, _z_endpoint_t endpoint);
 
-#endif /* Z_FEATURE_RAWETH_TRANSPORT */
 #endif /* ZENOH_PICO_LINK_CONFIG_RAWETH_H */

include/zenoh-pico/link/link.h

@@ -28,6 +28,10 @@
 #include "zenoh-pico/system/link/udp.h"
 #endif
 
+#if Z_FEATURE_RAWETH_TRANSPORT == 1
+#include "zenoh-pico/system/link/raweth.h"
+#endif
+
 #if Z_FEATURE_LINK_BLUETOOTH == 1
 #include "zenoh-pico/system/link/bt.h"
 #endif
@@ -52,6 +56,7 @@
 typedef enum {
     Z_LINK_CAP_TRANSPORT_UNICAST = 0,
     Z_LINK_CAP_TRANSPORT_MULTICAST = 1,
+    Z_LINK_CAP_TRANSPORT_RAWETH = 2,
 } _z_link_cap_transport_t;
 
 /**
@@ -73,7 +78,7 @@ typedef enum {
  *     transport: 2 bits, see _z_link_cap_transport_t enum.
  *     flow: 1 bit, see _z_link_cap_flow_t enum.
  *     reliable: 1 bit, 1 if the link is reliable (network definition)
- *     reserved: 4 bits, reserved for futur use
+ *     reserved: 4 bits, reserved for future use
  */
 typedef struct _z_link_capabilities_t {
     uint8_t _transport : 2;
@@ -111,6 +116,9 @@ typedef struct _z_link_t {
 #endif
 #if Z_FEATURE_LINK_WS == 1
         _z_ws_socket_t _ws;
+#endif
+#if Z_FEATURE_RAWETH_TRANSPORT == 1
+        _z_raweth_socket_t _raweth;
 #endif
     } _socket;
 

include/zenoh-pico/system/platform/unix.h

@@ -35,7 +35,8 @@ typedef struct timeval z_time_t;
 
 typedef struct {
     union {
-#if Z_FEATURE_LINK_TCP == 1 || Z_FEATURE_LINK_UDP_MULTICAST == 1 || Z_FEATURE_LINK_UDP_UNICAST == 1
+#if Z_FEATURE_LINK_TCP == 1 || Z_FEATURE_LINK_UDP_MULTICAST == 1 || Z_FEATURE_LINK_UDP_UNICAST == 1 || \
+    Z_FEATURE_RAWETH_TRANSPORT == 1
         int _fd;
 #endif
     };

include/zenoh-pico/transport/common/join.h

@@ -17,7 +17,7 @@
 
 #include "zenoh-pico/transport/transport.h"
 
+int8_t _z_send_join(_z_transport_t *zt);
 int8_t _z_send_keep_alive(_z_transport_t *zt);
-void *_zp_lease_task(void *zt_arg);  // The argument is void* to avoid incompatible pointer types in tasks
 
 #endif /* ZENOH_PICO_TRANSPORT_LEASE_H */

include/zenoh-pico/transport/common/lease.h

@@ -17,9 +17,10 @@
 
 #include "zenoh-pico/transport/transport.h"
 
+int8_t _zp_multicast_send_join(_z_transport_multicast_t *ztm);
 int8_t _zp_multicast_send_keep_alive(_z_transport_multicast_t *ztm);
-int8_t _zp_multicast_start_lease_task(_z_transport_t *zt, _z_task_attr_t *attr, _z_task_t *task);
-int8_t _zp_multicast_stop_lease_task(_z_transport_t *zt);
+int8_t _zp_multicast_start_lease_task(_z_transport_multicast_t *ztm, _z_task_attr_t *attr, _z_task_t *task);
+int8_t _zp_multicast_stop_lease_task(_z_transport_multicast_t *ztm);
 void *_zp_multicast_lease_task(void *ztm_arg);  // The argument is void* to avoid incompatible pointer types in tasks
 
 #endif /* ZENOH_PICO_MULTICAST_LEASE_H */

include/zenoh-pico/transport/multicast/join.h

@@ -18,7 +18,6 @@
 #include "zenoh-pico/transport/transport.h"
 
 int8_t _z_multicast_recv_t_msg(_z_transport_multicast_t *ztm, _z_transport_message_t *t_msg, _z_bytes_t *addr);
-int8_t _z_multicast_recv_t_msg_na(_z_transport_multicast_t *ztm, _z_transport_message_t *t_msg, _z_bytes_t *addr);
 int8_t _z_multicast_handle_transport_message(_z_transport_multicast_t *ztm, _z_transport_message_t *t_msg,
                                              _z_bytes_t *addr);
 

include/zenoh-pico/transport/multicast/lease.h

@@ -53,6 +53,11 @@ _Z_LIST_DEFINE(_z_transport_peer_entry, _z_transport_peer_entry_t)
 _z_transport_peer_entry_list_t *_z_transport_peer_entry_list_insert(_z_transport_peer_entry_list_t *root,
                                                                     _z_transport_peer_entry_t *entry);
 
+// Forward declaration to be used in _zp_f_send_tmsg*
+typedef struct _z_transport_multicast_t _z_transport_multicast_t;
+// Send function prototype
+typedef int8_t (*_zp_f_send_tmsg)(_z_transport_multicast_t *self, const _z_transport_message_t *t_msg);
+
 typedef struct {
     // Session associated to the transport
 
@@ -93,7 +98,7 @@ typedef struct {
     volatile _Bool _transmitted;
 } _z_transport_unicast_t;
 
-typedef struct {
+typedef struct _z_transport_multicast_t {
     // Session associated to the transport
     void *_session;
 
@@ -121,6 +126,9 @@ typedef struct {
     // Known valid peers
     _z_transport_peer_entry_list_t *_peers;
 
+    // T message send function
+    _zp_f_send_tmsg _send_f;
+
 #if Z_FEATURE_MULTI_THREAD == 1
     _z_task_t *_read_task;
     _z_task_t *_lease_task;
@@ -135,11 +143,13 @@ typedef struct {
     union {
         _z_transport_unicast_t _unicast;
         _z_transport_multicast_t _multicast;
+        _z_transport_multicast_t _raweth;
     } _transport;
 
     enum {
         _Z_TRANSPORT_UNICAST_TYPE,
         _Z_TRANSPORT_MULTICAST_TYPE,
+        _Z_TRANSPORT_RAWETH_TYPE,
     } _type;
 } _z_transport_t;
 

include/zenoh-pico/transport/multicast/rx.h

@@ -547,6 +547,7 @@ int8_t z_info_peers_zid(const z_session_t zs, z_owned_closure_zid_t *callback) {
     // Call transport function
     switch (zs._val->_tp._type) {
         case _Z_TRANSPORT_MULTICAST_TYPE:
+        case _Z_TRANSPORT_RAWETH_TYPE:
             _zp_multicast_fetch_zid(&zs._val->_tp, callback);
             break;
         default:
@@ -638,7 +639,7 @@ z_owned_publisher_t z_declare_publisher(z_session_t zs, z_keyexpr_t keyexpr, con
     // TODO: Currently, if resource declarations are done over multicast transports, the current protocol definition
     //       lacks a way to convey them to later-joining nodes. Thus, in the current version automatic
     //       resource declarations are only performed on unicast transports.
-    if (zs._val->_tp._type != _Z_TRANSPORT_MULTICAST_TYPE) {
+    if (zs._val->_tp._type == _Z_TRANSPORT_UNICAST_TYPE) {
         _z_resource_t *r = _z_get_resource_by_key(zs._val, &keyexpr);
         if (r == NULL) {
             uint16_t id = _z_declare_resource(zs._val, keyexpr);
@@ -794,7 +795,7 @@ z_owned_queryable_t z_declare_queryable(z_session_t zs, z_keyexpr_t keyexpr, z_o
     // TODO: Currently, if resource declarations are done over multicast transports, the current protocol definition
     //       lacks a way to convey them to later-joining nodes. Thus, in the current version automatic
     //       resource declarations are only performed on unicast transports.
-    if (zs._val->_tp._type != _Z_TRANSPORT_MULTICAST_TYPE) {
+    if (zs._val->_tp._type == _Z_TRANSPORT_UNICAST_TYPE) {
         _z_resource_t *r = _z_get_resource_by_key(zs._val, &keyexpr);
         if (r == NULL) {
             uint16_t id = _z_declare_resource(zs._val, keyexpr);
@@ -897,7 +898,7 @@ z_owned_subscriber_t z_declare_subscriber(z_session_t zs, z_keyexpr_t keyexpr, z
     // TODO: Currently, if resource declarations are done over multicast transports, the current protocol definition
     //       lacks a way to convey them to later-joining nodes. Thus, in the current version automatic
     //       resource declarations are only performed on unicast transports.
-    if (zs._val->_tp._type != _Z_TRANSPORT_MULTICAST_TYPE) {
+    if (zs._val->_tp._type == _Z_TRANSPORT_UNICAST_TYPE) {
         _z_resource_t *r = _z_get_resource_by_key(zs._val, &keyexpr);
         if (r == NULL) {
             char *wild = strpbrk(keyexpr._suffix, "*$");

include/zenoh-pico/transport/raweth/join.h

@@ -17,6 +17,7 @@
 #include <stddef.h>
 
 #include "zenoh-pico/config.h"
+#include "zenoh-pico/link/config/raweth.h"
 #include "zenoh-pico/link/manager.h"
 #include "zenoh-pico/utils/logging.h"
 
@@ -56,7 +57,6 @@ int8_t _z_open_link(_z_link_t *zl, const char *locator) {
         {
             ret = _Z_ERR_CONFIG_LOCATOR_SCHEMA_UNKNOWN;
         }
-
         if (ret == _Z_RES_OK) {
             // Open transport link for communication
             if (zl->_open_f(zl) != _Z_RES_OK) {
@@ -92,10 +92,11 @@ int8_t _z_listen_link(_z_link_t *zl, const char *locator) {
             ret = _z_new_link_bt(zl, ep);
         } else
 #endif
-        {
+            if (_z_endpoint_raweth_valid(&ep) == _Z_RES_OK) {
+            ret = _z_new_link_raweth(zl, ep);
+        } else {
             ret = _Z_ERR_CONFIG_LOCATOR_SCHEMA_UNKNOWN;
         }
-
         if (ret == _Z_RES_OK) {
             // Open transport link for listening
             if (zl->_listen_f(zl) != _Z_RES_OK) {