Multithread initial approach (#126)

* Static incoming mempools

Update


Parametrize lenght


Simplify xml/ref buffer


Uncrustified

Remove unused


Indent


Update


Update rmw_microxrcedds_c/test/test_sizes.cpp

Co-authored-by: Jose Antonio Moral <joseantoniomoralparras@gmail.com>
Update default history value


Initial multithread

Update

Add includes

Lock memory pools

Update


Update

* Uncrustified

Co-authored-by: GitHub Actions Bot <>

Author: pablogs9

rmw_microxrcedds_c/src/memory.c

@@ -18,15 +18,23 @@
 #include <rmw_microxrcedds_c/config.h>
 #include <rmw/allocators.h>
 
+#include <uxr/client/profile/multithread/multithread.h>
+
 bool has_memory(
         rmw_uxrce_mempool_t* mem)
 {
-    return mem->freeitems != NULL ? true : false;
+    UXR_LOCK(&mem->mutex);
+    bool rv = mem->freeitems != NULL ? true : false;
+    UXR_UNLOCK(&mem->mutex);
+
+    return rv;
 }
 
 rmw_uxrce_mempool_item_t* get_memory(
         rmw_uxrce_mempool_t* mem)
 {
+    UXR_LOCK(&mem->mutex);
+
     rmw_uxrce_mempool_item_t* item = NULL;
 
     if (has_memory(mem))
@@ -62,13 +70,18 @@ rmw_uxrce_mempool_item_t* get_memory(
         item = get_memory(mem);
 #endif /* ifdef RMW_UXRCE_ALLOW_DYNAMIC_ALLOCATIONS */
     }
+
+    UXR_UNLOCK(&mem->mutex);
+
     return item;
 }
 
 void put_memory(
         rmw_uxrce_mempool_t* mem,
         rmw_uxrce_mempool_item_t* item)
 {
+    UXR_LOCK(&mem->mutex);
+
     // Gets item from allocated pool
     if (item->prev)
     {
@@ -101,4 +114,6 @@ void put_memory(
     }
     item->prev     = NULL;
     mem->freeitems = item;
+
+    UXR_UNLOCK(&mem->mutex);
 }

rmw_microxrcedds_c/src/memory.h

@@ -18,6 +18,8 @@
 #include <stdbool.h>
 #include <stddef.h>
 
+#include <uxr/client/profile/multithread/multithread.h>
+
 typedef struct rmw_uxrce_mempool_item_t
 {
     bool is_dynamic_memory;
@@ -26,14 +28,18 @@ typedef struct rmw_uxrce_mempool_item_t
     void*                            data;
 } rmw_uxrce_mempool_item_t;
 
-
 typedef struct rmw_uxrce_mempool_t
 {
     struct rmw_uxrce_mempool_item_t* allocateditems;
     struct rmw_uxrce_mempool_item_t* freeitems;
 
     bool is_initialized;
     size_t element_size;
+
+#ifdef UCLIENT_PROFILE_MULTITHREAD
+    uxrMutex mutex;
+#endif // UCLIENT_PROFILE_MULTITHREAD
+
 } rmw_uxrce_mempool_t;
 
 bool has_memory(

rmw_microxrcedds_c/src/rmw_publish.c

@@ -20,6 +20,8 @@
 
 #include <rmw_uros/options.h>
 
+#include <uxr/client/profile/multithread/multithread.h>
+
 bool flush_session(
         uxrSession* session)
 {
@@ -83,6 +85,8 @@ rmw_publish(
                 custom_publisher->cs_cb_serialization(&mb);
             }
 
+            UXR_UNLOCK_STREAM_ID(&custom_publisher->owner_node->context->session, custom_publisher->stream_id);
+
             if (UXR_BEST_EFFORT_STREAM == custom_publisher->stream_id.type)
             {
                 uxr_flash_output_streams(&custom_publisher->owner_node->context->session);

rmw_microxrcedds_c/src/rmw_request.c

@@ -17,6 +17,8 @@
 #include <rmw/rmw.h>
 #include <rmw/error_handling.h>
 
+#include <uxr/client/profile/multithread/multithread.h>
+
 rmw_ret_t
 rmw_send_request(
         const rmw_client_t* client,
@@ -48,6 +50,8 @@ rmw_send_request(
 
     functions->cdr_serialize(ros_request, &mb);
 
+    UXR_UNLOCK_STREAM_ID(&custom_node->context->session, custom_client->stream_id);
+
     if (UXR_INVALID_REQUEST_ID == *sequence_id)
     {
         RMW_SET_ERROR_MSG("Micro XRCE-DDS service request error.");

rmw_microxrcedds_c/src/rmw_response.c

@@ -17,6 +17,8 @@
 #include <rmw/rmw.h>
 #include <rmw/error_handling.h>
 
+#include <uxr/client/profile/multithread/multithread.h>
+
 rmw_ret_t
 rmw_send_response(
         const rmw_service_t* service,
@@ -61,6 +63,8 @@ rmw_send_response(
     uxr_serialize_SampleIdentity(&mb, &sample_id);
     functions->cdr_serialize(ros_response, &mb);
 
+    UXR_UNLOCK_STREAM_ID(&custom_node->context->session, custom_service->stream_id);
+
     if (UXR_INVALID_REQUEST_ID == rc)
     {
         RMW_SET_ERROR_MSG("Micro XRCE-DDS service response error.");

rmw_microxrcedds_c/src/types.c

@@ -28,6 +28,8 @@
 #include "rmw/allocators.h"
 #include <rmw/error_handling.h>
 
+#include <uxr/client/profile/multithread/multithread.h>
+
 // Static memory pools
 
 char rmw_uxrce_entity_naming_buffer[RMW_UXRCE_ENTITY_NAMING_BUFFER_LENGTH];
@@ -65,6 +67,7 @@ rmw_uxrce_static_input_buffer_t custom_static_buffers[RMW_UXRCE_MAX_HISTORY];
         size_t size)                                    \
     {                                                   \
         if (size > 0 && !memory->is_initialized){       \
+            UXR_INIT_LOCK(&memory->mutex);              \
             memory->is_initialized = true;              \
             memory->element_size   = sizeof(*array);    \
             memory->allocateditems = NULL;              \