Compare commits

..

63 Commits

Author SHA1 Message Date
Yadunund
33a3545821 21.0.8 2024-11-08 22:04:11 +00:00
Yadunund
6a7eaca0e4 Update changelogs
Signed-off-by: Yadunund <yadunund@gmail.com>
2024-11-08 22:03:51 +00:00
mergify[bot]
21adb0ec1e associated clocks should be protected by mutex. (#2255) (#2258)
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
(cherry picked from commit 5d6e5fa766)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-11-07 13:39:37 -08:00
Cristóbal Arroyo
1f9e5e8992 Skip client_qos test (#2658)
* Skip client_qos test

Signed-off-by: Crola1702 <cristobal.arroyo@ekumenlabs.com>
Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-11-05 15:32:54 -05:00
mergify[bot]
34b3c99bd6 add logger level service to lifecycle node. (#2277) (#2288)
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
(cherry picked from commit e7f06398db)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-11-05 14:49:06 -05:00
Tomoya Fujita
531b2b1a08 Use the same context for the specified node in rclcpp::spin functions… (#2618)
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
Signed-off-by: Kotaro Yoshimoto <pythagora.yoshimoto@gmail.com>
2024-09-06 11:24:21 +02:00
Yadunund
2d34e13be2 21.0.7 2024-07-10 22:41:12 -07:00
Yadunund
627d91bef4 Update changelogs
Signed-off-by: Yadunund <yadunund@gmail.com>
2024-07-10 22:41:04 -07:00
jmachowinski
d588ccb562 fix: Fixed race condition in action server between is_ready and take"… (#2531)
* fix: Fixed race condition in action server between is_ready and take" (#2495)

Some background information: is_ready, take_data and execute data
may be called from different threads in any order. The code in the old
state expected them to be called in series, without interruption.
This lead to multiple race conditions, as the state of the pimpl objects
was altered by the three functions in a non thread safe way.

Co-authored-by: William Woodall <william@osrfoundation.org>
Signed-off-by: Janosch Machowinski <J.Machowinski@cellumation.com>

* fix: added workaround for call to double calls to take_data

This adds a workaround for a known bug in the executor in iron.

Signed-off-by: Janosch Machowinski <J.Machowinski@cellumation.com>

---------

Signed-off-by: Janosch Machowinski <J.Machowinski@cellumation.com>
Co-authored-by: Janosch Machowinski <J.Machowinski@cellumation.com>
Co-authored-by: William Woodall <william@osrfoundation.org>
2024-06-27 11:46:10 -07:00
Tomoya Fujita
c1a01fc08d revert call shutdown in LifecycleNode destructor (Iron) (#2559)
* Revert "lifecycle node dtor shutdown should be called only in primary state. (#2543)"

This reverts commit dceb612ef5.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

* Revert "call shutdown in LifecycleNode dtor to avoid leaving the device in un… (#2450) (#2490)"

This reverts commit f41a353b56.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

---------

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-06-11 07:41:30 -07:00
mergify[bot]
47adc8f0af Add test creating two content filter topics with the same topic name (#2546) (#2549) (#2550)
Signed-off-by: Mario-DL <mariodominguez@eprosima.com>
Co-authored-by: Mario Domínguez López <116071334+Mario-DL@users.noreply.github.com>
(cherry picked from commit 7c096888ca)

Co-authored-by: Alejandro Hernández Cordero <ahcorde@gmail.com>
2024-06-06 18:21:25 -07:00
Tomoya Fujita
dceb612ef5 lifecycle node dtor shutdown should be called only in primary state. (#2543)
* lifecycle node dtor shutdown should be called only in primary state.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

* LifecycleNode shutdown on dtor only with valid context. (#2545)

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

---------

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-06-06 11:05:00 -07:00
mergify[bot]
753a29b87f rclcpp::shutdown should not be called before LifecycleNode dtor. (#2527) (#2539)
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
(cherry picked from commit 22df1d593a)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-05-23 22:13:15 -07:00
mergify[bot]
d599f9e63b Do not generate the exception when action service response timeout. (#2464) (#2519)
* Do not generate the exception when action service response timeout.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

* address review comment.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

---------

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
(cherry picked from commit 6c7764e968)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-05-03 09:32:56 -07:00
mergify[bot]
2e8a23e09e Revise the description of service configure_introspection() (#2511) (#2514)
Signed-off-by: Barry Xu <barry.xu@sony.com>
(cherry picked from commit 55939613a0)

Co-authored-by: Barry Xu <barry.xu@sony.com>
2024-04-23 22:56:37 +02:00
Yadunund
ef85efaca2 21.0.6 2024-04-19 20:19:18 +08:00
Yadunund
a8f047d689 Update changelogs
Signed-off-by: Yadunund <yadunund@openrobotics.org>
2024-04-19 20:19:08 +08:00
mergify[bot]
f41a353b56 call shutdown in LifecycleNode dtor to avoid leaving the device in un… (#2450) (#2490)
* call shutdown in LifecycleNode dtor to avoid leaving the device in unknown state.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

* add test to verify LifecycleNode::shutdown is called on destructor.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

---------

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
(cherry picked from commit 04ea0bb002)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-04-07 14:58:59 -07:00
mergify[bot]
f80980b431 address ambiguous auto variable. (#2481) (#2486)
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
Signed-off-by: Steve Nogar <stephen.m.nogar.civ@army.mil>
(cherry picked from commit 3cdb25934e)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2024-04-04 16:59:23 -07:00
Yadunund
7907b2fee0 21.0.5 2024-02-07 17:29:33 +08:00
Yadunund
0dc2756dce Update changelogs
Signed-off-by: Yadunund <yadunund@openrobotics.org>
2024-02-07 17:29:13 +08:00
mergify[bot]
7877358e7b Fix data race in EventHandlerBase (#2349) (#2387)
Both the EventHandler and its associated pubs/subs share
the same underlying rmw event listener.
When a pub/sub is destroyed, the listener is destroyed.
There is a data race when the ~EventHandlerBase wants
to access the listener after it has been destroyed.

The EventHandler stores a shared_ptr of its associated pub/sub.
But since we were clearing the listener event callbacks on the
base class destructor ~EventHandlerBase, the pub/sub was
already destroyed, which means the rmw event listener was also
destroyed, thus causing a segfault when trying to obtain it
to clear the callbacks.

Clearing the callbacks on ~EventHandler instead of
~EventHandlerBase avoids the race, since the pub/sub are still valid.

Signed-off-by: Mauro Passerino <mpasserino@irobot.com>
(cherry picked from commit 411dbe8212)

Co-authored-by: mauropasse <mauropasse@hotmail.com>
2023-12-13 10:13:25 -08:00
mergify[bot]
dc6ac4e30f fix(rclcpp_components): increase the service queue sizes in component_container (#2363) (#2381)
* fix(rclcpp_components): increase the service queue sizes in component_container

Signed-off-by: M. Fatih Cırıt <mfc@leodrive.ai>
(cherry picked from commit 9c098e544e)

Co-authored-by: M. Fatih Cırıt <xmfcx@users.noreply.github.com>
2023-12-01 17:04:30 -05:00
Yadunund
da3d2f49b3 21.0.4 2023-11-17 11:27:34 +08:00
Yadunund
a043349ecc Update changelogs
Signed-off-by: Yadunund <yadunund@openrobotics.org>
2023-11-17 11:27:20 +08:00
mergify[bot]
abcdcc4ed7 Disable the loaned messages inside the executor. (#2335) (#2365)
* Disable the loaned messages inside the executor.

They are currently unsafe to use; see the comment in the
commit for more information.

Signed-off-by: Chris Lalancette <clalancette@gmail.com>
(cherry picked from commit f294488e17)

Co-authored-by: Chris Lalancette <clalancette@gmail.com>
2023-11-09 14:01:26 -05:00
mergify[bot]
613bcc52ba Add missing 'enable_rosout' comments (#2345) (#2346)
Signed-off-by: Jiaqi Li <ljq0831@qq.com>
(cherry picked from commit fff009a751)

Co-authored-by: Jiaqi Li <ljq0831@qq.com>
2023-10-31 08:17:46 -07:00
mergify[bot]
82a8dba6c3 address rate related flaky tests. (#2329) (#2341)
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
(cherry picked from commit fcbe64cff4)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2023-10-26 07:45:40 -07:00
Chris Lalancette
c67720e95c Add missing stdexcept include (#2186) (#2333)
Signed-off-by: Øystein Sture <os@skarvtech.com>
Co-authored-by: Øystein Sture <oysstu@users.noreply.github.com>
2023-10-10 15:26:10 -04:00
mergify[bot]
1ddf865efe Update SignalHandler get_global_signal_handler to avoid complex types in static memory (#2316) (#2322)
* Update SignalHandler get_global_signal_handler to avoid complex types in static memory

This was flagged by msan as a problem.

There's a description of why this is a potential problem here: https://google.github.io/styleguide/cppguide.html#Static_and_Global_Variables

Signed-off-by: Tully Foote <tullyfoote@intrinsic.ai>
Co-authored-by: William Woodall <william+github@osrfoundation.org>
(cherry picked from commit 7c1143dc15)
2023-10-04 14:56:06 -07:00
mergify[bot]
6af511f79f Add missing header required by the rclcpp::NodeOptions type (#2324) (#2325)
Signed-off-by: Ignacio Vizzo <ignacio@dexory.com>
(cherry picked from commit d6bd8baac5)

Co-authored-by: Ignacio Vizzo <ignaciovizzo@gmail.com>
2023-10-04 11:15:10 -04:00
mergify[bot]
fbb78ec975 Fix C++20 allocator construct deprecation (#2292) (#2318)
Signed-off-by: Guilherme Rodrigues <guilherme.rodrigues@ait.ac.at>
(cherry picked from commit fa732b9ee8)

Co-authored-by: AiVerisimilitude <133206333+AiVerisimilitude@users.noreply.github.com>
2023-09-27 20:39:53 -04:00
mergify[bot]
b82da1ade4 Topic correct typeadapter deduction (#2294) (#2298)
* fix TypeAdapter deduction

Signed-off-by: Chen Lihui <lihui.chen@sony.com>
(cherry picked from commit 5e152d77d8)

Co-authored-by: Chen Lihui <lihui.chen@sony.com>
2023-09-14 17:19:04 -07:00
Yadunund
45df3555d2 21.0.3 2023-09-08 13:44:12 +08:00
Yadunund
3fa5bc49d6 Update changelogs
Signed-off-by: Yadunund <yadunund@openrobotics.org>
2023-09-08 13:43:56 +08:00
mergify[bot]
5dcd5a39a4 Do not crash Executor when send_response fails due to client failure. (#2276) (#2279)
* Do not crash Executor when send_response fails due to client failure.

Related to https://github.com/ros2/ros2/issues/1253

It is not sane that a faulty client can crash our service Executor, as
discussed in the referred issue, if the client is not setup properly,
send_response may return RCL_RET_TIMEOUT, we should not throw an error
in this case.

Signed-off-by: Zang MingJie <zealot0630@gmail.com>

* Update rclcpp/include/rclcpp/service.hpp

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
Signed-off-by: Zang MingJie <zealot0630@gmail.com>

* address review comments.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

---------

Signed-off-by: Zang MingJie <zealot0630@gmail.com>
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
Co-authored-by: Zang MingJie <zealot0630@gmail.com>
(cherry picked from commit fbe8f28cd1)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2023-09-01 08:23:43 -07:00
mergify[bot]
7b7531bfd6 Switch lifecycle to use the RCLCPP macros. (backport #2233) (#2244)
* Switch lifecycle to use the RCLCPP macros. (#2233)

This ensures that they'll go out to /rosout and the disk.

Signed-off-by: Chris Lalancette <clalancette@gmail.com>
(cherry picked from commit 945d254e32)
2023-07-24 11:54:32 -04:00
Emerson Knapp
6d4a99f815 [Iron] Add new node interface TypeDescriptionsInterface to provide GetTypeDescription service (backport #2224) (#2236)
* Add new node interface TypeDescriptionsInterface to provide GetTypeDescription service (#2224)

* TypeDescriptions interface with readonly param configuration

* Add parameter descriptor, to make read only

* example of spinning in thread for get_type_description service

* Add a basic test for the new interface

* Fix tests with new parameter

* Add comments about builtin parameters

* Hide new member for ABI stability

* Add typedescription interface smoke test

Signed-off-by: Emerson Knapp <emerson.b.knapp@gmail.com>
Signed-off-by: William Woodall <william@osrfoundation.org>
2023-07-18 08:24:54 -04:00
Yadunund
052c075052 21.0.2 2023-07-14 01:21:56 +08:00
Yadunund
193c252036 Update Changelogs
Signed-off-by: Yadunund <yadunund@openrobotics.org>
2023-07-14 01:21:40 +08:00
mergify[bot]
efbb9b6c89 warning: comparison of integer expressions of different signedness (#2219) (#2222)
https://github.com/ros2/rclcpp/pull/2167#issuecomment-1597197552

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
(cherry picked from commit fe2e0e4c64)

Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2023-06-23 13:00:42 -04:00
mergify[bot]
3506dd1227 Fix race condition in events-executor (#2177) (#2191)
The initial implementation of the events-executor contained a bug where the executor
would end up in an inconsistent state and stop processing interrupt/shutdown notifications.
Manually adding a node to the executor results in a) producing a notify waitable event
and b) refreshing the executor collections.
The inconsistent state would happen if the event was processed before the collections were
finished to be refreshed: the executor would pick up the event but be unable to process it.
This would leave the `notify_waitable_event_pushed_` flag to true, preventing additional
notify waitable events to be pushed.
The behavior is observable only under heavy load.

Signed-off-by: Alberto Soragna <alberto.soragna@gmail.com>
(cherry picked from commit 6e1fea14e1)

Co-authored-by: Alberto Soragna <alberto.soragna@gmail.com>
2023-05-31 08:15:45 -05:00
Chris Lalancette
7a837496bd 21.0.1 2023-05-11 14:00:42 +00:00
Chris Lalancette
cd784f6612 Changelog.
Signed-off-by: Chris Lalancette <clalancette@gmail.com>
2023-05-11 14:00:36 +00:00
mergify[bot]
7aa390d5b1 Fix delivered message kind (#2175) (#2178)
Signed-off-by: methylDragon <methylDragon@gmail.com>
(cherry picked from commit 6ca1023ef7)

Co-authored-by: methylDragon <methylDragon@gmail.com>
2023-05-02 11:21:27 -04:00
Chris Lalancette
a431256383 21.0.0 2023-04-18 15:35:47 +00:00
Chris Lalancette
9d2849cb0a Changelog.
Signed-off-by: Chris Lalancette <clalancette@gmail.com>
2023-04-18 15:35:41 +00:00
Lei Liu
3610b68348 Add support for logging service. (#2122)
* Add support for logging service.

* Update to not modify interfaces and not change time_source

* Use unique_ptr for NodeBuiltinExecutorImpl

* Use user thread to run logger service

* Update code for lifecycle_node

Signed-off-by: Barry Xu <barry.xu@sony.com>
Signed-off-by: Lei Liu <Lei.Liu.AP@sony.com>
2023-04-18 11:30:00 -04:00
Michael Carroll
9c03a463c1 Picking ABI-incompatible executor changes (#2170)
* Picking ABI-incompatible executor changes

* Add PIMPL

Signed-off-by: Michael Carroll <mjcarroll@intrinsic.ai>
2023-04-18 11:29:30 -04:00
Alberto Soragna
a7048e115d add events-executor and timers-manager in rclcpp (#2155)
* add events-executor and timers-manager in rclcpp

fix check for execute_timers_separate_thread; ignore on_ready_callback if asked to execute a timer;  reduce usage of void pointers

* have the executor notify waitable fiddle with guard condition callbacks only if necessary

Signed-off-by: Alberto Soragna <alberto.soragna@gmail.com>
2023-04-17 14:33:58 -04:00
Michael Carroll
e3d9d819af Create common structures for executors to use (#2143)
* Deprecate callback_group call taking context

* Add base executor objects that can be used by implementors

* Template common operations

* Add callback to EntitiesCollector constructor
* Make function to check automatically added callback groups take a list

* Make executor own the notify waitable

* Add pending queue to collector, remove from waitable

Also change node's get_guard_condition to return shared_ptr

* Change interrupt guard condition to shared_ptr

Check if guard condition is valid before adding it to the waitable

* Make get_notify_guard_condition follow API tick-tock

* Improve callback group tick-tocking

* Add thread safety annotations and make locks consistent

* Remove the "add_valid_node" API

* Only notify if the trigger condition is valid

* Only trigger if valid and needed

Signed-off-by: Michael Carroll <michael@openrobotics.org>
2023-04-14 14:56:41 -04:00
Michael Babenko
6ffd54d61d Support publishing loaned messages in LifecyclePublisher (#2159)
* Support loaned messages in LifecyclePublisher

Signed-off-by: Michael Babenko <mbabenko@wayve.ai>
2023-04-14 13:15:03 -04:00
methylDragon
fd7a0dc219 Implement deliver message kind (#2168)
* Implement deliver message kind

Signed-off-by: methylDragon <methylDragon@gmail.com>

* Add examples to docstring

Signed-off-by: methylDragon <methylDragon@gmail.com>

---------

Signed-off-by: methylDragon <methylDragon@gmail.com>
2023-04-13 20:08:27 -07:00
Chris Lalancette
eaf6edd6b2 20.0.0 2023-04-13 19:48:07 +00:00
Chris Lalancette
d478525778 Changelog.
Signed-off-by: Chris Lalancette <clalancette@gmail.com>
2023-04-13 19:48:00 +00:00
ymski
82a693e028 applied tracepoints for ring_buffer (#2091)
applied tracepoints for intra_publish
add tracepoints for linking buffer and subscription

Signed-off-by: Kodai Yamasaki <114902604+ymski@users.noreply.github.com>
2023-04-13 15:44:02 -04:00
methylDragon
b8173e28c6 Dynamic Subscription (REP-2011 Subset): Stubs for rclcpp (#2165)
* Implement subscription base changes

* Add stubbed out classes

Signed-off-by: methylDragon <methylDragon@gmail.com>
2023-04-13 10:32:33 -04:00
Emerson Knapp
3088b536cc Add type_hash to cpp TopicEndpointInfo (#2137)
* Add type_hash to cpp TopicEndpointInfo

Signed-off-by: Emerson Knapp <emerson.b.knapp@gmail.com>
2023-04-12 08:57:57 -04:00
Michael Carroll
5f9695afb0 Trigger the intraprocess guard condition with data (#2164)
If the intraprocess buffer still has data after taking, re-trigger the
guard condition to ensure that the executor will continue to service it,
even if incoming publications stop.


Signed-off-by: Michael Carroll <michael@openrobotics.org>
2023-04-11 19:20:21 -05:00
Yadu
71a06985af Minor grammar fix (#2149)
Signed-off-by: Yadunund <yadunund@openrobotics.org>
2023-04-03 09:53:49 -07:00
Christopher Wecht
73d555b402 Fix unnecessary allocations in executor.cpp (#2135)
Signed-off-by: Christopher Wecht <cwecht@mailbox.org>
2023-04-03 09:01:39 -05:00
Tomoya Fujita
a5368e6fe4 add Logger::get_effective_level(). (#2141)
Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
2023-03-29 21:50:17 -07:00
Emerson Knapp
20e9cd17f6 Remove deprecated header (#2139)
Signed-off-by: Emerson Knapp <emerson.b.knapp@gmail.com>
2023-03-26 08:45:36 -04:00
119 changed files with 5613 additions and 1900 deletions

View File

@@ -2,6 +2,86 @@
Changelog for package rclcpp
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
21.0.8 (2024-11-08)
-------------------
* associated clocks should be protected by mutex. (`#2258 <https://github.com/ros2/rclcpp/issues/2258>`_)
* Skip client_qos test (`#2658 <https://github.com/ros2/rclcpp/issues/2658>`_)
* Use the same context for the specified node in rclcpp::spin functions. (`#2618 <https://github.com/ros2/rclcpp/issues/2618>`_)
* Contributors: Cristóbal Arroyo, Tomoya Fujita
21.0.7 (2024-07-10)
-------------------
* Add test creating two content filter topics with the same topic name (`#2550 <https://github.com/ros2/rclcpp/issues/2550>`_)
* Revise the description of service configure_introspection() (`#2514 <https://github.com/ros2/rclcpp/issues/2514>`_)
* Contributors: Alejandro Hernández Cordero, Barry Xu
21.0.6 (2024-04-19)
-------------------
* address ambiguous auto variable. (`#2486 <https://github.com/ros2/rclcpp/issues/2486>`_)
* Contributors: Tomoya Fujita
21.0.5 (2024-02-07)
-------------------
* Fix data race in EventHandlerBase (`#2387 <https://github.com/ros2/rclcpp/issues/2387>`_)
* Contributors: mauropasse
21.0.4 (2023-11-17)
-------------------
* Disable the loaned messages inside the executor. (`#2365 <https://github.com/ros2/rclcpp/issues/2365>`_)
* Add missing 'enable_rosout' comments (`#2346 <https://github.com/ros2/rclcpp/issues/2346>`_)
* Address rate related flaky tests. (`#2341 <https://github.com/ros2/rclcpp/issues/2341>`_)
* Add missing stdexcept include (`#2333 <https://github.com/ros2/rclcpp/issues/2333>`_)
* Update SignalHandler get_global_signal_handler to avoid complex types in static memory (`#2322 <https://github.com/ros2/rclcpp/issues/2322>`_)
* Fix C++20 allocator construct deprecation (`#2318 <https://github.com/ros2/rclcpp/issues/2318>`_)
* Topic correct typeadapter deduction (`#2298 <https://github.com/ros2/rclcpp/issues/2298>`_)
* Contributors: AiVerisimilitude, Chen Lihui, Chris Lalancette, Jiaqi Li, Øystein Sture, Tomoya Fujita, William Woodall
21.0.3 (2023-09-08)
-------------------
* Do not crash Executor when send_response fails due to client failure. (`#2279 <https://github.com/ros2/rclcpp/issues/2279>`_)
* Add new node interface TypeDescriptionsInterface to provide GetTypeDescription service (`#2236 <https://github.com/ros2/rclcpp/issues/2236>`_)
* Contributors: Emerson Knapp, Tomoya Fujita, Zang MingJie
21.0.2 (2023-07-14)
-------------------
* Fix warnings related to comparison of integer expressions of different signedness (`#2222 <https://github.com/ros2/rclcpp/issues/2222>`_)
* Fix race condition in events-executor (`#2191 <https://github.com/ros2/rclcpp/issues/2191>`_)
* Contributors: Alberto Soragna, Tomoya Fujita
21.0.1 (2023-05-11)
-------------------
* Fix delivered message kind (`#2175 <https://github.com/ros2/rclcpp/issues/2175>`_) (`#2178 <https://github.com/ros2/rclcpp/issues/2178>`_)
* Contributors: mergify[bot]
21.0.0 (2023-04-18)
-------------------
* Add support for logging service. (`#2122 <https://github.com/ros2/rclcpp/issues/2122>`_)
* Picking ABI-incompatible executor changes (`#2170 <https://github.com/ros2/rclcpp/issues/2170>`_)
* add events-executor and timers-manager in rclcpp (`#2155 <https://github.com/ros2/rclcpp/issues/2155>`_)
* Create common structures for executors to use (`#2143 <https://github.com/ros2/rclcpp/issues/2143>`_)
* Implement deliver message kind (`#2168 <https://github.com/ros2/rclcpp/issues/2168>`_)
* Contributors: Alberto Soragna, Lei Liu, Michael Carroll, methylDragon
20.0.0 (2023-04-13)
-------------------
* applied tracepoints for ring_buffer (`#2091 <https://github.com/ros2/rclcpp/issues/2091>`_)
* Dynamic Subscription (REP-2011 Subset): Stubs for rclcpp (`#2165 <https://github.com/ros2/rclcpp/issues/2165>`_)
* Add type_hash to cpp TopicEndpointInfo (`#2137 <https://github.com/ros2/rclcpp/issues/2137>`_)
* Trigger the intraprocess guard condition with data (`#2164 <https://github.com/ros2/rclcpp/issues/2164>`_)
* Minor grammar fix (`#2149 <https://github.com/ros2/rclcpp/issues/2149>`_)
* Fix unnecessary allocations in executor.cpp (`#2135 <https://github.com/ros2/rclcpp/issues/2135>`_)
* add Logger::get_effective_level(). (`#2141 <https://github.com/ros2/rclcpp/issues/2141>`_)
* Remove deprecated header (`#2139 <https://github.com/ros2/rclcpp/issues/2139>`_)
* Implement matched event (`#2105 <https://github.com/ros2/rclcpp/issues/2105>`_)
* use allocator via init_options argument. (`#2129 <https://github.com/ros2/rclcpp/issues/2129>`_)
* Fixes to silence some clang warnings. (`#2127 <https://github.com/ros2/rclcpp/issues/2127>`_)
* Documentation improvements on the executor (`#2125 <https://github.com/ros2/rclcpp/issues/2125>`_)
* Avoid losing waitable handles while using MultiThreadedExecutor (`#2109 <https://github.com/ros2/rclcpp/issues/2109>`_)
* Hook up the incompatible type event inside of rclcpp (`#2069 <https://github.com/ros2/rclcpp/issues/2069>`_)
* Update all rclcpp packages to C++17. (`#2121 <https://github.com/ros2/rclcpp/issues/2121>`_)
* Fix clang warning: bugprone-use-after-move (`#2116 <https://github.com/ros2/rclcpp/issues/2116>`_)
* Contributors: Barry Xu, Chris Lalancette, Christopher Wecht, Emerson Knapp, Michael Carroll, Tomoya Fujita, Yadu, mauropasse, methylDragon, ymski
19.3.0 (2023-03-01)
-------------------
* Fix memory leak in tracetools::get_symbol() (`#2104 <https://github.com/ros2/rclcpp/issues/2104>`_)

View File

@@ -49,19 +49,26 @@ set(${PROJECT_NAME}_SRCS
src/rclcpp/detail/rmw_implementation_specific_subscription_payload.cpp
src/rclcpp/detail/utilities.cpp
src/rclcpp/duration.cpp
src/rclcpp/dynamic_typesupport/dynamic_message.cpp
src/rclcpp/dynamic_typesupport/dynamic_message_type.cpp
src/rclcpp/dynamic_typesupport/dynamic_message_type_builder.cpp
src/rclcpp/dynamic_typesupport/dynamic_message_type_support.cpp
src/rclcpp/dynamic_typesupport/dynamic_serialization_support.cpp
src/rclcpp/event.cpp
src/rclcpp/exceptions/exceptions.cpp
src/rclcpp/experimental/executors/events_executor/events_executor_entities_collector.cpp
src/rclcpp/experimental/executors/events_executor/events_executor.cpp
src/rclcpp/experimental/timers_manager.cpp
src/rclcpp/executable_list.cpp
src/rclcpp/executor.cpp
src/rclcpp/executors.cpp
src/rclcpp/executors/executor_entities_collection.cpp
src/rclcpp/executors/executor_entities_collector.cpp
src/rclcpp/executors/executor_notify_waitable.cpp
src/rclcpp/executors/multi_threaded_executor.cpp
src/rclcpp/executors/single_threaded_executor.cpp
src/rclcpp/executors/static_executor_entities_collector.cpp
src/rclcpp/executors/static_single_threaded_executor.cpp
src/rclcpp/expand_topic_or_service_name.cpp
src/rclcpp/experimental/executors/events_executor/events_executor.cpp
src/rclcpp/experimental/timers_manager.cpp
src/rclcpp/future_return_code.cpp
src/rclcpp/generic_publisher.cpp
src/rclcpp/generic_subscription.cpp
@@ -85,6 +92,7 @@ set(${PROJECT_NAME}_SRCS
src/rclcpp/node_interfaces/node_time_source.cpp
src/rclcpp/node_interfaces/node_timers.cpp
src/rclcpp/node_interfaces/node_topics.cpp
src/rclcpp/node_interfaces/node_type_descriptions.cpp
src/rclcpp/node_interfaces/node_waitables.cpp
src/rclcpp/node_options.cpp
src/rclcpp/parameter.cpp

View File

@@ -93,11 +93,54 @@ public:
* determines whether a callback group is automatically added to an executor
* with the node with which it is associated.
*/
[[deprecated("Use CallbackGroup constructor with context function argument")]]
RCLCPP_PUBLIC
explicit CallbackGroup(
CallbackGroupType group_type,
bool automatically_add_to_executor_with_node = true);
/// Constructor for CallbackGroup.
/**
* Callback Groups have a type, either 'Mutually Exclusive' or 'Reentrant'
* and when creating one the type must be specified.
*
* Callbacks in Reentrant Callback Groups must be able to:
* - run at the same time as themselves (reentrant)
* - run at the same time as other callbacks in their group
* - run at the same time as other callbacks in other groups
*
* Callbacks in Mutually Exclusive Callback Groups:
* - will not be run multiple times simultaneously (non-reentrant)
* - will not be run at the same time as other callbacks in their group
* - but must run at the same time as callbacks in other groups
*
* Additionally, callback groups have a property which determines whether or
* not they are added to an executor with their associated node automatically.
* When creating a callback group the automatically_add_to_executor_with_node
* argument determines this behavior, and if true it will cause the newly
* created callback group to be added to an executor with the node when the
* Executor::add_node method is used.
* If false, this callback group will not be added automatically and would
* have to be added to an executor manually using the
* Executor::add_callback_group method.
*
* Whether the node was added to the executor before creating the callback
* group, or after, is irrelevant; the callback group will be automatically
* added to the executor in either case.
*
* \param[in] group_type The type of the callback group.
* \param[in] get_node_context Lambda to retrieve the node context when
* checking that the creating node is valid and using the guard condition.
* \param[in] automatically_add_to_executor_with_node A boolean that
* determines whether a callback group is automatically added to an executor
* with the node with which it is associated.
*/
RCLCPP_PUBLIC
explicit CallbackGroup(
CallbackGroupType group_type,
std::function<rclcpp::Context::SharedPtr(void)> get_node_context,
bool automatically_add_to_executor_with_node = true);
/// Default destructor.
RCLCPP_PUBLIC
~CallbackGroup();
@@ -137,6 +180,13 @@ public:
return _find_ptrs_if_impl<rclcpp::Waitable, Function>(func, waitable_ptrs_);
}
/// Get the total number of entities in this callback group.
/**
* \return the number of entities in the callback group.
*/
RCLCPP_PUBLIC
size_t size() const;
RCLCPP_PUBLIC
std::atomic_bool &
can_be_taken_from();
@@ -178,11 +228,24 @@ public:
bool
automatically_add_to_executor_with_node() const;
/// Defer creating the notify guard condition and return it.
/// Retrieve the guard condition used to signal changes to this callback group.
/**
* \param[in] context_ptr context to use when creating the guard condition
* \return guard condition if it is valid, otherwise nullptr.
*/
[[deprecated("Use get_notify_guard_condition() without arguments")]]
RCLCPP_PUBLIC
rclcpp::GuardCondition::SharedPtr
get_notify_guard_condition(const rclcpp::Context::SharedPtr context_ptr);
/// Retrieve the guard condition used to signal changes to this callback group.
/**
* \return guard condition if it is valid, otherwise nullptr.
*/
RCLCPP_PUBLIC
rclcpp::GuardCondition::SharedPtr
get_notify_guard_condition();
/// Trigger the notify guard condition.
RCLCPP_PUBLIC
void
@@ -234,6 +297,8 @@ protected:
std::shared_ptr<rclcpp::GuardCondition> notify_guard_condition_ = nullptr;
std::recursive_mutex notify_guard_condition_mutex_;
std::function<rclcpp::Context::SharedPtr(void)> get_context_;
private:
template<typename TypeT, typename Function>
typename TypeT::SharedPtr _find_ptrs_if_impl(

View File

@@ -845,7 +845,7 @@ protected:
"Received invalid sequence number. Ignoring...");
return std::nullopt;
}
auto value = std::move(it->second.second);
std::optional<CallbackInfoVariant> value = std::move(it->second.second);
this->pending_requests_.erase(request_number);
return value;
}

View File

@@ -26,6 +26,7 @@
#include <unordered_set>
#include <utility>
#include <vector>
#include <stdexcept>
#include "rcl/context.h"
#include "rcl/guard_condition.h"

View File

@@ -0,0 +1,70 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_HPP_
#define RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_HPP_
#include <rcl/allocator.h>
#include <rcl/types.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <memory>
#include <string>
#include "rclcpp/dynamic_typesupport/dynamic_message_type.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"
namespace rclcpp
{
namespace dynamic_typesupport
{
/// Utility wrapper class for rosidl_dynamic_typesupport_dynamic_data_t
/// STUBBED OUT
class DynamicMessage : public std::enable_shared_from_this<DynamicMessage>
{
public:
RCLCPP_SMART_PTR_ALIASES_ONLY(DynamicMessage)
RCLCPP_PUBLIC
virtual ~DynamicMessage();
protected:
// NOTE(methylDragon):
// This is just here to extend the lifetime of the serialization support
// It isn't actually used by the builder since the builder should compose its own support
//
// ... Though ideally it should be the exact same support as the one stored in the
// DynamicSerializationSupport
DynamicSerializationSupport::SharedPtr serialization_support_;
rosidl_dynamic_typesupport_dynamic_data_t rosidl_dynamic_data_;
bool is_loaned_;
// Used for returning the loaned value, and lifetime management
DynamicMessage::SharedPtr parent_data_;
private:
RCLCPP_DISABLE_COPY(DynamicMessage)
RCLCPP_PUBLIC
DynamicMessage();
};
} // namespace dynamic_typesupport
} // namespace rclcpp
#endif // RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_HPP_

View File

@@ -0,0 +1,64 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_HPP_
#define RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_HPP_
#include <rcl/allocator.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <memory>
#include <string>
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"
namespace rclcpp
{
namespace dynamic_typesupport
{
/// Utility wrapper class for `rosidl_dynamic_typesupport_dynamic_type_t`
/// STUBBED OUT
class DynamicMessageType : public std::enable_shared_from_this<DynamicMessageType>
{
public:
RCLCPP_SMART_PTR_ALIASES_ONLY(DynamicMessageType)
RCLCPP_PUBLIC
virtual ~DynamicMessageType();
protected:
// NOTE(methylDragon):
// This is just here to extend the lifetime of the serialization support
// It isn't actually used by the builder since the builder should compose its own support
//
// ... Though ideally it should be the exact same support as the one stored in the
// `DynamicSerializationSupport`
DynamicSerializationSupport::SharedPtr serialization_support_;
rosidl_dynamic_typesupport_dynamic_type_t rosidl_dynamic_type_;
private:
RCLCPP_DISABLE_COPY(DynamicMessageType)
RCLCPP_PUBLIC
DynamicMessageType();
};
} // namespace dynamic_typesupport
} // namespace rclcpp
#endif // RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_HPP_

View File

@@ -0,0 +1,65 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_BUILDER_HPP_
#define RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_BUILDER_HPP_
#include <rcl/allocator.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <memory>
#include <string>
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"
namespace rclcpp
{
namespace dynamic_typesupport
{
/// Utility wrapper class for `rosidl_dynamic_typesupport_dynamic_type_builder_t *`
/// STUBBED OUT
class DynamicMessageTypeBuilder : public std::enable_shared_from_this<DynamicMessageTypeBuilder>
{
public:
RCLCPP_SMART_PTR_ALIASES_ONLY(DynamicMessageTypeBuilder)
RCLCPP_PUBLIC
virtual ~DynamicMessageTypeBuilder();
protected:
// NOTE(methylDragon):
// This is just here to extend the lifetime of the serialization support
// It isn't actually used by the builder since the builder should compose its own support
//
// ... Though ideally it should be the exact same support as the one stored in the
// `DynamicSerializationSupport`
DynamicSerializationSupport::SharedPtr serialization_support_;
rosidl_dynamic_typesupport_dynamic_type_builder_t rosidl_dynamic_type_builder_;
private:
RCLCPP_DISABLE_COPY(DynamicMessageTypeBuilder)
RCLCPP_PUBLIC
DynamicMessageTypeBuilder();
};
} // namespace dynamic_typesupport
} // namespace rclcpp
#endif // RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_BUILDER_HPP_

View File

@@ -0,0 +1,67 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_SUPPORT_HPP_
#define RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_SUPPORT_HPP_
#include <rcl/allocator.h>
#include <rosidl_dynamic_typesupport/dynamic_message_type_support_struct.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <rosidl_runtime_c/message_type_support_struct.h>
#include <rosidl_runtime_c/type_description/type_description__struct.h>
#include <memory>
#include <string>
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"
namespace rclcpp
{
namespace dynamic_typesupport
{
/// Utility wrapper class for `rosidl_message_type_support_t` containing managed
/// STUBBED OUT
class DynamicMessageTypeSupport : public std::enable_shared_from_this<DynamicMessageTypeSupport>
{
public:
RCLCPP_SMART_PTR_ALIASES_ONLY(DynamicMessageTypeSupport)
RCLCPP_PUBLIC
virtual ~DynamicMessageTypeSupport();
protected:
DynamicSerializationSupport::SharedPtr serialization_support_;
DynamicMessageType::SharedPtr dynamic_message_type_;
DynamicMessage::SharedPtr dynamic_message_;
rosidl_message_type_support_t rosidl_message_type_support_;
private:
RCLCPP_DISABLE_COPY(DynamicMessageTypeSupport)
RCLCPP_PUBLIC
DynamicMessageTypeSupport();
};
} // namespace dynamic_typesupport
} // namespace rclcpp
#endif // RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_MESSAGE_TYPE_SUPPORT_HPP_

View File

@@ -0,0 +1,60 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_SERIALIZATION_SUPPORT_HPP_
#define RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_SERIALIZATION_SUPPORT_HPP_
#include <rcl/allocator.h>
#include <rosidl_dynamic_typesupport/api/serialization_support.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <memory>
#include <string>
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"
namespace rclcpp
{
namespace dynamic_typesupport
{
/// Utility wrapper class for rosidl_dynamic_typesupport_serialization_support_t
class DynamicSerializationSupport : public std::enable_shared_from_this<DynamicSerializationSupport>
{
public:
RCLCPP_SMART_PTR_ALIASES_ONLY(DynamicSerializationSupport)
RCLCPP_PUBLIC
explicit DynamicSerializationSupport(rcl_allocator_t allocator = rcl_get_default_allocator());
RCLCPP_PUBLIC
DynamicSerializationSupport(
const std::string & serialization_library_name,
rcl_allocator_t allocator = rcl_get_default_allocator());
RCLCPP_PUBLIC
virtual ~DynamicSerializationSupport();
protected:
rosidl_dynamic_typesupport_serialization_support_t rosidl_serialization_support_;
private:
RCLCPP_DISABLE_COPY(DynamicSerializationSupport)
};
} // namespace dynamic_typesupport
} // namespace rclcpp
#endif // RCLCPP__DYNAMIC_TYPESUPPORT__DYNAMIC_SERIALIZATION_SUPPORT_HPP_

View File

@@ -260,6 +260,16 @@ public:
}
}
~EventHandler()
{
// Since the rmw event listener holds a reference to the
// "on ready" callback, we need to clear it on destruction of this class.
// This clearing is not needed for other rclcpp entities like pub/subs, since
// they do own the underlying rmw entities, which are destroyed
// on their rclcpp destructors, thus no risk of dangling pointers.
clear_on_ready_callback();
}
/// Take data so that the callback cannot be scheduled again
std::shared_ptr<void>
take_data() override

View File

@@ -51,6 +51,7 @@ typedef std::map<rclcpp::CallbackGroup::WeakPtr,
// Forward declaration is used in convenience method signature.
class Node;
class ExecutorImplementation;
/// Coordinate the order and timing of available communication tasks.
/**
@@ -637,8 +638,9 @@ protected:
std::atomic_bool spinning;
/// Guard condition for signaling the rmw layer to wake up for special events.
rclcpp::GuardCondition interrupt_guard_condition_;
std::shared_ptr<rclcpp::GuardCondition> interrupt_guard_condition_;
/// Guard condition for signaling the rmw layer to wake up for system shutdown.
std::shared_ptr<rclcpp::GuardCondition> shutdown_guard_condition_;
/// Wait set for managing entities that the rmw layer waits on.
@@ -696,6 +698,9 @@ protected:
/// shutdown callback handle registered to Context
rclcpp::OnShutdownCallbackHandle shutdown_callback_handle_;
/// Pointer to implementation
std::unique_ptr<ExecutorImplementation> impl_;
};
} // namespace rclcpp

View File

@@ -108,7 +108,9 @@ spin_until_future_complete(
const FutureT & future,
std::chrono::duration<TimeRepT, TimeT> timeout = std::chrono::duration<TimeRepT, TimeT>(-1))
{
rclcpp::executors::SingleThreadedExecutor executor;
rclcpp::ExecutorOptions options;
options.context = node_ptr->get_context();
rclcpp::executors::SingleThreadedExecutor executor(options);
return executors::spin_node_until_future_complete<FutureT>(executor, node_ptr, future, timeout);
}

View File

@@ -0,0 +1,213 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__EXECUTORS__EXECUTOR_ENTITIES_COLLECTION_HPP_
#define RCLCPP__EXECUTORS__EXECUTOR_ENTITIES_COLLECTION_HPP_
#include <deque>
#include <functional>
#include <unordered_map>
#include <vector>
#include <rclcpp/any_executable.hpp>
#include <rclcpp/node_interfaces/node_base.hpp>
#include <rclcpp/callback_group.hpp>
#include <rclcpp/executors/executor_notify_waitable.hpp>
#include <rclcpp/visibility_control.hpp>
#include <rclcpp/wait_result.hpp>
#include <rclcpp/wait_set.hpp>
namespace rclcpp
{
namespace executors
{
/// Structure to represent a single entity's entry in a collection
template<typename EntityValueType>
struct CollectionEntry
{
/// Weak pointer to entity type
using EntityWeakPtr = typename EntityValueType::WeakPtr;
/// Shared pointer to entity type
using EntitySharedPtr = typename EntityValueType::SharedPtr;
/// The entity
EntityWeakPtr entity;
/// If relevant, the entity's corresponding callback_group
rclcpp::CallbackGroup::WeakPtr callback_group;
};
/// Update a collection based on another collection
/*
* Iterates update_from and update_to to see which entities have been added/removed between
* the two collections.
*
* For each new entry (in update_from, but not in update_to),
* add the entity and fire the on_added callback
* For each removed entry (in update_to, but not in update_from),
* remove the entity and fire the on_removed callback.
*
* \param[in] update_from The collection representing the next iteration's state
* \param[inout] update_to The collection representing the current iteration's state
* \param[in] on_added Callback fired when a new entity is detected
* \param[in] on_removed Callback fired when an entity is removed
*/
template<typename CollectionType>
void update_entities(
const CollectionType & update_from,
CollectionType & update_to,
std::function<void(const typename CollectionType::EntitySharedPtr &)> on_added,
std::function<void(const typename CollectionType::EntitySharedPtr &)> on_removed
)
{
for (auto it = update_to.begin(); it != update_to.end(); ) {
if (update_from.count(it->first) == 0) {
auto entity = it->second.entity.lock();
if (entity) {
on_removed(entity);
}
it = update_to.erase(it);
} else {
++it;
}
}
for (auto it = update_from.begin(); it != update_from.end(); ++it) {
if (update_to.count(it->first) == 0) {
auto entity = it->second.entity.lock();
if (entity) {
on_added(entity);
}
update_to.insert(*it);
}
}
}
/// A collection of entities, indexed by their corresponding handles
template<typename EntityKeyType, typename EntityValueType>
class EntityCollection
: public std::unordered_map<const EntityKeyType *, CollectionEntry<EntityValueType>>
{
public:
/// Key type of the map
using Key = const EntityKeyType *;
/// Weak pointer to entity type
using EntityWeakPtr = typename EntityValueType::WeakPtr;
/// Shared pointer to entity type
using EntitySharedPtr = typename EntityValueType::SharedPtr;
/// Update this collection based on the contents of another collection
/**
* Update the internal state of this collection, firing callbacks when entities have been
* added or removed.
*
* \param[in] other Collection to compare to
* \param[in] on_added Callback for when entities have been added
* \param[in] on_removed Callback for when entities have been removed
*/
void update(
const EntityCollection<EntityKeyType, EntityValueType> & other,
std::function<void(const EntitySharedPtr &)> on_added,
std::function<void(const EntitySharedPtr &)> on_removed)
{
update_entities(other, *this, on_added, on_removed);
}
};
/// Represent the total set of entities for a single executor
/**
* This allows the entities to be stored from ExecutorEntitiesCollector.
* The structure also makes in convenient to re-evaluate when entities have been added or removed.
*/
struct ExecutorEntitiesCollection
{
/// Collection type for timer entities
using TimerCollection = EntityCollection<rcl_timer_t, rclcpp::TimerBase>;
/// Collection type for subscription entities
using SubscriptionCollection = EntityCollection<rcl_subscription_t, rclcpp::SubscriptionBase>;
/// Collection type for client entities
using ClientCollection = EntityCollection<rcl_client_t, rclcpp::ClientBase>;
/// Collection type for service entities
using ServiceCollection = EntityCollection<rcl_service_t, rclcpp::ServiceBase>;
/// Collection type for waitable entities
using WaitableCollection = EntityCollection<rclcpp::Waitable, rclcpp::Waitable>;
/// Collection type for guard condition entities
using GuardConditionCollection = EntityCollection<rcl_guard_condition_t, rclcpp::GuardCondition>;
/// Collection of timers currently in use by the executor.
TimerCollection timers;
/// Collection of subscriptions currently in use by the executor.
SubscriptionCollection subscriptions;
/// Collection of clients currently in use by the executor.
ClientCollection clients;
/// Collection of services currently in use by the executor.
ServiceCollection services;
/// Collection of guard conditions currently in use by the executor.
GuardConditionCollection guard_conditions;
/// Collection of waitables currently in use by the executor.
WaitableCollection waitables;
/// Check if the entities collection is empty
/**
* \return true if all member collections are empty, false otherwise
*/
bool empty() const;
/// Clear the entities collection
void clear();
};
/// Build an entities collection from callback groups
/**
* Iterates a list of callback groups and adds entities from each valid group
*
* \param[in] callback_groups List of callback groups to check for entities
* \param[inout] colletion Entities collection to populate with found entities
*/
void
build_entities_collection(
const std::vector<rclcpp::CallbackGroup::WeakPtr> & callback_groups,
ExecutorEntitiesCollection & collection);
/// Build a queue of executables ready to be executed
/**
* Iterates a list of entities and adds them to a queue if they are ready.
*
* \param[in] collection Collection of entities corresponding to the current wait set.
* \param[in] wait_result Result of rclcpp::WaitSet::wait corresponding to the collection.
* \param[inout] queue of executables to append new ready executables to
* \return number of new ready executables
*/
size_t
ready_executables(
const ExecutorEntitiesCollection & collection,
rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
std::deque<rclcpp::AnyExecutable> & executables
);
} // namespace executors
} // namespace rclcpp
#endif // RCLCPP__EXECUTORS__EXECUTOR_ENTITIES_COLLECTION_HPP_

View File

@@ -0,0 +1,270 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__EXECUTORS__EXECUTOR_ENTITIES_COLLECTOR_HPP_
#define RCLCPP__EXECUTORS__EXECUTOR_ENTITIES_COLLECTOR_HPP_
#include <map>
#include <memory>
#include <mutex>
#include <set>
#include <vector>
#include "rcpputils/thread_safety_annotations.hpp"
#include <rclcpp/any_executable.hpp>
#include <rclcpp/node_interfaces/node_base.hpp>
#include <rclcpp/callback_group.hpp>
#include <rclcpp/executors/executor_notify_waitable.hpp>
#include <rclcpp/visibility_control.hpp>
#include <rclcpp/wait_set.hpp>
#include <rclcpp/wait_result.hpp>
namespace rclcpp
{
namespace executors
{
/// Class to monitor a set of nodes and callback groups for changes in entity membership
/**
* This is to be used with an executor to track the membership of various nodes, groups,
* and entities (timers, subscriptions, clients, services, etc) and report status to the
* executor.
*
* In general, users will add either nodes or callback groups to an executor.
* Each node may have callback groups that are automatically associated with executors,
* or callback groups that must be manually associated with an executor.
*
* This object tracks both types of callback groups as well as nodes that have been
* previously added to the executor.
* When a new callback group is added/removed or new entities are added/removed, the
* corresponding node or callback group will signal this to the executor so that the
* entity collection may be rebuilt according to that executor's implementation.
*
*/
class ExecutorEntitiesCollector
{
public:
/// Constructor
/**
* \param[in] notify_waitable Waitable that is used to signal to the executor
* when nodes or callback groups have been added or removed.
*/
RCLCPP_PUBLIC
explicit ExecutorEntitiesCollector(
std::shared_ptr<ExecutorNotifyWaitable> notify_waitable);
/// Destructor
RCLCPP_PUBLIC
~ExecutorEntitiesCollector();
/// Indicate if the entities collector has pending additions or removals.
/**
* \return true if there are pending additions or removals
*/
bool has_pending() const;
/// Add a node to the entity collector
/**
* \param[in] node_ptr a shared pointer that points to a node base interface
* \throw std::runtime_error if the node is associated with an executor
*/
RCLCPP_PUBLIC
void
add_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
/// Remove a node from the entity collector
/**
* \param[in] node_ptr a shared pointer that points to a node base interface
* \throw std::runtime_error if the node is associated with an executor
* \throw std::runtime_error if the node is associated with this executor
*/
RCLCPP_PUBLIC
void
remove_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
/// Add a callback group to the entity collector
/**
* \param[in] group_ptr a shared pointer that points to a callback group
* \throw std::runtime_error if the callback_group is associated with an executor
*/
RCLCPP_PUBLIC
void
add_callback_group(rclcpp::CallbackGroup::SharedPtr group_ptr);
/// Remove a callback group from the entity collector
/**
* \param[in] group_ptr a shared pointer that points to a callback group
* \throw std::runtime_error if the callback_group is not associated with an executor
* \throw std::runtime_error if the callback_group is not associated with this executor
*/
RCLCPP_PUBLIC
void
remove_callback_group(rclcpp::CallbackGroup::SharedPtr group_ptr);
/// Get all callback groups known to this entity collector
/**
* This will include manually added and automatically added (node associated) groups
* \return vector of all callback groups
*/
RCLCPP_PUBLIC
std::vector<rclcpp::CallbackGroup::WeakPtr>
get_all_callback_groups() const;
/// Get manually-added callback groups known to this entity collector
/**
* This will include callback groups that have been added via add_callback_group
* \return vector of manually-added callback groups
*/
RCLCPP_PUBLIC
std::vector<rclcpp::CallbackGroup::WeakPtr>
get_manually_added_callback_groups() const;
/// Get automatically-added callback groups known to this entity collector
/**
* This will include callback groups that are associated with nodes added via add_node
* \return vector of automatically-added callback groups
*/
RCLCPP_PUBLIC
std::vector<rclcpp::CallbackGroup::WeakPtr>
get_automatically_added_callback_groups() const;
/// Update the underlying collections
/**
* This will prune nodes and callback groups that are no longer valid as well
* as add new callback groups from any associated nodes.
*/
RCLCPP_PUBLIC
void
update_collections();
protected:
using NodeCollection = std::set<
rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
std::owner_less<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr>>;
using CallbackGroupCollection = std::set<
rclcpp::CallbackGroup::WeakPtr,
std::owner_less<rclcpp::CallbackGroup::WeakPtr>>;
using WeakNodesToGuardConditionsMap = std::map<
rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
rclcpp::GuardCondition::WeakPtr,
std::owner_less<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr>>;
using WeakGroupsToGuardConditionsMap = std::map<
rclcpp::CallbackGroup::WeakPtr,
rclcpp::GuardCondition::WeakPtr,
std::owner_less<rclcpp::CallbackGroup::WeakPtr>>;
/// Implementation of removing a node from the collector.
/**
* This will disassociate the node from the collector and remove any
* automatically-added callback groups
*
* This takes and returns an iterator so it may be used as:
*
* it = remove_weak_node(it);
*
* \param[in] weak_node iterator to the weak node to be removed
* \return Valid updated iterator in the same collection
*/
RCLCPP_PUBLIC
NodeCollection::iterator
remove_weak_node(NodeCollection::iterator weak_node) RCPPUTILS_TSA_REQUIRES(mutex_);
/// Implementation of removing a callback group from the collector.
/**
* This will disassociate the callback group from the collector
*
* This takes and returns an iterator so it may be used as:
*
* it = remove_weak_callback_group(it);
*
* \param[in] weak_group_it iterator to the weak group to be removed
* \param[in] collection the collection to remove the group from
* (manually or automatically added)
* \return Valid updated iterator in the same collection
*/
RCLCPP_PUBLIC
CallbackGroupCollection::iterator
remove_weak_callback_group(
CallbackGroupCollection::iterator weak_group_it,
CallbackGroupCollection & collection) RCPPUTILS_TSA_REQUIRES(mutex_);
/// Implementation of adding a callback group
/**
* \param[in] group_ptr the group to add
* \param[in] collection the collection to add the group to
*/
RCLCPP_PUBLIC
void
add_callback_group_to_collection(
rclcpp::CallbackGroup::SharedPtr group_ptr,
CallbackGroupCollection & collection) RCPPUTILS_TSA_REQUIRES(mutex_);
/// Iterate over queued added/remove nodes and callback_groups
RCLCPP_PUBLIC
void
process_queues() RCPPUTILS_TSA_REQUIRES(mutex_);
/// Check a collection of nodes and add any new callback_groups that
/// are set to be automatically associated via the node.
RCLCPP_PUBLIC
void
add_automatically_associated_callback_groups(
const NodeCollection & nodes_to_check) RCPPUTILS_TSA_REQUIRES(mutex_);
/// Check all nodes and group for expired weak pointers and remove them.
RCLCPP_PUBLIC
void
prune_invalid_nodes_and_groups() RCPPUTILS_TSA_REQUIRES(mutex_);
/// mutex to protect collections and pending queues
mutable std::mutex mutex_;
/// Callback groups that were added via `add_callback_group`
CallbackGroupCollection manually_added_groups_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// Callback groups that were added by their association with added nodes
CallbackGroupCollection automatically_added_groups_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// nodes that are associated with the executor
NodeCollection weak_nodes_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// Track guard conditions associated with added nodes
WeakNodesToGuardConditionsMap weak_nodes_to_guard_conditions_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// Track guard conditions associated with added callback groups
WeakGroupsToGuardConditionsMap weak_groups_to_guard_conditions_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// nodes that have been added since the last update.
NodeCollection pending_added_nodes_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// nodes that have been removed since the last update.
NodeCollection pending_removed_nodes_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// callback groups that have been added since the last update.
CallbackGroupCollection pending_manually_added_groups_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// callback groups that have been removed since the last update.
CallbackGroupCollection pending_manually_removed_groups_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
/// Waitable to add guard conditions to
std::shared_ptr<ExecutorNotifyWaitable> notify_waitable_;
};
} // namespace executors
} // namespace rclcpp
//
#endif // RCLCPP__EXECUTORS__EXECUTOR_ENTITIES_COLLECTOR_HPP_

View File

@@ -0,0 +1,158 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__EXECUTORS__EXECUTOR_NOTIFY_WAITABLE_HPP_
#define RCLCPP__EXECUTORS__EXECUTOR_NOTIFY_WAITABLE_HPP_
#include <functional>
#include <memory>
#include <mutex>
#include <set>
#include "rclcpp/guard_condition.hpp"
#include "rclcpp/waitable.hpp"
namespace rclcpp
{
namespace executors
{
/// Maintain a collection of guard conditions from associated nodes and callback groups
/// to signal to the executor when associated entities have changed.
class ExecutorNotifyWaitable : public rclcpp::Waitable
{
public:
RCLCPP_SMART_PTR_DEFINITIONS(ExecutorNotifyWaitable)
// Constructor
/**
* \param[in] on_execute_callback Callback to execute when one of the conditions
* of this waitable has signaled the wait_set.
*/
RCLCPP_PUBLIC
explicit ExecutorNotifyWaitable(std::function<void(void)> on_execute_callback = {});
// Destructor
RCLCPP_PUBLIC
~ExecutorNotifyWaitable() override = default;
RCLCPP_PUBLIC
ExecutorNotifyWaitable(const ExecutorNotifyWaitable & other);
RCLCPP_PUBLIC
ExecutorNotifyWaitable & operator=(const ExecutorNotifyWaitable & other);
/// Add conditions to the wait set
/**
* \param[inout] wait_set structure that conditions will be added to
*/
RCLCPP_PUBLIC
void
add_to_wait_set(rcl_wait_set_t * wait_set) override;
/// Check conditions against the wait set
/**
* \param[inout] wait_set structure that internal elements will be checked against.
* \return true if this waitable is ready to be executed, false otherwise.
*/
RCLCPP_PUBLIC
bool
is_ready(rcl_wait_set_t * wait_set) override;
/// Perform work associated with the waitable.
/**
* This will call the callback provided in the constructor.
* \param[in] data Data to be use for the execute, if available, else nullptr.
*/
RCLCPP_PUBLIC
void
execute(std::shared_ptr<void> & data) override;
/// Retrieve data to be used in the next execute call.
/**
* \return If available, data to be used, otherwise nullptr
*/
RCLCPP_PUBLIC
std::shared_ptr<void>
take_data() override;
/// Take the data from an entity ID so that it can be consumed with `execute`.
/**
* \param[in] id ID of the entity to take data from.
* \return If available, data to be used, otherwise nullptr
* \sa rclcpp::Waitable::take_data_by_entity_id
*/
RCLCPP_PUBLIC
std::shared_ptr<void>
take_data_by_entity_id(size_t id) override;
/// Set a callback to be called whenever the waitable becomes ready.
/**
* \param[in] callback callback to set
* \sa rclcpp::Waitable::set_on_ready_callback
*/
RCLCPP_PUBLIC
void
set_on_ready_callback(std::function<void(size_t, int)> callback) override;
/// Add a guard condition to be waited on.
/**
* \param[in] guard_condition The guard condition to add.
*/
RCLCPP_PUBLIC
void
add_guard_condition(rclcpp::GuardCondition::WeakPtr guard_condition);
/// Unset any callback registered via set_on_ready_callback.
/**
* \sa rclcpp::Waitable::clear_on_ready_callback
*/
RCLCPP_PUBLIC
void
clear_on_ready_callback() override;
/// Remove a guard condition from being waited on.
/**
* \param[in] weak_guard_condition The guard condition to remove.
*/
RCLCPP_PUBLIC
void
remove_guard_condition(rclcpp::GuardCondition::WeakPtr weak_guard_condition);
/// Get the number of ready guard_conditions
/**
* \return The number of guard_conditions associated with the Waitable.
*/
RCLCPP_PUBLIC
size_t
get_number_of_ready_guard_conditions() override;
private:
/// Callback to run when waitable executes
std::function<void(void)> execute_callback_;
std::mutex guard_condition_mutex_;
std::function<void(size_t)> on_ready_callback_;
/// The collection of guard conditions to be waited on.
std::set<rclcpp::GuardCondition::WeakPtr,
std::owner_less<rclcpp::GuardCondition::WeakPtr>> notify_guard_conditions_;
};
} // namespace executors
} // namespace rclcpp
#endif // RCLCPP__EXECUTORS__EXECUTOR_NOTIFY_WAITABLE_HPP_

View File

@@ -24,6 +24,7 @@
#include "rclcpp/allocator/allocator_deleter.hpp"
#include "rclcpp/experimental/buffers/buffer_implementation_base.hpp"
#include "rclcpp/macros.hpp"
#include "tracetools/tracetools.h"
namespace rclcpp
{
@@ -94,6 +95,10 @@ public:
buffer_ = std::move(buffer_impl);
TRACEPOINT(
rclcpp_buffer_to_ipb,
static_cast<const void *>(buffer_.get()),
static_cast<const void *>(this));
if (!allocator) {
message_allocator_ = std::make_shared<MessageAlloc>();
} else {

View File

@@ -25,6 +25,7 @@
#include "rclcpp/logging.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"
#include "tracetools/tracetools.h"
namespace rclcpp
{
@@ -51,6 +52,7 @@ public:
if (capacity == 0) {
throw std::invalid_argument("capacity must be a positive, non-zero value");
}
TRACEPOINT(rclcpp_construct_ring_buffer, static_cast<const void *>(this), capacity_);
}
virtual ~RingBufferImplementation() {}
@@ -67,6 +69,12 @@ public:
write_index_ = next_(write_index_);
ring_buffer_[write_index_] = std::move(request);
TRACEPOINT(
rclcpp_ring_buffer_enqueue,
static_cast<const void *>(this),
write_index_,
size_ + 1,
is_full_());
if (is_full_()) {
read_index_ = next_(read_index_);
@@ -90,6 +98,11 @@ public:
}
auto request = std::move(ring_buffer_[read_index_]);
TRACEPOINT(
rclcpp_ring_buffer_dequeue,
static_cast<const void *>(this),
read_index_,
size_ - 1);
read_index_ = next_(read_index_);
size_--;
@@ -135,7 +148,10 @@ public:
return is_full_();
}
void clear() {}
void clear()
{
TRACEPOINT(rclcpp_ring_buffer_clear, static_cast<const void *>(this));
}
private:
/// Get the next index value for the ring buffer

View File

@@ -1,70 +0,0 @@
// Copyright 2023 iRobot Corporation.
//
// 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 RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENT_WAITABLE_HPP_
#define RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENT_WAITABLE_HPP_
#include "rclcpp/waitable.hpp"
namespace rclcpp
{
namespace experimental
{
namespace executors
{
/**
* @brief This class provides a wrapper around the waitable object, that is
* meant to be used with the EventsExecutor.
* The waitset related methods are stubbed out as they should not be called.
* This class is abstract as the execute method of rclcpp::Waitable is not implemented.
* Nodes who want to implement a custom EventWaitable, can derive from this class and override
* the execute method.
*/
class EventWaitable : public rclcpp::Waitable
{
public:
// Constructor
RCLCPP_PUBLIC
EventWaitable() = default;
// Destructor
RCLCPP_PUBLIC
virtual ~EventWaitable() = default;
// Stub API: not used by EventsExecutor
RCLCPP_PUBLIC
bool
is_ready(rcl_wait_set_t * wait_set) final
{
(void)wait_set;
throw std::runtime_error("EventWaitable can't be checked if it's ready");
return false;
}
// Stub API: not used by EventsExecutor
RCLCPP_PUBLIC
void
add_to_wait_set(rcl_wait_set_t * wait_set) final
{
(void)wait_set;
throw std::runtime_error("EventWaitable can't be added to a wait_set");
}
};
} // namespace executors
} // namespace experimental
} // namespace rclcpp
#endif // RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENT_WAITABLE_HPP_

View File

@@ -15,15 +15,15 @@
#ifndef RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_HPP_
#define RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_HPP_
#include <atomic>
#include <chrono>
#include <memory>
#include <queue>
#include <vector>
#include "rclcpp/executor.hpp"
#include "rclcpp/experimental/executors/events_executor/events_executor_entities_collector.hpp"
#include "rclcpp/executors/executor_entities_collection.hpp"
#include "rclcpp/executors/executor_entities_collector.hpp"
#include "rclcpp/experimental/executors/events_executor/events_executor_event_types.hpp"
#include "rclcpp/experimental/executors/events_executor/events_executor_notify_waitable.hpp"
#include "rclcpp/experimental/executors/events_executor/events_queue.hpp"
#include "rclcpp/experimental/executors/events_executor/simple_events_queue.hpp"
#include "rclcpp/experimental/timers_manager.hpp"
@@ -40,7 +40,9 @@ namespace executors
/**
* This executor uses an events queue and a timers manager to execute entities from its
* associated nodes and callback groups.
* The RMW listener APIs are used to collect new events.
* ROS 2 entities allow to set callback functions that are invoked when the entity is triggered
* or has work to do. The events-executor sets these callbacks such that they push an
* event into its queue.
*
* This executor tries to reduce as much as possible the amount of maintenance operations.
* This allows to use customized `EventsQueue` classes to achieve different goals such
@@ -183,6 +185,10 @@ public:
rclcpp::CallbackGroup::SharedPtr group_ptr,
bool notify = true) override;
/// Get callback groups that belong to executor.
/**
* \sa rclcpp::Executor::get_all_callback_groups()
*/
RCLCPP_PUBLIC
std::vector<rclcpp::CallbackGroup::WeakPtr>
get_all_callback_groups() override;
@@ -204,10 +210,12 @@ public:
get_automatically_added_callback_groups_from_nodes() override;
protected:
/// Internal implementation of spin_once
RCLCPP_PUBLIC
void
spin_once_impl(std::chrono::nanoseconds timeout) override;
/// Internal implementation of spin_some
RCLCPP_PUBLIC
void
spin_some_impl(std::chrono::nanoseconds max_duration, bool exhaustive);
@@ -215,18 +223,64 @@ protected:
private:
RCLCPP_DISABLE_COPY(EventsExecutor)
// Execute a single event
RCLCPP_PUBLIC
/// Execute a provided executor event if its associated entities are available
void
execute_event(const ExecutorEvent & event);
// Queue where entities can push events
/// Collect entities from callback groups and refresh the current collection with them
void
refresh_current_collection_from_callback_groups();
/// Refresh the current collection using the provided new_collection
void
refresh_current_collection(const rclcpp::executors::ExecutorEntitiesCollection & new_collection);
/// Create a listener callback function for the provided entity
std::function<void(size_t)>
create_entity_callback(void * entity_key, ExecutorEventType type);
/// Create a listener callback function for the provided waitable entity
std::function<void(size_t, int)>
create_waitable_callback(const rclcpp::Waitable * waitable_id);
/// Utility to add the notify waitable to an entities collection
void
add_notify_waitable_to_collection(
rclcpp::executors::ExecutorEntitiesCollection::WaitableCollection & collection);
/// Searches for the provided entity_id in the collection and returns the entity if valid
template<typename CollectionType>
typename CollectionType::EntitySharedPtr
retrieve_entity(typename CollectionType::Key entity_id, CollectionType & collection)
{
// Check if the entity_id is in the collection
auto it = collection.find(entity_id);
if (it == collection.end()) {
return nullptr;
}
// Check if the entity associated with the entity_id is valid
// and remove it from the collection if it isn't
auto entity = it->second.entity.lock();
if (!entity) {
collection.erase(it);
}
// Return the retrieved entity (this can be a nullptr if the entity was not valid)
return entity;
}
/// Queue where entities can push events
rclcpp::experimental::executors::EventsQueue::UniquePtr events_queue_;
EventsExecutorEntitiesCollector::SharedPtr entities_collector_;
EventsExecutorNotifyWaitable::SharedPtr executor_notifier_;
std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollector> entities_collector_;
std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollection> current_entities_collection_;
std::shared_ptr<rclcpp::executors::ExecutorNotifyWaitable> notify_waitable_;
// Timers manager
/// Flag used to reduce the number of unnecessary waitable events
std::atomic<bool> notify_waitable_event_pushed_ {false};
/// Timers manager used to track and/or execute associated timers
std::shared_ptr<rclcpp::experimental::TimersManager> timers_manager_;
};

View File

@@ -1,348 +0,0 @@
// Copyright 2023 iRobot Corporation.
//
// 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 RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_ENTITIES_COLLECTOR_HPP_
#define RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_ENTITIES_COLLECTOR_HPP_
#include <chrono>
#include <list>
#include <map>
#include <memory>
#include <unordered_map>
#include <vector>
#include "rcl/guard_condition.h"
#include "rcl/wait.h"
#include "rclcpp/callback_group.hpp"
#include "rclcpp/experimental/executors/events_executor/events_executor_event_types.hpp"
#include "rclcpp/experimental/timers_manager.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/node_interfaces/node_base_interface.hpp"
#include "rclcpp/visibility_control.hpp"
#include "rclcpp/waitable.hpp"
namespace rclcpp
{
namespace experimental
{
namespace executors
{
// forward declaration of EventsExecutor to avoid circular dependency
class EventsExecutor;
typedef std::map<rclcpp::CallbackGroup::WeakPtr,
rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
std::owner_less<rclcpp::CallbackGroup::WeakPtr>> WeakCallbackGroupsToNodesMap;
/**
* @brief This class provides a waitable object that is used for managing the
* entities (i.e. nodes and their subscriptions, timers, services, etc)
* added to an EventsExecutor.
* The add/remove node APIs are used when a node is added/removed from
* the associated EventsExecutor and result in setting/unsetting the
* events callbacks and adding timers to the timers manager.
*
* Being this class derived from Waitable, it can be used to wake up an
* executor thread while it's spinning.
* When this occurs, the execute API takes care of handling changes
* in the entities currently used by the executor.
*/
class EventsExecutorEntitiesCollector final
: public rclcpp::Waitable,
public std::enable_shared_from_this<EventsExecutorEntitiesCollector>
{
public:
RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(EventsExecutorEntitiesCollector)
// Constructor
RCLCPP_PUBLIC
EventsExecutorEntitiesCollector(
EventsExecutor * executor);
// Destructor
RCLCPP_PUBLIC
~EventsExecutorEntitiesCollector() override;
// Initialize entities collector
RCLCPP_PUBLIC
void init();
/// Execute the waitable.
RCLCPP_PUBLIC
void
execute(std::shared_ptr<void> & data) override;
/// Function to add_handles_to_wait_set and wait for work and
/**
* block until the wait set is ready or until the timeout has been exceeded.
* \throws std::runtime_error if wait set couldn't be cleared or filled.
* \throws any rcl errors from rcl_wait, \see rclcpp::exceptions::throw_from_rcl_error()
*/
RCLCPP_PUBLIC
void
refresh_wait_set(std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));
/**
* \throws std::runtime_error if it couldn't add guard condition to wait set
*/
RCLCPP_PUBLIC
void
add_to_wait_set(rcl_wait_set_t * wait_set) override;
/// Complete all available queued work without blocking.
/**
* This function checks if after the guard condition was triggered
* (or a spurious wakeup happened) we are really ready to execute
* i.e. re-collect entities
*/
RCLCPP_PUBLIC
bool
is_ready(rcl_wait_set_t * wait_set) override;
RCLCPP_PUBLIC
std::shared_ptr<void>
take_data() override;
RCLCPP_PUBLIC
std::shared_ptr<void>
take_data_by_entity_id(size_t id) override;
/// Add a callback group to an executor.
/**
* \see rclcpp::Executor::add_callback_group
*/
RCLCPP_PUBLIC
bool
add_callback_group(
rclcpp::CallbackGroup::SharedPtr group_ptr,
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
/// Add a callback group to an executor.
/**
* \see rclcpp::Executor::add_callback_group
* \return boolean whether the node from the callback group is new
*/
RCLCPP_PUBLIC
bool
add_callback_group(
rclcpp::CallbackGroup::SharedPtr group_ptr,
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
WeakCallbackGroupsToNodesMap & weak_groups_to_nodes);
/// Remove a callback group from the executor.
/**
* \see rclcpp::Executor::remove_callback_group
*/
RCLCPP_PUBLIC
bool
remove_callback_group(
rclcpp::CallbackGroup::SharedPtr group_ptr);
/// Remove a callback group from the executor.
/**
* \see rclcpp::Executor::remove_callback_group_from_map
*/
RCLCPP_PUBLIC
bool
remove_callback_group_from_map(
rclcpp::CallbackGroup::SharedPtr group_ptr,
WeakCallbackGroupsToNodesMap & weak_groups_to_nodes);
/**
* \see rclcpp::Executor::add_node()
* \throw std::runtime_error if node was already added
*/
RCLCPP_PUBLIC
bool
add_node(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
/**
* \see rclcpp::Executor::remove_node()
* \throw std::runtime_error if no guard condition is associated with node.
*/
RCLCPP_PUBLIC
bool
remove_node(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
RCLCPP_PUBLIC
std::vector<rclcpp::CallbackGroup::WeakPtr>
get_all_callback_groups();
/// Get callback groups that belong to executor.
/**
* \see rclcpp::Executor::get_manually_added_callback_groups()
*/
RCLCPP_PUBLIC
std::vector<rclcpp::CallbackGroup::WeakPtr>
get_manually_added_callback_groups();
/// Get callback groups that belong to executor.
/**
* \see rclcpp::Executor::get_automatically_added_callback_groups_from_nodes()
*/
RCLCPP_PUBLIC
std::vector<rclcpp::CallbackGroup::WeakPtr>
get_automatically_added_callback_groups_from_nodes();
///
/**
* Get the subscription shared pointer corresponding
* to a subscription identifier
*/
RCLCPP_PUBLIC
rclcpp::SubscriptionBase::SharedPtr
get_subscription(const void * subscription_id);
///
/**
* Get the client shared pointer corresponding
* to a client identifier
*/
RCLCPP_PUBLIC
rclcpp::ClientBase::SharedPtr
get_client(const void * client_id);
///
/**
* Get the service shared pointer corresponding
* to a service identifier
*/
RCLCPP_PUBLIC
rclcpp::ServiceBase::SharedPtr
get_service(const void * service_id);
///
/**
* Get the waitable shared pointer corresponding
* to a waitable identifier
*/
RCLCPP_PUBLIC
rclcpp::Waitable::SharedPtr
get_waitable(const void * waitable_id);
///
/**
* Add a weak pointer to a waitable
*/
RCLCPP_PUBLIC
void
add_waitable(rclcpp::Waitable::SharedPtr waitable);
private:
/// Return true if the node belongs to the collector
/**
* \param[in] node_ptr a node base interface shared pointer
* \param[in] weak_groups_to_nodes map to nodes to lookup
* \return boolean whether a node belongs the collector
*/
bool
has_node(
const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes) const;
/// Add all callback groups that can be automatically added by any executor
/// and is not already associated with an executor from nodes
/// that are associated with executor
/**
* \see rclcpp::Executor::add_callback_groups_from_nodes_associated_to_executor()
*/
void
add_callback_groups_from_nodes_associated_to_executor();
void
callback_group_added_impl(
rclcpp::CallbackGroup::SharedPtr group);
void
node_added_impl(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node);
void
callback_group_removed_impl(
rclcpp::CallbackGroup::SharedPtr group);
void
node_removed_impl(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node);
void
set_entities_event_callbacks_from_map(
const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes);
void
set_callback_group_entities_callbacks(rclcpp::CallbackGroup::SharedPtr group);
void
unset_callback_group_entities_callbacks(rclcpp::CallbackGroup::SharedPtr group);
void
set_guard_condition_callback(rclcpp::GuardCondition * guard_condition);
void
unset_guard_condition_callback(rclcpp::GuardCondition * guard_condition);
std::function<void(size_t)>
create_entity_callback(void * exec_entity_id, ExecutorEventType type);
std::function<void(size_t, int)>
create_waitable_callback(void * waitable_id);
typedef std::map<rclcpp::CallbackGroup::WeakPtr,
rclcpp::GuardCondition *,
std::owner_less<rclcpp::CallbackGroup::WeakPtr>>
WeakCallbackGroupsToGuardConditionsMap;
typedef std::map<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
const rclcpp::GuardCondition *,
std::owner_less<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr>>
WeakNodesToGuardConditionsMap;
/// maps callback groups to guard conditions
WeakCallbackGroupsToGuardConditionsMap weak_groups_to_guard_conditions_;
WeakNodesToGuardConditionsMap weak_nodes_to_guard_conditions_;
// maps callback groups to nodes.
WeakCallbackGroupsToNodesMap weak_groups_associated_with_executor_to_nodes_;
// maps callback groups to nodes.
WeakCallbackGroupsToNodesMap weak_groups_to_nodes_associated_with_executor_;
/// List of weak nodes registered in the static executor
std::list<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr> weak_nodes_;
// Maps: entity identifiers to weak pointers from the entities registered in the executor
// so in the case of an event providing and ID, we can retrieve and own the corresponding
// entity while it performs work
std::unordered_map<const void *, rclcpp::SubscriptionBase::WeakPtr> weak_subscriptions_map_;
std::unordered_map<const void *, rclcpp::ServiceBase::WeakPtr> weak_services_map_;
std::unordered_map<const void *, rclcpp::ClientBase::WeakPtr> weak_clients_map_;
std::unordered_map<const void *, rclcpp::Waitable::WeakPtr> weak_waitables_map_;
/// Executor using this entities collector object
EventsExecutor * associated_executor_ = nullptr;
/// Instance of the timers manager used by the associated executor
rclcpp::experimental::TimersManager::SharedPtr timers_manager_;
// Mutex to protect vector of new nodes.
std::recursive_mutex reentrant_mutex_;
};
} // namespace executors
} // namespace experimental
} // namespace rclcpp
#endif // RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_ENTITIES_COLLECTOR_HPP_

View File

@@ -33,8 +33,8 @@ enum ExecutorEventType
struct ExecutorEvent
{
const void * exec_entity_id;
int gen_entity_id;
const void * entity_key;
int waitable_data;
ExecutorEventType type;
size_t num_events;
};

View File

@@ -1,103 +0,0 @@
// Copyright 2023 iRobot Corporation.
//
// 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 RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_NOTIFY_WAITABLE_HPP_
#define RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_NOTIFY_WAITABLE_HPP_
#include <list>
#include <memory>
#include "rcl/guard_condition.h"
#include "rclcpp/experimental/executors/events_executor/event_waitable.hpp"
namespace rclcpp
{
namespace experimental
{
namespace executors
{
/**
* @brief This class provides an EventWaitable that allows to
* wake up an EventsExecutor when a guard condition is notified.
*/
class EventsExecutorNotifyWaitable final : public EventWaitable
{
public:
RCLCPP_SMART_PTR_DEFINITIONS(EventsExecutorNotifyWaitable)
// Constructor
RCLCPP_PUBLIC
EventsExecutorNotifyWaitable() = default;
// Destructor
RCLCPP_PUBLIC
virtual ~EventsExecutorNotifyWaitable() = default;
// The function is a no-op, since we only care of waking up the executor
RCLCPP_PUBLIC
void
execute(std::shared_ptr<void> & data) override
{
(void)data;
}
RCLCPP_PUBLIC
void
add_guard_condition(rclcpp::GuardCondition * guard_condition)
{
notify_guard_conditions_.push_back(guard_condition);
}
RCLCPP_PUBLIC
void
set_on_ready_callback(std::function<void(size_t, int)> callback) override
{
// The second argument of the callback should identify which guard condition
// triggered the event. However it's not relevant here as we only
// care about waking up the executor, so we pass a 0.
auto gc_callback = [callback](size_t count) {
callback(count, 0);
};
for (auto gc : notify_guard_conditions_) {
gc->set_on_trigger_callback(gc_callback);
}
}
RCLCPP_PUBLIC
std::shared_ptr<void>
take_data() override
{
// This waitable doesn't handle any data
return nullptr;
}
RCLCPP_PUBLIC
std::shared_ptr<void>
take_data_by_entity_id(size_t id) override
{
(void) id;
return take_data();
}
private:
std::list<rclcpp::GuardCondition *> notify_guard_conditions_;
};
} // namespace executors
} // namespace experimental
} // namespace rclcpp
#endif // RCLCPP__EXPERIMENTAL__EXECUTORS__EVENTS_EXECUTOR__EVENTS_EXECUTOR_NOTIFY_WAITABLE_HPP_

View File

@@ -121,7 +121,7 @@ public:
private:
// The underlying queue implementation
std::queue<rclcpp::experimental::executors::ExecutorEvent> event_queue_;
// Mutex to protect the insertion/extraction of events in the queue
// Mutex to protect read/write access to the queue
mutable std::mutex mutex_;
// Variable used to notify when an event is added to the queue
std::condition_variable events_queue_cv_;

View File

@@ -481,13 +481,13 @@ private:
"subscription use different allocator types, which is not supported");
}
if constexpr (rclcpp::TypeAdapter<MessageT>::is_specialized::value) {
if constexpr (rclcpp::TypeAdapter<MessageT, ROSMessageType>::is_specialized::value) {
ROSMessageTypeAllocator ros_message_alloc(allocator);
auto ptr = ros_message_alloc.allocate(1);
ros_message_alloc.construct(ptr);
auto ptr = ROSMessageTypeAllocatorTraits::allocate(ros_message_alloc, 1);
ROSMessageTypeAllocatorTraits::construct(ros_message_alloc, ptr);
ROSMessageTypeDeleter deleter;
allocator::set_allocator_for_deleter(&deleter, &allocator);
rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*message, *ptr);
rclcpp::TypeAdapter<MessageT, ROSMessageType>::convert_to_ros_message(*message, *ptr);
auto ros_msg = std::unique_ptr<ROSMessageType, ROSMessageTypeDeleter>(ptr, deleter);
ros_message_subscription->provide_intra_process_message(std::move(ros_msg));
} else {

View File

@@ -118,6 +118,13 @@ public:
return nullptr;
}
}
if (this->buffer_->has_data()) {
// If there is data still to be processed, indicate to the
// executor or waitset by triggering the guard condition.
this->trigger_guard_condition();
}
return std::static_pointer_cast<void>(
std::make_shared<std::pair<ConstMessageSharedPtr, MessageUniquePtr>>(
std::pair<ConstMessageSharedPtr, MessageUniquePtr>(

View File

@@ -31,6 +31,8 @@
#include "rclcpp/qos.hpp"
#include "rclcpp/type_support_decl.hpp"
#include "tracetools/tracetools.h"
namespace rclcpp
{
namespace experimental
@@ -91,6 +93,10 @@ public:
buffer_type,
qos_profile,
std::make_shared<Alloc>(subscribed_type_allocator_));
TRACEPOINT(
rclcpp_ipb_to_subscription,
static_cast<const void *>(buffer_.get()),
static_cast<const void *>(this));
}
bool

View File

@@ -49,10 +49,13 @@ namespace experimental
* Because of this, they have a not-negligible impact on the performance.
*
* Timers execution
* The most efficient implementation consists in letting a TimersManager object
* to spawn a thread where timers are monitored and periodically executed.
* Besides this, other APIs allow to either execute a single timer or all the
* currently ready ones.
* The most efficient use of this class consists in letting a TimersManager object
* to spawn a thread where timers are monitored and optionally executed.
* This can be controlled via the `start` and `stop` methods.
* Ready timers can either be executed or an on_ready_callback can be used to notify
* other entities that they are ready and need to be executed.
* Other APIs allow to directly execute a given timer.
*
* This class assumes that the `execute_callback()` API of the stored timers is never
* called by other entities, but it can only be called from here.
* If this assumption is not respected, the heap property may be invalidated,
@@ -69,13 +72,21 @@ public:
*
* @param context custom context to be used.
* Shared ownership of the context is held until destruction.
* @param on_ready_callback The timers on ready callback. if not callable,
* this object will directly execute timers when they are ready.
* @param on_ready_callback The timers on ready callback. The presence of this function
* indicates what to do when the TimersManager is running and a timer becomes ready.
* The TimersManager is considered "running" when the `start` method has been called.
* If it's callable, it will be invoked instead of the timer callback.
* If it's not callable, then the TimersManager will
* directly execute timers when they are ready.
* All the methods that execute a given timer (e.g. `execute_head_timer`
* or `execute_ready_timer`) without the TimersManager being `running`, i.e.
* without actually explicitly waiting for the timer to become ready, will ignore this
* callback.
*/
RCLCPP_PUBLIC
TimersManager(
std::shared_ptr<rclcpp::Context> context,
std::function<void(void *)> on_ready_callback = nullptr);
std::function<void(const rclcpp::TimerBase *)> on_ready_callback = nullptr);
/**
* @brief Destruct the TimersManager object making sure to stop thread and release memory.
@@ -124,15 +135,6 @@ public:
RCLCPP_PUBLIC
void stop();
/**
* @brief Executes all the timers currently ready when the function was invoked.
* This function will lock all the stored timers throughout its duration.
* This function is thread safe.
* @throws std::runtime_error if the timers thread was already running.
*/
RCLCPP_PUBLIC
void execute_ready_timers();
/**
* @brief Get the number of timers that are currently ready.
* This function is thread safe.
@@ -146,6 +148,8 @@ public:
/**
* @brief Executes head timer if ready.
* This function is thread safe.
* This function will try to execute the timer callback regardless of whether
* the TimersManager on_ready_callback was passed during construction.
*
* @return true if head timer was ready.
* @throws std::runtime_error if the timers thread was already running.
@@ -156,11 +160,13 @@ public:
/**
* @brief Executes timer identified by its ID.
* This function is thread safe.
* This function will try to execute the timer callback regardless of whether
* the TimersManager on_ready_callback was passed during construction.
*
* @param timer_id the timer ID of the timer to execute
*/
RCLCPP_PUBLIC
void execute_ready_timer(const void * timer_id);
void execute_ready_timer(const rclcpp::TimerBase * timer_id);
/**
* @brief Get the amount of time before the next timer triggers.
@@ -233,7 +239,12 @@ public:
return removed;
}
TimerPtr get_timer(const void * timer_id)
/**
* @brief Retrieve the timer identified by the key
* @param timer_id The ID of the timer to retrieve.
* @return TimerPtr if there's a timer associated with the ID, nullptr otherwise
*/
TimerPtr get_timer(const rclcpp::TimerBase * timer_id)
{
for (auto & weak_timer : weak_heap_) {
auto timer = weak_timer.lock();
@@ -455,7 +466,10 @@ public:
std::make_heap(owned_heap_.begin(), owned_heap_.end(), timer_greater);
}
void clear_callbacks()
/**
* @brief Helper function to clear the "on_reset_callback" on all associated timers.
*/
void clear_timers_on_reset_callbacks()
{
for (TimerPtr & t : owned_heap_) {
t->clear_on_reset_callback();
@@ -481,7 +495,7 @@ private:
*/
static bool timer_greater(TimerPtr a, TimerPtr b)
{
// FIXME!
// TODO(alsora): this can cause an error if timers are using different clocks
return a->time_until_trigger() > b->time_until_trigger();
}
@@ -513,7 +527,7 @@ private:
void execute_ready_timers_unsafe();
// Callback to be called when timer is ready
std::function<void(void *)> on_ready_callback_ = nullptr;
std::function<void(const rclcpp::TimerBase *)> on_ready_callback_ = nullptr;
// Thread used to run the timers execution task
std::thread timers_thread_;

View File

@@ -84,7 +84,7 @@ public:
options.to_rcl_subscription_options(qos),
options.event_callbacks,
options.use_default_callbacks,
true),
DeliveredMessageKind::SERIALIZED_MESSAGE),
callback_(callback),
ts_lib_(ts_lib)
{}
@@ -123,6 +123,31 @@ public:
RCLCPP_PUBLIC
void return_serialized_message(std::shared_ptr<rclcpp::SerializedMessage> & message) override;
// DYNAMIC TYPE ==================================================================================
RCLCPP_PUBLIC
rclcpp::dynamic_typesupport::DynamicMessageType::SharedPtr get_shared_dynamic_message_type()
override;
RCLCPP_PUBLIC
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr get_shared_dynamic_message() override;
RCLCPP_PUBLIC
rclcpp::dynamic_typesupport::DynamicSerializationSupport::SharedPtr
get_shared_dynamic_serialization_support() override;
RCLCPP_PUBLIC
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr create_dynamic_message() override;
RCLCPP_PUBLIC
void return_dynamic_message(
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message) override;
RCLCPP_PUBLIC
void handle_dynamic_message(
const rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message,
const rclcpp::MessageInfo & message_info) override;
private:
RCLCPP_DISABLE_COPY(GenericSubscription)

View File

@@ -170,6 +170,24 @@ public:
RCLCPP_PUBLIC
void
set_level(Level level);
/// Get effective level for current logger.
/**
* The effective level is determined as the severity level of
* the logger if it is set, otherwise it is the first specified severity
* level of the logger's ancestors, starting with its closest ancestor.
* The ancestor hierarchy is signified by logger names being separated by dots:
* a logger named `x` is an ancestor of `x.y`, and both `x` and `x.y` are
* ancestors of `x.y.z`, etc.
* If the level has not been set for the logger nor any of its
* ancestors, the default level is used.
*
* \throws rclcpp::exceptions::RCLError if any error happens.
* \return Level for the current logger.
*/
RCLCPP_PUBLIC
Level
get_effective_level() const;
};
} // namespace rclcpp

View File

@@ -56,6 +56,7 @@
#include "rclcpp/node_interfaces/node_time_source_interface.hpp"
#include "rclcpp/node_interfaces/node_timers_interface.hpp"
#include "rclcpp/node_interfaces/node_topics_interface.hpp"
#include "rclcpp/node_interfaces/node_type_descriptions_interface.hpp"
#include "rclcpp/node_interfaces/node_waitables_interface.hpp"
#include "rclcpp/node_options.hpp"
#include "rclcpp/parameter.hpp"
@@ -1454,6 +1455,11 @@ public:
rclcpp::node_interfaces::NodeTimeSourceInterface::SharedPtr
get_node_time_source_interface();
/// Return the Node's internal NodeTypeDescriptionsInterface implementation.
RCLCPP_PUBLIC
rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
get_node_type_descriptions_interface();
/// Return the sub-namespace, if this is a sub-node, otherwise an empty string.
/**
* The returned sub-namespace is either the accumulated sub-namespaces which
@@ -1591,6 +1597,11 @@ private:
const rclcpp::NodeOptions node_options_;
const std::string sub_namespace_;
const std::string effective_namespace_;
/// Static map(s) containing extra member variables for Node without changing its ABI.
// See node.cpp for more details.
class BackportMembers;
static BackportMembers backport_members_;
};
} // namespace rclcpp

View File

@@ -121,10 +121,19 @@ public:
std::atomic_bool &
get_associated_with_executor_atomic() override;
[[deprecated("Use get_shared_notify_guard_condition or trigger_notify_guard_condition instead")]]
RCLCPP_PUBLIC
rclcpp::GuardCondition &
get_notify_guard_condition() override;
RCLCPP_PUBLIC
rclcpp::GuardCondition::SharedPtr
get_shared_notify_guard_condition() override;
RCLCPP_PUBLIC
void
trigger_notify_guard_condition() override;
RCLCPP_PUBLIC
bool
get_use_intra_process_default() const override;
@@ -153,7 +162,7 @@ private:
/// Guard condition for notifying the Executor of changes to this node.
mutable std::recursive_mutex notify_guard_condition_mutex_;
rclcpp::GuardCondition notify_guard_condition_;
std::shared_ptr<rclcpp::GuardCondition> notify_guard_condition_;
bool notify_guard_condition_is_valid_;
};

View File

@@ -148,13 +148,33 @@ public:
/**
* For example, this should be notified when a publisher is added or removed.
*
* \return the GuardCondition if it is valid, else thow runtime error
* \return the GuardCondition if it is valid, else throw runtime error
*/
RCLCPP_PUBLIC
virtual
rclcpp::GuardCondition &
get_notify_guard_condition() = 0;
/// Return a guard condition that should be notified when the internal node state changes.
/**
* For example, this should be notified when a publisher is added or removed.
*
* \return the GuardCondition if it is valid, else nullptr
*/
RCLCPP_PUBLIC
virtual
rclcpp::GuardCondition::SharedPtr
get_shared_notify_guard_condition() = 0;
/// Trigger the guard condition that notifies of internal node state changes.
/**
* For example, this should be notified when a publisher is added or removed.
*/
RCLCPP_PUBLIC
virtual
void
trigger_notify_guard_condition() = 0;
/// Return the default preference for using intra process communication.
RCLCPP_PUBLIC
virtual

View File

@@ -57,7 +57,8 @@ public:
node_namespace_(info.node_namespace),
topic_type_(info.topic_type),
endpoint_type_(static_cast<rclcpp::EndpointType>(info.endpoint_type)),
qos_profile_({info.qos_profile.history, info.qos_profile.depth}, info.qos_profile)
qos_profile_({info.qos_profile.history, info.qos_profile.depth}, info.qos_profile),
topic_type_hash_(info.topic_type_hash)
{
std::copy(info.endpoint_gid, info.endpoint_gid + RMW_GID_STORAGE_SIZE, endpoint_gid_.begin());
}
@@ -122,6 +123,16 @@ public:
const rclcpp::QoS &
qos_profile() const;
/// Get a mutable reference to the type hash of the topic endpoint.
RCLCPP_PUBLIC
rosidl_type_hash_t &
topic_type_hash();
/// Get a const reference to the type hash of the topic endpoint.
RCLCPP_PUBLIC
const rosidl_type_hash_t &
topic_type_hash() const;
private:
std::string node_name_;
std::string node_namespace_;
@@ -129,6 +140,7 @@ private:
rclcpp::EndpointType endpoint_type_;
std::array<uint8_t, RMW_GID_STORAGE_SIZE> endpoint_gid_;
rclcpp::QoS qos_profile_;
rosidl_type_hash_t topic_type_hash_;
};
namespace node_interfaces

View File

@@ -30,6 +30,7 @@
rclcpp::node_interfaces::NodeTimeSourceInterface, \
rclcpp::node_interfaces::NodeTimersInterface, \
rclcpp::node_interfaces::NodeTopicsInterface, \
rclcpp::node_interfaces::NodeTypeDescriptionsInterface, \
rclcpp::node_interfaces::NodeWaitablesInterface
@@ -118,6 +119,7 @@ public:
* - rclcpp::node_interfaces::NodeTimeSourceInterface
* - rclcpp::node_interfaces::NodeTimersInterface
* - rclcpp::node_interfaces::NodeTopicsInterface
* - rclcpp::node_interfaces::NodeTypeDescriptionsInterface
* - rclcpp::node_interfaces::NodeWaitablesInterface
*
* Or you use custom interfaces as long as you make a template specialization

View File

@@ -23,6 +23,9 @@
#include "rclcpp/node_interfaces/node_logging_interface.hpp"
#include "rclcpp/visibility_control.hpp"
#include "rcl_interfaces/srv/get_logger_levels.hpp"
#include "rcl_interfaces/srv/set_logger_levels.hpp"
namespace rclcpp
{
namespace node_interfaces
@@ -35,7 +38,7 @@ public:
RCLCPP_SMART_PTR_ALIASES_ONLY(NodeLoggingInterface)
RCLCPP_PUBLIC
explicit NodeLogging(rclcpp::node_interfaces::NodeBaseInterface * node_base);
explicit NodeLogging(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base);
RCLCPP_PUBLIC
virtual
@@ -49,13 +52,21 @@ public:
const char *
get_logger_name() const override;
RCLCPP_PUBLIC
void
create_logger_services(
node_interfaces::NodeServicesInterface::SharedPtr node_services) override;
private:
RCLCPP_DISABLE_COPY(NodeLogging)
/// Handle to the NodeBaseInterface given in the constructor.
rclcpp::node_interfaces::NodeBaseInterface * node_base_;
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_;
rclcpp::Logger logger_;
rclcpp::Service<rcl_interfaces::srv::GetLoggerLevels>::SharedPtr get_loggers_service_;
rclcpp::Service<rcl_interfaces::srv::SetLoggerLevels>::SharedPtr set_loggers_service_;
};
} // namespace node_interfaces

View File

@@ -19,6 +19,7 @@
#include "rclcpp/logger.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/node_interfaces/node_services_interface.hpp"
#include "rclcpp/node_interfaces/detail/node_interfaces_helpers.hpp"
#include "rclcpp/visibility_control.hpp"
@@ -54,6 +55,13 @@ public:
virtual
const char *
get_logger_name() const = 0;
/// create logger services
RCLCPP_PUBLIC
virtual
void
create_logger_services(
node_interfaces::NodeServicesInterface::SharedPtr node_services) = 0;
};
} // namespace node_interfaces

View File

@@ -0,0 +1,63 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_
#define RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_
#include <memory>
#include "rclcpp/macros.hpp"
#include "rclcpp/node_interfaces/node_base_interface.hpp"
#include "rclcpp/node_interfaces/node_logging_interface.hpp"
#include "rclcpp/node_interfaces/node_parameters_interface.hpp"
#include "rclcpp/node_interfaces/node_services_interface.hpp"
#include "rclcpp/node_interfaces/node_topics_interface.hpp"
#include "rclcpp/node_interfaces/node_type_descriptions_interface.hpp"
#include "rclcpp/visibility_control.hpp"
namespace rclcpp
{
namespace node_interfaces
{
/// Implementation of the NodeTypeDescriptions part of the Node API.
class NodeTypeDescriptions : public NodeTypeDescriptionsInterface
{
public:
RCLCPP_SMART_PTR_ALIASES_ONLY(NodeTypeDescriptions)
RCLCPP_PUBLIC
explicit NodeTypeDescriptions(
NodeBaseInterface::SharedPtr node_base,
NodeLoggingInterface::SharedPtr node_logging,
NodeParametersInterface::SharedPtr node_parameters,
NodeServicesInterface::SharedPtr node_services);
RCLCPP_PUBLIC
virtual
~NodeTypeDescriptions();
private:
RCLCPP_DISABLE_COPY(NodeTypeDescriptions)
// Pimpl hides helper types and functions used for wrapping a C service, which would be
// awkward to expose in this header.
class NodeTypeDescriptionsImpl;
std::unique_ptr<NodeTypeDescriptionsImpl> impl_;
};
} // namespace node_interfaces
} // namespace rclcpp
#endif // RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_

View File

@@ -0,0 +1,44 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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 RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_
#define RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_
#include "rclcpp/macros.hpp"
#include "rclcpp/node_interfaces/detail/node_interfaces_helpers.hpp"
#include "rclcpp/visibility_control.hpp"
namespace rclcpp
{
namespace node_interfaces
{
/// Pure virtual interface class for the NodeTypeDescriptions part of the Node API.
class NodeTypeDescriptionsInterface
{
public:
RCLCPP_SMART_PTR_ALIASES_ONLY(NodeTypeDescriptionsInterface)
RCLCPP_PUBLIC
virtual
~NodeTypeDescriptionsInterface() = default;
};
} // namespace node_interfaces
} // namespace rclcpp
RCLCPP_NODE_INTERFACE_HELPERS_SUPPORT(
rclcpp::node_interfaces::NodeTypeDescriptionsInterface, type_descriptions)
#endif // RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_

View File

@@ -43,6 +43,7 @@ public:
* - arguments = {}
* - parameter_overrides = {}
* - use_global_arguments = true
* - enable_rosout = true
* - use_intra_process_comms = false
* - enable_topic_statistics = false
* - start_parameter_services = true
@@ -50,6 +51,7 @@ public:
* - clock_type = RCL_ROS_TIME
* - clock_qos = rclcpp::ClockQoS()
* - use_clock_thread = true
* - enable_logger_service = false
* - rosout_qos = rclcpp::RosoutQoS()
* - parameter_event_qos = rclcpp::ParameterEventQoS
* - with history setting and depth from rmw_qos_profile_parameter_events
@@ -232,6 +234,24 @@ public:
NodeOptions &
start_parameter_services(bool start_parameter_services);
/// Return the enable_logger_service flag.
RCLCPP_PUBLIC
bool
enable_logger_service() const;
/// Set the enable_logger_service flag, return this for logger idiom.
/**
* If true, ROS services are created to allow external nodes to get
* and set logger levels of this node.
*
* If false, loggers will still be configured and set logger levels locally,
* but logger levels cannot be changed remotely .
*
*/
RCLCPP_PUBLIC
NodeOptions &
enable_logger_service(bool enable_log_service);
/// Return the start_parameter_event_publisher flag.
RCLCPP_PUBLIC
bool
@@ -421,6 +441,8 @@ private:
bool use_clock_thread_ {true};
bool enable_logger_service_ {false};
rclcpp::QoS parameter_event_qos_ = rclcpp::ParameterEventsQoS(
rclcpp::QoSInitialization::from_rmw(rmw_qos_profile_parameter_events)
);

View File

@@ -484,6 +484,10 @@ protected:
if (!msg) {
throw std::runtime_error("cannot publish msg which is a null pointer");
}
TRACEPOINT(
rclcpp_intra_publish,
static_cast<const void *>(publisher_handle_.get()),
msg.get());
ipm->template do_intra_process_publish<PublishedType, ROSMessageType, AllocatorT>(
intra_process_publisher_id_,
@@ -502,6 +506,10 @@ protected:
if (!msg) {
throw std::runtime_error("cannot publish msg which is a null pointer");
}
TRACEPOINT(
rclcpp_intra_publish,
static_cast<const void *>(publisher_handle_.get()),
msg.get());
ipm->template do_intra_process_publish<ROSMessageType, ROSMessageType, AllocatorT>(
intra_process_publisher_id_,
@@ -521,6 +529,10 @@ protected:
if (!msg) {
throw std::runtime_error("cannot publish msg which is a null pointer");
}
TRACEPOINT(
rclcpp_intra_publish,
static_cast<const void *>(publisher_handle_.get()),
msg.get());
return ipm->template do_intra_process_publish_and_return_shared<ROSMessageType, ROSMessageType,
AllocatorT>(

View File

@@ -117,6 +117,18 @@
* - Allocator related items:
* - rclcpp/allocator/allocator_common.hpp
* - rclcpp/allocator/allocator_deleter.hpp
* - Dynamic typesupport wrappers
* - rclcpp::dynamic_typesupport::DynamicMessage
* - rclcpp::dynamic_typesupport::DynamicMessageType
* - rclcpp::dynamic_typesupport::DynamicMessageTypeBuilder
* - rclcpp::dynamic_typesupport::DynamicSerializationSupport
* - rclcpp/dynamic_typesupport/dynamic_message.hpp
* - rclcpp/dynamic_typesupport/dynamic_message_type.hpp
* - rclcpp/dynamic_typesupport/dynamic_message_type_builder.hpp
* - rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp
* - Dynamic typesupport
* - rclcpp::dynamic_typesupport::DynamicMessageTypeSupport
* - rclcpp/dynamic_typesupport/dynamic_message_type_support.hpp
* - Generic publisher
* - rclcpp::Node::create_generic_publisher()
* - rclcpp::GenericPublisher

View File

@@ -1,56 +0,0 @@
// Copyright 2015 Open Source Robotics Foundation, Inc.
//
// 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.
// Based on: http://the-witness.net/news/2012/11/scopeexit-in-c11/
// But I changed the lambda to include by reference rather than value, see:
// http://the-witness.net/news/2012/11/scopeexit-in-c11/comment-page-1/#comment-86873
#ifndef RCLCPP__SCOPE_EXIT_HPP_
#define RCLCPP__SCOPE_EXIT_HPP_
// TODO(christophebedard) remove this header completely in I-turtle
#warning rclcpp/scope_exit.hpp has been deprecated, please use rcpputils/scope_exit.hpp instead
#include <functional>
#include "rclcpp/macros.hpp"
namespace rclcpp
{
template<typename Callable>
struct ScopeExit
{
explicit ScopeExit(Callable callable)
: callable_(callable) {}
~ScopeExit() {callable_();}
private:
Callable callable_;
};
template<typename Callable>
ScopeExit<Callable>
make_scope_exit(Callable callable)
{
return ScopeExit<Callable>(callable);
}
} // namespace rclcpp
#define RCLCPP_SCOPE_EXIT(code) \
auto RCLCPP_STRING_JOIN(scope_exit_, __LINE__) = rclcpp::make_scope_exit([&]() {code;})
#endif // RCLCPP__SCOPE_EXIT_HPP_

View File

@@ -482,12 +482,20 @@ public:
{
rcl_ret_t ret = rcl_send_response(get_service_handle().get(), &req_id, &response);
if (ret == RCL_RET_TIMEOUT) {
RCLCPP_WARN(
node_logger_.get_child("rclcpp"),
"failed to send response to %s (timeout): %s",
this->get_service_name(), rcl_get_error_string().str);
rcl_reset_error();
return;
}
if (ret != RCL_RET_OK) {
rclcpp::exceptions::throw_from_rcl_error(ret, "failed to send response");
}
}
/// Configure client introspection.
/// Configure service introspection.
/**
* \param[in] clock clock to use to generate introspection timestamps
* \param[in] qos_service_event_pub QoS settings to use when creating the introspection publisher

View File

@@ -144,7 +144,7 @@ public:
// NOTE(methylDragon): Passing these args separately is necessary for event binding
options.event_callbacks,
options.use_default_callbacks,
callback.is_serialized_message_callback()),
callback.is_serialized_message_callback() ? DeliveredMessageKind::SERIALIZED_MESSAGE : DeliveredMessageKind::ROS_MESSAGE), // NOLINT
any_callback_(callback),
options_(options),
message_memory_strategy_(message_memory_strategy)
@@ -388,6 +388,57 @@ public:
return any_callback_.use_take_shared_method();
}
// DYNAMIC TYPE ==================================================================================
// TODO(methylDragon): Reorder later
// TODO(methylDragon): Implement later...
rclcpp::dynamic_typesupport::DynamicMessageType::SharedPtr
get_shared_dynamic_message_type() override
{
throw rclcpp::exceptions::UnimplementedError(
"get_shared_dynamic_message_type is not implemented for Subscription");
}
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr
get_shared_dynamic_message() override
{
throw rclcpp::exceptions::UnimplementedError(
"get_shared_dynamic_message is not implemented for Subscription");
}
rclcpp::dynamic_typesupport::DynamicSerializationSupport::SharedPtr
get_shared_dynamic_serialization_support() override
{
throw rclcpp::exceptions::UnimplementedError(
"get_shared_dynamic_serialization_support is not implemented for Subscription");
}
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr
create_dynamic_message() override
{
throw rclcpp::exceptions::UnimplementedError(
"create_dynamic_message is not implemented for Subscription");
}
void
return_dynamic_message(
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message) override
{
(void) message;
throw rclcpp::exceptions::UnimplementedError(
"return_dynamic_message is not implemented for Subscription");
}
void
handle_dynamic_message(
const rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message,
const rclcpp::MessageInfo & message_info) override
{
(void) message;
(void) message_info;
throw rclcpp::exceptions::UnimplementedError(
"handle_dynamic_message is not implemented for Subscription");
}
private:
RCLCPP_DISABLE_COPY(Subscription)

View File

@@ -31,6 +31,9 @@
#include "rclcpp/any_subscription_callback.hpp"
#include "rclcpp/detail/cpp_callback_trampoline.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/experimental/intra_process_manager.hpp"
#include "rclcpp/experimental/subscription_intra_process_base.hpp"
#include "rclcpp/macros.hpp"
@@ -60,6 +63,27 @@ namespace experimental
class IntraProcessManager;
} // namespace experimental
/// The kind of message that the subscription delivers in its callback, used by the executor
/**
* This enum needs to exist because the callback handle is not accessible to the executor's scope.
*
* "Kind" is used since what is being delivered is a category of messages, for example, there are
* different ROS message types that can be delivered, but they're all ROS messages.
*
* As a concrete example, all of the following callbacks will be considered ROS_MESSAGE for
* DeliveredMessageKind:
* - void callback(const std_msgs::msg::String &)
* - void callback(const std::string &) // type adaption
* - void callback(std::unique_ptr<std_msgs::msg::String>)
*/
enum class DeliveredMessageKind : uint8_t
{
INVALID = 0,
ROS_MESSAGE = 1, // The subscription delivers a ROS message to its callback
SERIALIZED_MESSAGE = 2, // The subscription delivers a serialized message to its callback
DYNAMIC_MESSAGE = 3, // The subscription delivers a dynamic message to its callback
};
/// Virtual base class for subscriptions. This pattern allows us to iterate over different template
/// specializations of Subscription, among other things.
class SubscriptionBase : public std::enable_shared_from_this<SubscriptionBase>
@@ -76,7 +100,8 @@ public:
* \param[in] type_support_handle rosidl type support struct, for the Message type of the topic.
* \param[in] topic_name Name of the topic to subscribe to.
* \param[in] subscription_options Options for the subscription.
* \param[in] is_serialized is true if the message will be delivered still serialized
* \param[in] delivered_message_kind Enum flag to change how the message will be received and
* delivered
*/
RCLCPP_PUBLIC
SubscriptionBase(
@@ -86,7 +111,7 @@ public:
const rcl_subscription_options_t & subscription_options,
const SubscriptionEventCallbacks & event_callbacks,
bool use_default_callbacks,
bool is_serialized = false);
DeliveredMessageKind delivered_message_kind = DeliveredMessageKind::ROS_MESSAGE);
/// Destructor.
RCLCPP_PUBLIC
@@ -235,6 +260,14 @@ public:
bool
is_serialized() const;
/// Return the delivered message kind.
/**
* \return `DeliveredMessageKind`, which adjusts how messages are received and delivered.
*/
RCLCPP_PUBLIC
DeliveredMessageKind
get_delivered_message_kind() const;
/// Get matching publisher count.
/** \return The number of publishers on this topic. */
RCLCPP_PUBLIC
@@ -535,6 +568,49 @@ public:
rclcpp::ContentFilterOptions
get_content_filter() const;
// DYNAMIC TYPE ==================================================================================
// TODO(methylDragon): Reorder later
RCLCPP_PUBLIC
virtual
rclcpp::dynamic_typesupport::DynamicMessageType::SharedPtr
get_shared_dynamic_message_type() = 0;
RCLCPP_PUBLIC
virtual
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr
get_shared_dynamic_message() = 0;
RCLCPP_PUBLIC
virtual
rclcpp::dynamic_typesupport::DynamicSerializationSupport::SharedPtr
get_shared_dynamic_serialization_support() = 0;
/// Borrow a new serialized message (this clones!)
/** \return Shared pointer to a rclcpp::dynamic_typesupport::DynamicMessage. */
RCLCPP_PUBLIC
virtual
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr
create_dynamic_message() = 0;
RCLCPP_PUBLIC
virtual
void
return_dynamic_message(rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message) = 0;
RCLCPP_PUBLIC
virtual
void
handle_dynamic_message(
const rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message,
const rclcpp::MessageInfo & message_info) = 0;
RCLCPP_PUBLIC
bool
take_dynamic_message(
rclcpp::dynamic_typesupport::DynamicMessage & message_out,
rclcpp::MessageInfo & message_info_out);
// ===============================================================================================
protected:
template<typename EventCallbackT>
void
@@ -587,7 +663,7 @@ private:
RCLCPP_DISABLE_COPY(SubscriptionBase)
rosidl_message_type_support_t type_support_;
bool is_serialized_;
DeliveredMessageKind delivered_message_kind_;
std::atomic<bool> subscription_in_use_by_wait_set_{false};
std::atomic<bool> intra_process_subscription_waitable_in_use_by_wait_set_{false};

View File

@@ -2,7 +2,7 @@
<?xml-model href="http://download.ros.org/schema/package_format2.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="2">
<name>rclcpp</name>
<version>19.3.0</version>
<version>21.0.8</version>
<description>The ROS client library in C++.</description>
<maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>
@@ -39,6 +39,7 @@
<depend>rcpputils</depend>
<depend>rcutils</depend>
<depend>rmw</depend>
<depend>rosidl_dynamic_typesupport</depend>
<depend>statistics_msgs</depend>
<depend>tracetools</depend>

View File

@@ -31,10 +31,12 @@ using rclcpp::CallbackGroupType;
CallbackGroup::CallbackGroup(
CallbackGroupType group_type,
std::function<rclcpp::Context::SharedPtr(void)> get_context,
bool automatically_add_to_executor_with_node)
: type_(group_type), associated_with_executor_(false),
can_be_taken_from_(true),
automatically_add_to_executor_with_node_(automatically_add_to_executor_with_node)
automatically_add_to_executor_with_node_(automatically_add_to_executor_with_node),
get_context_(get_context)
{}
CallbackGroup::~CallbackGroup()
@@ -54,6 +56,16 @@ CallbackGroup::type() const
return type_;
}
size_t
CallbackGroup::size() const
{
return
subscription_ptrs_.size() +
service_ptrs_.size() +
client_ptrs_.size() +
timer_ptrs_.size() +
waitable_ptrs_.size();
}
void CallbackGroup::collect_all_ptrs(
std::function<void(const rclcpp::SubscriptionBase::SharedPtr &)> sub_func,
std::function<void(const rclcpp::ServiceBase::SharedPtr &)> service_func,
@@ -111,6 +123,7 @@ CallbackGroup::automatically_add_to_executor_with_node() const
return automatically_add_to_executor_with_node_;
}
// \TODO(mjcarroll) Deprecated, remove on tock
rclcpp::GuardCondition::SharedPtr
CallbackGroup::get_notify_guard_condition(const rclcpp::Context::SharedPtr context_ptr)
{
@@ -129,6 +142,29 @@ CallbackGroup::get_notify_guard_condition(const rclcpp::Context::SharedPtr conte
return notify_guard_condition_;
}
rclcpp::GuardCondition::SharedPtr
CallbackGroup::get_notify_guard_condition()
{
std::lock_guard<std::recursive_mutex> lock(notify_guard_condition_mutex_);
if (!this->get_context_) {
throw std::runtime_error("Callback group was created without context and not passed context");
}
auto context_ptr = this->get_context_();
if (context_ptr && context_ptr->is_valid()) {
if (notify_guard_condition_ && context_ptr != notify_guard_condition_->get_context()) {
if (associated_with_executor_) {
trigger_notify_guard_condition();
}
notify_guard_condition_ = nullptr;
}
if (!notify_guard_condition_) {
notify_guard_condition_ = std::make_shared<rclcpp::GuardCondition>(context_ptr);
}
return notify_guard_condition_;
}
return nullptr;
}
void
CallbackGroup::trigger_notify_guard_condition()
{

View File

@@ -496,7 +496,7 @@ Context::sleep_for(const std::chrono::nanoseconds & nanoseconds)
std::unique_lock<std::mutex> lock(interrupt_mutex_);
auto start = std::chrono::steady_clock::now();
// this will release the lock while waiting
interrupt_condition_variable_.wait_for(lock, nanoseconds);
interrupt_condition_variable_.wait_for(lock, time_left);
time_left -= std::chrono::steady_clock::now() - start;
}
} while (time_left > std::chrono::nanoseconds::zero() && this->is_valid());

View File

@@ -0,0 +1,40 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <rosidl_dynamic_typesupport/api/dynamic_type.h>
#include <rosidl_dynamic_typesupport/api/dynamic_data.h>
#include <rosidl_dynamic_typesupport/api/serialization_support.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <memory>
#include <string>
#include "rcl/allocator.h"
#include "rcl/types.h"
#include "rcutils/logging_macros.h"
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type_builder.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/exceptions.hpp"
using rclcpp::dynamic_typesupport::DynamicMessage;
using rclcpp::dynamic_typesupport::DynamicMessageType;
using rclcpp::dynamic_typesupport::DynamicMessageTypeBuilder;
using rclcpp::dynamic_typesupport::DynamicSerializationSupport;
DynamicMessage::~DynamicMessage()
{} // STUBBED

View File

@@ -0,0 +1,38 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <rosidl_dynamic_typesupport/api/dynamic_data.h>
#include <rosidl_dynamic_typesupport/api/dynamic_type.h>
#include <rosidl_dynamic_typesupport/api/serialization_support.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <memory>
#include <string>
#include "rcutils/logging_macros.h"
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type_builder.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/exceptions.hpp"
using rclcpp::dynamic_typesupport::DynamicMessage;
using rclcpp::dynamic_typesupport::DynamicMessageType;
using rclcpp::dynamic_typesupport::DynamicMessageTypeBuilder;
using rclcpp::dynamic_typesupport::DynamicSerializationSupport;
DynamicMessageType::~DynamicMessageType()
{} // STUBBED

View File

@@ -0,0 +1,37 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <rosidl_dynamic_typesupport/api/dynamic_data.h>
#include <rosidl_dynamic_typesupport/api/dynamic_type.h>
#include <rosidl_dynamic_typesupport/api/serialization_support.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <memory>
#include <string>
#include "rcutils/logging_macros.h"
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type_builder.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/exceptions.hpp"
using rclcpp::dynamic_typesupport::DynamicMessage;
using rclcpp::dynamic_typesupport::DynamicMessageType;
using rclcpp::dynamic_typesupport::DynamicMessageTypeBuilder;
using rclcpp::dynamic_typesupport::DynamicSerializationSupport;
DynamicMessageTypeBuilder::~DynamicMessageTypeBuilder()
{} // STUBBED

View File

@@ -0,0 +1,49 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <rosidl_dynamic_typesupport/identifier.h>
#include <rosidl_dynamic_typesupport/types.h>
#include <rosidl_runtime_c/message_type_support_struct.h>
#include <rosidl_runtime_c/type_description_utils.h>
#include <rosidl_runtime_c/type_description/type_description__functions.h>
#include <rosidl_runtime_c/type_description/type_description__struct.h>
#include <rosidl_runtime_c/type_description/type_source__functions.h>
#include <rosidl_runtime_c/type_description/type_source__struct.h>
#include <memory>
#include <string>
#include "rcl/allocator.h"
#include "rcl/dynamic_message_type_support.h"
#include "rcl/type_hash.h"
#include "rcl/types.h"
#include "rcutils/logging_macros.h"
#include "rcutils/types/rcutils_ret.h"
#include "rmw/dynamic_message_type_support.h"
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message_type_support.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/exceptions.hpp"
#include "rclcpp/macros.hpp"
#include "rclcpp/visibility_control.hpp"
using rclcpp::dynamic_typesupport::DynamicMessage;
using rclcpp::dynamic_typesupport::DynamicMessageType;
using rclcpp::dynamic_typesupport::DynamicMessageTypeSupport;
using rclcpp::dynamic_typesupport::DynamicSerializationSupport;
DynamicMessageTypeSupport::~DynamicMessageTypeSupport()
{} // STUBBED

View File

@@ -0,0 +1,46 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <rcl/allocator.h>
#include <rcutils/logging_macros.h>
#include <rmw/dynamic_message_type_support.h>
#include <rmw/ret_types.h>
#include <rosidl_dynamic_typesupport/api/serialization_support.h>
#include <memory>
#include <string>
#include "rclcpp/dynamic_typesupport/dynamic_serialization_support.hpp"
#include "rclcpp/exceptions.hpp"
using rclcpp::dynamic_typesupport::DynamicSerializationSupport;
// CONSTRUCTION ====================================================================================
DynamicSerializationSupport::DynamicSerializationSupport(rcl_allocator_t allocator)
: DynamicSerializationSupport::DynamicSerializationSupport("", allocator)
{
throw std::runtime_error("Unimplemented");
}
DynamicSerializationSupport::DynamicSerializationSupport(
const std::string & /*serialization_library_name*/,
rcl_allocator_t /*allocator*/)
: rosidl_serialization_support_(
rosidl_dynamic_typesupport_get_zero_initialized_serialization_support())
{
throw std::runtime_error("Unimplemented");
}
DynamicSerializationSupport::~DynamicSerializationSupport()
{} // STUBBED

View File

@@ -39,13 +39,6 @@ UnsupportedEventTypeException::UnsupportedEventTypeException(
EventHandlerBase::~EventHandlerBase()
{
// Since the rmw event listener holds a reference to
// this callback, we need to clear it on destruction of this class.
// This clearing is not needed for other rclcpp entities like pub/subs, since
// they do own the underlying rmw entities, which are destroyed
// on their rclcpp destructors, thus no risk of dangling pointers.
clear_on_ready_callback();
if (rcl_event_fini(&event_handle_) != RCL_RET_OK) {
RCUTILS_LOG_ERROR_NAMED(
"rclcpp",

View File

@@ -24,6 +24,7 @@
#include "rcl/error_handling.h"
#include "rcpputils/scope_exit.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/exceptions.hpp"
#include "rclcpp/executor.hpp"
#include "rclcpp/guard_condition.hpp"
@@ -38,16 +39,16 @@
using namespace std::chrono_literals;
using rclcpp::exceptions::throw_from_rcl_error;
using rclcpp::AnyExecutable;
using rclcpp::Executor;
using rclcpp::ExecutorOptions;
using rclcpp::FutureReturnCode;
class rclcpp::ExecutorImplementation {};
Executor::Executor(const rclcpp::ExecutorOptions & options)
: spinning(false),
interrupt_guard_condition_(options.context),
interrupt_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
shutdown_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
memory_strategy_(options.memory_strategy)
memory_strategy_(options.memory_strategy),
impl_(std::make_unique<rclcpp::ExecutorImplementation>())
{
// Store the context for later use.
context_ = options.context;
@@ -65,7 +66,7 @@ Executor::Executor(const rclcpp::ExecutorOptions & options)
memory_strategy_->add_guard_condition(*shutdown_guard_condition_.get());
// Put the executor's guard condition in
memory_strategy_->add_guard_condition(interrupt_guard_condition_);
memory_strategy_->add_guard_condition(*interrupt_guard_condition_.get());
rcl_allocator_t allocator = memory_strategy_->get_allocator();
rcl_ret_t ret = rcl_wait_set_init(
@@ -127,7 +128,7 @@ Executor::~Executor()
}
// Remove and release the sigint guard condition
memory_strategy_->remove_guard_condition(shutdown_guard_condition_.get());
memory_strategy_->remove_guard_condition(&interrupt_guard_condition_);
memory_strategy_->remove_guard_condition(interrupt_guard_condition_.get());
// Remove shutdown callback handle registered to Context
if (!context_->remove_on_shutdown_callback(shutdown_callback_handle_)) {
@@ -222,8 +223,7 @@ Executor::add_callback_group_to_map(
weak_groups_to_nodes_.insert(std::make_pair(weak_group_ptr, node_ptr));
if (node_ptr->get_context()->is_valid()) {
auto callback_group_guard_condition =
group_ptr->get_notify_guard_condition(node_ptr->get_context());
auto callback_group_guard_condition = group_ptr->get_notify_guard_condition();
weak_groups_to_guard_conditions_[weak_group_ptr] = callback_group_guard_condition.get();
// Add the callback_group's notify condition to the guard condition handles
memory_strategy_->add_guard_condition(*callback_group_guard_condition);
@@ -232,7 +232,7 @@ Executor::add_callback_group_to_map(
if (notify) {
// Interrupt waiting to handle new node
try {
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(
std::string(
@@ -280,10 +280,10 @@ Executor::add_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_pt
}
});
const auto & gc = node_ptr->get_notify_guard_condition();
weak_nodes_to_guard_conditions_[node_ptr] = &gc;
const auto gc = node_ptr->get_shared_notify_guard_condition();
weak_nodes_to_guard_conditions_[node_ptr] = gc.get();
// Add the node's notify condition to the guard condition handles
memory_strategy_->add_guard_condition(gc);
memory_strategy_->add_guard_condition(*gc);
weak_nodes_.push_back(node_ptr);
}
@@ -320,7 +320,7 @@ Executor::remove_callback_group_from_map(
if (notify) {
try {
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(
std::string(
@@ -388,7 +388,7 @@ Executor::remove_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node
}
}
memory_strategy_->remove_guard_condition(&node_ptr->get_notify_guard_condition());
memory_strategy_->remove_guard_condition(node_ptr->get_shared_notify_guard_condition().get());
weak_nodes_to_guard_conditions_.erase(node_ptr);
std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
@@ -501,7 +501,7 @@ Executor::cancel()
{
spinning.store(false);
try {
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(
std::string("Failed to trigger guard condition in cancel: ") + ex.what());
@@ -550,7 +550,7 @@ Executor::execute_any_executable(AnyExecutable & any_exec)
// Wake the wait, because it may need to be recalculated or work that
// was previously blocked is now available.
try {
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(
std::string(
@@ -558,13 +558,14 @@ Executor::execute_any_executable(AnyExecutable & any_exec)
}
}
template<typename Taker, typename Handler>
static
void
take_and_do_error_handling(
const char * action_description,
const char * topic_or_service_name,
std::function<bool()> take_action,
std::function<void()> handle_action)
Taker take_action,
Handler handle_action)
{
bool taken = false;
try {
@@ -597,70 +598,103 @@ take_and_do_error_handling(
void
Executor::execute_subscription(rclcpp::SubscriptionBase::SharedPtr subscription)
{
using rclcpp::dynamic_typesupport::DynamicMessage;
rclcpp::MessageInfo message_info;
message_info.get_rmw_message_info().from_intra_process = false;
if (subscription->is_serialized()) {
// This is the case where a copy of the serialized message is taken from
// the middleware via inter-process communication.
std::shared_ptr<SerializedMessage> serialized_msg = subscription->create_serialized_message();
take_and_do_error_handling(
"taking a serialized message from topic",
subscription->get_topic_name(),
[&]() {return subscription->take_serialized(*serialized_msg.get(), message_info);},
[&]()
switch (subscription->get_delivered_message_kind()) {
// Deliver ROS message
case rclcpp::DeliveredMessageKind::ROS_MESSAGE:
{
subscription->handle_serialized_message(serialized_msg, message_info);
});
subscription->return_serialized_message(serialized_msg);
} else if (subscription->can_loan_messages()) {
// This is the case where a loaned message is taken from the middleware via
// inter-process communication, given to the user for their callback,
// and then returned.
void * loaned_msg = nullptr;
// TODO(wjwwood): refactor this into methods on subscription when LoanedMessage
// is extened to support subscriptions as well.
take_and_do_error_handling(
"taking a loaned message from topic",
subscription->get_topic_name(),
[&]()
{
rcl_ret_t ret = rcl_take_loaned_message(
subscription->get_subscription_handle().get(),
&loaned_msg,
&message_info.get_rmw_message_info(),
nullptr);
if (RCL_RET_SUBSCRIPTION_TAKE_FAILED == ret) {
return false;
} else if (RCL_RET_OK != ret) {
rclcpp::exceptions::throw_from_rcl_error(ret);
if (subscription->can_loan_messages()) {
// This is the case where a loaned message is taken from the middleware via
// inter-process communication, given to the user for their callback,
// and then returned.
void * loaned_msg = nullptr;
// TODO(wjwwood): refactor this into methods on subscription when LoanedMessage
// is extened to support subscriptions as well.
take_and_do_error_handling(
"taking a loaned message from topic",
subscription->get_topic_name(),
[&]()
{
rcl_ret_t ret = rcl_take_loaned_message(
subscription->get_subscription_handle().get(),
&loaned_msg,
&message_info.get_rmw_message_info(),
nullptr);
if (RCL_RET_SUBSCRIPTION_TAKE_FAILED == ret) {
return false;
} else if (RCL_RET_OK != ret) {
rclcpp::exceptions::throw_from_rcl_error(ret);
}
return true;
},
[&]() {subscription->handle_loaned_message(loaned_msg, message_info);});
if (nullptr != loaned_msg) {
rcl_ret_t ret = rcl_return_loaned_message_from_subscription(
subscription->get_subscription_handle().get(), loaned_msg);
if (RCL_RET_OK != ret) {
RCLCPP_ERROR(
rclcpp::get_logger("rclcpp"),
"rcl_return_loaned_message_from_subscription() failed for subscription on topic "
"'%s': %s",
subscription->get_topic_name(), rcl_get_error_string().str);
}
loaned_msg = nullptr;
}
} else {
// This case is taking a copy of the message data from the middleware via
// inter-process communication.
std::shared_ptr<void> message = subscription->create_message();
take_and_do_error_handling(
"taking a message from topic",
subscription->get_topic_name(),
[&]() {return subscription->take_type_erased(message.get(), message_info);},
[&]() {subscription->handle_message(message, message_info);});
// TODO(clalancette): In the case that the user is using the MessageMemoryPool,
// and they take a shared_ptr reference to the message in the callback, this can
// inadvertently return the message to the pool when the user is still using it.
// This is a bug that needs to be fixed in the pool, and we should probably have
// a custom deleter for the message that actually does the return_message().
subscription->return_message(message);
}
return true;
},
[&]() {subscription->handle_loaned_message(loaned_msg, message_info);});
if (nullptr != loaned_msg) {
rcl_ret_t ret = rcl_return_loaned_message_from_subscription(
subscription->get_subscription_handle().get(),
loaned_msg);
if (RCL_RET_OK != ret) {
RCLCPP_ERROR(
rclcpp::get_logger("rclcpp"),
"rcl_return_loaned_message_from_subscription() failed for subscription on topic '%s': %s",
subscription->get_topic_name(), rcl_get_error_string().str);
break;
}
// Deliver serialized message
case rclcpp::DeliveredMessageKind::SERIALIZED_MESSAGE:
{
// This is the case where a copy of the serialized message is taken from
// the middleware via inter-process communication.
std::shared_ptr<SerializedMessage> serialized_msg =
subscription->create_serialized_message();
take_and_do_error_handling(
"taking a serialized message from topic",
subscription->get_topic_name(),
[&]() {return subscription->take_serialized(*serialized_msg.get(), message_info);},
[&]()
{
subscription->handle_serialized_message(serialized_msg, message_info);
});
subscription->return_serialized_message(serialized_msg);
break;
}
// DYNAMIC SUBSCRIPTION ========================================================================
// Deliver dynamic message
case rclcpp::DeliveredMessageKind::DYNAMIC_MESSAGE:
{
throw std::runtime_error("Unimplemented");
}
default:
{
throw std::runtime_error("Delivered message kind is not supported");
}
loaned_msg = nullptr;
}
} else {
// This case is taking a copy of the message data from the middleware via
// inter-process communication.
std::shared_ptr<void> message = subscription->create_message();
take_and_do_error_handling(
"taking a message from topic",
subscription->get_topic_name(),
[&]() {return subscription->take_type_erased(message.get(), message_info);},
[&]() {subscription->handle_message(message, message_info);});
subscription->return_message(message);
}
return;
}
void

View File

@@ -17,7 +17,9 @@
void
rclcpp::spin_some(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
{
rclcpp::executors::SingleThreadedExecutor exec;
rclcpp::ExecutorOptions options;
options.context = node_ptr->get_context();
rclcpp::executors::SingleThreadedExecutor exec(options);
exec.spin_node_some(node_ptr);
}
@@ -30,7 +32,9 @@ rclcpp::spin_some(rclcpp::Node::SharedPtr node_ptr)
void
rclcpp::spin(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
{
rclcpp::executors::SingleThreadedExecutor exec;
rclcpp::ExecutorOptions options;
options.context = node_ptr->get_context();
rclcpp::executors::SingleThreadedExecutor exec(options);
exec.add_node(node_ptr);
exec.spin();
exec.remove_node(node_ptr);

View File

@@ -0,0 +1,230 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include "rclcpp/executors/executor_entities_collection.hpp"
namespace rclcpp
{
namespace executors
{
bool ExecutorEntitiesCollection::empty() const
{
return
subscriptions.empty() &&
timers.empty() &&
guard_conditions.empty() &&
clients.empty() &&
services.empty() &&
waitables.empty();
}
void ExecutorEntitiesCollection::clear()
{
subscriptions.clear();
timers.clear();
guard_conditions.clear();
clients.clear();
services.clear();
waitables.clear();
}
void
build_entities_collection(
const std::vector<rclcpp::CallbackGroup::WeakPtr> & callback_groups,
ExecutorEntitiesCollection & collection)
{
collection.clear();
for (auto weak_group_ptr : callback_groups) {
auto group_ptr = weak_group_ptr.lock();
if (!group_ptr) {
continue;
}
if (group_ptr->can_be_taken_from().load()) {
group_ptr->collect_all_ptrs(
[&collection, weak_group_ptr](const rclcpp::SubscriptionBase::SharedPtr & subscription) {
collection.subscriptions.insert(
{
subscription->get_subscription_handle().get(),
{subscription, weak_group_ptr}
});
},
[&collection, weak_group_ptr](const rclcpp::ServiceBase::SharedPtr & service) {
collection.services.insert(
{
service->get_service_handle().get(),
{service, weak_group_ptr}
});
},
[&collection, weak_group_ptr](const rclcpp::ClientBase::SharedPtr & client) {
collection.clients.insert(
{
client->get_client_handle().get(),
{client, weak_group_ptr}
});
},
[&collection, weak_group_ptr](const rclcpp::TimerBase::SharedPtr & timer) {
collection.timers.insert(
{
timer->get_timer_handle().get(),
{timer, weak_group_ptr}
});
},
[&collection, weak_group_ptr](const rclcpp::Waitable::SharedPtr & waitable) {
collection.waitables.insert(
{
waitable.get(),
{waitable, weak_group_ptr}
});
}
);
}
}
}
size_t
ready_executables(
const ExecutorEntitiesCollection & collection,
rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
std::deque<rclcpp::AnyExecutable> & executables
)
{
size_t added = 0;
if (wait_result.kind() != rclcpp::WaitResultKind::Ready) {
return added;
}
auto rcl_wait_set = wait_result.get_wait_set().get_rcl_wait_set();
// Cache shared pointers to groups to avoid extra work re-locking them
std::map<rclcpp::CallbackGroup::WeakPtr,
rclcpp::CallbackGroup::SharedPtr,
std::owner_less<rclcpp::CallbackGroup::WeakPtr>> group_map;
auto group_cache = [&group_map](const rclcpp::CallbackGroup::WeakPtr & weak_cbg_ptr)
{
if (group_map.count(weak_cbg_ptr) == 0) {
group_map.insert({weak_cbg_ptr, weak_cbg_ptr.lock()});
}
return group_map.find(weak_cbg_ptr)->second;
};
for (size_t ii = 0; ii < rcl_wait_set.size_of_timers; ++ii) {
if (nullptr == rcl_wait_set.timers[ii]) {continue;}
auto entity_iter = collection.timers.find(rcl_wait_set.timers[ii]);
if (entity_iter != collection.timers.end()) {
auto entity = entity_iter->second.entity.lock();
if (!entity) {
continue;
}
auto group_info = group_cache(entity_iter->second.callback_group);
if (group_info && !group_info->can_be_taken_from().load()) {
continue;
}
if (!entity->call()) {
continue;
}
rclcpp::AnyExecutable exec;
exec.timer = entity;
exec.callback_group = group_info;
executables.push_back(exec);
added++;
}
}
for (size_t ii = 0; ii < rcl_wait_set.size_of_subscriptions; ++ii) {
if (nullptr == rcl_wait_set.subscriptions[ii]) {continue;}
auto entity_iter = collection.subscriptions.find(rcl_wait_set.subscriptions[ii]);
if (entity_iter != collection.subscriptions.end()) {
auto entity = entity_iter->second.entity.lock();
if (!entity) {
continue;
}
auto group_info = group_cache(entity_iter->second.callback_group);
if (group_info && !group_info->can_be_taken_from().load()) {
continue;
}
rclcpp::AnyExecutable exec;
exec.subscription = entity;
exec.callback_group = group_info;
executables.push_back(exec);
added++;
}
}
for (size_t ii = 0; ii < rcl_wait_set.size_of_services; ++ii) {
if (nullptr == rcl_wait_set.services[ii]) {continue;}
auto entity_iter = collection.services.find(rcl_wait_set.services[ii]);
if (entity_iter != collection.services.end()) {
auto entity = entity_iter->second.entity.lock();
if (!entity) {
continue;
}
auto group_info = group_cache(entity_iter->second.callback_group);
if (group_info && !group_info->can_be_taken_from().load()) {
continue;
}
rclcpp::AnyExecutable exec;
exec.service = entity;
exec.callback_group = group_info;
executables.push_back(exec);
added++;
}
}
for (size_t ii = 0; ii < rcl_wait_set.size_of_clients; ++ii) {
if (nullptr == rcl_wait_set.clients[ii]) {continue;}
auto entity_iter = collection.clients.find(rcl_wait_set.clients[ii]);
if (entity_iter != collection.clients.end()) {
auto entity = entity_iter->second.entity.lock();
if (!entity) {
continue;
}
auto group_info = group_cache(entity_iter->second.callback_group);
if (group_info && !group_info->can_be_taken_from().load()) {
continue;
}
rclcpp::AnyExecutable exec;
exec.client = entity;
exec.callback_group = group_info;
executables.push_back(exec);
added++;
}
}
for (auto & [handle, entry] : collection.waitables) {
auto waitable = entry.entity.lock();
if (!waitable) {
continue;
}
if (!waitable->is_ready(&rcl_wait_set)) {
continue;
}
auto group_info = group_cache(entry.callback_group);
if (group_info && !group_info->can_be_taken_from().load()) {
continue;
}
rclcpp::AnyExecutable exec;
exec.waitable = waitable;
exec.callback_group = group_info;
exec.data = waitable->take_data();
executables.push_back(exec);
added++;
}
return added;
}
} // namespace executors
} // namespace rclcpp

View File

@@ -0,0 +1,416 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <set>
#include "rclcpp/executors/executor_entities_collector.hpp"
#include "rclcpp/executors/executor_notify_waitable.hpp"
#include "rclcpp/node_interfaces/node_base_interface.hpp"
namespace rclcpp
{
namespace executors
{
ExecutorEntitiesCollector::ExecutorEntitiesCollector(
std::shared_ptr<ExecutorNotifyWaitable> notify_waitable)
: notify_waitable_(notify_waitable)
{
}
ExecutorEntitiesCollector::~ExecutorEntitiesCollector()
{
for (auto weak_node_it = weak_nodes_.begin(); weak_node_it != weak_nodes_.end(); ) {
weak_node_it = remove_weak_node(weak_node_it);
}
for (auto weak_group_it = automatically_added_groups_.begin();
weak_group_it != automatically_added_groups_.end(); )
{
weak_group_it = remove_weak_callback_group(weak_group_it, automatically_added_groups_);
}
for (auto weak_group_it = manually_added_groups_.begin();
weak_group_it != manually_added_groups_.end(); )
{
weak_group_it = remove_weak_callback_group(weak_group_it, manually_added_groups_);
}
for (auto weak_node_ptr : pending_added_nodes_) {
auto node_ptr = weak_node_ptr.lock();
if (node_ptr) {
node_ptr->get_associated_with_executor_atomic().store(false);
}
}
pending_added_nodes_.clear();
pending_removed_nodes_.clear();
for (auto weak_group_ptr : pending_manually_added_groups_) {
auto group_ptr = weak_group_ptr.lock();
if (group_ptr) {
group_ptr->get_associated_with_executor_atomic().store(false);
}
}
pending_manually_added_groups_.clear();
pending_manually_removed_groups_.clear();
}
bool
ExecutorEntitiesCollector::has_pending() const
{
std::lock_guard<std::mutex> lock(mutex_);
return pending_manually_added_groups_.size() != 0 ||
pending_manually_removed_groups_.size() != 0 ||
pending_added_nodes_.size() != 0 ||
pending_removed_nodes_.size() != 0;
}
void
ExecutorEntitiesCollector::add_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
{
// If the node already has an executor
std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
if (has_executor.exchange(true)) {
throw std::runtime_error(
std::string("Node '") + node_ptr->get_fully_qualified_name() +
"' has already been added to an executor.");
}
std::lock_guard<std::mutex> lock(mutex_);
bool associated = weak_nodes_.count(node_ptr) != 0;
bool add_queued = pending_added_nodes_.count(node_ptr) != 0;
bool remove_queued = pending_removed_nodes_.count(node_ptr) != 0;
if ((associated || add_queued) && !remove_queued) {
throw std::runtime_error(
std::string("Node '") + node_ptr->get_fully_qualified_name() +
"' has already been added to this executor.");
}
this->pending_added_nodes_.insert(node_ptr);
}
void
ExecutorEntitiesCollector::remove_node(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
{
if (!node_ptr->get_associated_with_executor_atomic().load()) {
throw std::runtime_error(
std::string("Node '") + node_ptr->get_fully_qualified_name() +
"' needs to be associated with an executor.");
}
std::lock_guard<std::mutex> lock(mutex_);
bool associated = weak_nodes_.count(node_ptr) != 0;
bool add_queued = pending_added_nodes_.count(node_ptr) != 0;
bool remove_queued = pending_removed_nodes_.count(node_ptr) != 0;
if (!(associated || add_queued) || remove_queued) {
throw std::runtime_error(
std::string("Node '") + node_ptr->get_fully_qualified_name() +
"' needs to be associated with this executor.");
}
this->pending_removed_nodes_.insert(node_ptr);
}
void
ExecutorEntitiesCollector::add_callback_group(rclcpp::CallbackGroup::SharedPtr group_ptr)
{
std::atomic_bool & has_executor = group_ptr->get_associated_with_executor_atomic();
if (has_executor.exchange(true)) {
throw std::runtime_error("Callback group has already been added to an executor.");
}
std::lock_guard<std::mutex> lock(mutex_);
bool associated = manually_added_groups_.count(group_ptr) != 0;
bool add_queued = pending_manually_added_groups_.count(group_ptr) != 0;
bool remove_queued = pending_manually_removed_groups_.count(group_ptr) != 0;
if ((associated || add_queued) && !remove_queued) {
throw std::runtime_error("Callback group has already been added to this executor.");
}
this->pending_manually_added_groups_.insert(group_ptr);
}
void
ExecutorEntitiesCollector::remove_callback_group(rclcpp::CallbackGroup::SharedPtr group_ptr)
{
if (!group_ptr->get_associated_with_executor_atomic().load()) {
throw std::runtime_error("Callback group needs to be associated with an executor.");
}
/**
* TODO(mjcarroll): The callback groups, being created by a node, should never outlive
* the node. Since we haven't historically enforced this, turning this on may cause
* previously-functional code to fail.
* Consider re-enablng this check (along with corresponding CallbackGroup::has_valid_node),
* when we can guarantee node/group lifetimes.
if (!group_ptr->has_valid_node()) {
throw std::runtime_error("Node must not be deleted before its callback group(s).");
}
*/
auto weak_group_ptr = rclcpp::CallbackGroup::WeakPtr(group_ptr);
std::lock_guard<std::mutex> lock(mutex_);
bool associated = manually_added_groups_.count(group_ptr) != 0;
bool add_queued = pending_manually_added_groups_.count(group_ptr) != 0;
bool remove_queued = pending_manually_removed_groups_.count(group_ptr) != 0;
if (!(associated || add_queued) || remove_queued) {
throw std::runtime_error("Callback group needs to be associated with this executor.");
}
this->pending_manually_removed_groups_.insert(group_ptr);
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
ExecutorEntitiesCollector::get_all_callback_groups() const
{
std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
std::lock_guard<std::mutex> lock(mutex_);
for (const auto & group_ptr : manually_added_groups_) {
groups.push_back(group_ptr);
}
for (auto const & group_ptr : automatically_added_groups_) {
groups.push_back(group_ptr);
}
return groups;
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
ExecutorEntitiesCollector::get_manually_added_callback_groups() const
{
std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
std::lock_guard<std::mutex> lock(mutex_);
for (const auto & group_ptr : manually_added_groups_) {
groups.push_back(group_ptr);
}
return groups;
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
ExecutorEntitiesCollector::get_automatically_added_callback_groups() const
{
std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
std::lock_guard<std::mutex> lock(mutex_);
for (auto const & group_ptr : automatically_added_groups_) {
groups.push_back(group_ptr);
}
return groups;
}
void
ExecutorEntitiesCollector::update_collections()
{
std::lock_guard<std::mutex> lock(mutex_);
this->process_queues();
this->add_automatically_associated_callback_groups(this->weak_nodes_);
this->prune_invalid_nodes_and_groups();
}
ExecutorEntitiesCollector::NodeCollection::iterator
ExecutorEntitiesCollector::remove_weak_node(NodeCollection::iterator weak_node)
{
// Disassociate the guard condition from the executor notify waitable
auto guard_condition_it = weak_nodes_to_guard_conditions_.find(*weak_node);
if (guard_condition_it != weak_nodes_to_guard_conditions_.end()) {
this->notify_waitable_->remove_guard_condition(guard_condition_it->second);
weak_nodes_to_guard_conditions_.erase(guard_condition_it);
}
// Mark the node as disassociated (if the node is still valid)
auto node_ptr = weak_node->lock();
if (node_ptr) {
std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
has_executor.store(false);
}
// Remove the node from tracked nodes
return weak_nodes_.erase(weak_node);
}
ExecutorEntitiesCollector::CallbackGroupCollection::iterator
ExecutorEntitiesCollector::remove_weak_callback_group(
CallbackGroupCollection::iterator weak_group_it,
CallbackGroupCollection & collection
)
{
// Disassociate the guard condition from the executor notify waitable
auto guard_condition_it = weak_groups_to_guard_conditions_.find(*weak_group_it);
if (guard_condition_it != weak_groups_to_guard_conditions_.end()) {
this->notify_waitable_->remove_guard_condition(guard_condition_it->second);
weak_groups_to_guard_conditions_.erase(guard_condition_it);
}
// Mark the node as disassociated (if the group is still valid)
auto group_ptr = weak_group_it->lock();
if (group_ptr) {
/**
* TODO(mjcarroll): The callback groups, being created by a node, should never outlive
* the node. Since we haven't historically enforced this, turning this on may cause
* previously-functional code to fail.
* Consider re-enablng this check (along with corresponding CallbackGroup::has_valid_node),
* when we can guarantee node/group lifetimes.
if (!group_ptr->has_valid_node()) {
throw std::runtime_error("Node must not be deleted before its callback group(s).");
}
*/
std::atomic_bool & has_executor = group_ptr->get_associated_with_executor_atomic();
has_executor.store(false);
}
// Remove the node from tracked nodes
return collection.erase(weak_group_it);
}
void
ExecutorEntitiesCollector::add_callback_group_to_collection(
rclcpp::CallbackGroup::SharedPtr group_ptr,
CallbackGroupCollection & collection)
{
auto iter = collection.insert(group_ptr);
if (iter.second == false) {
throw std::runtime_error("Callback group has already been added to this executor.");
}
// Store node guard condition in map and add it to the notify waitable
auto group_guard_condition = group_ptr->get_notify_guard_condition();
weak_groups_to_guard_conditions_.insert({group_ptr, group_guard_condition});
this->notify_waitable_->add_guard_condition(group_guard_condition);
}
void
ExecutorEntitiesCollector::process_queues()
{
for (auto weak_node_ptr : pending_added_nodes_) {
auto node_ptr = weak_node_ptr.lock();
if (!node_ptr) {
continue;
}
weak_nodes_.insert(weak_node_ptr);
this->add_automatically_associated_callback_groups({weak_node_ptr});
// Store node guard condition in map and add it to the notify waitable
auto node_guard_condition = node_ptr->get_shared_notify_guard_condition();
weak_nodes_to_guard_conditions_.insert({weak_node_ptr, node_guard_condition});
this->notify_waitable_->add_guard_condition(node_guard_condition);
}
pending_added_nodes_.clear();
for (auto weak_node_ptr : pending_removed_nodes_) {
auto node_it = weak_nodes_.find(weak_node_ptr);
if (node_it != weak_nodes_.end()) {
remove_weak_node(node_it);
} else {
throw std::runtime_error("Node needs to be associated with this executor.");
}
auto node_ptr = weak_node_ptr.lock();
if (node_ptr) {
for (auto group_it = automatically_added_groups_.begin();
group_it != automatically_added_groups_.end(); )
{
auto group_ptr = group_it->lock();
if (node_ptr->callback_group_in_node(group_ptr)) {
group_it = remove_weak_callback_group(group_it, automatically_added_groups_);
} else {
++group_it;
}
}
}
}
pending_removed_nodes_.clear();
for (auto weak_group_ptr : pending_manually_added_groups_) {
auto group_ptr = weak_group_ptr.lock();
if (group_ptr) {
this->add_callback_group_to_collection(group_ptr, manually_added_groups_);
}
}
pending_manually_added_groups_.clear();
for (auto weak_group_ptr : pending_manually_removed_groups_) {
auto group_ptr = weak_group_ptr.lock();
if (group_ptr) {
auto group_it = manually_added_groups_.find(group_ptr);
if (group_it != manually_added_groups_.end()) {
remove_weak_callback_group(group_it, manually_added_groups_);
} else {
throw std::runtime_error(
"Attempting to remove a callback group not added to this executor.");
}
}
}
pending_manually_removed_groups_.clear();
}
void
ExecutorEntitiesCollector::add_automatically_associated_callback_groups(
const NodeCollection & nodes_to_check)
{
for (auto & weak_node : nodes_to_check) {
auto node = weak_node.lock();
if (node) {
node->for_each_callback_group(
[this, node](rclcpp::CallbackGroup::SharedPtr group_ptr)
{
if (!group_ptr->get_associated_with_executor_atomic().load() &&
group_ptr->automatically_add_to_executor_with_node())
{
std::atomic_bool & has_executor = group_ptr->get_associated_with_executor_atomic();
if (has_executor.exchange(true)) {
throw std::runtime_error("Callback group has already been added to an executor.");
}
this->add_callback_group_to_collection(group_ptr, this->automatically_added_groups_);
}
});
}
}
}
void
ExecutorEntitiesCollector::prune_invalid_nodes_and_groups()
{
for (auto node_it = weak_nodes_.begin();
node_it != weak_nodes_.end(); )
{
if (node_it->expired()) {
node_it = remove_weak_node(node_it);
} else {
node_it++;
}
}
for (auto group_it = automatically_added_groups_.begin();
group_it != automatically_added_groups_.end(); )
{
if (group_it->expired()) {
group_it = remove_weak_callback_group(group_it, automatically_added_groups_);
} else {
group_it++;
}
}
for (auto group_it = manually_added_groups_.begin();
group_it != manually_added_groups_.end(); )
{
if (group_it->expired()) {
group_it = remove_weak_callback_group(group_it, manually_added_groups_);
} else {
group_it++;
}
}
}
} // namespace executors
} // namespace rclcpp

View File

@@ -0,0 +1,183 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <iostream>
#include "rclcpp/exceptions.hpp"
#include "rclcpp/executors/executor_notify_waitable.hpp"
namespace rclcpp
{
namespace executors
{
ExecutorNotifyWaitable::ExecutorNotifyWaitable(std::function<void(void)> on_execute_callback)
: execute_callback_(on_execute_callback)
{
}
ExecutorNotifyWaitable::ExecutorNotifyWaitable(const ExecutorNotifyWaitable & other)
: ExecutorNotifyWaitable(other.execute_callback_)
{
this->notify_guard_conditions_ = other.notify_guard_conditions_;
}
ExecutorNotifyWaitable & ExecutorNotifyWaitable::operator=(const ExecutorNotifyWaitable & other)
{
if (this != &other) {
this->execute_callback_ = other.execute_callback_;
this->notify_guard_conditions_ = other.notify_guard_conditions_;
}
return *this;
}
void
ExecutorNotifyWaitable::add_to_wait_set(rcl_wait_set_t * wait_set)
{
std::lock_guard<std::mutex> lock(guard_condition_mutex_);
for (auto weak_guard_condition : this->notify_guard_conditions_) {
auto guard_condition = weak_guard_condition.lock();
if (guard_condition) {
auto rcl_guard_condition = &guard_condition->get_rcl_guard_condition();
rcl_ret_t ret = rcl_wait_set_add_guard_condition(
wait_set,
rcl_guard_condition, NULL);
if (RCL_RET_OK != ret) {
rclcpp::exceptions::throw_from_rcl_error(
ret, "failed to add guard condition to wait set");
}
}
}
}
bool
ExecutorNotifyWaitable::is_ready(rcl_wait_set_t * wait_set)
{
std::lock_guard<std::mutex> lock(guard_condition_mutex_);
bool any_ready = false;
for (size_t ii = 0; ii < wait_set->size_of_guard_conditions; ++ii) {
auto rcl_guard_condition = wait_set->guard_conditions[ii];
if (nullptr == rcl_guard_condition) {
continue;
}
for (auto weak_guard_condition : this->notify_guard_conditions_) {
auto guard_condition = weak_guard_condition.lock();
if (guard_condition && &guard_condition->get_rcl_guard_condition() == rcl_guard_condition) {
any_ready = true;
}
}
}
return any_ready;
}
void
ExecutorNotifyWaitable::execute(std::shared_ptr<void> & data)
{
(void) data;
this->execute_callback_();
}
std::shared_ptr<void>
ExecutorNotifyWaitable::take_data()
{
return nullptr;
}
std::shared_ptr<void>
ExecutorNotifyWaitable::take_data_by_entity_id(size_t id)
{
(void) id;
return nullptr;
}
void
ExecutorNotifyWaitable::set_on_ready_callback(std::function<void(size_t, int)> callback)
{
// The second argument of the callback could be used to identify which guard condition
// triggered the event.
// We could indicate which of the guard conditions was triggered, but the executor
// is already going to check that.
auto gc_callback = [callback](size_t count) {
callback(count, 0);
};
std::lock_guard<std::mutex> lock(guard_condition_mutex_);
on_ready_callback_ = gc_callback;
for (auto weak_gc : notify_guard_conditions_) {
auto gc = weak_gc.lock();
if (!gc) {
continue;
}
gc->set_on_trigger_callback(on_ready_callback_);
}
}
RCLCPP_PUBLIC
void
ExecutorNotifyWaitable::clear_on_ready_callback()
{
std::lock_guard<std::mutex> lock(guard_condition_mutex_);
on_ready_callback_ = nullptr;
for (auto weak_gc : notify_guard_conditions_) {
auto gc = weak_gc.lock();
if (!gc) {
continue;
}
gc->set_on_trigger_callback(nullptr);
}
}
void
ExecutorNotifyWaitable::add_guard_condition(rclcpp::GuardCondition::WeakPtr weak_guard_condition)
{
std::lock_guard<std::mutex> lock(guard_condition_mutex_);
auto guard_condition = weak_guard_condition.lock();
if (guard_condition && notify_guard_conditions_.count(weak_guard_condition) == 0) {
notify_guard_conditions_.insert(weak_guard_condition);
if (on_ready_callback_) {
guard_condition->set_on_trigger_callback(on_ready_callback_);
}
}
}
void
ExecutorNotifyWaitable::remove_guard_condition(rclcpp::GuardCondition::WeakPtr weak_guard_condition)
{
std::lock_guard<std::mutex> lock(guard_condition_mutex_);
if (notify_guard_conditions_.count(weak_guard_condition) != 0) {
notify_guard_conditions_.erase(weak_guard_condition);
auto guard_condition = weak_guard_condition.lock();
// If this notify waitable doesn't have an on_ready_callback, then there's nothing to unset
if (guard_condition && on_ready_callback_) {
guard_condition->set_on_trigger_callback(nullptr);
}
}
}
size_t
ExecutorNotifyWaitable::get_number_of_ready_guard_conditions()
{
std::lock_guard<std::mutex> lock(guard_condition_mutex_);
return notify_guard_conditions_.size();
}
} // namespace executors
} // namespace rclcpp

View File

@@ -109,8 +109,8 @@ StaticExecutorEntitiesCollector::execute(std::shared_ptr<void> & data)
std::lock_guard<std::mutex> guard{new_nodes_mutex_};
for (const auto & weak_node : new_nodes_) {
if (auto node_ptr = weak_node.lock()) {
const auto & gc = node_ptr->get_notify_guard_condition();
weak_nodes_to_guard_conditions_[node_ptr] = &gc;
weak_nodes_to_guard_conditions_[node_ptr] =
node_ptr->get_shared_notify_guard_condition().get();
}
}
new_nodes_.clear();

View File

@@ -139,7 +139,7 @@ StaticSingleThreadedExecutor::add_callback_group(
bool is_new_node = entities_collector_->add_callback_group(group_ptr, node_ptr);
if (is_new_node && notify) {
// Interrupt waiting to handle new node
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
}
}
@@ -150,7 +150,7 @@ StaticSingleThreadedExecutor::add_node(
bool is_new_node = entities_collector_->add_node(node_ptr);
if (is_new_node && notify) {
// Interrupt waiting to handle new node
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
}
}
@@ -167,7 +167,7 @@ StaticSingleThreadedExecutor::remove_callback_group(
bool node_removed = entities_collector_->remove_callback_group(group_ptr);
// If the node was matched and removed, interrupt waiting
if (node_removed && notify) {
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
}
}
@@ -181,7 +181,7 @@ StaticSingleThreadedExecutor::remove_node(
}
// If the node was matched and removed, interrupt waiting
if (notify) {
interrupt_guard_condition_.trigger();
interrupt_guard_condition_->trigger();
}
}

View File

@@ -15,15 +15,11 @@
#include "rclcpp/experimental/executors/events_executor/events_executor.hpp"
#include <memory>
#include <queue>
#include <string>
#include <utility>
#include <vector>
#include "rcpputils/scope_exit.hpp"
#include "rclcpp/exceptions/exceptions.hpp"
using namespace std::chrono_literals;
using rclcpp::experimental::executors::EventsExecutor;
@@ -41,9 +37,12 @@ EventsExecutor::EventsExecutor(
events_queue_ = std::move(events_queue);
// Create timers manager
std::function<void(void *)> timer_on_ready_cb = nullptr;
if (execute_timers_separate_thread) {
timer_on_ready_cb = [this](const void * timer_id) {
// The timers manager can be used either to only track timers (in this case an expired
// timer will generate an executor event and then it will be executed by the executor thread)
// or it can also take care of executing expired timers in its dedicated thread.
std::function<void(const rclcpp::TimerBase *)> timer_on_ready_cb = nullptr;
if (!execute_timers_separate_thread) {
timer_on_ready_cb = [this](const rclcpp::TimerBase * timer_id) {
ExecutorEvent event = {timer_id, -1, ExecutorEventType::TIMER_EVENT, 1};
this->events_queue_->enqueue(event);
};
@@ -51,22 +50,57 @@ EventsExecutor::EventsExecutor(
timers_manager_ =
std::make_shared<rclcpp::experimental::TimersManager>(context_, timer_on_ready_cb);
// Create entities collector
entities_collector_ = std::make_shared<EventsExecutorEntitiesCollector>(this);
entities_collector_->init();
this->current_entities_collection_ =
std::make_shared<rclcpp::executors::ExecutorEntitiesCollection>();
// Setup the executor notifier to wake up the executor when some guard conditions are tiggered.
// The added guard conditions are guaranteed to not go out of scope before the executor itself.
executor_notifier_ = std::make_shared<EventsExecutorNotifyWaitable>();
executor_notifier_->add_guard_condition(shutdown_guard_condition_.get());
executor_notifier_->add_guard_condition(&interrupt_guard_condition_);
notify_waitable_ = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(
[this]() {
// This callback is invoked when:
// - the interrupt or shutdown guard condition is triggered:
// ---> we need to wake up the executor so that it can terminate
// - a node or callback group guard condition is triggered:
// ---> the entities collection is changed, we need to update callbacks
notify_waitable_event_pushed_ = false;
this->refresh_current_collection_from_callback_groups();
});
entities_collector_->add_waitable(executor_notifier_);
// Make sure that the notify waitable is immediately added to the collection
// to avoid missing events
this->add_notify_waitable_to_collection(current_entities_collection_->waitables);
notify_waitable_->add_guard_condition(interrupt_guard_condition_);
notify_waitable_->add_guard_condition(shutdown_guard_condition_);
notify_waitable_->set_on_ready_callback(
this->create_waitable_callback(notify_waitable_.get()));
auto notify_waitable_entity_id = notify_waitable_.get();
notify_waitable_->set_on_ready_callback(
[this, notify_waitable_entity_id](size_t num_events, int waitable_data) {
// The notify waitable has a special callback.
// We don't care about how many events as when we wake up the executor we are going to
// process everything regardless.
// For the same reason, if an event of this type has already been pushed but it has not been
// processed yet, we avoid pushing additional events.
(void)num_events;
if (notify_waitable_event_pushed_.exchange(true)) {
return;
}
ExecutorEvent event =
{notify_waitable_entity_id, waitable_data, ExecutorEventType::WAITABLE_EVENT, 1};
this->events_queue_->enqueue(event);
});
this->entities_collector_ =
std::make_shared<rclcpp::executors::ExecutorEntitiesCollector>(notify_waitable_);
}
EventsExecutor::~EventsExecutor()
{
spinning.store(false);
notify_waitable_->clear_on_ready_callback();
this->refresh_current_collection({});
}
void
@@ -200,7 +234,9 @@ EventsExecutor::add_node(
(void) notify;
// Add node to entities collector
entities_collector_->add_node(node_ptr);
this->entities_collector_->add_node(node_ptr);
this->refresh_current_collection_from_callback_groups();
}
void
@@ -220,7 +256,9 @@ EventsExecutor::remove_node(
// This will result in un-setting all the event callbacks from its entities.
// After this function returns, this executor will not receive any more events associated
// to these entities.
entities_collector_->remove_node(node_ptr);
this->entities_collector_->remove_node(node_ptr);
this->refresh_current_collection_from_callback_groups();
}
void
@@ -235,19 +273,23 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
switch (event.type) {
case ExecutorEventType::CLIENT_EVENT:
{
auto client = entities_collector_->get_client(event.exec_entity_id);
auto client = this->retrieve_entity(
static_cast<const rcl_client_t *>(event.entity_key),
current_entities_collection_->clients);
if (client) {
for (size_t i = 0; i < event.num_events; i++) {
execute_client(client);
}
}
break;
}
case ExecutorEventType::SUBSCRIPTION_EVENT:
{
auto subscription = entities_collector_->get_subscription(event.exec_entity_id);
auto subscription = this->retrieve_entity(
static_cast<const rcl_subscription_t *>(event.entity_key),
current_entities_collection_->subscriptions);
if (subscription) {
for (size_t i = 0; i < event.num_events; i++) {
execute_subscription(subscription);
@@ -257,27 +299,32 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
}
case ExecutorEventType::SERVICE_EVENT:
{
auto service = entities_collector_->get_service(event.exec_entity_id);
auto service = this->retrieve_entity(
static_cast<const rcl_service_t *>(event.entity_key),
current_entities_collection_->services);
if (service) {
for (size_t i = 0; i < event.num_events; i++) {
execute_service(service);
}
}
break;
}
case ExecutorEventType::TIMER_EVENT:
{
timers_manager_->execute_ready_timer(event.exec_entity_id);
timers_manager_->execute_ready_timer(
static_cast<const rclcpp::TimerBase *>(event.entity_key));
break;
}
case ExecutorEventType::WAITABLE_EVENT:
{
auto waitable = entities_collector_->get_waitable(event.exec_entity_id);
auto waitable = this->retrieve_entity(
static_cast<const rclcpp::Waitable *>(event.entity_key),
current_entities_collection_->waitables);
if (waitable) {
for (size_t i = 0; i < event.num_events; i++) {
auto data = waitable->take_data_by_entity_id(event.gen_entity_id);
auto data = waitable->take_data_by_entity_id(event.waitable_data);
waitable->execute(data);
}
}
@@ -295,7 +342,11 @@ EventsExecutor::add_callback_group(
// This field is unused because we don't have to wake up
// the executor when a callback group is added.
(void)notify;
entities_collector_->add_callback_group(group_ptr, node_ptr);
(void)node_ptr;
this->entities_collector_->add_callback_group(group_ptr);
this->refresh_current_collection_from_callback_groups();
}
void
@@ -305,23 +356,140 @@ EventsExecutor::remove_callback_group(
// This field is unused because we don't have to wake up
// the executor when a callback group is removed.
(void)notify;
entities_collector_->remove_callback_group(group_ptr);
this->entities_collector_->remove_callback_group(group_ptr);
this->refresh_current_collection_from_callback_groups();
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
EventsExecutor::get_all_callback_groups()
{
return entities_collector_->get_all_callback_groups();
this->entities_collector_->update_collections();
return this->entities_collector_->get_all_callback_groups();
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
EventsExecutor::get_manually_added_callback_groups()
{
return entities_collector_->get_manually_added_callback_groups();
this->entities_collector_->update_collections();
return this->entities_collector_->get_manually_added_callback_groups();
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
EventsExecutor::get_automatically_added_callback_groups_from_nodes()
{
return entities_collector_->get_automatically_added_callback_groups_from_nodes();
this->entities_collector_->update_collections();
return this->entities_collector_->get_automatically_added_callback_groups();
}
void
EventsExecutor::refresh_current_collection_from_callback_groups()
{
this->entities_collector_->update_collections();
auto callback_groups = this->entities_collector_->get_all_callback_groups();
rclcpp::executors::ExecutorEntitiesCollection new_collection;
rclcpp::executors::build_entities_collection(callback_groups, new_collection);
// TODO(alsora): this may be implemented in a better way.
// We need the notify waitable to be included in the executor "current_collection"
// because we need to be able to retrieve events for it.
// We could explicitly check for the notify waitable ID when we receive a waitable event
// but I think that it's better if the waitable was in the collection and it could be
// retrieved in the "standard" way.
// To do it, we need to add the notify waitable as an entry in both the new and
// current collections such that it's neither added or removed.
this->add_notify_waitable_to_collection(new_collection.waitables);
this->add_notify_waitable_to_collection(current_entities_collection_->waitables);
this->refresh_current_collection(new_collection);
}
void
EventsExecutor::refresh_current_collection(
const rclcpp::executors::ExecutorEntitiesCollection & new_collection)
{
current_entities_collection_->timers.update(
new_collection.timers,
[this](rclcpp::TimerBase::SharedPtr timer) {timers_manager_->add_timer(timer);},
[this](rclcpp::TimerBase::SharedPtr timer) {timers_manager_->remove_timer(timer);});
current_entities_collection_->subscriptions.update(
new_collection.subscriptions,
[this](auto subscription) {
subscription->set_on_new_message_callback(
this->create_entity_callback(
subscription->get_subscription_handle().get(), ExecutorEventType::SUBSCRIPTION_EVENT));
},
[](auto subscription) {subscription->clear_on_new_message_callback();});
current_entities_collection_->clients.update(
new_collection.clients,
[this](auto client) {
client->set_on_new_response_callback(
this->create_entity_callback(
client->get_client_handle().get(), ExecutorEventType::CLIENT_EVENT));
},
[](auto client) {client->clear_on_new_response_callback();});
current_entities_collection_->services.update(
new_collection.services,
[this](auto service) {
service->set_on_new_request_callback(
this->create_entity_callback(
service->get_service_handle().get(), ExecutorEventType::SERVICE_EVENT));
},
[](auto service) {service->clear_on_new_request_callback();});
// DO WE NEED THIS? WE ARE NOT DOING ANYTHING WITH GUARD CONDITIONS
/*
current_entities_collection_->guard_conditions.update(new_collection.guard_conditions,
[](auto guard_condition) {(void)guard_condition;},
[](auto guard_condition) {guard_condition->set_on_trigger_callback(nullptr);});
*/
current_entities_collection_->waitables.update(
new_collection.waitables,
[this](auto waitable) {
waitable->set_on_ready_callback(
this->create_waitable_callback(waitable.get()));
},
[](auto waitable) {waitable->clear_on_ready_callback();});
}
std::function<void(size_t)>
EventsExecutor::create_entity_callback(
void * entity_key, ExecutorEventType event_type)
{
std::function<void(size_t)>
callback = [this, entity_key, event_type](size_t num_events) {
ExecutorEvent event = {entity_key, -1, event_type, num_events};
this->events_queue_->enqueue(event);
};
return callback;
}
std::function<void(size_t, int)>
EventsExecutor::create_waitable_callback(const rclcpp::Waitable * entity_key)
{
std::function<void(size_t, int)>
callback = [this, entity_key](size_t num_events, int waitable_data) {
ExecutorEvent event =
{entity_key, waitable_data, ExecutorEventType::WAITABLE_EVENT, num_events};
this->events_queue_->enqueue(event);
};
return callback;
}
void
EventsExecutor::add_notify_waitable_to_collection(
rclcpp::executors::ExecutorEntitiesCollection::WaitableCollection & collection)
{
// The notify waitable is not associated to any group, so use an invalid one
rclcpp::CallbackGroup::WeakPtr weak_group_ptr;
collection.insert(
{
this->notify_waitable_.get(),
{this->notify_waitable_, weak_group_ptr}
});
}

View File

@@ -1,699 +0,0 @@
// Copyright 2023 iRobot Corporation.
//
// 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.
#include "rclcpp/experimental/executors/events_executor/events_executor_entities_collector.hpp"
#include <algorithm>
#include <memory>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
#include "rclcpp/memory_strategy.hpp"
#include "rclcpp/detail/add_guard_condition_to_rcl_wait_set.hpp"
#include "rclcpp/experimental/executors/events_executor/events_executor.hpp"
using rclcpp::experimental::executors::ExecutorEvent;
using rclcpp::experimental::executors::ExecutorEventType;
using rclcpp::experimental::executors::EventsExecutorEntitiesCollector;
EventsExecutorEntitiesCollector::EventsExecutorEntitiesCollector(
rclcpp::experimental::executors::EventsExecutor * executor)
{
if (executor == nullptr) {
throw std::runtime_error("Received nullptr executor in EventsExecutorEntitiesCollector.");
}
associated_executor_ = executor;
timers_manager_ = associated_executor_->timers_manager_;
}
void
EventsExecutorEntitiesCollector::init()
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
// Add the EventsExecutorEntitiesCollector shared_ptr to waitables map
weak_waitables_map_.emplace(this, this->shared_from_this());
}
EventsExecutorEntitiesCollector::~EventsExecutorEntitiesCollector()
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
// Disassociate all callback groups and thus nodes.
for (const auto & pair : weak_groups_associated_with_executor_to_nodes_) {
auto group = pair.first.lock();
if (group) {
std::atomic_bool & has_executor = group->get_associated_with_executor_atomic();
has_executor.store(false);
callback_group_removed_impl(group);
}
}
for (const auto & pair : weak_groups_to_nodes_associated_with_executor_) {
auto group = pair.first.lock();
if (group) {
std::atomic_bool & has_executor = group->get_associated_with_executor_atomic();
has_executor.store(false);
callback_group_removed_impl(group);
}
}
// Disassociate all nodes
for (const auto & weak_node : weak_nodes_) {
auto node = weak_node.lock();
if (node) {
std::atomic_bool & has_executor = node->get_associated_with_executor_atomic();
has_executor.store(false);
node_removed_impl(node);
}
}
// Unset callback group notify guard condition executor callback
for (const auto & pair : weak_groups_to_guard_conditions_) {
auto group = pair.first.lock();
if (group) {
auto & group_gc = pair.second;
unset_guard_condition_callback(group_gc);
}
}
weak_clients_map_.clear();
weak_services_map_.clear();
weak_waitables_map_.clear();
weak_subscriptions_map_.clear();
weak_nodes_to_guard_conditions_.clear();
weak_groups_to_guard_conditions_.clear();
weak_groups_associated_with_executor_to_nodes_.clear();
weak_groups_to_nodes_associated_with_executor_.clear();
weak_nodes_.clear();
}
void
EventsExecutorEntitiesCollector::execute(std::shared_ptr<void> & data)
{
// This function is called when the associated executor is notified that something changed.
// We do not know if an entity has been added or removed so we have to rebuild everything.
(void)data;
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
timers_manager_->clear();
// If a registered node has a new callback group, register the group.
add_callback_groups_from_nodes_associated_to_executor();
// For all groups registered in the executor, set their event callbacks.
set_entities_event_callbacks_from_map(weak_groups_associated_with_executor_to_nodes_);
set_entities_event_callbacks_from_map(weak_groups_to_nodes_associated_with_executor_);
}
std::shared_ptr<void>
EventsExecutorEntitiesCollector::take_data()
{
return nullptr;
}
std::shared_ptr<void>
EventsExecutorEntitiesCollector::take_data_by_entity_id(size_t id)
{
(void)id;
return nullptr;
}
void
EventsExecutorEntitiesCollector::add_to_wait_set(rcl_wait_set_t * wait_set)
{
(void)wait_set;
}
bool
EventsExecutorEntitiesCollector::is_ready(rcl_wait_set_t * p_wait_set)
{
(void)p_wait_set;
return false;
}
bool
EventsExecutorEntitiesCollector::add_node(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
{
bool is_new_node = false;
// If the node already has an executor
std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
if (has_executor.exchange(true)) {
throw std::runtime_error("Node has already been added to an executor.");
}
node_ptr->for_each_callback_group(
[this, node_ptr, &is_new_node](rclcpp::CallbackGroup::SharedPtr group_ptr)
{
if (
!group_ptr->get_associated_with_executor_atomic().load() &&
group_ptr->automatically_add_to_executor_with_node())
{
is_new_node = (add_callback_group(
group_ptr,
node_ptr,
weak_groups_to_nodes_associated_with_executor_) ||
is_new_node);
}
});
weak_nodes_.push_back(node_ptr);
return is_new_node;
}
bool
EventsExecutorEntitiesCollector::add_callback_group(
rclcpp::CallbackGroup::SharedPtr group_ptr,
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap & weak_groups_to_nodes)
{
// If the callback_group already has an executor
std::atomic_bool & has_executor = group_ptr->get_associated_with_executor_atomic();
if (has_executor.exchange(true)) {
throw std::runtime_error("Callback group has already been added to an executor.");
}
bool is_new_node = !has_node(node_ptr, weak_groups_associated_with_executor_to_nodes_) &&
!has_node(node_ptr, weak_groups_to_nodes_associated_with_executor_);
rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group_ptr;
auto insert_info = weak_groups_to_nodes.insert(
std::make_pair(weak_group_ptr, node_ptr));
bool was_inserted = insert_info.second;
if (!was_inserted) {
throw std::runtime_error("Callback group was already added to executor.");
}
if (is_new_node) {
node_added_impl(node_ptr);
}
if (node_ptr->get_context()->is_valid()) {
auto callback_group_guard_condition =
group_ptr->get_notify_guard_condition(node_ptr->get_context());
rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group_ptr;
weak_groups_to_guard_conditions_[weak_group_ptr] = callback_group_guard_condition.get();
}
callback_group_added_impl(group_ptr);
return is_new_node;
}
bool
EventsExecutorEntitiesCollector::add_callback_group(
rclcpp::CallbackGroup::SharedPtr group_ptr,
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
{
return add_callback_group(group_ptr, node_ptr, weak_groups_associated_with_executor_to_nodes_);
}
bool
EventsExecutorEntitiesCollector::remove_callback_group(
rclcpp::CallbackGroup::SharedPtr group_ptr)
{
return this->remove_callback_group_from_map(
group_ptr,
weak_groups_associated_with_executor_to_nodes_);
}
bool
EventsExecutorEntitiesCollector::remove_callback_group_from_map(
rclcpp::CallbackGroup::SharedPtr group_ptr,
rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap & weak_groups_to_nodes)
{
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr;
rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group_ptr;
auto iter = weak_groups_to_nodes.find(weak_group_ptr);
if (iter != weak_groups_to_nodes.end()) {
node_ptr = iter->second.lock();
if (node_ptr == nullptr) {
throw std::runtime_error("Node must not be deleted before its callback group(s).");
}
weak_groups_to_nodes.erase(iter);
callback_group_removed_impl(group_ptr);
} else {
throw std::runtime_error("Callback group needs to be associated with executor.");
}
// If the node was matched and removed, interrupt waiting.
bool node_removed = false;
if (!has_node(node_ptr, weak_groups_associated_with_executor_to_nodes_) &&
!has_node(node_ptr, weak_groups_to_nodes_associated_with_executor_))
{
node_removed_impl(node_ptr);
node_removed = true;
}
weak_groups_to_guard_conditions_.erase(weak_group_ptr);
return node_removed;
}
bool
EventsExecutorEntitiesCollector::remove_node(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
{
if (!node_ptr->get_associated_with_executor_atomic().load()) {
return false;
}
bool node_found = false;
auto node_it = weak_nodes_.begin();
while (node_it != weak_nodes_.end()) {
bool matched = (node_it->lock() == node_ptr);
if (matched) {
weak_nodes_.erase(node_it);
node_found = true;
break;
}
++node_it;
}
if (!node_found) {
return false;
}
std::vector<rclcpp::CallbackGroup::SharedPtr> found_group_ptrs;
std::for_each(
weak_groups_to_nodes_associated_with_executor_.begin(),
weak_groups_to_nodes_associated_with_executor_.end(),
[&found_group_ptrs, node_ptr](std::pair<rclcpp::CallbackGroup::WeakPtr,
rclcpp::node_interfaces::NodeBaseInterface::WeakPtr> key_value_pair) {
auto & weak_node_ptr = key_value_pair.second;
auto shared_node_ptr = weak_node_ptr.lock();
auto group_ptr = key_value_pair.first.lock();
if (shared_node_ptr == node_ptr) {
found_group_ptrs.push_back(group_ptr);
}
});
std::for_each(
found_group_ptrs.begin(), found_group_ptrs.end(), [this]
(rclcpp::CallbackGroup::SharedPtr group_ptr) {
this->remove_callback_group_from_map(
group_ptr,
weak_groups_to_nodes_associated_with_executor_);
});
std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
has_executor.store(false);
return true;
}
// Returns true iff the weak_groups_to_nodes map has node_ptr as the value in any of its entry.
bool
EventsExecutorEntitiesCollector::has_node(
const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
const rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap &
weak_groups_to_nodes) const
{
return std::find_if(
weak_groups_to_nodes.begin(),
weak_groups_to_nodes.end(),
[&](const WeakCallbackGroupsToNodesMap::value_type & other) -> bool {
auto other_ptr = other.second.lock();
return other_ptr == node_ptr;
}) != weak_groups_to_nodes.end();
}
void
EventsExecutorEntitiesCollector::add_callback_groups_from_nodes_associated_to_executor()
{
for (const auto & weak_node : weak_nodes_) {
auto node = weak_node.lock();
if (node) {
node->for_each_callback_group(
[this, node](rclcpp::CallbackGroup::SharedPtr shared_group_ptr)
{
if (shared_group_ptr->automatically_add_to_executor_with_node() &&
!shared_group_ptr->get_associated_with_executor_atomic().load())
{
add_callback_group(
shared_group_ptr,
node,
weak_groups_to_nodes_associated_with_executor_);
}
});
}
}
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
EventsExecutorEntitiesCollector::get_all_callback_groups()
{
std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
for (const auto & group_node_ptr : weak_groups_associated_with_executor_to_nodes_) {
groups.push_back(group_node_ptr.first);
}
for (const auto & group_node_ptr : weak_groups_to_nodes_associated_with_executor_) {
groups.push_back(group_node_ptr.first);
}
return groups;
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
EventsExecutorEntitiesCollector::get_manually_added_callback_groups()
{
std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
for (const auto & group_node_ptr : weak_groups_associated_with_executor_to_nodes_) {
groups.push_back(group_node_ptr.first);
}
return groups;
}
std::vector<rclcpp::CallbackGroup::WeakPtr>
EventsExecutorEntitiesCollector::get_automatically_added_callback_groups_from_nodes()
{
std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
for (const auto & group_node_ptr : weak_groups_to_nodes_associated_with_executor_) {
groups.push_back(group_node_ptr.first);
}
return groups;
}
void
EventsExecutorEntitiesCollector::callback_group_added_impl(
rclcpp::CallbackGroup::SharedPtr group)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group;
auto iter = weak_groups_to_guard_conditions_.find(weak_group_ptr);
if (iter != weak_groups_to_guard_conditions_.end()) {
// Set an event callback for the group's notify guard condition, so if new entities are added
// or removed to this node we will receive an event.
set_guard_condition_callback(iter->second);
}
// For all entities in the callback group, set their event callback
set_callback_group_entities_callbacks(group);
}
void
EventsExecutorEntitiesCollector::node_added_impl(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
auto notify_guard_condition = &(node->get_notify_guard_condition());
// Set an event callback for the node's notify guard condition, so if new entities are added
// or removed to this node we will receive an event.
set_guard_condition_callback(notify_guard_condition);
// Store node's notify guard condition
weak_nodes_to_guard_conditions_[node] = notify_guard_condition;
}
void
EventsExecutorEntitiesCollector::callback_group_removed_impl(
rclcpp::CallbackGroup::SharedPtr group)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
// For all the entities in the group, unset their callbacks
unset_callback_group_entities_callbacks(group);
}
void
EventsExecutorEntitiesCollector::node_removed_impl(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
// Node doesn't have more callback groups associated to the executor.
// Unset the event callback for the node's notify guard condition, to stop
// receiving events if entities are added or removed to this node.
unset_guard_condition_callback(&(node->get_notify_guard_condition()));
// Remove guard condition from list
rclcpp::node_interfaces::NodeBaseInterface::WeakPtr weak_node_ptr(node);
weak_nodes_to_guard_conditions_.erase(weak_node_ptr);
}
void
EventsExecutorEntitiesCollector::set_entities_event_callbacks_from_map(
const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes)
{
for (const auto & pair : weak_groups_to_nodes) {
auto group = pair.first.lock();
auto node = pair.second.lock();
if (!node || !group || !group->can_be_taken_from().load()) {
continue;
}
set_callback_group_entities_callbacks(group);
}
}
void
EventsExecutorEntitiesCollector::set_callback_group_entities_callbacks(
rclcpp::CallbackGroup::SharedPtr group)
{
// Timers are handled by the timers manager
group->find_timer_ptrs_if(
[this](const rclcpp::TimerBase::SharedPtr & timer) {
if (timer) {
timers_manager_->add_timer(timer);
}
return false;
});
// Set callbacks for all other entity types
group->find_subscription_ptrs_if(
[this](const rclcpp::SubscriptionBase::SharedPtr & subscription) {
if (subscription) {
weak_subscriptions_map_.emplace(subscription.get(), subscription);
subscription->set_on_new_message_callback(
create_entity_callback(subscription.get(), ExecutorEventType::SUBSCRIPTION_EVENT));
}
return false;
});
group->find_service_ptrs_if(
[this](const rclcpp::ServiceBase::SharedPtr & service) {
if (service) {
weak_services_map_.emplace(service.get(), service);
service->set_on_new_request_callback(
create_entity_callback(service.get(), ExecutorEventType::SERVICE_EVENT));
}
return false;
});
group->find_client_ptrs_if(
[this](const rclcpp::ClientBase::SharedPtr & client) {
if (client) {
weak_clients_map_.emplace(client.get(), client);
client->set_on_new_response_callback(
create_entity_callback(client.get(), ExecutorEventType::CLIENT_EVENT));
}
return false;
});
group->find_waitable_ptrs_if(
[this](const rclcpp::Waitable::SharedPtr & waitable) {
if (waitable) {
weak_waitables_map_.emplace(waitable.get(), waitable);
waitable->set_on_ready_callback(
create_waitable_callback(waitable.get()));
}
return false;
});
}
void
EventsExecutorEntitiesCollector::unset_callback_group_entities_callbacks(
rclcpp::CallbackGroup::SharedPtr group)
{
auto iter = weak_groups_to_guard_conditions_.find(group);
if (iter != weak_groups_to_guard_conditions_.end()) {
unset_guard_condition_callback(iter->second);
}
// Timers are handled by the timers manager
group->find_timer_ptrs_if(
[this](const rclcpp::TimerBase::SharedPtr & timer) {
if (timer) {
timers_manager_->remove_timer(timer);
}
return false;
});
// Unset callbacks for all other entity types
group->find_subscription_ptrs_if(
[this](const rclcpp::SubscriptionBase::SharedPtr & subscription) {
if (subscription) {
subscription->clear_on_new_message_callback();
weak_subscriptions_map_.erase(subscription.get());
}
return false;
});
group->find_service_ptrs_if(
[this](const rclcpp::ServiceBase::SharedPtr & service) {
if (service) {
service->clear_on_new_request_callback();
weak_services_map_.erase(service.get());
}
return false;
});
group->find_client_ptrs_if(
[this](const rclcpp::ClientBase::SharedPtr & client) {
if (client) {
client->clear_on_new_response_callback();
weak_clients_map_.erase(client.get());
}
return false;
});
group->find_waitable_ptrs_if(
[this](const rclcpp::Waitable::SharedPtr & waitable) {
if (waitable) {
waitable->clear_on_ready_callback();
weak_waitables_map_.erase(waitable.get());
}
return false;
});
}
void
EventsExecutorEntitiesCollector::set_guard_condition_callback(
rclcpp::GuardCondition * guard_condition)
{
auto gc_callback = [this](size_t num_events) {
// Override num events (we don't care more than a single event)
num_events = 1;
int gc_id = -1;
ExecutorEvent event = {this, gc_id, ExecutorEventType::WAITABLE_EVENT, num_events};
associated_executor_->events_queue_->enqueue(event);
};
guard_condition->set_on_trigger_callback(gc_callback);
}
void
EventsExecutorEntitiesCollector::unset_guard_condition_callback(
rclcpp::GuardCondition * guard_condition)
{
guard_condition->set_on_trigger_callback(nullptr);
}
rclcpp::SubscriptionBase::SharedPtr
EventsExecutorEntitiesCollector::get_subscription(const void * subscription_id)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
auto it = weak_subscriptions_map_.find(subscription_id);
if (it != weak_subscriptions_map_.end()) {
auto subscription_weak_ptr = it->second;
auto subscription_shared_ptr = subscription_weak_ptr.lock();
if (subscription_shared_ptr) {
return subscription_shared_ptr;
}
// The subscription expired, remove from map
weak_subscriptions_map_.erase(it);
}
return nullptr;
}
rclcpp::ClientBase::SharedPtr
EventsExecutorEntitiesCollector::get_client(const void * client_id)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
auto it = weak_clients_map_.find(client_id);
if (it != weak_clients_map_.end()) {
auto client_weak_ptr = it->second;
auto client_shared_ptr = client_weak_ptr.lock();
if (client_shared_ptr) {
return client_shared_ptr;
}
// The client expired, remove from map
weak_clients_map_.erase(it);
}
return nullptr;
}
rclcpp::ServiceBase::SharedPtr
EventsExecutorEntitiesCollector::get_service(const void * service_id)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
auto it = weak_services_map_.find(service_id);
if (it != weak_services_map_.end()) {
auto service_weak_ptr = it->second;
auto service_shared_ptr = service_weak_ptr.lock();
if (service_shared_ptr) {
return service_shared_ptr;
}
// The service expired, remove from map
weak_services_map_.erase(it);
}
return nullptr;
}
rclcpp::Waitable::SharedPtr
EventsExecutorEntitiesCollector::get_waitable(const void * waitable_id)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
auto it = weak_waitables_map_.find(waitable_id);
if (it != weak_waitables_map_.end()) {
auto waitable_weak_ptr = it->second;
auto waitable_shared_ptr = waitable_weak_ptr.lock();
if (waitable_shared_ptr) {
return waitable_shared_ptr;
}
// The waitable expired, remove from map
weak_waitables_map_.erase(it);
}
return nullptr;
}
void
EventsExecutorEntitiesCollector::add_waitable(rclcpp::Waitable::SharedPtr waitable)
{
std::lock_guard<std::recursive_mutex> lock(reentrant_mutex_);
weak_waitables_map_.emplace(waitable.get(), waitable);
waitable->set_on_ready_callback(
create_waitable_callback(waitable.get()));
}
std::function<void(size_t)>
EventsExecutorEntitiesCollector::create_entity_callback(
void * exec_entity_id, ExecutorEventType event_type)
{
std::function<void(size_t)>
callback = [this, exec_entity_id, event_type](size_t num_events) {
ExecutorEvent event = {exec_entity_id, -1, event_type, num_events};
associated_executor_->events_queue_->enqueue(event);
};
return callback;
}
std::function<void(size_t, int)>
EventsExecutorEntitiesCollector::create_waitable_callback(void * exec_entity_id)
{
std::function<void(size_t, int)>
callback = [this, exec_entity_id](size_t num_events, int gen_entity_id) {
ExecutorEvent event =
{exec_entity_id, gen_entity_id, ExecutorEventType::WAITABLE_EVENT, num_events};
associated_executor_->events_queue_->enqueue(event);
};
return callback;
}

View File

@@ -21,11 +21,13 @@
#include <memory>
#include <stdexcept>
#include "rcpputils/scope_exit.hpp"
using rclcpp::experimental::TimersManager;
TimersManager::TimersManager(
std::shared_ptr<rclcpp::Context> context,
std::function<void(void *)> on_ready_callback)
std::function<void(const rclcpp::TimerBase *)> on_ready_callback)
{
context_ = context;
on_ready_callback_ = on_ready_callback;
@@ -123,18 +125,6 @@ size_t TimersManager::get_number_ready_timers()
return locked_heap.get_number_ready_timers();
}
void TimersManager::execute_ready_timers()
{
// Do not allow to interfere with the thread running
if (running_) {
throw std::runtime_error(
"execute_ready_timers() can't be used while timers thread is running");
}
std::unique_lock<std::mutex> lock(timers_mutex_);
this->execute_ready_timers_unsafe();
}
bool TimersManager::execute_head_timer()
{
// Do not allow to interfere with the thread running
@@ -156,12 +146,10 @@ bool TimersManager::execute_head_timer()
const bool timer_ready = head_timer->is_ready();
if (timer_ready) {
// NOTE: here we always execute the timer, regardless of whether the
// on_ready_callback is set or not.
head_timer->call();
if (on_ready_callback_) {
on_ready_callback_(head_timer.get());
} else {
head_timer->execute_callback();
}
head_timer->execute_callback();
timers_heap.heapify_root();
weak_timers_heap_.store(timers_heap);
}
@@ -169,7 +157,7 @@ bool TimersManager::execute_head_timer()
return timer_ready;
}
void TimersManager::execute_ready_timer(const void * timer_id)
void TimersManager::execute_ready_timer(const rclcpp::TimerBase * timer_id)
{
TimerPtr ready_timer;
{
@@ -246,6 +234,10 @@ void TimersManager::execute_ready_timers_unsafe()
void TimersManager::run_timers()
{
// Make sure the running flag is set to false when we exit from this function
// to allow restarting the timers thread.
RCPPUTILS_SCOPE_EXIT(this->running_.store(false); );
while (rclcpp::ok(context_) && running_) {
// Lock mutex
std::unique_lock<std::mutex> lock(timers_mutex_);
@@ -269,10 +261,6 @@ void TimersManager::run_timers()
// Execute timers
this->execute_ready_timers_unsafe();
}
// Make sure the running flag is set to false when we exit from this function
// to allow restarting the timers thread.
running_ = false;
}
void TimersManager::clear()
@@ -282,7 +270,7 @@ void TimersManager::clear()
std::unique_lock<std::mutex> lock(timers_mutex_);
TimersHeap locked_heap = weak_timers_heap_.validate_and_lock();
locked_heap.clear_callbacks();
locked_heap.clear_timers_on_reset_callbacks();
weak_timers_heap_.clear();

View File

@@ -25,17 +25,20 @@
namespace rclcpp
{
std::shared_ptr<void> GenericSubscription::create_message()
std::shared_ptr<void>
GenericSubscription::create_message()
{
return create_serialized_message();
}
std::shared_ptr<rclcpp::SerializedMessage> GenericSubscription::create_serialized_message()
std::shared_ptr<rclcpp::SerializedMessage>
GenericSubscription::create_serialized_message()
{
return std::make_shared<rclcpp::SerializedMessage>(0);
}
void GenericSubscription::handle_message(
void
GenericSubscription::handle_message(
std::shared_ptr<void> &,
const rclcpp::MessageInfo &)
{
@@ -51,7 +54,8 @@ GenericSubscription::handle_serialized_message(
callback_(message);
}
void GenericSubscription::handle_loaned_message(
void
GenericSubscription::handle_loaned_message(
void * message, const rclcpp::MessageInfo & message_info)
{
(void) message;
@@ -60,16 +64,69 @@ void GenericSubscription::handle_loaned_message(
"handle_loaned_message is not implemented for GenericSubscription");
}
void GenericSubscription::return_message(std::shared_ptr<void> & message)
void
GenericSubscription::return_message(std::shared_ptr<void> & message)
{
auto typed_message = std::static_pointer_cast<rclcpp::SerializedMessage>(message);
return_serialized_message(typed_message);
}
void GenericSubscription::return_serialized_message(
void
GenericSubscription::return_serialized_message(
std::shared_ptr<rclcpp::SerializedMessage> & message)
{
message.reset();
}
// DYNAMIC TYPE ====================================================================================
// TODO(methylDragon): Reorder later
rclcpp::dynamic_typesupport::DynamicMessageType::SharedPtr
GenericSubscription::get_shared_dynamic_message_type()
{
throw rclcpp::exceptions::UnimplementedError(
"get_shared_dynamic_message_type is not implemented for GenericSubscription");
}
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr
GenericSubscription::get_shared_dynamic_message()
{
throw rclcpp::exceptions::UnimplementedError(
"get_shared_dynamic_message is not implemented for GenericSubscription");
}
rclcpp::dynamic_typesupport::DynamicSerializationSupport::SharedPtr
GenericSubscription::get_shared_dynamic_serialization_support()
{
throw rclcpp::exceptions::UnimplementedError(
"get_shared_dynamic_serialization_support is not implemented for GenericSubscription");
}
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr
GenericSubscription::create_dynamic_message()
{
throw rclcpp::exceptions::UnimplementedError(
"create_dynamic_message is not implemented for GenericSubscription");
}
void
GenericSubscription::return_dynamic_message(
rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message)
{
(void) message;
throw rclcpp::exceptions::UnimplementedError(
"return_dynamic_message is not implemented for GenericSubscription");
}
void
GenericSubscription::handle_dynamic_message(
const rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message,
const rclcpp::MessageInfo & message_info)
{
(void) message;
(void) message_info;
throw rclcpp::exceptions::UnimplementedError(
"handle_dynamic_message is not implemented for GenericSubscription");
}
} // namespace rclcpp

View File

@@ -125,4 +125,18 @@ Logger::set_level(Level level)
}
}
Logger::Level
Logger::get_effective_level() const
{
int logger_level = rcutils_logging_get_logger_effective_level(get_name());
if (logger_level < 0) {
exceptions::throw_from_rcl_error(
RCL_RET_ERROR, "Couldn't get logger level",
rcutils_get_error_state(), rcutils_reset_error);
}
return static_cast<Level>(logger_level);
}
} // namespace rclcpp

View File

@@ -14,6 +14,7 @@
#include <memory>
#include <mutex>
#include <stdexcept>
#include "rcutils/macros.h"

View File

@@ -17,7 +17,10 @@
#include <limits>
#include <map>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>
@@ -36,6 +39,7 @@
#include "rclcpp/node_interfaces/node_time_source.hpp"
#include "rclcpp/node_interfaces/node_timers.hpp"
#include "rclcpp/node_interfaces/node_topics.hpp"
#include "rclcpp/node_interfaces/node_type_descriptions.hpp"
#include "rclcpp/node_interfaces/node_waitables.hpp"
#include "rclcpp/qos_overriding_options.hpp"
@@ -109,6 +113,72 @@ create_effective_namespace(const std::string & node_namespace, const std::string
} // namespace
/// \brief Associate new extra member variables with instances of Node without changing ABI.
/**
* It is used only for bugfixes or backported features that require new members.
* Atomically constructs/destroys all extra members.
* Node instance will register and remove itself, and use its methods to retrieve members.
* Note for performance consideration that accessing these members uses a map lookup.
*/
class Node::BackportMembers
{
public:
BackportMembers() = default;
~BackportMembers() = default;
/// \brief Add all backported members for a new Node.
/**
* \param[in] key Raw pointer to the Node instance that will use new members.
*/
void add(Node * key)
{
// Adding a new instance to the maps requires exclusive access
std::unique_lock lock(map_access_mutex_);
type_descriptions_map_.emplace(
key,
std::make_shared<rclcpp::node_interfaces::NodeTypeDescriptions>(
key->get_node_base_interface(),
key->get_node_logging_interface(),
key->get_node_parameters_interface(),
key->get_node_services_interface()));
}
/// \brief Remove the members for an instance of Node
/**
* \param[in] key Raw pointer to the Node
*/
void remove(const Node * key)
{
// Removing an instance from the maps requires exclusive access
std::unique_lock lock(map_access_mutex_);
type_descriptions_map_.erase(key);
}
/// \brief Retrieve the NodeTypeDescriptionsInterface for a Node.
/**
* \param[in] key Raw pointer to an instance of Node.
* \return A shared ptr to this Node's NodeTypeDescriptionsInterface instance.
*/
rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
get_node_type_descriptions_interface(const Node * key) const
{
// Multiple threads can retrieve from the maps at the same time
std::shared_lock lock(map_access_mutex_);
return type_descriptions_map_.at(key);
}
private:
/// \brief Map that stored TypeDescriptionsInterface members
std::unordered_map<
const Node *, rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
> type_descriptions_map_;
/// \brief Controls access to all private maps
mutable std::shared_mutex map_access_mutex_;
};
// Definition of static member declaration
Node::BackportMembers Node::backport_members_;
Node::Node(
const std::string & node_name,
const NodeOptions & options)
@@ -167,7 +237,7 @@ Node::Node(
options.use_intra_process_comms(),
options.enable_topic_statistics())),
node_graph_(new rclcpp::node_interfaces::NodeGraph(node_base_.get())),
node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_.get())),
node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_)),
node_timers_(new rclcpp::node_interfaces::NodeTimers(node_base_.get())),
node_topics_(new rclcpp::node_interfaces::NodeTopics(node_base_.get(), node_timers_.get())),
node_services_(new rclcpp::node_interfaces::NodeServices(node_base_.get())),
@@ -211,6 +281,8 @@ Node::Node(
sub_namespace_(""),
effective_namespace_(create_effective_namespace(this->get_namespace(), sub_namespace_))
{
backport_members_.add(this);
// we have got what we wanted directly from the overrides,
// but declare the parameters anyway so they are visible.
rclcpp::detail::declare_qos_parameters(
@@ -225,6 +297,10 @@ Node::Node(
node_topics_->resolve_topic_name("/parameter_events"),
options.parameter_event_qos(),
rclcpp::detail::PublisherQosParametersTraits{});
if (options.enable_logger_service()) {
node_logging_->create_logger_services(node_services_);
}
}
Node::Node(
@@ -268,6 +344,7 @@ Node::Node(
Node::~Node()
{
// release sub-interfaces in an order that allows them to consult with node_base during tear-down
backport_members_.remove(this);
node_waitables_.reset();
node_time_source_.reset();
node_parameters_.reset();
@@ -587,6 +664,12 @@ Node::get_node_topics_interface()
return node_topics_;
}
rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
Node::get_node_type_descriptions_interface()
{
return backport_members_.get_node_type_descriptions_interface(this);
}
rclcpp::node_interfaces::NodeServicesInterface::SharedPtr
Node::get_node_services_interface()
{

View File

@@ -45,7 +45,7 @@ NodeBase::NodeBase(
node_handle_(nullptr),
default_callback_group_(default_callback_group),
associated_with_executor_(false),
notify_guard_condition_(context),
notify_guard_condition_(std::make_shared<rclcpp::GuardCondition>(context)),
notify_guard_condition_is_valid_(false)
{
// Create the rcl node and store it in a shared_ptr with a custom destructor.
@@ -132,8 +132,10 @@ NodeBase::NodeBase(
// Create the default callback group, if needed.
if (nullptr == default_callback_group_) {
using rclcpp::CallbackGroupType;
// Default callback group is mutually exclusive and automatically associated with
// any executors that this node is added to.
default_callback_group_ =
NodeBase::create_callback_group(CallbackGroupType::MutuallyExclusive);
NodeBase::create_callback_group(CallbackGroupType::MutuallyExclusive, true);
}
// Indicate the notify_guard_condition is now valid.
@@ -202,11 +204,27 @@ NodeBase::create_callback_group(
rclcpp::CallbackGroupType group_type,
bool automatically_add_to_executor_with_node)
{
auto weak_context = this->get_context()->weak_from_this();
auto get_node_context = [weak_context]() -> rclcpp::Context::SharedPtr {
return weak_context.lock();
};
auto group = std::make_shared<rclcpp::CallbackGroup>(
group_type,
get_node_context,
automatically_add_to_executor_with_node);
std::lock_guard<std::mutex> lock(callback_groups_mutex_);
callback_groups_.push_back(group);
// This guard condition is generally used to signal to this node's executor that a callback
// group has been added that should be considered for new entities.
// If this is creating the default callback group, then the notify guard condition won't be
// ready or needed yet, as the node is not done being constructed and therefore cannot be added.
// If the callback group is not automatically associated with this node's executors, then
// triggering the guard condition is also unnecessary, it will be manually added to an exector.
if (notify_guard_condition_is_valid_ && automatically_add_to_executor_with_node) {
this->trigger_notify_guard_condition();
}
return group;
}
@@ -253,9 +271,29 @@ NodeBase::get_notify_guard_condition()
if (!notify_guard_condition_is_valid_) {
throw std::runtime_error("failed to get notify guard condition because it is invalid");
}
return *notify_guard_condition_;
}
rclcpp::GuardCondition::SharedPtr
NodeBase::get_shared_notify_guard_condition()
{
std::lock_guard<std::recursive_mutex> notify_condition_lock(notify_guard_condition_mutex_);
if (!notify_guard_condition_is_valid_) {
return nullptr;
}
return notify_guard_condition_;
}
void
NodeBase::trigger_notify_guard_condition()
{
std::lock_guard<std::recursive_mutex> notify_condition_lock(notify_guard_condition_mutex_);
if (!notify_guard_condition_is_valid_) {
throw std::runtime_error("failed to trigger notify guard condition because it is invalid");
}
notify_guard_condition_->trigger();
}
bool
NodeBase::get_use_intra_process_default() const
{

View File

@@ -533,9 +533,8 @@ NodeGraph::notify_graph_change()
}
}
graph_cv_.notify_all();
auto & node_gc = node_base_->get_notify_guard_condition();
try {
node_gc.trigger();
node_base_->trigger_notify_guard_condition();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(
std::string("failed to notify wait set on graph change: ") + ex.what());
@@ -789,3 +788,15 @@ rclcpp::TopicEndpointInfo::qos_profile() const
{
return qos_profile_;
}
rosidl_type_hash_t &
rclcpp::TopicEndpointInfo::topic_type_hash()
{
return topic_type_hash_;
}
const rosidl_type_hash_t &
rclcpp::TopicEndpointInfo::topic_type_hash() const
{
return topic_type_hash_;
}

View File

@@ -12,11 +12,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include "rclcpp/node_impl.hpp"
#include "rclcpp/node_interfaces/node_logging.hpp"
#include "rclcpp/node_interfaces/node_services_interface.hpp"
using rclcpp::node_interfaces::NodeLogging;
NodeLogging::NodeLogging(rclcpp::node_interfaces::NodeBaseInterface * node_base)
NodeLogging::NodeLogging(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base)
: node_base_(node_base)
{
logger_ = rclcpp::get_logger(NodeLogging::get_logger_name());
@@ -37,3 +39,55 @@ NodeLogging::get_logger_name() const
{
return rcl_node_get_logger_name(node_base_->get_rcl_node_handle());
}
void NodeLogging::create_logger_services(
node_interfaces::NodeServicesInterface::SharedPtr node_services)
{
rclcpp::ServicesQoS qos_profile;
const std::string node_name = node_base_->get_name();
auto callback_group = node_base_->get_default_callback_group();
get_loggers_service_ = rclcpp::create_service<rcl_interfaces::srv::GetLoggerLevels>(
node_base_, node_services,
node_name + "/get_logger_levels",
[](
const std::shared_ptr<rmw_request_id_t>,
const std::shared_ptr<rcl_interfaces::srv::GetLoggerLevels::Request> request,
std::shared_ptr<rcl_interfaces::srv::GetLoggerLevels::Response> response)
{
for (auto & name : request->names) {
rcl_interfaces::msg::LoggerLevel logger_level;
logger_level.name = name;
auto ret = rcutils_logging_get_logger_level(name.c_str());
if (ret < 0) {
logger_level.level = 0;
} else {
logger_level.level = static_cast<uint8_t>(ret);
}
response->levels.push_back(std::move(logger_level));
}
},
qos_profile, callback_group);
set_loggers_service_ = rclcpp::create_service<rcl_interfaces::srv::SetLoggerLevels>(
node_base_, node_services,
node_name + "/set_logger_levels",
[](
const std::shared_ptr<rmw_request_id_t>,
const std::shared_ptr<rcl_interfaces::srv::SetLoggerLevels::Request> request,
std::shared_ptr<rcl_interfaces::srv::SetLoggerLevels::Response> response)
{
rcl_interfaces::msg::SetLoggerLevelsResult result;
for (auto & level : request->levels) {
auto ret = rcutils_logging_set_logger_level(level.name.c_str(), level.level);
if (ret != RCUTILS_RET_OK) {
result.successful = false;
result.reason = rcutils_get_error_string().str;
} else {
result.successful = true;
}
response->results.push_back(std::move(result));
}
},
qos_profile, callback_group);
}

View File

@@ -651,7 +651,7 @@ NodeParameters::undeclare_parameter(const std::string & name)
}
if (!parameter_info->second.descriptor.dynamic_typing) {
throw rclcpp::exceptions::InvalidParameterTypeException{
name, "cannot undeclare an statically typed parameter"};
name, "cannot undeclare a statically typed parameter"};
}
parameters_.erase(parameter_info);
@@ -824,7 +824,7 @@ NodeParameters::set_parameters_atomically(const std::vector<rclcpp::Parameter> &
auto it = parameters_.find(parameter.get_name());
if (it != parameters_.end() && rclcpp::PARAMETER_NOT_SET != it->second.value.get_type()) {
if (!it->second.descriptor.dynamic_typing) {
result.reason = "cannot undeclare an statically typed parameter";
result.reason = "cannot undeclare a statically typed parameter";
result.successful = false;
return result;
}

View File

@@ -42,9 +42,8 @@ NodeServices::add_service(
group->add_service(service_base_ptr);
// Notify the executor that a new service was created using the parent Node.
auto & node_gc = node_base_->get_notify_guard_condition();
try {
node_gc.trigger();
node_base_->trigger_notify_guard_condition();
group->trigger_notify_guard_condition();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(
@@ -69,9 +68,8 @@ NodeServices::add_client(
group->add_client(client_base_ptr);
// Notify the executor that a new client was created using the parent Node.
auto & node_gc = node_base_->get_notify_guard_condition();
try {
node_gc.trigger();
node_base_->trigger_notify_guard_condition();
group->trigger_notify_guard_condition();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(

View File

@@ -42,9 +42,8 @@ NodeTimers::add_timer(
}
callback_group->add_timer(timer);
auto & node_gc = node_base_->get_notify_guard_condition();
try {
node_gc.trigger();
node_base_->trigger_notify_guard_condition();
callback_group->trigger_notify_guard_condition();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(

View File

@@ -70,9 +70,8 @@ NodeTopics::add_publisher(
}
// Notify the executor that a new publisher was created using the parent Node.
auto & node_gc = node_base_->get_notify_guard_condition();
try {
node_gc.trigger();
node_base_->trigger_notify_guard_condition();
callback_group->trigger_notify_guard_condition();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(
@@ -119,9 +118,8 @@ NodeTopics::add_subscription(
}
// Notify the executor that a new subscription was created using the parent Node.
auto & node_gc = node_base_->get_notify_guard_condition();
try {
node_gc.trigger();
node_base_->trigger_notify_guard_condition();
callback_group->trigger_notify_guard_condition();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(

View File

@@ -0,0 +1,157 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <memory>
#include <string>
#include <thread>
#include "rclcpp/node_interfaces/node_type_descriptions.hpp"
#include "rclcpp/parameter_client.hpp"
#include "type_description_interfaces/srv/get_type_description.h"
namespace
{
// Helper wrapper for rclcpp::Service to access ::Request and ::Response types for allocation.
struct GetTypeDescription__C
{
using Request = type_description_interfaces__srv__GetTypeDescription_Request;
using Response = type_description_interfaces__srv__GetTypeDescription_Response;
using Event = type_description_interfaces__srv__GetTypeDescription_Event;
};
} // namespace
// Helper function for C typesupport.
namespace rosidl_typesupport_cpp
{
template<>
rosidl_service_type_support_t const *
get_service_type_support_handle<GetTypeDescription__C>()
{
return ROSIDL_GET_SRV_TYPE_SUPPORT(type_description_interfaces, srv, GetTypeDescription);
}
} // namespace rosidl_typesupport_cpp
namespace rclcpp
{
namespace node_interfaces
{
class NodeTypeDescriptions::NodeTypeDescriptionsImpl
{
public:
using ServiceT = GetTypeDescription__C;
rclcpp::Logger logger_;
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_;
rclcpp::Service<ServiceT>::SharedPtr type_description_srv_;
NodeTypeDescriptionsImpl(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base,
rclcpp::node_interfaces::NodeLoggingInterface::SharedPtr node_logging,
rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters,
rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services)
: logger_(node_logging->get_logger()),
node_base_(node_base)
{
const std::string enable_param_name = "start_type_description_service";
bool enabled = false;
try {
auto enable_param = node_parameters->declare_parameter(
enable_param_name,
rclcpp::ParameterValue(true),
rcl_interfaces::msg::ParameterDescriptor()
.set__name(enable_param_name)
.set__type(rclcpp::PARAMETER_BOOL)
.set__description("Start the ~/get_type_description service for this node.")
.set__read_only(true));
enabled = enable_param.get<bool>();
} catch (const rclcpp::exceptions::InvalidParameterTypeException & exc) {
RCLCPP_ERROR(logger_, "%s", exc.what());
throw;
}
if (enabled) {
auto rcl_node = node_base->get_rcl_node_handle();
rcl_ret_t rcl_ret = rcl_node_type_description_service_init(rcl_node);
if (rcl_ret != RCL_RET_OK) {
RCLCPP_ERROR(
logger_, "Failed to initialize ~/get_type_description_service: %s",
rcl_get_error_string().str);
throw std::runtime_error(
"Failed to initialize ~/get_type_description service.");
}
rcl_service_t * rcl_srv = nullptr;
rcl_ret = rcl_node_get_type_description_service(rcl_node, &rcl_srv);
if (rcl_ret != RCL_RET_OK) {
throw std::runtime_error(
"Failed to get initialized ~/get_type_description service from rcl.");
}
rclcpp::AnyServiceCallback<ServiceT> cb;
cb.set(
[this](
std::shared_ptr<rmw_request_id_t> header,
std::shared_ptr<ServiceT::Request> request,
std::shared_ptr<ServiceT::Response> response
) {
rcl_node_type_description_service_handle_request(
node_base_->get_rcl_node_handle(),
header.get(),
request.get(),
response.get());
});
type_description_srv_ = std::make_shared<Service<ServiceT>>(
node_base_->get_shared_rcl_node_handle(),
rcl_srv,
cb);
node_services->add_service(
std::dynamic_pointer_cast<ServiceBase>(type_description_srv_),
nullptr);
}
}
~NodeTypeDescriptionsImpl()
{
if (
type_description_srv_ &&
RCL_RET_OK != rcl_node_type_description_service_fini(node_base_->get_rcl_node_handle()))
{
RCLCPP_ERROR(
logger_,
"Error in shutdown of get_type_description service: %s", rcl_get_error_string().str);
}
}
};
NodeTypeDescriptions::NodeTypeDescriptions(
rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base,
rclcpp::node_interfaces::NodeLoggingInterface::SharedPtr node_logging,
rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters,
rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services)
: impl_(new NodeTypeDescriptionsImpl(
node_base,
node_logging,
node_parameters,
node_services))
{}
NodeTypeDescriptions::~NodeTypeDescriptions()
{}
} // namespace node_interfaces
} // namespace rclcpp

View File

@@ -42,9 +42,8 @@ NodeWaitables::add_waitable(
group->add_waitable(waitable_ptr);
// Notify the executor that a new waitable was created using the parent Node.
auto & node_gc = node_base_->get_notify_guard_condition();
try {
node_gc.trigger();
node_base_->trigger_notify_guard_condition();
group->trigger_notify_guard_condition();
} catch (const rclcpp::exceptions::RCLError & ex) {
throw std::runtime_error(

View File

@@ -248,6 +248,19 @@ NodeOptions::start_parameter_services(bool start_parameter_services)
return *this;
}
bool
NodeOptions::enable_logger_service() const
{
return this->enable_logger_service_;
}
NodeOptions &
NodeOptions::enable_logger_service(bool enable_logger_service)
{
this->enable_logger_service_ = enable_logger_service;
return *this;
}
bool
NodeOptions::start_parameter_event_publisher() const
{

View File

@@ -113,7 +113,7 @@ SignalHandler::get_logger()
SignalHandler &
SignalHandler::get_global_signal_handler()
{
static SignalHandler signal_handler;
static SignalHandler & signal_handler = *new SignalHandler();
return signal_handler;
}

View File

@@ -22,6 +22,7 @@
#include "rcpputils/scope_exit.hpp"
#include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
#include "rclcpp/exceptions.hpp"
#include "rclcpp/expand_topic_or_service_name.hpp"
#include "rclcpp/experimental/intra_process_manager.hpp"
@@ -32,6 +33,8 @@
#include "rmw/error_handling.h"
#include "rmw/rmw.h"
#include "rosidl_dynamic_typesupport/types.h"
using rclcpp::SubscriptionBase;
SubscriptionBase::SubscriptionBase(
@@ -41,7 +44,7 @@ SubscriptionBase::SubscriptionBase(
const rcl_subscription_options_t & subscription_options,
const SubscriptionEventCallbacks & event_callbacks,
bool use_default_callbacks,
bool is_serialized)
DeliveredMessageKind delivered_message_kind)
: node_base_(node_base),
node_handle_(node_base_->get_shared_rcl_node_handle()),
node_logger_(rclcpp::get_node_logger(node_handle_.get())),
@@ -49,7 +52,7 @@ SubscriptionBase::SubscriptionBase(
intra_process_subscription_id_(0),
event_callbacks_(event_callbacks),
type_support_(type_support_handle),
is_serialized_(is_serialized)
delivered_message_kind_(delivered_message_kind)
{
auto custom_deletor = [node_handle = this->node_handle_](rcl_subscription_t * rcl_subs)
{
@@ -258,7 +261,13 @@ SubscriptionBase::get_message_type_support_handle() const
bool
SubscriptionBase::is_serialized() const
{
return is_serialized_;
return delivered_message_kind_ == rclcpp::DeliveredMessageKind::SERIALIZED_MESSAGE;
}
rclcpp::DeliveredMessageKind
SubscriptionBase::get_delivered_message_kind() const
{
return delivered_message_kind_;
}
size_t
@@ -289,7 +298,20 @@ SubscriptionBase::setup_intra_process(
bool
SubscriptionBase::can_loan_messages() const
{
return rcl_subscription_can_loan_messages(subscription_handle_.get());
bool retval = rcl_subscription_can_loan_messages(subscription_handle_.get());
if (retval) {
// TODO(clalancette): The loaned message interface is currently not safe to use with
// shared_ptr callbacks. If a user takes a copy of the shared_ptr, it can get freed from
// underneath them via rcl_return_loaned_message_from_subscription(). The correct solution is
// to return the loaned message in a custom deleter, but that needs to be carefully handled
// with locking. Warn the user about this until we fix it.
RCLCPP_WARN_ONCE(
this->node_logger_,
"Loaned messages are only safe with const ref subscription callbacks. "
"If you are using any other kind of subscriptions, "
"set the ROS_DISABLE_LOANED_MESSAGES environment variable to 1 (the default).");
}
return retval;
}
rclcpp::Waitable::SharedPtr
@@ -442,8 +464,7 @@ SubscriptionBase::set_content_filter(
rcl_subscription_content_filter_options_t options =
rcl_get_zero_initialized_subscription_content_filter_options();
std::vector<const char *> cstrings =
get_c_vector_string(expression_parameters);
std::vector<const char *> cstrings = get_c_vector_string(expression_parameters);
rcl_ret_t ret = rcl_subscription_content_filter_options_init(
subscription_handle_.get(),
get_c_string(filter_expression),
@@ -515,3 +536,14 @@ SubscriptionBase::get_content_filter() const
return ret_options;
}
// DYNAMIC TYPE ==================================================================================
bool
SubscriptionBase::take_dynamic_message(
rclcpp::dynamic_typesupport::DynamicMessage & /*message_out*/,
rclcpp::MessageInfo & /*message_info_out*/)
{
throw std::runtime_error("Unimplemented");
return false;
}

View File

@@ -54,9 +54,7 @@ public:
ros_time_active_ = true;
// Update all attached clocks to zero or last recorded time
for (auto it = associated_clocks_.begin(); it != associated_clocks_.end(); ++it) {
set_clock(last_time_msg_, true, *it);
}
set_all_clocks(last_time_msg_, true);
}
// An internal method to use in the clock callback that iterates and disables all clocks
@@ -71,11 +69,8 @@ public:
ros_time_active_ = false;
// Update all attached clocks
std::lock_guard<std::mutex> guard(clock_list_lock_);
for (auto it = associated_clocks_.begin(); it != associated_clocks_.end(); ++it) {
auto msg = std::make_shared<builtin_interfaces::msg::Time>();
set_clock(msg, false, *it);
}
auto msg = std::make_shared<builtin_interfaces::msg::Time>();
set_all_clocks(msg, false);
}
// Check if ROS time is active

View File

@@ -262,6 +262,11 @@ if(TARGET test_node_interfaces__node_topics)
"test_msgs")
target_link_libraries(test_node_interfaces__node_topics ${PROJECT_NAME} mimick)
endif()
ament_add_gtest(test_node_interfaces__node_type_descriptions
node_interfaces/test_node_type_descriptions.cpp)
if(TARGET test_node_interfaces__node_type_descriptions)
target_link_libraries(test_node_interfaces__node_type_descriptions ${PROJECT_NAME} mimick)
endif()
ament_add_gtest(test_node_interfaces__node_waitables
node_interfaces/test_node_waitables.cpp)
if(TARGET test_node_interfaces__node_waitables)
@@ -649,6 +654,13 @@ if(TARGET test_wait_for_message)
target_link_libraries(test_wait_for_message ${PROJECT_NAME})
endif()
ament_add_gtest(test_logger_service test_logger_service.cpp)
if(TARGET test_logger_service)
ament_target_dependencies(test_logger_service
"rcl_interfaces")
target_link_libraries(test_logger_service ${PROJECT_NAME})
endif()
ament_add_gtest(test_interface_traits test_interface_traits.cpp
APPEND_LIBRARY_DIRS "${append_library_dirs}")
if(TARGET test_interface_traits)
@@ -696,8 +708,25 @@ if(TARGET test_static_executor_entities_collector)
target_link_libraries(test_static_executor_entities_collector ${PROJECT_NAME} mimick)
endif()
ament_add_gtest(test_events_executor executors/test_events_executor.cpp
APPEND_LIBRARY_DIRS "${append_library_dirs}")
ament_add_gtest(test_entities_collector executors/test_entities_collector.cpp
APPEND_LIBRARY_DIRS "${append_library_dirs}" TIMEOUT 120)
if(TARGET test_entities_collector)
ament_target_dependencies(test_entities_collector
"rcl"
"test_msgs")
target_link_libraries(test_entities_collector ${PROJECT_NAME} mimick)
endif()
ament_add_gtest(test_executor_notify_waitable executors/test_executor_notify_waitable.cpp
APPEND_LIBRARY_DIRS "${append_library_dirs}" TIMEOUT 120)
if(TARGET test_executor_notify_waitable)
ament_target_dependencies(test_executor_notify_waitable
"rcl"
"test_msgs")
target_link_libraries(test_executor_notify_waitable ${PROJECT_NAME} mimick)
endif()
ament_add_gtest(test_events_executor executors/test_events_executor.cpp TIMEOUT 5)
if(TARGET test_events_executor)
ament_target_dependencies(test_events_executor
"test_msgs")

View File

@@ -0,0 +1,320 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <gtest/gtest.h>
#include "rclcpp/executors/executor_notify_waitable.hpp"
#include "rclcpp/rclcpp.hpp"
#include "rclcpp/executors/executor_entities_collector.hpp"
#include "../../utils/rclcpp_gtest_macros.hpp"
class TestExecutorEntitiesCollector : public ::testing::Test
{
public:
void SetUp()
{
rclcpp::init(0, nullptr);
notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
entities_collector = std::make_shared<rclcpp::executors::ExecutorEntitiesCollector>(
notify_waitable);
}
void TearDown()
{
rclcpp::shutdown();
}
std::shared_ptr<rclcpp::executors::ExecutorNotifyWaitable> notify_waitable;
std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollector> entities_collector;
};
TEST_F(TestExecutorEntitiesCollector, add_remove_node) {
auto node1 = std::make_shared<rclcpp::Node>("node1", "ns");
// Add a node
EXPECT_NO_THROW(entities_collector->add_node(node1->get_node_base_interface()));
EXPECT_NO_THROW(entities_collector->update_collections());
// Remove a node
EXPECT_NO_THROW(entities_collector->remove_node(node1->get_node_base_interface()));
EXPECT_NO_THROW(entities_collector->update_collections());
}
TEST_F(TestExecutorEntitiesCollector, add_node_twice) {
auto node1 = std::make_shared<rclcpp::Node>("node1", "ns");
EXPECT_NO_THROW(entities_collector->add_node(node1->get_node_base_interface()));
RCLCPP_EXPECT_THROW_EQ(
entities_collector->add_node(node1->get_node_base_interface()),
std::runtime_error("Node '/ns/node1' has already been added to an executor."));
EXPECT_NO_THROW(entities_collector->update_collections());
}
TEST_F(TestExecutorEntitiesCollector, add_associated_node) {
auto node1 = std::make_shared<rclcpp::Node>("node1", "ns");
// Simulate node being associated somewhere else
auto & has_executor = node1->get_node_base_interface()->get_associated_with_executor_atomic();
has_executor.store(true);
// Add an already-associated node
RCLCPP_EXPECT_THROW_EQ(
entities_collector->remove_node(node1->get_node_base_interface()),
std::runtime_error("Node '/ns/node1' needs to be associated with this executor."));
has_executor.store(false);
}
TEST_F(TestExecutorEntitiesCollector, remove_unassociated_node) {
auto node1 = std::make_shared<rclcpp::Node>("node1", "ns");
// Add an already-associated node
RCLCPP_EXPECT_THROW_EQ(
entities_collector->remove_node(node1->get_node_base_interface()),
std::runtime_error("Node '/ns/node1' needs to be associated with an executor."));
// Simulate node being associated somewhere else
auto & has_executor = node1->get_node_base_interface()->get_associated_with_executor_atomic();
has_executor.store(true);
// Add an already-associated node
RCLCPP_EXPECT_THROW_EQ(
entities_collector->remove_node(node1->get_node_base_interface()),
std::runtime_error("Node '/ns/node1' needs to be associated with this executor."));
}
TEST_F(TestExecutorEntitiesCollector, add_remove_node_before_update) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node1 = std::make_shared<rclcpp::Node>("node1", "ns");
auto node2 = std::make_shared<rclcpp::Node>("node2", "ns");
// Add and remove nodes without running updatenode
EXPECT_NO_THROW(entities_collector.add_node(node1->get_node_base_interface()));
EXPECT_NO_THROW(entities_collector.add_node(node2->get_node_base_interface()));
EXPECT_NO_THROW(entities_collector.remove_node(node1->get_node_base_interface()));
EXPECT_NO_THROW(entities_collector.remove_node(node2->get_node_base_interface()));
EXPECT_NO_THROW(entities_collector.update_collections());
}
TEST_F(TestExecutorEntitiesCollector, add_callback_group) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node = std::make_shared<rclcpp::Node>("node1", "ns");
rclcpp::CallbackGroup::SharedPtr cb_group = node->create_callback_group(
rclcpp::CallbackGroupType::MutuallyExclusive);
// Add a callback group and update
entities_collector.add_callback_group(cb_group);
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
// Remove callback group and update
entities_collector.remove_callback_group(cb_group);
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
}
TEST_F(TestExecutorEntitiesCollector, add_node_default_callback_group) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node = std::make_shared<rclcpp::Node>("node1", "ns");
entities_collector.add_node(node->get_node_base_interface());
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 1u);
}
TEST_F(TestExecutorEntitiesCollector, add_callback_group_after_add_node) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node = std::make_shared<rclcpp::Node>("node1", "ns");
rclcpp::CallbackGroup::SharedPtr cb_group = node->create_callback_group(
rclcpp::CallbackGroupType::MutuallyExclusive);
entities_collector.add_node(node->get_node_base_interface());
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 2u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 2u);
RCLCPP_EXPECT_THROW_EQ(
entities_collector.add_callback_group(cb_group),
std::runtime_error("Callback group has already been added to an executor."));
}
TEST_F(TestExecutorEntitiesCollector, add_callback_group_twice) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node = std::make_shared<rclcpp::Node>("node1", "ns");
rclcpp::CallbackGroup::SharedPtr cb_group = node->create_callback_group(
rclcpp::CallbackGroupType::MutuallyExclusive);
entities_collector.add_callback_group(cb_group);
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
cb_group->get_associated_with_executor_atomic().exchange(false);
RCLCPP_EXPECT_THROW_EQ(
entities_collector.add_callback_group(cb_group),
std::runtime_error("Callback group has already been added to this executor."));
}
TEST_F(TestExecutorEntitiesCollector, remove_callback_group_after_node) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node = std::make_shared<rclcpp::Node>("node1", "ns");
rclcpp::CallbackGroup::SharedPtr cb_group = node->create_callback_group(
rclcpp::CallbackGroupType::MutuallyExclusive);
entities_collector.add_callback_group(cb_group);
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
node.reset();
/**
* TODO(mjcarroll): Assert this when we are enforcing that nodes must be destroyed
* after their created callback groups.
RCLCPP_EXPECT_THROW_EQ(
entities_collector.remove_callback_group(cb_group),
std::runtime_error("Node must not be deleted before its callback group(s)."));
*/
EXPECT_NO_THROW(entities_collector.update_collections());
}
TEST_F(TestExecutorEntitiesCollector, remove_callback_group_after_node2) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node = std::make_shared<rclcpp::Node>("node1", "ns");
rclcpp::CallbackGroup::SharedPtr cb_group = node->create_callback_group(
rclcpp::CallbackGroupType::MutuallyExclusive);
entities_collector.add_callback_group(cb_group);
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 0u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
EXPECT_NO_THROW(entities_collector.remove_callback_group(cb_group));
node.reset();
/**
* TODO(mjcarroll): Assert this when we are enforcing that nodes must be destroyed
* after their created callback groups.
RCLCPP_EXPECT_THROW_EQ(
entities_collector.remove_callback_group(cb_group),
std::runtime_error("Node must not be deleted before its callback group(s)."));
*/
EXPECT_NO_THROW(entities_collector.update_collections());
}
TEST_F(TestExecutorEntitiesCollector, remove_callback_group_twice) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node = std::make_shared<rclcpp::Node>("node1", "ns");
rclcpp::CallbackGroup::SharedPtr cb_group = node->create_callback_group(
rclcpp::CallbackGroupType::MutuallyExclusive);
entities_collector.add_callback_group(cb_group);
entities_collector.update_collections();
ASSERT_EQ(entities_collector.get_all_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_manually_added_callback_groups().size(), 1u);
ASSERT_EQ(entities_collector.get_automatically_added_callback_groups().size(), 0u);
entities_collector.remove_callback_group(cb_group);
entities_collector.update_collections();
RCLCPP_EXPECT_THROW_EQ(
entities_collector.remove_callback_group(cb_group),
std::runtime_error("Callback group needs to be associated with an executor."));
}
TEST_F(TestExecutorEntitiesCollector, remove_node_opposite_order) {
auto notify_waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
auto entities_collector = rclcpp::executors::ExecutorEntitiesCollector(notify_waitable);
auto node1 = std::make_shared<rclcpp::Node>("node1", "ns");
EXPECT_NO_THROW(entities_collector.add_node(node1->get_node_base_interface()));
auto node2 = std::make_shared<rclcpp::Node>("node2", "ns");
EXPECT_NO_THROW(entities_collector.add_node(node2->get_node_base_interface()));
EXPECT_NO_THROW(entities_collector.remove_node(node2->get_node_base_interface()));
}

View File

@@ -41,6 +41,58 @@ public:
}
};
TEST_F(TestEventsExecutor, run_pub_sub)
{
// rmw_connextdds doesn't support events-executor
if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
GTEST_SKIP();
}
auto node = std::make_shared<rclcpp::Node>("node");
bool msg_received = false;
auto subscription = node->create_subscription<test_msgs::msg::Empty>(
"topic", rclcpp::SensorDataQoS(),
[&msg_received](test_msgs::msg::Empty::ConstSharedPtr msg)
{
(void)msg;
msg_received = true;
});
auto publisher = node->create_publisher<test_msgs::msg::Empty>("topic", rclcpp::SensorDataQoS());
EventsExecutor executor;
executor.add_node(node);
bool spin_exited = false;
std::thread spinner([&spin_exited, &executor, this]() {
executor.spin();
spin_exited = true;
});
auto msg = std::make_unique<test_msgs::msg::Empty>();
publisher->publish(std::move(msg));
// Wait some time for the subscription to receive the message
auto start = std::chrono::high_resolution_clock::now();
while (
!msg_received &&
!spin_exited &&
(std::chrono::high_resolution_clock::now() - start < 1s))
{
auto time = std::chrono::high_resolution_clock::now() - start;
auto time_msec = std::chrono::duration_cast<std::chrono::milliseconds>(time);
std::this_thread::sleep_for(25ms);
}
executor.cancel();
spinner.join();
executor.remove_node(node);
EXPECT_TRUE(msg_received);
EXPECT_TRUE(spin_exited);
}
TEST_F(TestEventsExecutor, run_clients_servers)
{
// rmw_connextdds doesn't support events-executor
@@ -136,8 +188,6 @@ TEST_F(TestEventsExecutor, spin_once_max_duration_timeout)
EXPECT_TRUE(std::chrono::steady_clock::now() - start < 200ms);
}
// FIX THIS TEST! The entities collector is being called too many times!
/*
TEST_F(TestEventsExecutor, spin_once_max_duration_timer)
{
// rmw_connextdds doesn't support events-executor
@@ -173,7 +223,6 @@ TEST_F(TestEventsExecutor, spin_once_max_duration_timer)
EXPECT_EQ(1u, t_runs);
EXPECT_TRUE(std::chrono::steady_clock::now() - start < 200ms);
}
*/
TEST_F(TestEventsExecutor, spin_some_max_duration)
{
@@ -383,6 +432,9 @@ TEST_F(TestEventsExecutor, cancel_while_timers_waiting)
TEST_F(TestEventsExecutor, destroy_entities)
{
// This test fails on Windows! We skip it for now
GTEST_SKIP();
// rmw_connextdds doesn't support events-executor
if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
GTEST_SKIP();
@@ -427,9 +479,7 @@ TEST_F(TestEventsExecutor, destroy_entities)
spinner.join();
}
/*
Testing construction of a subscriptions with QoS event callback functions.
*/
// Testing construction of a subscriptions with QoS event callback functions.
std::string * g_pub_log_msg;
std::string * g_sub_log_msg;
std::promise<void> * g_log_msgs_promise;

View File

@@ -75,8 +75,8 @@ TEST(TestEventsQueue, SimpleQueueTest)
simple_queue->enqueue(push_event);
ret = simple_queue->dequeue(event);
EXPECT_TRUE(ret);
EXPECT_EQ(push_event.exec_entity_id, event.exec_entity_id);
EXPECT_EQ(push_event.gen_entity_id, event.gen_entity_id);
EXPECT_EQ(push_event.entity_key, event.entity_key);
EXPECT_EQ(push_event.waitable_data, event.waitable_data);
EXPECT_EQ(push_event.type, event.type);
EXPECT_EQ(push_event.num_events, event.num_events);
}

View File

@@ -0,0 +1,97 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <gtest/gtest.h>
#include <chrono>
#include <memory>
#include <stdexcept>
#include "rclcpp/executors/single_threaded_executor.hpp"
#include "rclcpp/rclcpp.hpp"
#include "rcpputils/scope_exit.hpp"
#include "rclcpp/executors/executor_notify_waitable.hpp"
#include "../../utils/rclcpp_gtest_macros.hpp"
class TestExecutorNotifyWaitable : public ::testing::Test
{
public:
void SetUp()
{
rclcpp::init(0, nullptr);
}
void TearDown()
{
rclcpp::shutdown();
}
};
TEST_F(TestExecutorNotifyWaitable, construct_destruct) {
{
auto waitable = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>();
waitable.reset();
}
{
auto on_execute_callback = []() {};
auto waitable =
std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(on_execute_callback);
waitable.reset();
}
}
TEST_F(TestExecutorNotifyWaitable, add_remove_guard_conditions) {
auto on_execute_callback = []() {};
auto waitable =
std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(on_execute_callback);
auto node = std::make_shared<rclcpp::Node>("my_node", "/ns");
auto notify_guard_condition =
node->get_node_base_interface()->get_shared_notify_guard_condition();
EXPECT_NO_THROW(waitable->add_guard_condition(notify_guard_condition));
EXPECT_NO_THROW(waitable->remove_guard_condition(notify_guard_condition));
}
TEST_F(TestExecutorNotifyWaitable, wait) {
int on_execute_calls = 0;
auto on_execute_callback = [&on_execute_calls]() {on_execute_calls++;};
auto waitable =
std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(on_execute_callback);
auto node = std::make_shared<rclcpp::Node>("my_node", "/ns");
auto notify_guard_condition =
node->get_node_base_interface()->get_shared_notify_guard_condition();
EXPECT_NO_THROW(waitable->add_guard_condition(notify_guard_condition));
auto default_cbg = node->get_node_base_interface()->get_default_callback_group();
ASSERT_NE(nullptr, default_cbg->get_notify_guard_condition());
auto waitables = node->get_node_waitables_interface();
waitables->add_waitable(std::static_pointer_cast<rclcpp::Waitable>(waitable), default_cbg);
rclcpp::executors::SingleThreadedExecutor executor;
executor.add_node(node);
executor.spin_all(std::chrono::seconds(1));
EXPECT_EQ(1u, on_execute_calls);
// on_execute_callback doesn't change if the topology doesn't change
executor.spin_all(std::chrono::seconds(1));
EXPECT_EQ(1u, on_execute_calls);
}

View File

@@ -20,12 +20,15 @@
#include <gtest/gtest.h>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <future>
#include <limits>
#include <memory>
#include <string>
#include <thread>
#include <utility>
#include <vector>
#include "rcl/error_handling.h"
#include "rcl/time.h"
@@ -43,18 +46,10 @@ template<typename T>
class TestExecutors : public ::testing::Test
{
public:
static void SetUpTestCase()
{
rclcpp::init(0, nullptr);
}
static void TearDownTestCase()
{
rclcpp::shutdown();
}
void SetUp()
{
rclcpp::init(0, nullptr);
const auto test_info = ::testing::UnitTest::GetInstance()->current_test_info();
std::stringstream test_name;
test_name << test_info->test_case_name() << "_" << test_info->name();
@@ -75,6 +70,8 @@ public:
publisher.reset();
subscription.reset();
node.reset();
rclcpp::shutdown();
}
rclcpp::Node::SharedPtr node;
@@ -187,6 +184,25 @@ TYPED_TEST(TestExecutorsStable, addTemporaryNode)
spinner.join();
}
// Make sure that a spinning empty executor can be cancelled
TYPED_TEST(TestExecutors, emptyExecutor)
{
using ExecutorType = TypeParam;
// rmw_connextdds doesn't support events-executor
if (
std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
{
GTEST_SKIP();
}
ExecutorType executor;
std::thread spinner([&]() {EXPECT_NO_THROW(executor.spin());});
std::this_thread::sleep_for(50ms);
executor.cancel();
spinner.join();
}
// Check executor throws properly if the same node is added a second time
TYPED_TEST(TestExecutors, addNodeTwoExecutors)
{
@@ -599,8 +615,9 @@ TYPED_TEST(TestExecutors, spinSome)
this->publisher->publish(test_msgs::msg::Empty());
std::this_thread::sleep_for(1ms);
}
EXPECT_EQ(1u, my_waitable->get_count());
// The count of "execute" depends on whether the executor starts spinning before (1) or after (0)
// the first iteration of the while loop
EXPECT_LE(1u, my_waitable->get_count());
waitable_interfaces->remove_waitable(my_waitable, nullptr);
EXPECT_TRUE(spin_exited);
// Cancel if it hasn't exited already.
@@ -709,6 +726,77 @@ TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteInterrupted)
spinner.join();
}
// This test verifies that the add_node operation is robust wrt race conditions.
// It's mostly meant to prevent regressions in the events-executor, but the operation should be
// thread-safe in all executor implementations.
// The initial implementation of the events-executor contained a bug where the executor
// would end up in an inconsistent state and stop processing interrupt/shutdown notifications.
// Manually adding a node to the executor results in a) producing a notify waitable event
// and b) refreshing the executor collections.
// The inconsistent state would happen if the event was processed before the collections were
// finished to be refreshed: the executor would pick up the event but be unable to process it.
// This would leave the `notify_waitable_event_pushed_` flag to true, preventing additional
// notify waitable events to be pushed.
// The behavior is observable only under heavy load, so this test spawns several worker
// threads. Due to the nature of the bug, this test may still succeed even if the
// bug is present. However repeated runs will show its flakiness nature and indicate
// an eventual regression.
TYPED_TEST(TestExecutors, testRaceConditionAddNode)
{
using ExecutorType = TypeParam;
// rmw_connextdds doesn't support events-executor
if (
std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
{
GTEST_SKIP();
}
// Spawn some threads to do some heavy work
std::atomic<bool> should_cancel = false;
std::vector<std::thread> stress_threads;
for (size_t i = 0; i < 5 * std::thread::hardware_concurrency(); i++) {
stress_threads.emplace_back(
[&should_cancel, i]() {
// This is just some arbitrary heavy work
volatile size_t total = 0;
for (size_t k = 0; k < 549528914167; k++) {
if (should_cancel) {
break;
}
total += k * (i + 42);
}
});
}
// Create an executor
auto executor = std::make_shared<ExecutorType>();
// Start spinning
auto executor_thread = std::thread(
[executor]() {
executor->spin();
});
// Add a node to the executor
executor->add_node(this->node);
// Cancel the executor (make sure that it's already spinning first)
while (!executor->is_spinning() && rclcpp::ok()) {
continue;
}
executor->cancel();
// Try to join the thread after cancelling the executor
// This is the "test". We want to make sure that we can still cancel the executor
// regardless of the presence of race conditions
executor_thread.join();
// The test is now completed: we can join the stress threads
should_cancel = true;
for (auto & t : stress_threads) {
t.join();
}
}
// Check spin_until_future_complete with node base pointer (instantiates its own executor)
TEST(TestExecutors, testSpinUntilFutureCompleteNodeBasePtr)
{
@@ -749,3 +837,134 @@ TEST(TestExecutors, testSpinUntilFutureCompleteNodePtr)
rclcpp::shutdown();
}
template<typename T>
class TestIntraprocessExecutors : public ::testing::Test
{
public:
static void SetUpTestCase()
{
rclcpp::init(0, nullptr);
}
static void TearDownTestCase()
{
rclcpp::shutdown();
}
void SetUp()
{
const auto test_info = ::testing::UnitTest::GetInstance()->current_test_info();
std::stringstream test_name;
test_name << test_info->test_case_name() << "_" << test_info->name();
node = std::make_shared<rclcpp::Node>("node", test_name.str());
callback_count = 0u;
const std::string topic_name = std::string("topic_") + test_name.str();
rclcpp::PublisherOptions po;
po.use_intra_process_comm = rclcpp::IntraProcessSetting::Enable;
publisher = node->create_publisher<test_msgs::msg::Empty>(topic_name, rclcpp::QoS(1), po);
auto callback = [this](test_msgs::msg::Empty::ConstSharedPtr) {
this->callback_count.fetch_add(1u);
};
rclcpp::SubscriptionOptions so;
so.use_intra_process_comm = rclcpp::IntraProcessSetting::Enable;
subscription =
node->create_subscription<test_msgs::msg::Empty>(
topic_name, rclcpp::QoS(kNumMessages), std::move(callback), so);
}
void TearDown()
{
publisher.reset();
subscription.reset();
node.reset();
}
const size_t kNumMessages = 100;
rclcpp::Node::SharedPtr node;
rclcpp::Publisher<test_msgs::msg::Empty>::SharedPtr publisher;
rclcpp::Subscription<test_msgs::msg::Empty>::SharedPtr subscription;
std::atomic_size_t callback_count;
};
TYPED_TEST_SUITE(TestIntraprocessExecutors, ExecutorTypes, ExecutorTypeNames);
TYPED_TEST(TestIntraprocessExecutors, testIntraprocessRetrigger) {
// This tests that executors will continue to service intraprocess subscriptions in the case
// that publishers aren't continuing to publish.
// This was previously broken in that intraprocess guard conditions were only triggered on
// publish and the test was added to prevent future regressions.
const size_t kNumMessages = 100;
using ExecutorType = TypeParam;
ExecutorType executor;
executor.add_node(this->node);
EXPECT_EQ(0u, this->callback_count.load());
this->publisher->publish(test_msgs::msg::Empty());
// Wait for up to 5 seconds for the first message to come available.
const std::chrono::milliseconds sleep_per_loop(10);
int loops = 0;
while (1u != this->callback_count.load() && loops < 500) {
rclcpp::sleep_for(sleep_per_loop);
executor.spin_some();
loops++;
}
EXPECT_EQ(1u, this->callback_count.load());
// reset counter
this->callback_count.store(0u);
for (size_t ii = 0; ii < kNumMessages; ++ii) {
this->publisher->publish(test_msgs::msg::Empty());
}
// Fire a timer every 10ms up to 5 seconds waiting for subscriptions to be read.
loops = 0;
auto timer = this->node->create_wall_timer(
std::chrono::milliseconds(10), [this, &executor, &loops, &kNumMessages]() {
loops++;
if (kNumMessages == this->callback_count.load() ||
loops == 500)
{
executor.cancel();
}
});
executor.spin();
EXPECT_EQ(kNumMessages, this->callback_count.load());
}
// Check spin functions with non default context
TEST(TestExecutors, testSpinWithNonDefaultContext)
{
auto non_default_context = std::make_shared<rclcpp::Context>();
non_default_context->init(0, nullptr);
{
auto node =
std::make_unique<rclcpp::Node>("node", rclcpp::NodeOptions().context(non_default_context));
EXPECT_NO_THROW(rclcpp::spin_some(node->get_node_base_interface()));
auto check_spin_until_future_complete = [&]() {
std::promise<bool> promise;
std::future<bool> future = promise.get_future();
promise.set_value(true);
auto shared_future = future.share();
auto ret = rclcpp::spin_until_future_complete(
node->get_node_base_interface(), shared_future, 1s);
EXPECT_EQ(rclcpp::FutureReturnCode::SUCCESS, ret);
};
EXPECT_NO_THROW(check_spin_until_future_complete());
}
rclcpp::shutdown(non_default_context);
}

View File

@@ -28,6 +28,7 @@
#include "rclcpp/node_interfaces/node_graph.hpp"
#include "rclcpp/rclcpp.hpp"
#include "rcutils/strdup.h"
#include "test_msgs/msg/empty.h"
#include "test_msgs/msg/empty.hpp"
#include "test_msgs/srv/empty.hpp"
@@ -599,6 +600,18 @@ TEST_F(TestNodeGraph, get_info_by_topic)
rclcpp::QoS const_actual_qos = const_publisher_endpoint_info.qos_profile();
EXPECT_EQ(const_actual_qos.reliability(), rclcpp::ReliabilityPolicy::Reliable);
const rosidl_type_hash_t expected_type_hash = *test_msgs__msg__Empty__get_type_hash(nullptr);
EXPECT_EQ(
0, memcmp(
&publisher_endpoint_info.topic_type_hash(),
&expected_type_hash,
sizeof(rosidl_type_hash_t)));
EXPECT_EQ(
0, memcmp(
&const_publisher_endpoint_info.topic_type_hash(),
&expected_type_hash,
sizeof(rosidl_type_hash_t)));
auto endpoint_gid = publisher_endpoint_info.endpoint_gid();
auto const_endpoint_gid = const_publisher_endpoint_info.endpoint_gid();
bool endpoint_gid_is_all_zeros = true;

View File

@@ -77,9 +77,9 @@ TEST_F(TestNodeParameters, list_parameters)
std::vector<std::string> prefixes;
const auto list_result = node_parameters->list_parameters(prefixes, 1u);
// Currently the only default parameter is 'use_sim_time', but that may change.
// Currently the default parameters are 'use_sim_time' and 'start_type_description_service'
size_t number_of_parameters = list_result.names.size();
EXPECT_GE(1u, number_of_parameters);
EXPECT_GE(2u, number_of_parameters);
const std::string parameter_name = "new_parameter";
const rclcpp::ParameterValue value(true);

View File

@@ -28,6 +28,11 @@
#include "../../mocking_utils/patch.hpp"
#include "../../utils/rclcpp_gtest_macros.hpp"
using rclcpp::dynamic_typesupport::DynamicMessage;
using rclcpp::dynamic_typesupport::DynamicMessageType;
using rclcpp::dynamic_typesupport::DynamicSerializationSupport;
namespace
{
@@ -77,6 +82,18 @@ public:
const std::shared_ptr<rclcpp::SerializedMessage> &, const rclcpp::MessageInfo &) override {}
void return_message(std::shared_ptr<void> &) override {}
void return_serialized_message(std::shared_ptr<rclcpp::SerializedMessage> &) override {}
DynamicMessageType::SharedPtr get_shared_dynamic_message_type() override {return nullptr;}
DynamicMessage::SharedPtr get_shared_dynamic_message() override {return nullptr;}
DynamicSerializationSupport::SharedPtr get_shared_dynamic_serialization_support() override
{
return nullptr;
}
DynamicMessage::SharedPtr create_dynamic_message() override {return nullptr;}
void return_dynamic_message(DynamicMessage::SharedPtr &) override {}
void handle_dynamic_message(
const DynamicMessage::SharedPtr &,
const rclcpp::MessageInfo &) override {}
};
class TestNodeTopics : public ::testing::Test

View File

@@ -0,0 +1,63 @@
// Copyright 2023 Open Source Robotics Foundation, Inc.
//
// 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.
#include <gtest/gtest.h>
#include "rclcpp/node.hpp"
#include "rclcpp/node_interfaces/node_type_descriptions.hpp"
class TestNodeTypeDescriptions : public ::testing::Test
{
public:
void SetUp()
{
rclcpp::init(0, nullptr);
}
void TearDown()
{
rclcpp::shutdown();
}
};
TEST_F(TestNodeTypeDescriptions, interface_created)
{
rclcpp::Node node{"node", "ns"};
ASSERT_NE(nullptr, node.get_node_type_descriptions_interface());
}
TEST_F(TestNodeTypeDescriptions, disabled_no_service)
{
rclcpp::NodeOptions node_options;
node_options.append_parameter_override("start_type_description_service", false);
rclcpp::Node node{"node", "ns", node_options};
rcl_node_t * rcl_node = node.get_node_base_interface()->get_rcl_node_handle();
rcl_service_t * srv = nullptr;
rcl_ret_t ret = rcl_node_get_type_description_service(rcl_node, &srv);
ASSERT_EQ(RCL_RET_NOT_INIT, ret);
}
TEST_F(TestNodeTypeDescriptions, enabled_creates_service)
{
rclcpp::NodeOptions node_options;
node_options.append_parameter_override("start_type_description_service", true);
rclcpp::Node node{"node", "ns", node_options};
rcl_node_t * rcl_node = node.get_node_base_interface()->get_rcl_node_handle();
rcl_service_t * srv = nullptr;
rcl_ret_t ret = rcl_node_get_type_description_service(rcl_node, &srv);
ASSERT_EQ(RCL_RET_OK, ret);
ASSERT_NE(nullptr, srv);
}

View File

@@ -521,6 +521,10 @@ TEST_F(TestClient, rcl_client_response_subscription_get_actual_qos_error) {
}
TEST_F(TestClient, client_qos) {
if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
GTEST_SKIP() << "Skipping. The tests is known to be flaky in the rmw_connextdds.";
}
rclcpp::ServicesQoS qos_profile;
qos_profile.liveliness(rclcpp::LivelinessPolicy::Automatic);
rclcpp::Duration duration(std::chrono::nanoseconds(1));

View File

@@ -160,6 +160,50 @@ TEST(TestLogger, set_level) {
EXPECT_EQ(RCUTILS_RET_OK, rcutils_logging_shutdown());
}
TEST(TestLogger, get_effective_level) {
ASSERT_EQ(RCUTILS_RET_OK, rcutils_logging_initialize());
rclcpp::Logger logger = rclcpp::get_logger("test_logger");
rclcpp::Logger child_logger = rclcpp::get_logger("test_logger.child");
// set child logger level unset to test effective level
child_logger.set_level(rclcpp::Logger::Level::Unset);
// default
EXPECT_EQ(rclcpp::Logger::Level::Info, logger.get_effective_level());
EXPECT_EQ(rclcpp::Logger::Level::Info, child_logger.get_effective_level());
// unset
logger.set_level(rclcpp::Logger::Level::Unset);
EXPECT_EQ(rclcpp::Logger::Level::Info, logger.get_effective_level());
EXPECT_EQ(rclcpp::Logger::Level::Info, child_logger.get_effective_level());
// debug
logger.set_level(rclcpp::Logger::Level::Debug);
EXPECT_EQ(rclcpp::Logger::Level::Debug, logger.get_effective_level());
EXPECT_EQ(rclcpp::Logger::Level::Debug, child_logger.get_effective_level());
// info
logger.set_level(rclcpp::Logger::Level::Info);
EXPECT_EQ(rclcpp::Logger::Level::Info, logger.get_effective_level());
EXPECT_EQ(rclcpp::Logger::Level::Info, child_logger.get_effective_level());
// warn
logger.set_level(rclcpp::Logger::Level::Warn);
EXPECT_EQ(rclcpp::Logger::Level::Warn, logger.get_effective_level());
EXPECT_EQ(rclcpp::Logger::Level::Warn, child_logger.get_effective_level());
// error
logger.set_level(rclcpp::Logger::Level::Error);
EXPECT_EQ(rclcpp::Logger::Level::Error, logger.get_effective_level());
EXPECT_EQ(rclcpp::Logger::Level::Error, child_logger.get_effective_level());
// fatal
logger.set_level(rclcpp::Logger::Level::Fatal);
EXPECT_EQ(rclcpp::Logger::Level::Fatal, logger.get_effective_level());
EXPECT_EQ(rclcpp::Logger::Level::Fatal, child_logger.get_effective_level());
}
TEST(TestLogger, get_logging_directory) {
ASSERT_EQ(true, rcutils_set_env("HOME", "/fake_home_dir"));
ASSERT_EQ(true, rcutils_set_env("USERPROFILE", nullptr));

View File

@@ -0,0 +1,214 @@
// Copyright 2023 Sony Group Corporation.
//
// 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.
#include <gtest/gtest.h>
#include <memory>
#include <string>
#include <vector>
#include "rclcpp/rclcpp.hpp"
#include "rclcpp/node.hpp"
#include "rcl_interfaces/srv/get_logger_levels.hpp"
#include "rcl_interfaces/srv/set_logger_levels.hpp"
using namespace std::chrono_literals;
class TestLoggerService : public ::testing::Test
{
protected:
void SetUp()
{
rclcpp::init(0, nullptr);
rclcpp::NodeOptions options = rclcpp::NodeOptions();
options.enable_logger_service(true);
node_ = std::make_shared<rclcpp::Node>("test_logger_service", "/test", options);
}
void TearDown()
{
rclcpp::shutdown();
}
rclcpp::Node::SharedPtr node_;
std::thread thread_;
};
TEST_F(TestLoggerService, check_connect_get_logger_service) {
auto client = node_->create_client<rcl_interfaces::srv::GetLoggerLevels>(
"/test/test_logger_service/get_logger_levels");
ASSERT_TRUE(client->wait_for_service(2s));
}
TEST_F(TestLoggerService, check_connect_set_logger_service) {
auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
"/test/test_logger_service/set_logger_levels");
ASSERT_TRUE(client->wait_for_service(2s));
}
TEST_F(TestLoggerService, test_set_and_get_one_logging_level) {
std::string test_logger_name = "rcl";
uint8_t test_logger_level = 20;
{
auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
"/test/test_logger_service/set_logger_levels");
ASSERT_TRUE(client->wait_for_service(1s));
auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
auto level = rcl_interfaces::msg::LoggerLevel();
level.name = test_logger_name;
level.level = test_logger_level;
request->levels.push_back(level);
auto result = client->async_send_request(request);
ASSERT_EQ(
rclcpp::spin_until_future_complete(node_, result),
rclcpp::FutureReturnCode::SUCCESS);
auto result_get = result.get();
ASSERT_EQ(result_get->results.size(), 1u);
ASSERT_TRUE(result_get->results[0].successful);
ASSERT_STREQ(result_get->results[0].reason.c_str(), "");
}
{
auto client = node_->create_client<rcl_interfaces::srv::GetLoggerLevels>(
"/test/test_logger_service/get_logger_levels");
ASSERT_TRUE(client->wait_for_service(1s));
auto request = std::make_shared<rcl_interfaces::srv::GetLoggerLevels::Request>();
request->names.emplace_back(test_logger_name);
auto result = client->async_send_request(request);
ASSERT_EQ(
rclcpp::spin_until_future_complete(node_, result),
rclcpp::FutureReturnCode::SUCCESS);
auto result_get = result.get();
ASSERT_EQ(result_get->levels.size(), 1u);
ASSERT_STREQ(result_get->levels[0].name.c_str(), test_logger_name.c_str());
ASSERT_EQ(result_get->levels[0].level, test_logger_level);
}
}
TEST_F(TestLoggerService, test_set_and_get_multi_logging_level) {
std::vector<std::pair<std::string, uint8_t>> test_data {
{"rcl", 30},
{"rclcpp", 40},
{"/test/test_logger_service", 50}
};
// Set multi log levels
{
auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
"/test/test_logger_service/set_logger_levels");
ASSERT_TRUE(client->wait_for_service(1s));
auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
for (auto & set_level : test_data) {
auto level = rcl_interfaces::msg::LoggerLevel();
level.name = std::get<0>(set_level);
level.level = std::get<1>(set_level);
request->levels.push_back(level);
}
auto result = client->async_send_request(request);
ASSERT_EQ(
rclcpp::spin_until_future_complete(node_, result),
rclcpp::FutureReturnCode::SUCCESS);
auto result_get = result.get();
ASSERT_EQ(result_get->results.size(), test_data.size());
for (uint32_t i = 0; i < test_data.size(); i++) {
ASSERT_TRUE(result_get->results[0].successful);
}
}
// Get multi log levels
{
auto client = node_->create_client<rcl_interfaces::srv::GetLoggerLevels>(
"/test/test_logger_service/get_logger_levels");
ASSERT_TRUE(client->wait_for_service(1s));
auto request = std::make_shared<rcl_interfaces::srv::GetLoggerLevels::Request>();
for (auto & set_level : test_data) {
request->names.emplace_back(std::get<0>(set_level));
}
auto result = client->async_send_request(request);
ASSERT_EQ(
rclcpp::spin_until_future_complete(node_, result),
rclcpp::FutureReturnCode::SUCCESS);
auto result_get = result.get();
ASSERT_EQ(result_get->levels.size(), test_data.size());
for (uint32_t i = 0; i < test_data.size(); i++) {
ASSERT_STREQ(result_get->levels[i].name.c_str(), std::get<0>(test_data[i]).c_str());
ASSERT_EQ(result_get->levels[i].level, std::get<1>(test_data[i]));
}
}
}
TEST_F(TestLoggerService, test_set_logging_level_with_invalid_param) {
std::vector<std::pair<std::string, uint8_t>> test_data {
{"rcl", 12},
{"/test/test_logger_service", 22}
};
// Set multi log levels
{
auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
"/test/test_logger_service/set_logger_levels");
ASSERT_TRUE(client->wait_for_service(1s));
auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
for (auto & set_level : test_data) {
auto level = rcl_interfaces::msg::LoggerLevel();
level.name = std::get<0>(set_level);
level.level = std::get<1>(set_level);
request->levels.push_back(level);
}
auto result = client->async_send_request(request);
ASSERT_EQ(
rclcpp::spin_until_future_complete(node_, result),
rclcpp::FutureReturnCode::SUCCESS);
auto result_get = result.get();
ASSERT_EQ(result_get->results.size(), test_data.size());
for (uint32_t i = 0; i < test_data.size(); i++) {
ASSERT_FALSE(result_get->results[i].successful);
// Check string starts with prefix
ASSERT_EQ(
result_get->results[i].reason.rfind("Unable to determine severity_string for severity", 0),
0);
}
}
}
TEST_F(TestLoggerService, test_set_logging_level_with_partial_invalid_param) {
std::vector<std::pair<std::string, uint8_t>> test_data {
{"rcl", 20},
{"rclcpp", 22},
{"/test/test_logger_service", 30}
};
// Set multi log levels
{
auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
"/test/test_logger_service/set_logger_levels");
ASSERT_TRUE(client->wait_for_service(1s));
auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
for (auto & set_level : test_data) {
auto level = rcl_interfaces::msg::LoggerLevel();
level.name = std::get<0>(set_level);
level.level = std::get<1>(set_level);
request->levels.push_back(level);
}
auto result = client->async_send_request(request);
ASSERT_EQ(
rclcpp::spin_until_future_complete(node_, result),
rclcpp::FutureReturnCode::SUCCESS);
auto result_get = result.get();
ASSERT_EQ(result_get->results.size(), test_data.size());
ASSERT_TRUE(result_get->results[0].successful);
ASSERT_FALSE(result_get->results[1].successful);
ASSERT_TRUE(result_get->results[2].successful);
}
}

View File

@@ -78,6 +78,7 @@ TEST_F(TestNode, construction_and_destruction) {
EXPECT_NE(nullptr, node->get_node_options().get_rcl_node_options());
EXPECT_NE(nullptr, node->get_graph_event());
EXPECT_NE(nullptr, node->get_clock());
EXPECT_NE(nullptr, node->get_node_type_descriptions_interface());
}
{

View File

@@ -266,6 +266,11 @@ TEST(TestNodeOptions, bool_setters_and_getters) {
EXPECT_FALSE(options.automatically_declare_parameters_from_overrides());
options.automatically_declare_parameters_from_overrides(true);
EXPECT_TRUE(options.automatically_declare_parameters_from_overrides());
options.enable_logger_service(false);
EXPECT_FALSE(options.enable_logger_service());
options.enable_logger_service(true);
EXPECT_TRUE(options.enable_logger_service());
}
TEST(TestNodeOptions, parameter_event_qos) {

View File

@@ -59,6 +59,8 @@ protected:
node_with_option.reset();
}
// "start_type_description_service" and "use_sim_time"
const uint64_t builtin_param_count = 2;
rclcpp::Node::SharedPtr node;
rclcpp::Node::SharedPtr node_with_option;
};
@@ -925,6 +927,7 @@ TEST_F(TestParameterClient, sync_parameter_delete_parameters) {
Coverage for async load_parameters
*/
TEST_F(TestParameterClient, async_parameter_load_parameters) {
const uint64_t expected_param_count = 4 + builtin_param_count;
auto load_node = std::make_shared<rclcpp::Node>(
"load_node",
"namespace",
@@ -944,12 +947,13 @@ TEST_F(TestParameterClient, async_parameter_load_parameters) {
auto list_parameters = asynchronous_client->list_parameters({}, 3);
rclcpp::spin_until_future_complete(
load_node, list_parameters, std::chrono::milliseconds(100));
ASSERT_EQ(list_parameters.get().names.size(), static_cast<uint64_t>(5));
ASSERT_EQ(list_parameters.get().names.size(), expected_param_count);
}
/*
Coverage for sync load_parameters
*/
TEST_F(TestParameterClient, sync_parameter_load_parameters) {
const uint64_t expected_param_count = 4 + builtin_param_count;
auto load_node = std::make_shared<rclcpp::Node>(
"load_node",
"namespace",
@@ -964,13 +968,14 @@ TEST_F(TestParameterClient, sync_parameter_load_parameters) {
ASSERT_EQ(load_future[0].successful, true);
// list parameters
auto list_parameters = synchronous_client->list_parameters({}, 3);
ASSERT_EQ(list_parameters.names.size(), static_cast<uint64_t>(5));
ASSERT_EQ(list_parameters.names.size(), static_cast<uint64_t>(expected_param_count));
}
/*
Coverage for async load_parameters with complicated regex expression
*/
TEST_F(TestParameterClient, async_parameter_load_parameters_complicated_regex) {
const uint64_t expected_param_count = 5 + builtin_param_count;
auto load_node = std::make_shared<rclcpp::Node>(
"load_node",
"namespace",
@@ -990,7 +995,7 @@ TEST_F(TestParameterClient, async_parameter_load_parameters_complicated_regex) {
auto list_parameters = asynchronous_client->list_parameters({}, 3);
rclcpp::spin_until_future_complete(
load_node, list_parameters, std::chrono::milliseconds(100));
ASSERT_EQ(list_parameters.get().names.size(), static_cast<uint64_t>(6));
ASSERT_EQ(list_parameters.get().names.size(), expected_param_count);
// to check the parameter "a_value"
std::string param_name = "a_value";
auto param = load_node->get_parameter(param_name);
@@ -1020,6 +1025,7 @@ TEST_F(TestParameterClient, async_parameter_load_no_valid_parameter) {
Coverage for async load_parameters from maps with complicated regex expression
*/
TEST_F(TestParameterClient, async_parameter_load_parameters_from_map) {
const uint64_t expected_param_count = 5 + builtin_param_count;
auto load_node = std::make_shared<rclcpp::Node>(
"load_node",
"namespace",
@@ -1068,7 +1074,7 @@ TEST_F(TestParameterClient, async_parameter_load_parameters_from_map) {
auto list_parameters = asynchronous_client->list_parameters({}, 3);
rclcpp::spin_until_future_complete(
load_node, list_parameters, std::chrono::milliseconds(100));
ASSERT_EQ(list_parameters.get().names.size(), static_cast<uint64_t>(6));
ASSERT_EQ(list_parameters.get().names.size(), expected_param_count);
// to check the parameter "a_value"
std::string param_name = "a_value";
auto param = load_node->get_parameter(param_name);

View File

@@ -152,33 +152,54 @@ TEST_F(TestPublisher, conversion_exception_is_passed_up) {
}
}
using UseTakeSharedMethod = bool;
class TestPublisherFixture
: public TestPublisher,
public ::testing::WithParamInterface<UseTakeSharedMethod>
{
};
/*
* Testing that publisher sends type adapted types and ROS message types with intra proccess communications.
*/
TEST_F(
TestPublisher,
TEST_P(
TestPublisherFixture,
check_type_adapted_message_is_sent_and_received_intra_process) {
using StringTypeAdapter = rclcpp::TypeAdapter<std::string, rclcpp::msg::String>;
const std::string message_data = "Message Data";
const std::string topic_name = "topic_name";
bool is_received;
auto callback =
[message_data, &is_received](
const rclcpp::msg::String::ConstSharedPtr msg,
const rclcpp::MessageInfo & message_info
) -> void
{
is_received = true;
ASSERT_STREQ(message_data.c_str(), msg->data.c_str());
ASSERT_TRUE(message_info.get_rmw_message_info().from_intra_process);
};
auto node = rclcpp::Node::make_shared(
"test_intra_process",
rclcpp::NodeOptions().use_intra_process_comms(true));
auto pub = node->create_publisher<StringTypeAdapter>(topic_name, 10);
auto sub = node->create_subscription<rclcpp::msg::String>(topic_name, 1, callback);
rclcpp::Subscription<rclcpp::msg::String>::SharedPtr sub;
if (GetParam()) {
auto callback =
[message_data, &is_received](
const rclcpp::msg::String::ConstSharedPtr msg,
const rclcpp::MessageInfo & message_info
) -> void
{
is_received = true;
ASSERT_STREQ(message_data.c_str(), msg->data.c_str());
ASSERT_TRUE(message_info.get_rmw_message_info().from_intra_process);
};
sub = node->create_subscription<rclcpp::msg::String>(topic_name, 1, callback);
} else {
auto callback_unique =
[message_data, &is_received](
rclcpp::msg::String::UniquePtr msg,
const rclcpp::MessageInfo & message_info
) -> void
{
is_received = true;
ASSERT_STREQ(message_data.c_str(), msg->data.c_str());
ASSERT_TRUE(message_info.get_rmw_message_info().from_intra_process);
};
sub = node->create_subscription<rclcpp::msg::String>(topic_name, 1, callback_unique);
}
auto wait_for_message_to_be_received = [&is_received, &node]() {
rclcpp::executors::SingleThreadedExecutor executor;
@@ -239,6 +260,14 @@ TEST_F(
}
}
INSTANTIATE_TEST_SUITE_P(
TestPublisherFixtureWithParam,
TestPublisherFixture,
::testing::Values(
true, // use take shared method
false // not use take shared method
));
/*
* Testing that publisher sends type adapted types and ROS message types with inter proccess communications.
*/

View File

@@ -18,10 +18,24 @@
#include "rclcpp/rate.hpp"
class TestRate : public ::testing::Test
{
public:
void SetUp()
{
rclcpp::init(0, nullptr);
}
void TearDown()
{
rclcpp::shutdown();
}
};
/*
Basic tests for the Rate and WallRate classes.
*/
TEST(TestRate, rate_basics) {
TEST_F(TestRate, rate_basics) {
auto period = std::chrono::milliseconds(1000);
auto offset = std::chrono::milliseconds(500);
auto epsilon = std::chrono::milliseconds(100);
@@ -61,7 +75,7 @@ TEST(TestRate, rate_basics) {
ASSERT_TRUE(epsilon > delta);
}
TEST(TestRate, wall_rate_basics) {
TEST_F(TestRate, wall_rate_basics) {
auto period = std::chrono::milliseconds(100);
auto offset = std::chrono::milliseconds(50);
auto epsilon = std::chrono::milliseconds(1);
@@ -101,7 +115,7 @@ TEST(TestRate, wall_rate_basics) {
EXPECT_GT(epsilon, delta);
}
TEST(TestRate, from_double) {
TEST_F(TestRate, from_double) {
{
rclcpp::WallRate rate(1.0);
EXPECT_EQ(std::chrono::seconds(1), rate.period());

Some files were not shown because too many files have changed in this diff Show More