Dispatcher Executor

[JanStaschulat]

Implements a multi-threaded Executor.

New API Feature:

• A scheduling policy (e.g. priority) can be configured for a subscription callback.

Implementation:

• The user specifies a scheduling policy for each subscription.
• For each subscription a worker thread is created by the Executor.
• The Executor accesses the DDS and checks for new data.
• When new data for a subscription is available, the Executor dispatches it to the corresponding thread, hence its name.
• finally, the subscription callback is executed by the corresponding worker thread.

Limitation:

• only implemented for subscriptions
• trigger condition not supported

Example:

• Feature/foxy rbs executor demo micro-ROS/nuttx_apps#33

Publication:

• J. Staschulat, R. Lange, D.N. Dasari: Budget-based real-time Executor for Micro-ROS, arXiv preprint arXiv:2105.05590, 2021

[codecov-commenter]

Codecov Report

Merging #87 (723505f) into master (243ee63) will decrease coverage by 0.06%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##           master      #87      +/-   ##
==========================================
- Coverage   69.20%   69.14%   -0.07%     
==========================================
  Files          16       16              
  Lines        2715     2716       +1     
  Branches      765      765              
==========================================
- Hits         1879     1878       -1     
- Misses        450      451       +1     
- Partials      386      387       +1     

Impacted Files	Coverage Δ
rclc/src/rclc/executor_handle.c	69.86% <ø> (ø)
rclc/src/rclc/executor.c	61.94% <100.00%> (+0.03%)	⬆️
rclc/src/rclc/action_goal_handle.c	59.87% <0.00%> (-1.28%)	⬇️

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

[JanStaschulat]

Created this branch feature/dispatcher-executor-with-guard-condition-check on github/micro-ros including the guard_condition check (if a thread becomes available).

[ralph-lange]

LGTM

[pablogs9]

Hello @JanStaschulat, I guess that we need a refactor here in order to merge into main since I'm afraid that this will break almost all micro-ROS build systems.

My proposals are:

• Conditional CMake to avoid compiling new (thread-specific) .c files
• This CMake flag enables a compilation flag (i.e: RCLC_MULTITHREADED_EXECUTOR_ENABLED) that is used to:
  • Guard the new includes in the .h files
  • Guard the new structures in the .h files
  • Guard new code sections in .c files

Ideally, I would like to see this as another package in this RCLC repo and outside of rclc main package, but I know that this implies rearchitecting the executor.

So, ideally for me, this contribution shall add the new functionality letting the old mode by default.

[JanStaschulat]

Hi @pablogs9, thanks for pointing that out. I would propose

• to add a void * mt pointer for a multi-threaded-interface in rclc_executor_handle_t and rclc_executor_t struct.
• create a new package for multi-threaded executor
• at initialization of the multi-threaded executor this interface will point to a dynamically-allocated struct containing the necessary variables.

Benefit:

• minimal change for rclc_executor (only two pointer variables are added in the structs)
• no additional includes for <pthread.h> <sched.h>.

rclc_executor.h

/// Container for RCLC-Executor
typedef struct
{
 // ...
 /// interface for multi-threaded executor
 void * mt;
} rclc_executor_t;

rclc_executor_handle.h

/// Container for a handle.
typedef struct
{
  // ...
 /// interface for multi-threaded executor handle
 void * mt;
} rclc_executor_handle_t;

rclc_multi_threaded_executor.h

#include <pthread.h>
#include <sched.h>

typedef struct
{
  pthread_mutex_t * thread_state_mutex;
  pthread_mutex_t * micro_ros_mutex;
  rclc_executor_handle_t * handle;
}
rclc_executor_worker_thread_param_t;

/// Implementation for sporadic server scheduler
typedef enum
{
  RCLC_THREAD_NONE,
  RCLC_THREAD_READY,
  RCLC_THREAD_BUSY
} rclc_executor_thread_state_t;

/// Scheduling policy (SCHED_FIFO, SCHED_SPORADIC) and sched_param
typedef struct
{
  int policy;
  struct sched_param param;
} rclc_executor_sched_parameter_t;

.... // all the other strucs and #includes

typedef struct 
{
  pthread_mutex_t thread_state_mutex;
  /// multi-threaded executor: mutex for RCL layer
  pthread_mutex_t micro_ros_mutex;
} rclc_executor_multi_threaded_interface_t;

typedef struct 
{
  /// variables for multi-threading
  /// worker thread
  pthread_t worker_thread;
  /// worker thread state and its mutex
  rclc_executor_thread_state_t worker_thread_state;
  /// signaling condition variable and its mutex
  pthread_cond_t new_msg_cond;
  pthread_mutex_t new_msg_mutex;
  bool new_msg_avail;
  /// scheduling parameter
  rclc_executor_sched_parameter_t * sparam;
  pthread_attr_t tattr;
} rclc_executor_handle_multi_threaded_interface_t;

void rclc_executor_multi_threaded_init(rclc_executor_t * executor, ...)
{
 rclc_executor_multi_threaded_interface_t * mt_interface = executor->allocator->allocate(
    (sizeof(rclc_executor_multi_threaded_interface_t)),
    executor->allocator->state);
  executor->mt = mt_interface;
}

// similar add the struct to each rclc_executor_handle

[pablogs9]

That approach looks great to me. Please use void * multithread_handle; or something more verbose.

Just for curiosity, that would enable us to create independent packages with different multithread executors targets such as rclc_multithread_executor_freertos or rclc_multithread_executor_pthread?

[JanStaschulat]

That approach looks great to me. Please use void * multithread_handle; or something more verbose.

Just for curiosity, that would enable us to create independent packages with different multithread executors targets such as rclc_multithread_executor_freertos or rclc_multithread_executor_pthread?

Yes, absolutly.

We were even thinking about some intermediate executor interface functions to abstract the actual operating system function calls; similar to the approach for custom_transport layer in xrce-dds.

[pablogs9]

That abstraction layer is interesting, ideally, IMO it shall be implemented in RCL or RMW. But it can be such a big business.

[JanStaschulat]

Hi @pablogs9 @ralph-lange , I moved all the code of the multi-threaded executor to a new package rclc_dispatcher_executor. The structs rclc_executor_t and rclc_executor_handle_t have now only one additional void pointer multi_threaded instead of all the variables. This pointer will be used by the Dispatcher Executor to store the additional variables.

I added also an example program rclc_dispatcher_executor.c in rclc_examples package.

There are still some issues open:

• how shall the inialization be handled (calling rclc_executor_init as well as rclc_dispatcher_executor_init)
• people could still use the rclc_executor API with the rclc_dispatcher_executor. This will lead to false usage.
• need to write some unit tests (which requires a spin_some() function which could be called in a while-loop - instead of only a spin-function (that never ends).

Note: the build job on the OpenRobotics build farm (Rpr__rclc__ubuntu_jammy_amd64) fails only because of a timeout of rclc_parameter unit test.

[JanStaschulat]

will be part of the pull request #339