xiahb
/
Micro-XRCE-DDS-Client
mirror of https://github.com/eProsima/Micro-XRCE-DDS-Client.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Multithread and shared memory (#216) 

* Initial


Update


Revert "Initial"

This reverts commit bbef97f3599786be2febc15f8e8b474a4006234d.

Initial


Update


Update


Initial


Update


Revert example


Uncrustify fix


Add include


Fix header


Fix headers


Fix


Uncrus


Update to const char


Update


Update


Revert "Update"

This reverts commit 3a05e402e5.

Update


Multithread initial


Interprocess initial


Matching initial


Fix interprocess indexing


Update multithread public API


Initial DLL


Double linked list for ucdrBuffers


renaming


Update


Update


Indent


Deinit


Add shared memory tests (#221)

* Initial

* Update

* Update

* Update

* Update

* Fix leak
Update on request reply shared mem

Initial changes req/res


Update


Initial simple API


Working


Renaming


Fix example pthread header

Create shared memory profile folder


Fix warning


Fix warnings


Fix warnings


Multithread test


Update


Add tests


Style


Uncrustify


Uncrustify examples


Fix


Update


Fix


Add test multithread request reply

Update SharedMemory.cpp

Fix build warning
Include requester info on reply SampleIdentity info

Uncrustify

Fix warnings

Add request reply fragmentation test

Update


Update shared memory engine to bin


Update


Uncrustify


Fix

* Remove examples

* Fix naming

* Uncrus
This commit is contained in:
Pablo Garrido 2021-04-30 12:03:19 +02:00 committed by GitHub
parent 206e7bf1b3
commit 527b4dd2f0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
32 changed files with 2430 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
CMakeLists.txt
View File

@ -62,6 +62,12 @@ set(UCLIENT_SERIAL_TRANSPORT_MTU 512 CACHE STRING "Set the Serial transport MTU.
option(UCLIENT_PROFILE_CUSTOM_TRANSPORT "Enable Custom transport." ON)
set(UCLIENT_CUSTOM_TRANSPORT_MTU 512 CACHE STRING "Set the Custom transport MTU.")
option(UCLIENT_PROFILE_MULTITHREAD "Enable multithread support." OFF)
option(UCLIENT_PROFILE_SHARED_MEMORY "Enable shared memory transport support." OFF)
option(UCLIENT_PROFILE_MATCHING "Enable QoS matching support." OFF)
set(UCLIENT_SHARED_MEMORY_MAX_ENTITIES 4 CACHE STRING "Max number of entities involved in shared memory.")
set(UCLIENT_SHARED_MEMORY_STATIC_MEM_SIZE 10 CACHE STRING "Max number data buffers stored in shared memory")
###############################################################################
# Dependencies
###############################################################################
@ -215,6 +221,14 @@ if(UCLIENT_PROFILE_CUSTOM_TRANSPORT)
list(APPEND _transport_src src/c/profile/transport/custom/custom_transport.c)
endif()
if(UCLIENT_PROFILE_MULTITHREAD)
set(UCLIENT_PROFILE_SHARED_MEMORY ON)
endif()
if(UCLIENT_PROFILE_SHARED_MEMORY)
set(UCLIENT_PROFILE_MATCHING ON)
endif()
# Other sources
set(SRCS
src/c/core/session/stream/input_best_effort_stream.c
@ -239,6 +253,9 @@ set(SRCS
src/c/core/session/create_entities_bin.c
src/c/core/session/read_access.c
src/c/core/session/write_access.c
$<$<BOOL:${UCLIENT_PROFILE_MULTITHREAD}>:src/c/profile/multithread/multithread.c>
$<$<BOOL:${UCLIENT_PROFILE_SHARED_MEMORY}>:src/c/profile/shared_memory/shared_memory.c>
$<$<BOOL:${UCLIENT_PROFILE_MATCHING}>:src/c/profile/matching/matching.c>
$<$<BOOL:${UCLIENT_PROFILE_STREAM_FRAMING}>:src/c/profile/transport/stream_framing/stream_framing_protocol.c>
$<$<OR:$<BOOL:${UCLIENT_VERBOSE_MESSAGE}>,$<BOOL:${UCLIENT_VERBOSE_SERIALIZATION}>>:src/c/core/log/log.c>
${_transport_src}
@ -308,12 +325,17 @@ if((CMAKE_SYSTEM_NAME STREQUAL "Windows") AND (UCLIENT_TARGET_TYPE STREQUAL "SHA
)
endif()
if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
find_package(Threads REQUIRED)
endif()
target_link_libraries(${PROJECT_NAME}
PUBLIC
microcdr
$<$<BOOL:$<PLATFORM_ID:Windows>>:ws2_32>
PRIVATE
$<$<BOOL:$<PLATFORM_ID:Linux>>:rt>
$<$<BOOL:$<PLATFORM_ID:Linux>>:${CMAKE_THREAD_LIBS_INIT}>
)
target_include_directories(${PROJECT_NAME}
PUBLIC
@ -367,6 +389,7 @@ if(UCLIENT_BUILD_EXAMPLES)
add_subdirectory(examples/ContinuousFragment)
add_subdirectory(examples/PublishHelloWorldBestEffort)
add_subdirectory(examples/SubscribeHelloWorldBestEffort)
if(UCLIENT_PLATFORM_LINUX)
add_subdirectory(examples/CustomTransports)
endif()
@ -392,6 +415,7 @@ if(UCLIENT_BUILD_TESTS)
if(UCLIENT_PLATFORM_LINUX)
add_subdirectory(test/transport/custom_comm)
add_subdirectory(test/transport/serial_comm)
add_subdirectory(test/shared_memory)
endif()
endif()

1
include/uxr/client/client.h
View File

@ -29,5 +29,6 @@
#include <uxr/client/core/session/create_entities_bin.h>
#include <uxr/client/transport.h>
#include <uxr/client/profile/multithread/multithread.h>
#endif // _UXR_CLIENT_CLIENT_H_

9
include/uxr/client/config.h.in
View File

@ -26,6 +26,8 @@
#cmakedefine UCLIENT_PROFILE_TCP
#cmakedefine UCLIENT_PROFILE_SERIAL
#cmakedefine UCLIENT_PROFILE_CUSTOM_TRANSPORT
#cmakedefine UCLIENT_PROFILE_MULTITHREAD
#cmakedefine UCLIENT_PROFILE_SHARED_MEMORY
#cmakedefine UCLIENT_PLATFORM_POSIX
#cmakedefine UCLIENT_PLATFORM_POSIX_NOPOLL
@ -56,7 +58,12 @@
#define UXR_CONFIG_SERIAL_TRANSPORT_MTU @UCLIENT_SERIAL_TRANSPORT_MTU@
#endif
#ifdef UCLIENT_PROFILE_CUSTOM_TRANSPORT
#define UXR_CONFIG_CUSTOM_TRANSPORT_MTU @UCLIENT_CUSTOM_TRANSPORT_MTU@
#define UXR_CONFIG_CUSTOM_TRANSPORT_MTU @UCLIENT_CUSTOM_TRANSPORT_MTU@
#endif
#ifdef UCLIENT_PROFILE_SHARED_MEMORY
#define UXR_CONFIG_SHARED_MEMORY_MAX_ENTITIES @UCLIENT_SHARED_MEMORY_MAX_ENTITIES@
#define UXR_CONFIG_SHARED_MEMORY_STATIC_MEM_SIZE @UCLIENT_SHARED_MEMORY_STATIC_MEM_SIZE@
#endif
#endif // _UXR_CLIENT_CONFIG_H_

3
include/uxr/client/core/session/create_entities_bin.h
View File

@ -80,9 +80,6 @@ UXRDLLAPI uint16_t uxr_buffer_create_topic_bin(
* As a result of the reception of this submessage, the Agent will create an XRCE Publisher according to
* the binary provided in the CREATE submessage.
* @param session A uxrSession structure previously initialized.
* @param stream_id The output stream identifier where the CREATE submessage will be buffered.
* @param object_id The identifier of the XRCE Publisher.
* @param participant_id The identifier of the associated XRCE Participant.
* @param mode The set of flags that determines the entity creation mode.
* The Creation Mode Table describes the entities creation behaviour according to the
* `UXR_REUSE` and `UXR_REPLACE` flags.

9
include/uxr/client/core/session/session.h
View File

@ -24,10 +24,15 @@ extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/core/session/session_info.h>
#include <uxr/client/core/session/stream/stream_storage.h>
#include <uxr/client/core/type/xrce_types.h>
#ifdef UCLIENT_PROFILE_MULTITHREAD
#include <uxr/client/profile/multithread/multithread.h>
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
#define UXR_TIMEOUT_INF -1
struct uxrSession;
@ -188,6 +193,10 @@ typedef struct uxrSession
bool on_data_flag;
uxrContinuousArgs continuous_args;
#ifdef UCLIENT_PROFILE_MULTITHREAD
uxrMutex mutex;
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
#ifdef PERFORMANCE_TESTING
uxrOnPerformanceFunc on_performance;
void* on_performance_args;

9
include/uxr/client/core/session/stream/input_best_effort_stream.h
View File

@ -20,8 +20,13 @@ extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/core/session/stream/seq_num.h>
#ifdef UCLIENT_PROFILE_MULTITHREAD
#include <uxr/client/profile/multithread/multithread.h>
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
#include <stdint.h>
#include <stdbool.h>
@ -29,6 +34,10 @@ typedef struct uxrInputBestEffortStream
{
uxrSeqNum last_handled;
#ifdef UCLIENT_PROFILE_MULTITHREAD
uxrMutex mutex;
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
} uxrInputBestEffortStream;
#ifdef __cplusplus

9
include/uxr/client/core/session/stream/input_reliable_stream.h
View File

@ -20,9 +20,14 @@ extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/core/session/stream/reliable_stream.h>
#include <uxr/client/core/session/stream/seq_num.h>
#ifdef UCLIENT_PROFILE_MULTITHREAD
#include <uxr/client/profile/multithread/multithread.h>
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
#include <stdbool.h>
struct ucdrBuffer;
@ -49,6 +54,10 @@ typedef struct uxrInputReliableStream
bool cleanup_flag;
#ifdef UCLIENT_PROFILE_MULTITHREAD
uxrMutex mutex;
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
} uxrInputReliableStream;
#ifdef __cplusplus

9
include/uxr/client/core/session/stream/output_best_effort_stream.h
View File

@ -20,8 +20,13 @@ extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/core/session/stream/seq_num.h>
#ifdef UCLIENT_PROFILE_MULTITHREAD
#include <uxr/client/profile/multithread/multithread.h>
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
#include <stddef.h>
#include <stdbool.h>
@ -35,6 +40,10 @@ typedef struct uxrOutputBestEffortStream
uxrSeqNum last_send;
#ifdef UCLIENT_PROFILE_MULTITHREAD
uxrMutex mutex;
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
} uxrOutputBestEffortStream;
#ifdef __cplusplus

9
include/uxr/client/core/session/stream/output_reliable_stream.h
View File

@ -20,9 +20,14 @@ extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/core/session/stream/reliable_stream.h>
#include <uxr/client/core/session/stream/seq_num.h>
#ifdef UCLIENT_PROFILE_MULTITHREAD
#include <uxr/client/profile/multithread/multithread.h>
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
#include <stdbool.h>
struct ucdrBuffer;
@ -45,6 +50,10 @@ typedef struct uxrOutputReliableStream
uint8_t next_heartbeat_tries;
bool send_lost;
#ifdef UCLIENT_PROFILE_MULTITHREAD
uxrMutex mutex;
#endif // ifdef UCLIENT_PROFILE_MULTITHREAD
} uxrOutputReliableStream;
#ifdef __cplusplus

4
include/uxr/client/core/session/stream/stream_id.h
View File

@ -53,7 +53,9 @@ typedef enum uxrStreamType
/** Identifies a best-effort stream. */
UXR_BEST_EFFORT_STREAM,
/** Identifies a reliable stream. */
UXR_RELIABLE_STREAM
UXR_RELIABLE_STREAM,
/** Identifies a shared memory stream. */
UXR_SHARED_MEMORY_STREAM
} uxrStreamType;

14
include/uxr/client/core/session/write_access.h
View File

@ -97,9 +97,9 @@ uint16_t uxr_buffer_topic(
* As a result of the reception of this submessage, the Agent will write a topic into the DDS Global-Data-Space.
* @param session A uxrSession structure previously initialized.
* @param stream_id The output stream identifier where the WRITE_DATA submessage will be buffered.
* @param entity_id The identifier of the XRCE DataWriter that will write the topic into the DDS GDS.
* @param ub_topic The ucdrBuffer structure used for serializing the topic.
* @param topic_size The size of the topic in bytes.
* @param entity_id The identifier of the XRCE DataWriter that will write the topic into the DDS GDS.
* @param ub The ucdrBuffer structure used for serializing the topic.
* @param len The size of the topic in bytes.
* @return A `request_id` that identifies the XRCE request made by the Entity.
* This could be used in the `uxr_run_session_until_one_status` or `uxr_run_session_until_all_status` functions.
* */
@ -107,8 +107,8 @@ UXRDLLAPI uint16_t uxr_prepare_output_stream(
uxrSession* session,
uxrStreamId stream_id,
uxrObjectId entity_id,
struct ucdrBuffer* ub_topic,
uint32_t topic_size);
struct ucdrBuffer* ub,
uint32_t len);
/**
@ -120,7 +120,7 @@ UXRDLLAPI uint16_t uxr_prepare_output_stream(
* @param stream_id The output stream identifier where the WRITE_DATA submessage will be buffered.
* @param datawriter_id The identifier of the XRCE DataWriter that will write the topic into the DDS GDS.
* @param ub The ucdrBuffer structure used for serializing the topic.
* @param topic_size The size of the topic in bytes.
* @param data_size The size of the topic in bytes.
* @param flush_callback Callback that is call by the library when user should flush output buffers.
* @return A `request_id` that identifies the XRCE request made by the Entity.
* This could be used in the `uxr_run_session_until_one_status` or `uxr_run_session_until_all_status` functions.
@ -131,7 +131,7 @@ UXRDLLAPI uint16_t uxr_prepare_output_stream_fragmented(
uxrStreamId stream_id,
uxrObjectId datawriter_id,
ucdrBuffer* ub,
size_t topic_size,
size_t data_size,
uxrOnBuffersFull flush_callback);
/** @}*/

135
include/uxr/client/profile/multithread/multithread.h Normal file
View File

@ -0,0 +1,135 @@
// Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef UXR_CLIENT_PROFILE_MULTITHREAD_H_
#define UXR_CLIENT_PROFILE_MULTITHREAD_H_
#ifdef __cplusplus
extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/visibility.h>
#include <uxr/client/core/session/stream/stream_id.h>
struct uxrSession;
#ifdef UCLIENT_PROFILE_MULTITHREAD
#ifdef WIN32
#elif defined(PLATFORM_NAME_FREERTOS)
#include <freertos/FreeRTOS.h>
#include <freertos/semphr.h>
#elif defined(UCLIENT_PLATFORM_ZEPHYR)
#elif defined(UCLIENT_PLATFORM_POSIX)
#include <pthread.h>
#endif // ifdef WIN32
// Micro XRCE-DDS Client mutex implementation
typedef struct uxrMutex
{
#ifdef WIN32
#elif defined(PLATFORM_NAME_FREERTOS)
SemaphoreHandle_t impl;
StaticSemaphore_t xMutexBuffer;
#elif defined(UCLIENT_PLATFORM_ZEPHYR)
struct k_mutex impl;
#elif defined(UCLIENT_PLATFORM_POSIX)
pthread_mutex_t impl;
#endif // ifdef WIN32
} uxrMutex;
/**
* @brief
* TODO
*/
UXRDLLAPI uxrMutex* uxr_get_stream_mutex_from_id(
struct uxrSession* session,
uxrStreamId stream_id);
/**
* @brief
* TODO
*/
UXRDLLAPI void uxr_init_lock(
uxrMutex* mutex);
/**
* @brief
* TODO
*/
UXRDLLAPI void uxr_lock(
uxrMutex* mutex);
/**
* @brief
* TODO
*/
UXRDLLAPI void uxr_unlock(
uxrMutex* mutex);
// Conditional defines
#define UXR_INIT_LOCK(X) uxr_init_lock(X)
#define UXR_LOCK(X) uxr_lock(X)
#define UXR_UNLOCK(X) uxr_unlock(X)
#define UXR_INIT_LOCK_SESSION uxr_init_lock(&session->mutex)
#define UXR_LOCK_SESSION(session) uxr_lock(&session->mutex)
#define UXR_UNLOCK_SESSION(session) uxr_unlock(&session->mutex)
#define UXR_LOCK_STREAM_ID(session, stream_id) uxr_lock(uxr_get_stream_mutex_from_id(session, stream_id))
#define UXR_UNLOCK_STREAM_ID(session, stream_id) uxr_unlock(uxr_get_stream_mutex_from_id(session, stream_id))
#define UXR_LOCK_ALL_INPUT_STREAMS(session) \
for (uint8_t i = 0; i < session->streams.input_best_effort_size; \
++i){ uxr_lock(&session->streams.input_best_effort[i].mutex); } \
for (uint8_t i = 0; i < session->streams.input_reliable_size; ++i){ uxr_lock( \
&session->streams.input_reliable[i].mutex); \
}
#define UXR_UNLOCK_ALL_INPUT_STREAMS(session) \
for (uint8_t i = 0; i < session->streams.input_best_effort_size; ++i){ uxr_unlock( \
&session->streams.input_best_effort[i].mutex); \
} \
for (uint8_t i = 0; i < session->streams.input_reliable_size; ++i){ uxr_unlock( \
&session->streams.input_reliable[i].mutex); \
}
#else // UCLIENT_PROFILE_MULTITHREAD
#define UXR_INIT_LOCK(X)
#define UXR_LOCK(X)
#define UXR_UNLOCK(X)
#define UXR_INIT_LOCK_SESSION
#define UXR_LOCK_SESSION(session)
#define UXR_UNLOCK_SESSION(session)
#define UXR_LOCK_STREAM_ID(session, stream_id)
#define UXR_UNLOCK_STREAM_ID(session, stream_id)
#define UXR_LOCK_ALL_INPUT_STREAMS(session)
#define UXR_UNLOCK_ALL_INPUT_STREAMS(session)
#endif // UCLIENT_PROFILE_MULTITHREAD
#ifdef __cplusplus
}
#endif // ifdef __cplusplus
#endif // UXR_CLIENT_PROFILE_MULTITHREAD_H_

11
src/c/core/session/common_create_entities.c
View File

@ -4,6 +4,7 @@
#include "session_internal.h"
#include "session_info_internal.h"
#include "submessage_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
//==================================================================
// PUBLIC
@ -22,12 +23,17 @@ uint16_t uxr_buffer_delete_entity(
payload_length = (uint16_t)(payload_length + 4); // delete payload (request id + object_id), no padding.
ucdrBuffer ub;
UXR_LOCK_STREAM_ID(session, stream_id);
if (uxr_prepare_stream_to_write_submessage(session, stream_id, payload_length, &ub, SUBMESSAGE_ID_DELETE, 0))
{
request_id = uxr_init_base_object_request(&session->info, object_id, &payload.base);
(void) uxr_serialize_DELETE_Payload(&ub, &payload);
}
UXR_UNLOCK_STREAM_ID(session, stream_id);
return request_id;
}
@ -55,11 +61,16 @@ uint16_t uxr_common_create_entity(
payload_length = (uint16_t)(payload_length + 2); //object id ref
ucdrBuffer ub;
UXR_LOCK_STREAM_ID(session, stream_id);
if (uxr_prepare_stream_to_write_submessage(session, stream_id, payload_length, &ub, SUBMESSAGE_ID_CREATE, mode))
{
request_id = uxr_init_base_object_request(&session->info, object_id, &payload->base);
(void) uxr_serialize_CREATE_Payload(&ub, payload);
}
UXR_UNLOCK_STREAM_ID(session, stream_id);
return request_id;
}

11
src/c/core/session/create_entities_bin.c
View File

@ -1,6 +1,7 @@
#include <uxr/client/core/type/xrce_types.h>
#include "common_create_entities_internal.h"
#include "../../profile/shared_memory/shared_memory_internal.h"
#include <string.h>
@ -58,6 +59,8 @@ uint16_t uxr_buffer_create_topic_bin(
uxr_serialize_OBJK_Topic_Binary(&ub, &topic);
payload.object_representation._.topic.base.representation._.binary_representation.size = (uint32_t) ub.offset;
UXR_ADD_SHARED_MEMORY_ENTITY_BIN(session, object_id, &topic);
return uxr_common_create_entity(session, stream_id, object_id, (uint16_t) ub.offset, mode, &payload);
}
@ -150,6 +153,8 @@ uint16_t uxr_buffer_create_datawriter_bin(
datawriter.qos.base.qos_flags |= is_durability_transient_local;
}
UXR_ADD_SHARED_MEMORY_ENTITY_BIN(session, object_id, &datawriter);
ucdrBuffer ub;
ucdr_init_buffer(&ub, payload.object_representation._.data_writer.base.representation._.binary_representation.data,
UXR_BINARY_SEQUENCE_MAX);
@ -199,6 +204,8 @@ uint16_t uxr_buffer_create_datareader_bin(
datareader.qos.base.qos_flags |= is_durability_transient_local;
}
UXR_ADD_SHARED_MEMORY_ENTITY_BIN(session, object_id, &datareader);
ucdrBuffer ub;
ucdr_init_buffer(&ub, payload.object_representation._.data_reader.base.representation._.binary_representation.data,
UXR_BINARY_SEQUENCE_MAX);
@ -234,6 +241,8 @@ uint16_t uxr_buffer_create_requester_bin(
requester.optional_request_topic_name = true;
requester.request_topic_name = (char*) request_topic_name;
UXR_ADD_SHARED_MEMORY_ENTITY_BIN(session, object_id, &requester);
ucdrBuffer ub;
ucdr_init_buffer(&ub, payload.object_representation._.requester.base.representation._.binary_representation.data,
UXR_BINARY_SEQUENCE_MAX);
@ -269,6 +278,8 @@ uint16_t uxr_buffer_create_replier_bin(
replier.optional_request_topic_name = true;
replier.request_topic_name = (char*) request_topic_name;
UXR_ADD_SHARED_MEMORY_ENTITY_BIN(session, object_id, &replier);
ucdrBuffer ub;
ucdr_init_buffer(&ub, payload.object_representation._.replier.base.representation._.binary_representation.data,
UXR_BINARY_SEQUENCE_MAX);

9
src/c/core/session/create_entities_xml.c
View File

@ -2,6 +2,7 @@
#include <uxr/client/core/type/xrce_types.h>
#include "common_create_entities_internal.h"
#include "../../profile/shared_memory/shared_memory_internal.h"
#include <string.h>
@ -94,6 +95,8 @@ uint16_t uxr_buffer_create_datawriter_xml(
{
//assert with the object_id type
UXR_ADD_SHARED_MEMORY_ENTITY_XML(session, object_id, xml);
CREATE_Payload payload;
payload.object_representation.kind = DDS_XRCE_OBJK_DATAWRITER;
uxr_object_id_to_raw(publisher_id, payload.object_representation._.data_writer.publisher_id.data);
@ -111,6 +114,8 @@ uint16_t uxr_buffer_create_datareader_xml(
{
//assert with the object_id type
UXR_ADD_SHARED_MEMORY_ENTITY_XML(session, object_id, xml);
CREATE_Payload payload;
payload.object_representation.kind = DDS_XRCE_OBJK_DATAREADER;
uxr_object_id_to_raw(subscriber_id, payload.object_representation._.data_reader.subscriber_id.data);
@ -126,6 +131,8 @@ uint16_t uxr_buffer_create_requester_xml(
const char* xml,
uint8_t mode)
{
UXR_ADD_SHARED_MEMORY_ENTITY_XML(session, object_id, xml);
CREATE_Payload payload;
payload.object_representation.kind = DDS_XRCE_OBJK_REQUESTER;
uxr_object_id_to_raw(participant_id, payload.object_representation._.requester.participant_id.data);
@ -141,6 +148,8 @@ uint16_t uxr_buffer_create_replier_xml(
const char* xml,
uint8_t mode)
{
UXR_ADD_SHARED_MEMORY_ENTITY_XML(session, object_id, xml);
CREATE_Payload payload;
payload.object_representation.kind = DDS_XRCE_OBJK_REPLIER;
uxr_object_id_to_raw(participant_id, payload.object_representation._.replier.participant_id.data);

6
src/c/core/session/read_access.c
View File

@ -4,6 +4,7 @@
#include "session_internal.h"
#include "session_info_internal.h"
#include "submessage_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
extern void read_submessage_format(
uxrSession* session,
@ -87,12 +88,17 @@ uint16_t uxr_buffer_request_data(
payload_length += (control != NULL) ? 8 : 0; // delivery control
ucdrBuffer ub;
UXR_LOCK_STREAM_ID(session, stream_id);
if (uxr_prepare_stream_to_write_submessage(session, stream_id, payload_length, &ub, SUBMESSAGE_ID_READ_DATA, 0))
{
request_id = uxr_init_base_object_request(&session->info, datareader_id, &payload.base);
(void) uxr_serialize_READ_DATA_Payload(&ub, &payload);
}
UXR_UNLOCK_STREAM_ID(session, stream_id);
return request_id;
}

83
src/c/core/session/session.c
View File

@ -16,8 +16,17 @@
#include "stream/seq_num_internal.h"
#include "../log/log_internal.h"
#include "../../util/time_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
#include "../../profile/shared_memory/shared_memory_internal.h"
#ifdef UCLIENT_PROFILE_SHARED_MEMORY
#define CREATE_SESSION_PROPERTIES_MAX_SIZE 21
#define CREATE_SESSION_MAX_MSG_SIZE (MAX_HEADER_SIZE + SUBHEADER_SIZE + CREATE_CLIENT_PAYLOAD_SIZE + \
CREATE_SESSION_PROPERTIES_MAX_SIZE)
#else
#define CREATE_SESSION_MAX_MSG_SIZE (MAX_HEADER_SIZE + SUBHEADER_SIZE + CREATE_CLIENT_PAYLOAD_SIZE)
#endif /* ifdef UCLIENT_PROFILE_SHARED_MEMORY */
#define DELETE_SESSION_MAX_MSG_SIZE (MAX_HEADER_SIZE + SUBHEADER_SIZE + DELETE_CLIENT_PAYLOAD_SIZE)
#define HEARTBEAT_MAX_MSG_SIZE (MAX_HEADER_SIZE + SUBHEADER_SIZE + HEARTBEAT_PAYLOAD_SIZE)
#define ACKNACK_MAX_MSG_SIZE (MAX_HEADER_SIZE + SUBHEADER_SIZE + ACKNACK_PAYLOAD_SIZE)
@ -124,6 +133,8 @@ void uxr_init_session(
uxrCommunication* comm,
uint32_t key)
{
UXR_INIT_LOCK(&session->mutex);
session->comm = comm;
session->request_list = NULL;
@ -230,6 +241,8 @@ bool uxr_delete_session_retries(
uxrSession* session,
size_t retries)
{
UXR_CLEAN_SHARED_MEMORY();
uint8_t delete_session_buffer[DELETE_SESSION_MAX_MSG_SIZE];
ucdrBuffer ub;
ucdr_init_buffer_origin_offset(&ub, delete_session_buffer, DELETE_SESSION_MAX_MSG_SIZE, 0u, uxr_session_header_offset(
@ -286,6 +299,8 @@ bool uxr_run_session_time(
uxrSession* session,
int timeout_ms)
{
UXR_LOCK_SESSION(session);
uxr_flash_output_streams(session);
bool timeout = false;
@ -294,13 +309,18 @@ bool uxr_run_session_time(
timeout = !listen_message_reliably(session, timeout_ms);
}
return uxr_output_streams_confirmed(&session->streams);
bool ret = uxr_output_streams_confirmed(&session->streams);
UXR_UNLOCK_SESSION(session);
return ret;
}
bool uxr_run_session_timeout(
uxrSession* session,
int timeout_ms)
{
UXR_LOCK_SESSION(session);
int64_t start_timestamp = uxr_millis();
int remaining_time = timeout_ms;
@ -312,13 +332,19 @@ bool uxr_run_session_timeout(
remaining_time = timeout_ms - (int)(uxr_millis() - start_timestamp);
}
while (remaining_time > 0);
return uxr_output_streams_confirmed(&session->streams);
bool ret = uxr_output_streams_confirmed(&session->streams);
UXR_UNLOCK_SESSION(session);
return ret;
}
bool uxr_run_session_until_data(
uxrSession* session,
int timeout_ms)
{
UXR_LOCK_SESSION(session);
int64_t start_timestamp = uxr_millis();
int remaining_time = timeout_ms;
@ -335,6 +361,9 @@ bool uxr_run_session_until_data(
remaining_time = timeout_ms - (int)(uxr_millis() - start_timestamp);
}
while (remaining_time > 0);
UXR_UNLOCK_SESSION(session);
return session->on_data_flag;
}
@ -342,15 +371,22 @@ bool uxr_run_session_until_timeout(
uxrSession* session,
int timeout_ms)
{
UXR_LOCK_SESSION(session);
uxr_flash_output_streams(session);
return listen_message_reliably(session, timeout_ms);
bool ret = listen_message_reliably(session, timeout_ms);
UXR_UNLOCK_SESSION(session);
return ret;
}
bool uxr_run_session_until_confirm_delivery(
uxrSession* session,
int timeout_ms)
{
UXR_LOCK_SESSION(session);
uxr_flash_output_streams(session);
bool timeout = false;
@ -359,7 +395,10 @@ bool uxr_run_session_until_confirm_delivery(
timeout = !listen_message_reliably(session, timeout_ms);
}
return uxr_output_streams_confirmed(&session->streams);
bool ret = uxr_output_streams_confirmed(&session->streams);
UXR_UNLOCK_SESSION(session);
return ret;
}
bool uxr_run_session_until_all_status(
@ -369,6 +408,8 @@ bool uxr_run_session_until_all_status(
uint8_t* status_list,
size_t list_size)
{
UXR_LOCK_SESSION(session);
uxr_flash_output_streams(session);
for (unsigned i = 0; i < list_size; ++i)
@ -376,6 +417,7 @@ bool uxr_run_session_until_all_status(
status_list[i] = UXR_STATUS_NONE;
}
session->request_list = request_list;
session->status_list = status_list;
session->request_status_list_size = list_size;
@ -403,6 +445,8 @@ bool uxr_run_session_until_all_status(
status_ok = status_list[i] == UXR_STATUS_OK || status_list[i] == UXR_STATUS_OK_MATCHED;
}
UXR_UNLOCK_SESSION(session);
return status_ok;
}
@ -413,6 +457,8 @@ bool uxr_run_session_until_one_status(
uint8_t* status_list,
size_t list_size)
{
UXR_LOCK_SESSION(session);
uxr_flash_output_streams(session);
for (unsigned i = 0; i < list_size; ++i)
@ -438,6 +484,7 @@ bool uxr_run_session_until_one_status(
while (!timeout && !status_confirmed);
session->request_status_list_size = 0;
UXR_UNLOCK_SESSION(session);
return status_confirmed;
}
@ -446,6 +493,8 @@ bool uxr_sync_session(
uxrSession* session,
int time)
{
UXR_LOCK_SESSION(session);
uint8_t timestamp_buffer[TIMESTAMP_MAX_MSG_SIZE];
ucdrBuffer ub;
ucdr_init_buffer_origin_offset(&ub, timestamp_buffer, sizeof(timestamp_buffer), 0u,
@ -460,7 +509,10 @@ bool uxr_sync_session(
uxr_stamp_session_header(&session->info, 0, 0, ub.init);
send_message(session, timestamp_buffer, ucdr_buffer_length(&ub));
return run_session_until_sync(session, time);
bool ret = run_session_until_sync(session, time);
UXR_UNLOCK_SESSION(session);
return ret;
}
int64_t uxr_epoch_millis(
@ -510,17 +562,24 @@ bool uxr_buffer_performance(
void uxr_flash_output_streams(
uxrSession* session)
{
UXR_HANDLE_SHARED_MEMORY();
for (uint8_t i = 0; i < session->streams.output_best_effort_size; ++i)
{
uxrOutputBestEffortStream* stream = &session->streams.output_best_effort[i];
uxrStreamId id = uxr_stream_id(i, UXR_BEST_EFFORT_STREAM, UXR_OUTPUT_STREAM);
uint8_t* buffer; size_t length; uxrSeqNum seq_num;
UXR_LOCK_STREAM_ID(session, id);
if (uxr_prepare_best_effort_buffer_to_send(stream, &buffer, &length, &seq_num))
{
uxr_stamp_session_header(&session->info, id.raw, seq_num, buffer);
send_message(session, buffer, length);
}
UXR_UNLOCK_STREAM_ID(session, id);
}
for (uint8_t i = 0; i < session->streams.output_reliable_size; ++i)
@ -529,11 +588,16 @@ void uxr_flash_output_streams(
uxrStreamId id = uxr_stream_id(i, UXR_RELIABLE_STREAM, UXR_OUTPUT_STREAM);
uint8_t* buffer; size_t length; uxrSeqNum seq_num;
UXR_LOCK_STREAM_ID(session, id);
while (uxr_prepare_next_reliable_buffer_to_send(stream, &buffer, &length, &seq_num))
{
uxr_stamp_session_header(&session->info, id.raw, seq_num, buffer);
send_message(session, buffer, length);
}
UXR_UNLOCK_STREAM_ID(session, id);
}
}
@ -545,6 +609,8 @@ bool listen_message(
int poll_ms)
{
uint8_t* data; size_t length;
UXR_LOCK_ALL_INPUT_STREAMS(session);
bool must_be_read = recv_message(session, &data, &length, poll_ms);
if (must_be_read)
{
@ -552,6 +618,7 @@ bool listen_message(
ucdr_init_buffer(&ub, data, (uint32_t)length);
read_message(session, &ub);
}
UXR_UNLOCK_ALL_INPUT_STREAMS(session);
return must_be_read;
}
@ -571,6 +638,7 @@ bool listen_message_reliably(
uxrOutputReliableStream* stream = &session->streams.output_reliable[i];
uxrStreamId id = uxr_stream_id(i, UXR_RELIABLE_STREAM, UXR_OUTPUT_STREAM);
UXR_LOCK_STREAM_ID(session, id);
if (uxr_update_output_stream_heartbeat_timestamp(stream, timestamp))
{
write_submessage_heartbeat(session, id);
@ -580,6 +648,7 @@ bool listen_message_reliably(
{
next_heartbeat_timestamp = stream->next_heartbeat_timestamp;
}
UXR_UNLOCK_STREAM_ID(session, id);
}
int32_t poll_to_next_heartbeat =
@ -893,6 +962,8 @@ void read_submessage_acknack(
uxrOutputReliableStream* stream = uxr_get_output_reliable_stream(&session->streams, id.index);
if (stream)
{
UXR_LOCK_STREAM_ID(session, id);
uint16_t nack_bitmap = (uint16_t)(((uint16_t)acknack.nack_bitmap[0] << 8) + acknack.nack_bitmap[1]);
uxr_process_acknack(stream, nack_bitmap, acknack.first_unacked_seq_num);
@ -902,6 +973,8 @@ void read_submessage_acknack(
{
send_message(session, buffer, length);
}
UXR_UNLOCK_STREAM_ID(session, id);
}
}

7
src/c/core/session/session_info.c
View File

@ -1,6 +1,7 @@
#include <uxr/client/defines.h>
#include <uxr/client/core/session/object_id.h>
#include <uxr/client/core/type/xrce_types.h>
#include <uxr/client/config.h>
#include "session_info_internal.h"
#include "submessage_internal.h"
@ -53,6 +54,12 @@ void uxr_buffer_create_session(
payload.client_representation.client_key.data[3] = info->key[3];
payload.client_representation.session_id = info->id;
payload.client_representation.optional_properties = false;
#ifdef UCLIENT_PROFILE_SHARED_MEMORY
payload.client_representation.optional_properties = true;
payload.client_representation.properties.size = 1;
payload.client_representation.properties.data[0].name = "uxr_sm";
payload.client_representation.properties.data[0].value = "1";
#endif /* ifdef UCLIENT_PROFILE_SHARED_MEMORY */
payload.client_representation.mtu = mtu;
info->last_request_id = UXR_REQUEST_LOGIN;

3
src/c/core/session/stream/input_best_effort_stream.c
View File

@ -1,5 +1,6 @@
#include "input_best_effort_stream_internal.h"
#include "seq_num_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
//==================================================================
// PUBLIC
@ -8,6 +9,8 @@ void uxr_init_input_best_effort_stream(
uxrInputBestEffortStream* stream)
{
stream->last_handled = SEQ_NUM_MAX;
UXR_INIT_LOCK(&stream->mutex);
}
void uxr_reset_input_best_effort_stream(

3
src/c/core/session/stream/input_reliable_stream.c
View File

@ -4,6 +4,7 @@
#include "./input_reliable_stream_internal.h"
#include "./common_reliable_stream_internal.h"
#include "../submessage_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
#include <string.h>
@ -33,6 +34,8 @@ void uxr_init_input_reliable_stream(
stream->on_get_fragmentation_info = on_get_fragmentation_info;
stream->cleanup_flag = false;
UXR_INIT_LOCK(&stream->mutex);
uxr_reset_input_reliable_stream(stream);
}

3
src/c/core/session/stream/output_best_effort_stream.c
View File

@ -2,6 +2,7 @@
#include "../submessage_internal.h"
#include "seq_num_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
#include <ucdr/microcdr.h>
@ -18,6 +19,8 @@ void uxr_init_output_best_effort_stream(
stream->offset = offset;
stream->size = size;
UXR_INIT_LOCK(&stream->mutex);
uxr_reset_output_best_effort_stream(stream);
}

3
src/c/core/session/stream/output_reliable_stream.c
View File

@ -8,6 +8,7 @@
#include "./output_reliable_stream_internal.h"
#include "./common_reliable_stream_internal.h"
#include "../submessage_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
#define MIN_HEARTBEAT_TIME_INTERVAL ((int64_t) UXR_CONFIG_MIN_HEARTBEAT_TIME_INTERVAL) // ms
#define MAX_HEARTBEAT_TRIES (sizeof(int64_t) * 8 - 1)
@ -29,6 +30,8 @@ void uxr_init_output_reliable_stream(
stream->base.history = history;
stream->offset = header_offset;
UXR_INIT_LOCK(&stream->mutex);
uxr_reset_output_reliable_stream(stream);
}

2
src/c/core/session/stream/stream_id.c
View File

@ -29,6 +29,8 @@ uxrStreamId uxr_stream_id(
case UXR_RELIABLE_STREAM:
stream_id.raw = (uint8_t)(index + RELIABLE_STREAM_THRESHOLD);
break;
default:
break;
}
return stream_id;

3
src/c/core/session/stream/stream_storage.c
View File

@ -3,6 +3,7 @@
#include "input_reliable_stream_internal.h"
#include "output_best_effort_stream_internal.h"
#include "output_reliable_stream_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
//==================================================================
// PUBLIC
@ -141,7 +142,9 @@ bool uxr_output_streams_confirmed(
bool up_to_date = true;
for (unsigned i = 0; i < storage->output_reliable_size && up_to_date; ++i)
{
UXR_LOCK((uxrMutex*) &storage->output_reliable[i].mutex);
up_to_date = uxr_is_output_up_to_date(&storage->output_reliable[i]);
UXR_UNLOCK((uxrMutex*) &storage->output_reliable[i].mutex);
}
return up_to_date;
}

30
src/c/core/session/write_access.c
View File

@ -8,6 +8,8 @@
#include "./stream/common_reliable_stream_internal.h"
#include "./stream/stream_storage_internal.h"
#include "./stream/seq_num_internal.h"
#include <uxr/client/profile/multithread/multithread.h>
#include "../../profile/shared_memory/shared_memory_internal.h"
#define WRITE_DATA_PAYLOAD_SIZE 4
#define SAMPLE_IDENTITY_SIZE 24
@ -26,6 +28,8 @@ uint16_t uxr_buffer_request(
ucdrBuffer ub;
size_t payload_size = WRITE_DATA_PAYLOAD_SIZE + len;
UXR_LOCK_STREAM_ID(session, stream_id);
ub.error = !uxr_prepare_stream_to_write_submessage(session, stream_id, payload_size, &ub, SUBMESSAGE_ID_WRITE_DATA,
FORMAT_DATA);
if (!ub.error)
@ -33,9 +37,12 @@ uint16_t uxr_buffer_request(
WRITE_DATA_Payload_Data payload;
rv = uxr_init_base_object_request(&session->info, requester_id, &payload.base);
uxr_serialize_WRITE_DATA_Payload_Data(&ub, &payload);
UXR_PREPARE_SHARED_MEMORY(session, requester_id, &ub, (uint16_t) len, rv);
ucdr_serialize_array_uint8_t(&ub, buffer, len);
}
UXR_UNLOCK_STREAM_ID(session, stream_id);
return rv;
}
@ -51,6 +58,8 @@ uint16_t uxr_buffer_reply(
ucdrBuffer ub;
size_t payload_size = WRITE_DATA_PAYLOAD_SIZE + SAMPLE_IDENTITY_SIZE + len;
UXR_LOCK_STREAM_ID(session, stream_id);
ub.error = !uxr_prepare_stream_to_write_submessage(session, stream_id, payload_size, &ub, SUBMESSAGE_ID_WRITE_DATA,
FORMAT_DATA);
if (!ub.error)
@ -58,10 +67,13 @@ uint16_t uxr_buffer_reply(
WRITE_DATA_Payload_Data payload;
rv = uxr_init_base_object_request(&session->info, replier_id, &payload.base);
uxr_serialize_WRITE_DATA_Payload_Data(&ub, &payload);
UXR_PREPARE_SHARED_MEMORY(session, replier_id, &ub, (uint16_t) (SAMPLE_IDENTITY_SIZE + len), rv);
uxr_serialize_SampleIdentity(&ub, sample_id);
ucdr_serialize_array_uint8_t(&ub, buffer, len);
}
UXR_UNLOCK_STREAM_ID(session, stream_id);
return rv;
}
@ -76,6 +88,8 @@ uint16_t uxr_buffer_topic(
ucdrBuffer ub;
size_t payload_size = WRITE_DATA_PAYLOAD_SIZE + len;
UXR_LOCK_STREAM_ID(session, stream_id);
ub.error = !uxr_prepare_stream_to_write_submessage(session, stream_id, payload_size, &ub, SUBMESSAGE_ID_WRITE_DATA,
FORMAT_DATA);
if (!ub.error)
@ -83,9 +97,12 @@ uint16_t uxr_buffer_topic(
WRITE_DATA_Payload_Data payload;
rv = uxr_init_base_object_request(&session->info, datawriter_id, &payload.base);
uxr_serialize_WRITE_DATA_Payload_Data(&ub, &payload);
UXR_PREPARE_SHARED_MEMORY(session, datawriter_id, &ub, (uint16_t) len, rv);
ucdr_serialize_array_uint8_t(&ub, buffer, len);
}
UXR_UNLOCK_STREAM_ID(session, stream_id);
return rv;
}
@ -94,11 +111,14 @@ uint16_t uxr_prepare_output_stream(
uxrStreamId stream_id,
uxrObjectId entity_id,
ucdrBuffer* ub,
uint32_t data_size)
uint32_t len)
{
uint16_t rv = UXR_INVALID_REQUEST_ID;
size_t payload_size = WRITE_DATA_PAYLOAD_SIZE + data_size;
UXR_LOCK_STREAM_ID(session, stream_id);
size_t payload_size = WRITE_DATA_PAYLOAD_SIZE + len;
ub->error = !uxr_prepare_stream_to_write_submessage(session, stream_id, payload_size, ub, SUBMESSAGE_ID_WRITE_DATA,
FORMAT_DATA);
if (!ub->error)
@ -111,6 +131,8 @@ uint16_t uxr_prepare_output_stream(
void* args = ub->args;
ucdr_init_buffer(ub, ub->iterator, (size_t)(ub->final - ub->iterator));
ucdr_set_on_full_buffer_callback(ub, on_full_buffer, args);
UXR_PREPARE_SHARED_MEMORY(session, entity_id, ub, (uint16_t) len, rv);
}
return rv;
@ -174,6 +196,10 @@ uint16_t uxr_prepare_output_stream_fragmented(
{
uint16_t rv = UXR_INVALID_REQUEST_ID;
#ifdef UCLIENT_PROFILE_MULTITHREAD
return rv;
#endif /* ifdef UCLIENT_PROFILE_MULTITHREAD */
size_t user_required_space = data_size + SUBHEADER_SIZE + WRITE_DATA_PAYLOAD_SIZE;
uxrOutputReliableStream* stream = uxr_get_output_reliable_stream(&session->streams, stream_id.index);

256
src/c/profile/matching/matching.c Normal file
View File

@ -0,0 +1,256 @@
#include "matching_internal.h"
#include <uxr/client/core/type/xrce_types.h>
#include <stdint.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <stdarg.h>
//==================================================================
// PRIVATE
//==================================================================
// djb2 by Dan Bernstein: http://www.cse.yorku.ca/~oz/hash.html
void hash_string(
unsigned char* str,
char* hash,
bool initial)
{
hash_int_t int_hash = 5381;
if (initial)
{
int_hash = *((hash_int_t*) hash);
}
int c;
do {
c = *str++;
int_hash = ((int_hash << 5) + int_hash) + (hash_int_t)c; /* hash * 33 + c */
} while (c);
for (size_t i = 0; i < UXR_MATCHING_HASH_SIZE; i++)
{
hash[i] = ((char*)&int_hash)[i];
}
}
// Find first occurrence of tag in XML
bool find_tag_xml(
const char* xml,
size_t len,
char* tag,
const char** content,
size_t* content_len)
{
size_t tag_len = strlen(tag);
bool found_begin = false;
bool found_end = false;
for (size_t i = 1; i < len; i++)
{
if (!found_begin && 0 == memcmp(&xml[i], tag, tag_len) && xml[i - 1] == '<')
{
size_t tag_opener_len = 0;
while (xml[i + tag_opener_len] != '>')
{
tag_opener_len++;
}
*content = &xml[i + tag_opener_len + 1];
found_begin = true;
}
else if (found_begin && 0 == memcmp(&xml[i], tag, tag_len) && xml[i - 1] == '/')
{
*content_len = (size_t)(&xml[i - 2] - *content);
found_end = true;
break;
}
}
return found_begin && found_end;
}
// Find property in first occurrence of tag in XML
bool find_tag_property(
const char* xml,
size_t len,
const char* tag,
char* property,
const char** content,
size_t* content_len)
{
size_t tag_len = strlen(tag);
size_t property_len = strlen(property);
bool found_tag = false;
bool found_property = false;
for (size_t i = 1; i < len; i++)
{
if (!found_tag && 0 == memcmp(&xml[i], tag, tag_len) && xml[i - 1] == '<')
{
found_tag = true;
}
else if (found_tag && 0 == memcmp(&xml[i], property, property_len))
{
*content = &xml[i + property_len + 2];
i += property_len + 2;
*content_len = 0;
while (xml[i + (*content_len)] != '"')
{
*content_len += 1;
}
found_property = true;
break;
}
}
return found_tag && found_property;
}
bool uxr_generate_hash_from_xml(
const char* xml,
uxrObjectId id,
char* hash)
{
bool found = true;
char name_type_buffer[100];
switch (id.type)
{
case UXR_DATAREADER_ID:
case UXR_DATAWRITER_ID:
{
const char* data_reader_or_writer = (id.type == UXR_DATAREADER_ID) ? "data_reader\0" : "data_writer\0";
char xml_strings[3][12] =
{
"dds",
"",
"topic"
};
memmove(xml_strings[1], data_reader_or_writer, 12);
const char* content_in = xml;
char* content_out;
size_t content_len_in = strlen(content_in);
size_t content_len_out;
for (size_t i = 0; i < 3; i++)
{
if (find_tag_xml(content_in, content_len_in, xml_strings[i], (const char**)&content_out,
&content_len_out))
{
content_in = content_out;
content_len_in = content_len_out;
}
else
{
return false;
}
}
size_t topic_name_len;
size_t type_name_len;
found &= find_tag_xml(content_in, content_len_in, "name", (const char**)&content_out, &topic_name_len);
memcpy(name_type_buffer, content_out, topic_name_len);
found &= find_tag_xml(content_in, content_len_in, "dataType", (const char**)&content_out, &type_name_len);
memcpy(&name_type_buffer[topic_name_len], content_out, type_name_len);
name_type_buffer[topic_name_len + type_name_len] = '\0';
hash_string((unsigned char*) name_type_buffer, hash, false);
break;
}
case UXR_REQUESTER_ID:
case UXR_REPLIER_ID:
{
char* content_out;
size_t service_name_len;
size_t request_type_name_len;
size_t reply_type_name_len;
found &= find_tag_property(xml,
strlen(xml),
(id.type == UXR_REQUESTER_ID) ? "requester" : "replier",
"service_name",
(const char**)&content_out,
&service_name_len);
if (found)
{
memcpy(name_type_buffer, content_out, service_name_len);
}
found &= find_tag_property(xml,
strlen(xml),
(id.type == UXR_REQUESTER_ID) ? "requester" : "replier",
"request_type",
(const char**)&content_out,
&request_type_name_len);
if (found)
{
memcpy(&name_type_buffer[service_name_len], content_out, request_type_name_len);
}
found &= find_tag_property(xml,
strlen(xml),
(id.type == UXR_REQUESTER_ID) ? "requester" : "replier",
"reply_type",
(const char**)&content_out,
&reply_type_name_len);
if (found)
{
memcpy(&name_type_buffer[service_name_len + request_type_name_len], content_out, reply_type_name_len);
name_type_buffer[service_name_len + request_type_name_len + reply_type_name_len] = '\0';
}
if (found)
{
hash_string((unsigned char*) name_type_buffer, hash, false);
}
break;
}
}
return found;
}
bool uxr_generate_hash_from_ref(
const char* ref,
char* hash)
{
hash_string((unsigned char*) ref, hash, false);
return true;
}
bool uxr_generate_hash_from_strings(
char* hash,
size_t number_strings,
...)
{
va_list args;
va_start(args, number_strings);
char* s = va_arg(args, char*);
hash_string((unsigned char*) s, hash, false);
for (size_t i = 1; i < number_strings; i++)
{
s = va_arg(args, char*);
hash_string((unsigned char*) s, hash, true);
}
va_end(args);
return true;
}
bool uxr_match_endpoint_qosbinary(
const OBJK_Endpoint_QosBinary* entity_1,
const OBJK_Endpoint_QosBinary* entity_2)
{
bool matched = true;
matched &= entity_1->qos_flags == entity_2->qos_flags;
return matched;
}

58
src/c/profile/matching/matching_internal.h Normal file
View File

@ -0,0 +1,58 @@
// Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef UXR_CLIENT_PROFILE_MATCHING_INTERNAL_H_
#define UXR_CLIENT_PROFILE_MATCHING_INTERNAL_H_
#ifdef __cplusplus
extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/visibility.h>
#include <uxr/client/core/session/object_id.h>
#include <uxr/client/core/type/xrce_types.h>
#include <stdbool.h>
typedef uint32_t hash_int_t;
#define UXR_MATCHING_HASH_SIZE sizeof(hash_int_t)
bool uxr_generate_hash_from_xml(
const char* xml,
uxrObjectId id,
char* hash);
bool uxr_generate_hash_from_ref(
const char* ref,
char* hash);
bool uxr_generate_hash_from_strings(
char* hash,
size_t number_strings,
...);
bool uxr_match_endpoint_qosbinary(
const OBJK_Endpoint_QosBinary* entity_1,
const OBJK_Endpoint_QosBinary* entity_2);
#ifdef __cplusplus
}
#endif // ifdef __cplusplus
#endif // UXR_CLIENT_PROFILE_MATCHING_INTERNAL_H_

83
src/c/profile/multithread/multithread.c Normal file
View File

@ -0,0 +1,83 @@
#include <uxr/client/profile/multithread/multithread.h>
#include <uxr/client/core/session/session.h>
#include "../../core/session/stream/stream_storage_internal.h"
//==================================================================
// PRIVATE
//==================================================================
void uxr_init_lock(
uxrMutex* mutex)
{
#if defined(PLATFORM_NAME_FREERTOS)
mutex->impl = xSemaphoreCreateRecursiveMutexStatic( &mutex->xMutexBuffer );
#elif defined(UCLIENT_PLATFORM_ZEPHYR)
k_mutex_init(&mutex->impl);
#elif defined(UCLIENT_PLATFORM_POSIX)
pthread_mutexattr_t mat;
pthread_mutexattr_init(&mat);
pthread_mutexattr_settype(&mat, PTHREAD_MUTEX_RECURSIVE);
pthread_mutex_init(&mutex->impl, &mat);
#else
#error XRCE multithreading not supported for this platform.
#endif /* if defined(PLATFORM_NAME_FREERTOS) */
}
void uxr_lock(
uxrMutex* mutex)
{
#if defined(PLATFORM_NAME_FREERTOS)
xSemaphoreTakeRecursive(mutex->impl, portMAX_DELAY);
#elif defined(UCLIENT_PLATFORM_ZEPHYR)
k_mutex_lock(&mutex->impl, K_FOREVER);
#elif defined(UCLIENT_PLATFORM_POSIX)
pthread_mutex_lock(&mutex->impl);
#else
#error XRCE multithreading not supported for this platform.
#endif /* if defined(PLATFORM_NAME_FREERTOS) */
}
void uxr_unlock(
uxrMutex* mutex)
{
#if defined(PLATFORM_NAME_FREERTOS)
xSemaphoreGiveRecursive(mutex->impl);
#elif defined(UCLIENT_PLATFORM_ZEPHYR)
k_mutex_unlock(&mutex->impl);
#elif defined(UCLIENT_PLATFORM_POSIX)
pthread_mutex_unlock(&mutex->impl);
#else
#error XRCE multithreading not supported for this platform.
#endif /* if defined(PLATFORM_NAME_FREERTOS) */
}
uxrMutex* uxr_get_stream_mutex_from_id(
struct uxrSession* session,
uxrStreamId stream_id)
{
uxrMutex* mutex = NULL;
if (stream_id.type == UXR_BEST_EFFORT_STREAM && stream_id.direction == UXR_OUTPUT_STREAM)
{
uxrOutputBestEffortStream* stream = uxr_get_output_best_effort_stream(&session->streams, stream_id.index);
mutex = (stream == NULL) ? NULL : &stream->mutex;
}
else if (stream_id.type == UXR_BEST_EFFORT_STREAM && stream_id.direction == UXR_INPUT_STREAM)
{
uxrInputBestEffortStream* stream = uxr_get_input_best_effort_stream(&session->streams, stream_id.index);
mutex = (stream == NULL) ? NULL : &stream->mutex;
}
else if (stream_id.type == UXR_RELIABLE_STREAM && stream_id.direction == UXR_OUTPUT_STREAM)
{
uxrOutputReliableStream* stream = uxr_get_output_reliable_stream(&session->streams, stream_id.index);
mutex = (stream == NULL) ? NULL : &stream->mutex;
}
else if (stream_id.type == UXR_RELIABLE_STREAM && stream_id.direction == UXR_INPUT_STREAM)
{
uxrInputReliableStream* stream = uxr_get_input_reliable_stream(&session->streams, stream_id.index);
mutex = (stream == NULL) ? NULL : &stream->mutex;
}
return mutex;
}

482
src/c/profile/shared_memory/shared_memory.c Normal file
View File

@ -0,0 +1,482 @@
#include <uxr/client/config.h>
#include <uxr/client/core/session/session.h>
#include <uxr/client/core/type/xrce_types.h>
#include <uxr/client/profile/multithread/multithread.h>
#include <uxr/client/core/type/xrce_types.h>
#include "../../core/session/stream/stream_storage_internal.h"
#include "../matching/matching_internal.h"
#include <string.h>
#if defined(WIN32)
#include <BaseTsd.h>
typedef SSIZE_T ssize_t;
#else
#include <sys/types.h>
#endif /* if defined(WIN32) */
//==================================================================
// PRIVATE
//==================================================================
typedef struct uxr_shared_memory_buffer_item_t
{
struct uxr_shared_memory_buffer_item_t* next;
ucdrBuffer data;
uint16_t data_size;
uint16_t request_id;
} uxr_shared_memory_buffer_item_t;
typedef enum entity_descriptor_type_t
{
UXR_SHARED_MEMORY_HASH_ENTITY,
UXR_SHARED_MEMORY_BIN_ENTITY
} entity_descriptor_type_t;
typedef struct uxr_shared_memory_entity_t
{
uxrObjectId object_id;
uxrSession* session;
entity_descriptor_type_t type;
union
{
char hash[UXR_MATCHING_HASH_SIZE];
OBJK_DataWriter_Binary datawriter;
OBJK_DataReader_Binary datareader;
} data;
struct uxr_shared_memory_entity_t* related_topic;
} uxr_shared_memory_entity_t;
typedef struct uxr_shared_memory_matrix_block_t
{
bool matched;
struct uxr_shared_memory_buffer_item_t* list;
} uxr_shared_memory_matrix_block_t;
typedef struct uxr_shared_memory_map_t
{
uxr_shared_memory_entity_t entities[UXR_CONFIG_SHARED_MEMORY_MAX_ENTITIES];
size_t entities_len;
uxr_shared_memory_matrix_block_t matrix[UXR_CONFIG_SHARED_MEMORY_MAX_ENTITIES][UXR_CONFIG_SHARED_MEMORY_MAX_ENTITIES
];
uxr_shared_memory_buffer_item_t mem_pool[UXR_CONFIG_SHARED_MEMORY_STATIC_MEM_SIZE];
uxr_shared_memory_buffer_item_t* mempool_freeitems;
bool init;
#ifdef UCLIENT_PROFILE_MULTITHREAD
uxrMutex lock;
#endif /* ifdef UCLIENT_PROFILE_MULTITHREAD */
} uxr_shared_memory_map_t;
static uxr_shared_memory_map_t uxr_sm_map = {
0
};
// Double linked pool for ucdrBuffers
void uxr_prepend_to_list(
uxr_shared_memory_buffer_item_t* member,
uxr_shared_memory_buffer_item_t** list)
{
member->next = *list;
*list = member;
}
uxr_shared_memory_buffer_item_t* uxr_pop_head_from_list(
uxr_shared_memory_buffer_item_t** list)
{
uxr_shared_memory_buffer_item_t* member = *list;
*list = (*list == NULL ) ? NULL : (*list)->next;
return member;
}
#define UXR_SHARED_MEMORY_INIT() if (!uxr_sm_map.init){uxr_init_shared_memory();}
void uxr_init_shared_memory()
{
uxr_sm_map.init = true;
UXR_INIT_LOCK(&uxr_sm_map.lock);
for (size_t i = 0; i < UXR_CONFIG_SHARED_MEMORY_STATIC_MEM_SIZE; i++)
{
uxr_prepend_to_list(&uxr_sm_map.mem_pool[i], &uxr_sm_map.mempool_freeitems);
}
}
void uxr_clean_shared_memory()
{
memset(&uxr_sm_map, 0, sizeof(uxr_sm_map));
}
// API
bool uxr_shared_memory_entity_compare(
uxr_shared_memory_entity_t* e1,
uxr_shared_memory_entity_t* e2)
{
bool ret = true;
for (uint8_t i = 0; i < sizeof(e1->session->info.key); i++)
{
ret &= e1->session->info.key[i] == e2->session->info.key[i];
}
ret &= e1->object_id.id == e2->object_id.id;
ret &= e1->object_id.type == e2->object_id.type;
return ret;
}
ssize_t uxr_shared_memory_get_entity_index(
uxrSession* session,
uxrObjectId* entity_id)
{
uxr_shared_memory_entity_t aux;
aux.session = session;
aux.object_id = *entity_id;
for (size_t i = 0; i < uxr_sm_map.entities_len; i++)
{
if (uxr_shared_memory_entity_compare(&aux, &uxr_sm_map.entities[i]))
{
return (ssize_t)i;
}
}
return -1;
}
uxr_shared_memory_entity_t* uxr_shared_memory_get_entity(
uxrSession* session,
uxrObjectId* entity_id)
{
uxr_shared_memory_entity_t aux;
aux.session = session;
aux.object_id = *entity_id;
for (size_t i = 0; i < uxr_sm_map.entities_len; i++)
{
if (uxr_shared_memory_entity_compare(&aux, &uxr_sm_map.entities[i]))
{
return &uxr_sm_map.entities[i];
}
}
return NULL;
}
void uxr_prepare_shared_memory(
uxrSession* session,
uxrObjectId entity_id,
ucdrBuffer* ub,
uint16_t data_size,
uint16_t request_id)
{
UXR_SHARED_MEMORY_INIT();
UXR_LOCK(&uxr_sm_map.lock);
ssize_t entity_index = uxr_shared_memory_get_entity_index(session, &entity_id);
if (-1 == entity_index)
{
return;
}
for (size_t i = 0; i < uxr_sm_map.entities_len; i++)
{
if (uxr_sm_map.matrix[entity_index][i].matched)
{
uxr_shared_memory_buffer_item_t* item = uxr_pop_head_from_list(&uxr_sm_map.mempool_freeitems);
if (item != NULL)
{
item->data = *ub;
item->data_size = data_size;
item->request_id = request_id;
uxr_prepend_to_list(item, &uxr_sm_map.matrix[entity_index][i].list);
}
}
}
UXR_UNLOCK(&uxr_sm_map.lock);
}
void uxr_handle_shared_memory()
{
UXR_SHARED_MEMORY_INIT();
UXR_LOCK(&uxr_sm_map.lock);
for (size_t i = 0; i < uxr_sm_map.entities_len; i++)
{
if (uxr_sm_map.entities[i].object_id.type != UXR_DATAWRITER_ID
&& uxr_sm_map.entities[i].object_id.type != UXR_REPLIER_ID
&& uxr_sm_map.entities[i].object_id.type != UXR_REQUESTER_ID
)
{
continue;
}
for (size_t j = 0; j < uxr_sm_map.entities_len; j++)
{
while (uxr_sm_map.matrix[i][j].list != NULL)
{
uxrStreamId stream_id = {
.type = UXR_SHARED_MEMORY_STREAM
};
uxr_shared_memory_buffer_item_t* item = uxr_pop_head_from_list(&uxr_sm_map.matrix[i][j].list);
switch (uxr_sm_map.entities[j].object_id.type)
{
case UXR_DATAREADER_ID:
{
if (NULL != uxr_sm_map.entities[j].session->on_topic)
{
uxr_sm_map.entities[j].session->on_topic(
uxr_sm_map.entities[j].session,
uxr_sm_map.entities[j].object_id,
0, // Req ID = 0 means shared memory?
stream_id,
&item->data,
item->data_size,
uxr_sm_map.entities[j].session->on_topic_args
);
}
break;
}
case UXR_REPLIER_ID:
{
if (NULL != uxr_sm_map.entities[j].session->on_request)
{
SampleIdentity sample_identity;
sample_identity.sequence_number.low = item->request_id;
// Save session key
memcpy(sample_identity.writer_guid.guidPrefix.data,
uxr_sm_map.entities[i].session->info.key,
sizeof(uxr_sm_map.entities[i].session->info.key));
// Save source uxrObjectId info
sample_identity.writer_guid.entityId.entityKey[1] =
(uint8_t) (uxr_sm_map.entities[i].object_id.id >> 8) & 0xFF;
sample_identity.writer_guid.entityId.entityKey[0] =
(uint8_t) uxr_sm_map.entities[i].object_id.id &
0xFF;
sample_identity.writer_guid.entityId.entityKind = uxr_sm_map.entities[i].object_id.type;
uxr_sm_map.entities[j].session->on_request(
uxr_sm_map.entities[j].session,
uxr_sm_map.entities[j].object_id,
0, // Req ID = 0 means shared memory?
&sample_identity,
&item->data,
item->data_size,
uxr_sm_map.entities[j].session->on_topic_args);
}
break;
}
case UXR_REQUESTER_ID:
{
if (NULL != uxr_sm_map.entities[j].session->on_reply)
{
SampleIdentity sample_identity;
size_t offset = item->data.offset;
uxr_deserialize_SampleIdentity(&item->data, &sample_identity);
// Check destination session key and uxrObjectId info
if (!memcmp(uxr_sm_map.entities[j].session->info.key,
sample_identity.writer_guid.guidPrefix.data,
sizeof(uxr_sm_map.entities[j].session->info.key))
&& uxr_sm_map.entities[j].object_id.id ==
(sample_identity.writer_guid.entityId.entityKey[1] >> 8) +
sample_identity.writer_guid.entityId.entityKey[0]
&& sample_identity.writer_guid.entityId.entityKind ==
uxr_sm_map.entities[j].object_id.type)
{
// TODO: Check this size calculation taking alignment
item->data_size = (uint16_t)(item->data_size - (item->data.offset - offset));
uxr_sm_map.entities[j].session->on_reply(
uxr_sm_map.entities[j].session,
uxr_sm_map.entities[j].object_id,
0, // Req ID = 0 means shared memory?
(uint16_t)sample_identity.sequence_number.low,
&item->data,
(size_t)item->data_size,
uxr_sm_map.entities[j].session->on_topic_args);
}
}
break;
}
default:
break;
}
uxr_prepend_to_list(item, &uxr_sm_map.mempool_freeitems);
}
}
}
UXR_UNLOCK(&uxr_sm_map.lock);
}
bool uxr_match_binary_entities(
uxr_shared_memory_entity_t* entity_1,
uxr_shared_memory_entity_t* entity_2)
{
if (!(entity_1->object_id.type == UXR_DATAREADER_ID && entity_2->object_id.type == UXR_DATAWRITER_ID) &&
!(entity_1->object_id.type == UXR_DATAWRITER_ID && entity_2->object_id.type == UXR_DATAREADER_ID) &&
!(entity_1->object_id.type == UXR_REQUESTER_ID && entity_2->object_id.type == UXR_REPLIER_ID) &&
!(entity_1->object_id.type == UXR_REPLIER_ID && entity_2->object_id.type == UXR_REQUESTER_ID))
{
return false;
}
bool matched = true;
switch (entity_1->object_id.type)
{
case UXR_DATAREADER_ID:
{
OBJK_DataReader_Binary* dr = (OBJK_DataReader_Binary*) &entity_1->data.datareader;
OBJK_DataWriter_Binary* dw = (OBJK_DataWriter_Binary*) &entity_2->data.datawriter;
matched &= uxr_match_endpoint_qosbinary(&dr->qos.base, &dw->qos.base);
matched &= 0 == strcmp(entity_1->related_topic->data.hash, entity_2->related_topic->data.hash);
break;
}
case UXR_DATAWRITER_ID:
{
OBJK_DataWriter_Binary* dw = (OBJK_DataWriter_Binary*) &entity_1->data.datawriter;
OBJK_DataReader_Binary* dr = (OBJK_DataReader_Binary*) &entity_2->data.datareader;
matched &= uxr_match_endpoint_qosbinary(&dr->qos.base, &dw->qos.base);
matched &= 0 == strcmp(entity_1->related_topic->data.hash, entity_2->related_topic->data.hash);
break;
}
case UXR_REQUESTER_ID:
case UXR_REPLIER_ID:
default:
matched = false;
break;
}
return matched;
}
void uxr_update_shared_memory_matching()
{
UXR_LOCK(&uxr_sm_map.lock);
for (size_t i = 0; i < uxr_sm_map.entities_len; i++)
{
for (size_t j = 0; j < uxr_sm_map.entities_len; j++)
{
if (i != j &&
uxr_sm_map.entities[i].type == uxr_sm_map.entities[j].type &&
uxr_sm_map.entities[i].object_id.type != uxr_sm_map.entities[j].object_id.type &&
uxr_sm_map.entities[i].object_id.type != UXR_TOPIC_ID &&
uxr_sm_map.entities[j].object_id.type != UXR_TOPIC_ID)
{
switch (uxr_sm_map.entities[i].type)
{
case UXR_SHARED_MEMORY_HASH_ENTITY:
uxr_sm_map.matrix[i][j].matched =
0 == memcmp(uxr_sm_map.entities[i].data.hash, uxr_sm_map.entities[j].data.hash,
UXR_MATCHING_HASH_SIZE);
break;
case UXR_SHARED_MEMORY_BIN_ENTITY:
uxr_sm_map.matrix[i][j].matched = uxr_match_binary_entities(&uxr_sm_map.entities[i],
&uxr_sm_map.entities[j]);
break;
default:
uxr_sm_map.matrix[i][j].matched = 0;
}
}
}
}
UXR_UNLOCK(&uxr_sm_map.lock);
}
void uxr_add_shared_memory_entity_xml(
uxrSession* session,
uxrObjectId entity_id,
const char* xml)
{
UXR_SHARED_MEMORY_INIT();
UXR_LOCK(&uxr_sm_map.lock);
if (uxr_sm_map.entities_len <= UXR_CONFIG_SHARED_MEMORY_MAX_ENTITIES - 1)
{
uxr_sm_map.entities[uxr_sm_map.entities_len].object_id = entity_id;
uxr_sm_map.entities[uxr_sm_map.entities_len].session = session;
uxr_sm_map.entities[uxr_sm_map.entities_len].type = UXR_SHARED_MEMORY_HASH_ENTITY;
uxr_generate_hash_from_xml(xml, entity_id, uxr_sm_map.entities[uxr_sm_map.entities_len].data.hash);
uxr_sm_map.entities_len++;
uxr_update_shared_memory_matching();
}
UXR_UNLOCK(&uxr_sm_map.lock);
}
void uxr_add_shared_memory_entity_bin(
uxrSession* session,
uxrObjectId entity_id,
const void* entity)
{
UXR_SHARED_MEMORY_INIT();
UXR_LOCK(&uxr_sm_map.lock);
if (uxr_sm_map.entities_len <= UXR_CONFIG_SHARED_MEMORY_MAX_ENTITIES - 1)
{
uxr_sm_map.entities[uxr_sm_map.entities_len].object_id = entity_id;
uxr_sm_map.entities[uxr_sm_map.entities_len].session = session;
uxr_sm_map.entities[uxr_sm_map.entities_len].type = UXR_SHARED_MEMORY_BIN_ENTITY;
uxrObjectId related_object_id;
switch (entity_id.type)
{
case UXR_DATAWRITER_ID:
related_object_id = uxr_object_id_from_raw(((OBJK_DataWriter_Binary*) entity)->topic_id.data);
uxr_sm_map.entities[uxr_sm_map.entities_len].data.datawriter = *((OBJK_DataWriter_Binary*) entity);
uxr_sm_map.entities[uxr_sm_map.entities_len].related_topic = uxr_shared_memory_get_entity(session,
&related_object_id);
break;
case UXR_DATAREADER_ID:
related_object_id = uxr_object_id_from_raw(((OBJK_DataReader_Binary*) entity)->topic_id.data);
uxr_sm_map.entities[uxr_sm_map.entities_len].data.datareader = *((OBJK_DataReader_Binary*) entity);
uxr_sm_map.entities[uxr_sm_map.entities_len].related_topic = uxr_shared_memory_get_entity(session,
&related_object_id);
break;
case UXR_REQUESTER_ID:
uxr_sm_map.entities[uxr_sm_map.entities_len].type = UXR_SHARED_MEMORY_HASH_ENTITY;
uxr_generate_hash_from_strings(
uxr_sm_map.entities[uxr_sm_map.entities_len].data.hash, 5,
((OBJK_Requester_Binary*) entity)->service_name,
((OBJK_Requester_Binary*) entity)->request_type,
((OBJK_Requester_Binary*) entity)->reply_type,
((OBJK_Requester_Binary*) entity)->request_topic_name,
((OBJK_Requester_Binary*) entity)->reply_topic_name);
uxr_sm_map.entities[uxr_sm_map.entities_len].related_topic = NULL;
break;
case UXR_REPLIER_ID:
uxr_sm_map.entities[uxr_sm_map.entities_len].type = UXR_SHARED_MEMORY_HASH_ENTITY;
uxr_generate_hash_from_strings(
uxr_sm_map.entities[uxr_sm_map.entities_len].data.hash, 5,
((OBJK_Replier_Binary*) entity)->service_name,
((OBJK_Replier_Binary*) entity)->request_type,
((OBJK_Replier_Binary*) entity)->reply_type,
((OBJK_Replier_Binary*) entity)->request_topic_name,
((OBJK_Replier_Binary*) entity)->reply_topic_name);
uxr_sm_map.entities[uxr_sm_map.entities_len].related_topic = NULL;
break;
case UXR_TOPIC_ID:
uxr_sm_map.entities[uxr_sm_map.entities_len].type = UXR_SHARED_MEMORY_HASH_ENTITY;
uxr_generate_hash_from_strings(
uxr_sm_map.entities[uxr_sm_map.entities_len].data.hash, 2,
((OBJK_Topic_Binary*) entity)->topic_name,
((OBJK_Topic_Binary*) entity)->type_name);
uxr_sm_map.entities[uxr_sm_map.entities_len].related_topic = NULL;
default:
break;
}
uxr_sm_map.entities_len++;
uxr_update_shared_memory_matching();
}
UXR_UNLOCK(&uxr_sm_map.lock);
}

72
src/c/profile/shared_memory/shared_memory_internal.h Normal file
View File

@ -0,0 +1,72 @@
// Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef UXR_CLIENT_PROFILE_SHARED_MEMORY_INTERNAL_H_
#define UXR_CLIENT_PROFILE_SHARED_MEMORY_INTERNAL_H_
#ifdef __cplusplus
extern "C"
{
#endif // ifdef __cplusplus
#include <uxr/client/config.h>
#include <uxr/client/visibility.h>
#include <uxr/client/core/session/session.h>
#include <uxr/client/profile/multithread/multithread.h>
#ifdef UCLIENT_PROFILE_SHARED_MEMORY
#define UXR_PREPARE_SHARED_MEMORY(a, b, c, d, e) uxr_prepare_shared_memory(a, b, c, d, e)
#define UXR_HANDLE_SHARED_MEMORY() uxr_handle_shared_memory()
#define UXR_ADD_SHARED_MEMORY_ENTITY_XML(a, b, c) uxr_add_shared_memory_entity_xml(a, b, c)
#define UXR_ADD_SHARED_MEMORY_ENTITY_BIN(a, b, c) uxr_add_shared_memory_entity_bin(a, b, (void*) c)
#define UXR_CLEAN_SHARED_MEMORY() uxr_clean_shared_memory()
#else // UCLIENT_PROFILE_SHARED_MEMORY
#define UXR_PREPARE_SHARED_MEMORY(a, b, c, d, e)
#define UXR_HANDLE_SHARED_MEMORY()
#define UXR_ADD_SHARED_MEMORY_ENTITY_XML(a, b, c)
#define UXR_ADD_SHARED_MEMORY_ENTITY_BIN(a, b, c)
#define UXR_CLEAN_SHARED_MEMORY()
#endif // UCLIENT_PROFILE_SHARED_MEMORY
void uxr_prepare_shared_memory(
uxrSession* session,
uxrObjectId entity_id,
ucdrBuffer* ub,
uint16_t data_size,
uint16_t request_id);
void uxr_handle_shared_memory();
void uxr_add_shared_memory_entity_xml(
uxrSession* session,
uxrObjectId entity_id,
const char* xml);
void uxr_add_shared_memory_entity_bin(
uxrSession* session,
uxrObjectId entity_id,
const void* entity);
void uxr_clean_shared_memory();
#ifdef __cplusplus
}
#endif // ifdef __cplusplus
#endif // UXR_CLIENT_PROFILE_SHARED_MEMORY_INTERNAL_H_

80
test/shared_memory/CMakeLists.txt Normal file
View File

@ -0,0 +1,80 @@
# Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
project(uxr_sharedmem_unitary_tests LANGUAGES C CXX)
include(ExternalProject)
ExternalProject_Add(uxr_sharedmem_project
SOURCE_DIR
${CMAKE_CURRENT_SOURCE_DIR}/../../
BINARY_DIR
${PROJECT_BINARY_DIR}/microxrcedds_client_sharedmem
INSTALL_DIR
${PROJECT_BINARY_DIR}/temp_install/sharedmem
CMAKE_CACHE_ARGS
-DUCLIENT_PROFILE_SHARED_MEMORY:BOOL=ON
-DUCLIENT_PROFILE_MULTITHREAD:BOOL=ON
-DUCLIENT_PROFILE_MATCHING:BOOL=ON
-DUCLIENT_SHARED_MEMORY_MAX_ENTITIES:STRING=30
-DUCLIENT_SHARED_MEMORY_STATIC_MEM_SIZE:STRING=30
-DCMAKE_INSTALL_PREFIX:PATH=<INSTALL_DIR>
-DCMAKE_BUILD_TYPE:STRING=${CMAKE_BUILD_TYPE}
BUILD_BYPRODUCTS ${PROJECT_BINARY_DIR}/temp_install/lib/sharedmem/libmicroxrcedds_client.a
BUILD_BYPRODUCTS ${PROJECT_BINARY_DIR}/temp_install/lib/sharedmem/libmicrocdr.a
)
ExternalProject_Get_Property(uxr_sharedmem_project INSTALL_DIR)
add_library(ucdr_sharedmem STATIC IMPORTED GLOBAL)
add_dependencies(ucdr_sharedmem uxr_sharedmem_project)
set_property(TARGET ucdr_sharedmem PROPERTY IMPORTED_LOCATION ${PROJECT_BINARY_DIR}/temp_install/sharedmem/lib/libmicrocdr.a)
add_library(uxr_sharedmem STATIC IMPORTED GLOBAL)
add_dependencies(uxr_sharedmem uxr_sharedmem_project)
target_link_libraries(uxr_sharedmem INTERFACE ucdr_sharedmem)
set_property(TARGET uxr_sharedmem PROPERTY IMPORTED_LOCATION ${PROJECT_BINARY_DIR}/temp_install/sharedmem/lib/libmicroxrcedds_client.a)
add_executable(sharedmem_test SharedMemory.cpp)
set_common_compile_options(sharedmem_test)
if(MSVC OR MSVC_IDE)
target_compile_options(sharedmem_test PRIVATE /wd4996)
endif()
add_gtest(sharedmem_test SOURCES SharedMemory.cpp)
target_link_libraries(sharedmem_test
PUBLIC
ucdr_sharedmem
uxr_sharedmem
PRIVATE
${GTEST_BOTH_LIBRARIES}
${CMAKE_THREAD_LIBS_INIT}
)
target_include_directories(sharedmem_test
PUBLIC
${PROJECT_BINARY_DIR}/temp_install/sharedmem/include
PRIVATE
${GTEST_INCLUDE_DIRS}
)
set_target_properties(sharedmem_test PROPERTIES
CXX_STANDARD
11
CXX_STANDARD_REQUIRED
YES
)

1009
test/shared_memory/SharedMemory.cpp Normal file
View File

File diff suppressed because it is too large Load Diff