Merge pull request #424 from elfenpiff/iox2-390-waitset

[#390] waitset

Author: elfenpiff

README.md

@@ -123,7 +123,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let publisher = service.publisher_builder().create()?;
 
-    while let NodeEvent::Tick = node.wait(CYCLE_TIME) {
+    while node.wait(CYCLE_TIME) != WaitEvent::TerminationRequest {
         let sample = publisher.loan_uninit()?;
         let sample = sample.write_payload(1234);
         sample.send()?;
@@ -150,7 +150,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let subscriber = service.subscriber_builder().create()?;
 
-    while let NodeEvent::Tick = node.wait(CYCLE_TIME) {
+    while node.wait(CYCLE_TIME) != WaitEvent::TerminationRequest {
         while let Some(sample) = subscriber.receive()? {
             println!("received: {:?}", *sample);
         }
@@ -200,7 +200,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let notifier = event.notifier_builder().create()?;
 
     let id = EventId::new(12);
-    while let NodeEvent::Tick = node.wait(CYCLE_TIME) {
+    while node.wait(CYCLE_TIME) != WaitEvent::TerminationRequest {
         notifier.notify_with_custom_event_id(id)?;
 
         println!("Trigger event with id {:?} ...", id);
@@ -227,7 +227,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let listener = event.listener_builder().create()?;
 
-    while let NodeEvent::Tick = node.wait(Duration::ZERO) {
+    while node.wait(Duration::ZERO) != WaitEvent::TerminationRequest {
         if let Ok(Some(event_id)) = listener.timed_wait_one(CYCLE_TIME) {
             println!("event was triggered with id: {:?}", event_id);
         }
@@ -254,7 +254,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let listener = event.listener_builder().create()?;
 
-    while let NodeEvent::Tick = node.wait(Duration::ZERO) {
+    while node.wait(Duration::ZERO) != WaitEvent::TerminationRequest {
         listener.timed_wait_all(
             |event_id| {
                 println!("event was triggered with id: {:?}", event_id);

doc/release-notes/iceoryx2-unreleased.md

@@ -11,6 +11,7 @@
     conflicts when merging.
 -->
 
+* Add Event-Multiplexer `WaitSet` [#390](https://github.com/eclipse-iceoryx/iceoryx2/issues/390)
 * Add `PeriodicTimer` into POSIX building blocks [#425](https://github.com/eclipse-iceoryx/iceoryx2/issues/425)
 
 ### Bugfixes
@@ -29,7 +30,7 @@
     conflicts when merging.
 -->
 
-* Example text [#1](https://github.com/eclipse-iceoryx/iceoryx2/issues/1)
+* Rename `NodeEvent` into `WaitEvent` [#390](https://github.com/eclipse-iceoryx/iceoryx2/issues/390)
 
 ### Workflow
 
@@ -51,12 +52,16 @@
 
 ### API Breaking Changes
 
-1. Example
+1. Renamed `NodeEvent` into `WaitEvent`
 
    ```rust
    // old
-   let fuu = hello().is_it_me_you_re_looking_for()
+   while node.wait(CYCLE_TIME) != NodeEvent::TerminationRequest {
+    // ...
+   }
 
    // new
-   let fuu = hypnotoad().all_glory_to_the_hypnotoad()
+   while node.wait(CYCLE_TIME) != WaitEvent::TerminationRequest {
+    // ...
+   }
    ```

examples/Cargo.toml

@@ -43,6 +43,16 @@ path = "rust/event/listener.rs"
 name = "event_notifier"
 path = "rust/event/notifier.rs"
 
+# event multiplexing
+
+[[example]]
+name = "event_multiplexing_notifier"
+path = "rust/event_multiplexing/notifier.rs"
+
+[[example]]
+name = "event_multiplexing_wait"
+path = "rust/event_multiplexing/wait.rs"
+
 # publish_subscribe
 
 [[example]]

examples/README.md

@@ -49,6 +49,7 @@ and exchange data.
 | docker                             | [all](rust/docker)                                                                                                                      | Communicate between different docker containers and the host.                                                                                                                                                   |
 | domains                            | [C](c/domains) [C++](cxx/domains) [Rust](rust/domains)                                                                                  | Establish separate domains that operate independently from one another.                                                                                                                                         |
 | event                              | [C](c/event) [C++](cxx/event) [Rust](rust/event)                                                                                        | Push notifications - send event signals to wakeup processes that are waiting for them.                                                                                                                          |
+| event multiplexing                 | [Rust](rust/event_multiplexing)                                                                                                         | Wait on multiple listeners or sockets with a single call. The WaitSet demultiplexes incoming events and notifies the user.                                                                                      |
 | publish subscribe                  | [C](c/publish_subscribe) [C++](cxx/publish_subscribe) [Rust](rust/publish_subscribe)                                                    | Communication between multiple processes with a [publish subscribe messaging pattern](https://en.wikipedia.org/wiki/Publish–subscribe_pattern).                                                                 |
 | publish subscribe dynamic data     | [Rust](rust/publish_subscribe_dynamic_data)                                                                                             | Communication between multiple processes with a [publish subscribe messaging pattern](https://en.wikipedia.org/wiki/Publish–subscribe_pattern) and payload data that has a dynamic size.                        |
 | publish subscribe with user header | [C](c/publish_subscribe_with_user_header) [C++](cxx/publish_subscribe_with_user_header) [Rust](rust/publish_subscribe_with_user_header) | Add a user header to the payload (samples) to transfer additional information.                                                                                                                                  |

examples/c/domains/src/publisher.c

@@ -94,7 +94,7 @@ int main(int argc, char** argv) {
     }
 
     int32_t counter = 0;
-    while (iox2_node_wait(&node_handle, 1, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 1, 0) == IOX2_OK) {
         counter += 1;
 
         // loan sample

examples/c/domains/src/subscriber.c

@@ -96,7 +96,7 @@ int main(int argc, char** argv) {
 
     uint64_t counter = 0;
     printf("subscribed to: [domain: \"%s\", service: \"%s\"]\n", argv[1], argv[2]);
-    while (iox2_node_wait(&node_handle, 1, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 1, 0) == IOX2_OK) {
         counter += 1;
 
         // receive sample

examples/c/event/src/listener.c

@@ -52,7 +52,7 @@ int main(void) {
     }
     iox2_event_id_t event_id;
 
-    while (iox2_node_wait(&node_handle, 0, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 0, 0) == IOX2_OK) {
         bool has_received_one = false;
         if (iox2_listener_timed_wait_one(&listener, &event_id, &has_received_one, 1, 0) != IOX2_OK) {
             printf("Unable to wait for notification!\n");

examples/c/event/src/notifier.c

@@ -59,7 +59,7 @@ int main(void) {
     }
 
     uint64_t counter = 0;
-    while (iox2_node_wait(&node_handle, 0, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 0, 0) == IOX2_OK) {
         counter += 1;
         iox2_event_id_t event_id = { .value = counter % 12 }; // NOLINT
         if (iox2_notifier_notify_with_custom_event_id(&notifier, &event_id, NULL) != IOX2_OK) {

examples/c/publish_subscribe/src/publisher.c

@@ -74,7 +74,7 @@ int main(void) {
     }
 
     int32_t counter = 0;
-    while (iox2_node_wait(&node_handle, 1, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 1, 0) == IOX2_OK) {
         counter += 1;
 
         // loan sample

examples/c/publish_subscribe/src/subscriber.c

@@ -74,7 +74,7 @@ int main(void) {
     }
 
     uint64_t counter = 0;
-    while (iox2_node_wait(&node_handle, 1, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 1, 0) == IOX2_OK) {
         counter += 1;
 
         // receive sample

examples/c/publish_subscribe_with_user_header/src/publisher.c

@@ -87,7 +87,7 @@ int main(void) {
     }
 
     int32_t counter = 0;
-    while (iox2_node_wait(&node_handle, 1, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 1, 0) == IOX2_OK) {
         counter += 1;
 
         // loan sample

examples/c/publish_subscribe_with_user_header/src/subscriber.c

@@ -87,7 +87,7 @@ int main(void) {
     }
 
     uint64_t counter = 0;
-    while (iox2_node_wait(&node_handle, 1, 0) == iox2_node_event_e_TICK) {
+    while (iox2_node_wait(&node_handle, 1, 0) == IOX2_OK) {
         counter += 1;
 
         // receive sample

examples/cxx/complex_data_types/src/complex_data_types.cpp

@@ -51,7 +51,7 @@ auto main() -> int {
     auto subscriber = service.subscriber_builder().create().expect("successful subscriber creation");
 
     auto counter = 0;
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         counter += 1;
         auto sample = publisher.loan_uninit().expect("acquire sample");
         new (&sample.payload_mut()) ComplexDataType {};

examples/cxx/domains/src/publisher.cpp

@@ -61,7 +61,7 @@ auto main(int argc, char** argv) -> int {
     auto publisher = service.publisher_builder().create().expect("successful publisher creation");
 
     auto counter = 0;
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         counter += 1;
 
         auto sample = publisher.loan_uninit().expect("acquire sample");

examples/cxx/domains/src/subscriber.cpp

@@ -58,7 +58,7 @@ auto main(int argc, char** argv) -> int {
 
     std::cout << "subscribed to: [domain: \"" << args.domain() << "\", service: \"" << args.service() << "\"]"
               << std::endl;
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         auto sample = subscriber.receive().expect("receive succeeds");
         while (sample.has_value()) {
             std::cout << "received: " << sample->payload() << std::endl;

examples/cxx/event/src/listener.cpp

@@ -30,7 +30,7 @@ auto main() -> int {
 
     auto listener = service.listener_builder().create().expect("successful listener creation");
 
-    while (node.wait(iox::units::Duration::zero()) == NodeEvent::Tick) {
+    while (node.wait(iox::units::Duration::zero()).has_value()) {
         listener.timed_wait_one(CYCLE_TIME).and_then([](auto maybe_event_id) {
             maybe_event_id.and_then(
                 [](auto event_id) { std::cout << "event was triggered with id: " << event_id << std::endl; });

examples/cxx/event/src/notifier.cpp

@@ -33,7 +33,7 @@ auto main() -> int {
     auto notifier = service.notifier_builder().create().expect("successful notifier creation");
 
     auto counter = 0;
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         counter += 1;
         const auto event_id = EventId(counter % max_event_id);
         notifier.notify_with_custom_event_id(event_id).expect("notification");

examples/cxx/publish_subscribe/src/publisher.cpp

@@ -34,7 +34,7 @@ auto main() -> int {
     auto publisher = service.publisher_builder().create().expect("successful publisher creation");
 
     auto counter = 0;
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         counter += 1;
 
         auto sample = publisher.loan_uninit().expect("acquire sample");

examples/cxx/publish_subscribe/src/subscriber.cpp

@@ -31,7 +31,7 @@ auto main() -> int {
 
     auto subscriber = service.subscriber_builder().create().expect("successful subscriber creation");
 
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         auto sample = subscriber.receive().expect("receive succeeds");
         while (sample.has_value()) {
             std::cout << "received: " << sample->payload() << std::endl;

examples/cxx/publish_subscribe_dynamic_data/src/publisher.cpp

@@ -40,7 +40,7 @@ auto main() -> int {
 
     auto counter = 1;
 
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         counter += 1;
 
         auto required_memory_size = (8 + counter) % 16; // NOLINT

examples/cxx/publish_subscribe_dynamic_data/src/subscriber.cpp

@@ -32,7 +32,7 @@ auto main() -> int {
 
     auto subscriber = service.subscriber_builder().create().expect("successful subscriber creation");
 
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         auto sample = subscriber.receive().expect("receive succeeds");
         while (sample.has_value()) {
             std::cout << "received " << sample->payload().size() << " bytes: ";

examples/cxx/publish_subscribe_with_user_header/src/publisher.cpp

@@ -38,7 +38,7 @@ auto main() -> int {
     auto publisher = service.publisher_builder().create().expect("successful publisher creation");
 
     auto counter = 0;
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         counter += 1;
         auto sample = publisher.loan_uninit().expect("acquire sample");
 

examples/cxx/publish_subscribe_with_user_header/src/subscriber.cpp

@@ -33,7 +33,7 @@ auto main() -> int {
 
     auto subscriber = service.subscriber_builder().create().expect("successful subscriber creation");
 
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         auto sample = subscriber.receive().expect("receive succeeds");
         while (sample.has_value()) {
             std::cout << "received: " << sample->payload() << ", user_header: " << sample->user_header() << std::endl;

examples/cxx/service_attributes/src/creator.cpp

@@ -43,7 +43,7 @@ auto main() -> int {
 
     std::cout << "defined service attributes: " << service.attributes() << std::endl;
 
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         auto sample = publisher.loan().expect("acquire sample");
         sample.payload_mut() = 0;
         send(std::move(sample)).expect("send successful");

examples/cxx/service_attributes/src/opener.cpp

@@ -36,7 +36,7 @@ auto main() -> int {
 
     std::cout << "defined service attributes: " << service.attributes() << std::endl;
 
-    while (node.wait(CYCLE_TIME) == NodeEvent::Tick) {
+    while (node.wait(CYCLE_TIME).has_value()) {
         auto sample = subscriber.receive().expect("receive succeeds");
         while (sample.has_value()) {
             std::cout << "received: " << sample->payload() << std::endl;

examples/rust/complex_data_types/complex_data_types.rs

@@ -54,7 +54,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let subscriber = service.subscriber_builder().create()?;
     let mut counter = 0;
 
-    while let NodeEvent::Tick = node.wait(CYCLE_TIME) {
+    while node.wait(CYCLE_TIME).is_ok() {
         // ComplexDataType as a size of over 30MB, we need to perform a placement new
         // otherwise we will encounter a stack overflow in debug builds.
         // Therefore, we acquire an uninitialized sample, use the PlacementDefault

examples/rust/domains/publisher.rs

@@ -47,7 +47,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut counter: u64 = 0;
 
-    while let NodeEvent::Tick = node.wait(CYCLE_TIME) {
+    while node.wait(CYCLE_TIME).is_ok() {
         counter += 1;
         let sample = publisher.loan_uninit()?;
 

examples/rust/domains/subscriber.rs

@@ -46,7 +46,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         "subscribed to: [domain: \"{}\", service: \"{}\"]",
         args.domain, args.service
     );
-    while let NodeEvent::Tick = node.wait(CYCLE_TIME) {
+    while node.wait(CYCLE_TIME).is_ok() {
         while let Some(sample) = subscriber.receive()? {
             println!("received: {:?}", *sample);
         }

examples/rust/event/listener.rs

@@ -25,7 +25,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let listener = event.listener_builder().create()?;
 
-    while let NodeEvent::Tick = node.wait(Duration::ZERO) {
+    while node.wait(Duration::ZERO).is_ok() {
         if let Ok(Some(event_id)) = listener.timed_wait_one(CYCLE_TIME) {
             println!("event was triggered with id: {:?}", event_id);
         }

examples/rust/event/notifier.rs

@@ -27,7 +27,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let notifier = event.notifier_builder().create()?;
 
     let mut counter: usize = 0;
-    while let NodeEvent::Tick = node.wait(CYCLE_TIME) {
+    while node.wait(CYCLE_TIME).is_ok() {
         counter += 1;
         notifier.notify_with_custom_event_id(EventId::new(counter % max_event_id))?;
 

examples/rust/event_multiplexing/README.md

@@ -0,0 +1,53 @@
+# Event Multiplexing
+
+## Running The Example
+
+This example demonstrates iceoryx2's event multiplexing mechanism,
+called the `WaitSet`. It allows waiting, with a single call, on
+multiple `Listener` ports as well as external file descriptor-based
+events such as `sockets`.
+
+In this setup, the `wait` process monitors an arbitrary number of
+services, which the user can specify via the command line option `-s`.
+The `notifier` can define the service to which it will send event
+notifications using the `-s` option and specify the event ID with
+the `-e` option.
+
+In the example below, we are waiting for events on the services `fuu` and
+`bar`. Service `fuu` is notified with event ID `123`, and service `bar` is
+notified with event ID `456`.
+
+### Terminal 1
+
+```sh
+cargo run --example event_multiplexing_wait -- -s "fuu" -s "bar"
+```
+
+### Terminal 2
+
+```sh
+cargo run --example event_multiplexing_notifier -- -s "fuu" -e 123
+```
+
+### Terminal 3
+
+```sh
+cargo run --example event_multiplexing_notifier -- -s "bar" -e 456
+```
+
+Feel free to instantiate multiple notifiers for the same service with the same
+or different event id's. Or to for different services.
+
+## Technical Details
+
+The `WaitSet` utilizes `epoll`, `select`, or other event-multiplexing
+mechanisms. Before the `WaitSet` can monitor a specific event, it must first be
+attached using `WaitSet::attach()`, which returns a RAII `Guard`. This `Guard`
+automatically detaches the attachment when it goes out of scope.
+
+The `WaitSet::**_wait()` calls require a closure that is invoked for each
+triggered attachment and provides the `AttachmentId`. The user can either use
+`AttachmentId::originates_from($ATTACHED_OBJECT$)` to identify the object
+associated with the `AttachmentId`, or set up a
+`HashMap::<AttachmentId, Listener<ipc::Service>>` to quickly access the
+corresponding object.