Reactive client–server concept

[TE-2]

Hi everyone,

based on a previous discussion I’ve been experimenting with a small reactive client–server example and I’d like to get your thoughts on the general design approach.

In my setup, both the client and the server are implemented as custom classes, each with their own listener and notifier. The communication flow works as follows:

1. The client sends a request and then immediately sends a notification to the server using the service name <generic_service_name>/Request.
2. The server’s internal listener wakes up on that notification, processes the pending request, and sends a response back.
3. Once the response has been sent, the server sends a notification to <generic_service_name>/Response to wake up the client, which then reads and handles the response.

My goal is to have a fully event-driven system where both sides react instantly to incoming data, but without any busy waiting. So far, this setup seems to work, but I’m not sure if this is an idiomatic or efficient way to use the Iceoryx2 tools.

I’m mainly uncertain about the following points:

• Whether separating the notification channels (/Request and /Response) is a good design choice.
• How to ensure proper synchronization between event notifications and the corresponding request/response messages.
  For example, what happens if clientA fills the server’s request queue but never sends a corresponding notification?
  When a clientB sends both a request and a notification, from my understanding the server could processes a mismatched request, since the two are not directly linked.
  Is there a native way to ensure that the notification and the message originate from the same client (apart from manually sending some metadata with the rewuest and the notification)? I noticed that requests contain a unique ID that can be retrieved, but I couldn’t find an equivalent identifier for notifications.

I’d appreciate any feedback or ideas on improving this design, especially regarding synchronization or best practices with Iceoryx2.

I’ve included my custom class implementations and main programs below for reference (mainly based on the example cxx_event_based_communication)

---

custom_client.hpp

#include "iox2/iceoryx2.hpp"
#include "transmission_data.hpp"

class CustomClient : public iox2::FileDescriptorBased {
  public:
    static auto create(iox2::Node<iox2::ServiceType::Ipc>& node, const std::string service_name)
        -> CustomClient {
        
        auto service_rr = node.service_builder(iox2::ServiceName::create(service_name.c_str()).expect("valid"))
                  .request_response<uint64_t, TransmissionData>()
                  .open_or_create()
                  .expect("open_or_create");

        // Event used to wake up the server
        auto service_request = node.service_builder(iox2::ServiceName::create((service_name + "/Request").c_str()).expect("valid"))
                    .event()
                    .open_or_create()
                    .expect("open_or_create");

        auto service_response = node.service_builder(iox2::ServiceName::create((service_name + "/Response").c_str()).expect("valid"))
            .event()
            .open_or_create()
            .expect("open_or_create");

        auto client = service_rr.client_builder().create().expect("client");
        auto notifier = service_request.notifier_builder().create().expect("notifier");
        auto listener = service_response.listener_builder().create().expect("");

        return CustomClient { std::move(client), std::move(notifier), std::move(listener) };
    }

    auto file_descriptor() const -> iox2::FileDescriptorView override {
        return m_listener.file_descriptor();
    }

    void handle_event() {
        std::cout << "handle_event() noch nicht implementiert -> TODO für non busy Wait auf Antwort" << std::endl;
    }

    auto sendAndReceive(uint64_t& request_data) -> TransmissionData
    {
        auto request = m_client.loan_uninit().expect("loan request");
        auto initialized_request = request.write_payload(std::move(request_data));

        // Senden
        auto pending = iox2::send(std::move(initialized_request)).expect("send request");
        m_notifier.notify().expect("notify server");

        m_listener.timed_wait_one(iox::units::Duration::fromMilliseconds(100)).and_then([](auto maybe_event_id) -> auto {
            maybe_event_id.and_then(
                [](auto event_id) -> auto { std::cout << "event was triggered with id: " << event_id << std::endl; });
            });

        // Receive all responses for this request
        auto response = pending.receive().expect("receive");
        return response->payload();
    }

    CustomClient(const CustomClient&) = delete;
    CustomClient(CustomClient&&) = default;
    auto operator=(const CustomClient&) -> CustomClient& = delete;
    auto operator=(CustomClient&&) -> CustomClient& = default;

  private:
    CustomClient(iox2::Client<iox2::ServiceType::Ipc, uint64_t, void, TransmissionData, void>&& client,
                 iox2::Notifier<iox2::ServiceType::Ipc>&& notifier,   
                 iox2::Listener<iox2::ServiceType::Ipc>&& listener)
            : m_client { std::move(client) }
            , m_notifier { std::move(notifier) } 
            , m_listener { std::move(listener) }
    {
    }

    mutable iox2::Client<iox2::ServiceType::Ipc, uint64_t, void, TransmissionData, void> m_client;
    iox2::Notifier<iox2::ServiceType::Ipc> m_notifier;
    iox2::Listener<iox2::ServiceType::Ipc> m_listener;
};

custom_server.hpp

#include "iox2/iceoryx2.hpp"
#include "transmission_data.hpp"


class CustomServer : public iox2::FileDescriptorBased { 
  public:
    CustomServer(const CustomServer&) = delete;
    CustomServer(CustomServer&&) = default;

    auto operator=(const CustomServer&) -> CustomServer& = delete;
    auto operator=(CustomServer&&) -> CustomServer& = default;

    static auto create(iox2::Node<iox2::ServiceType::Ipc>& node, const std::string& service_name)
        -> CustomServer {
        
        auto service_rr = node.service_builder(iox2::ServiceName::create(service_name.c_str()).expect("valid"))
                                  .request_response<uint64_t, TransmissionData>()
                                  .open_or_create()
                                  .expect("");

        auto service_request = node.service_builder(iox2::ServiceName::create((service_name + "/Request").c_str()).expect("valid"))
                                 .event()
                                 .open_or_create()
                                 .expect("");

        auto service_response = node.service_builder(iox2::ServiceName::create((service_name + "/Response").c_str()).expect("valid"))
                                 .event()
                                 .open_or_create()
                                 .expect("");

        auto server = service_rr.server_builder().create().expect("cannot create Server");
        auto listener = service_request.listener_builder().create().expect("");
        auto notifier = service_response.notifier_builder().create().expect("");     

        return CustomServer { std::move(server), std::move(notifier), std::move(listener) };
    }

    auto file_descriptor() const -> iox2::FileDescriptorView override {
        return m_listener.file_descriptor();
    }

    void handle_event() {
        for(auto event = m_listener.try_wait_one();     
            event.has_value() && event->has_value();
            event = m_listener.try_wait_one())
        {
            std::cout << "handle_event()" << std::endl;

            auto request = m_server.receive().expect("receive successful");
            std::cout << "received request: " << request->payload() << std::endl;

            auto response = request->loan_uninit().expect("loan successful");
            auto initialized_response = response.write_payload(
                TransmissionData{(request->payload() + 1), 5, 0.1234});
            std::cout << "send response: " << initialized_response.payload() << std::endl;

            iox2::send(std::move(initialized_response)).expect("send successful");
            m_notifier.notify().expect("Response Notification for client");
        }

        for (auto request = m_server.receive().expect("receive successful"); 
            request.has_value(); 
            request = m_server.receive().expect("receive successful")) {
            std::cout << "Abfanglogik -> sind noch weiter Anfragen da?" << std::endl;
        }
    }

    auto receive() -> iox::optional<iox2::ActiveRequest<iox2::ServiceType::Ipc, uint64_t, void, TransmissionData, void>> {
        std::cout << "Implementierung folgt noch zu kompliziert" << std::endl;
        return m_server.receive().expect("receive successful");
    }

  private:
    CustomServer(iox2::Server<iox2::ServiceType::Ipc, uint64_t, void, TransmissionData, void>&& server,
                     iox2::Notifier<iox2::ServiceType::Ipc>&& notifier,
                     iox2::Listener<iox2::ServiceType::Ipc>&& listener)
        : m_server { std::move(server) }
        , m_notifier { std::move(notifier) }
        , m_listener { std::move(listener) } {
    }

    iox2::Server<iox2::ServiceType::Ipc, uint64_t, void, TransmissionData, void> m_server;
    iox2::Notifier<iox2::ServiceType::Ipc> m_notifier;
    iox2::Listener<iox2::ServiceType::Ipc> m_listener;
};

main_client.cpp

#include "custom_client.hpp"
#include <csignal>

constexpr iox::units::Duration PERIOD = iox::units::Duration::fromMilliseconds(1);

bool running = true;

int main() {
    set_log_level_from_env_or(iox2::LogLevel::Info);

    auto node = iox2::NodeBuilder().create<iox2::ServiceType::Ipc>().expect("node creation");

    auto iox2_client_browse = CustomClient::create(node, "ServiceSet/Service1");

    uint64_t counter = 0;

    while (node.wait(PERIOD).has_value()) {
        auto start = std::chrono::steady_clock::now();

        auto result = iox2_client_browse.sendAndReceive(counter);

        auto end = std::chrono::steady_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
        std::cout << "receive: " << result << " | duration: " << duration << " µs" << std::endl;
        counter += 1;
    }
}

main_server.cpp

#include "custom_server.hpp"
//#include <iostream>

constexpr iox::units::Duration DEADLINE = iox::units::Duration::fromSeconds(2);

using namespace iox2;

auto main() -> int {
    set_log_level_from_env_or(LogLevel::Info);
    auto node = NodeBuilder().create<ServiceType::Ipc>().expect("successful node creation");

    auto server1 = CustomServer::create(node, "ServiceSet/Service1");
    auto server2 = CustomServer::create(node, "ServiceSet/Service2");

    auto waitset = WaitSetBuilder().create<ServiceType::Ipc>().expect("");

    auto server1_guard = waitset.attach_deadline(server1, DEADLINE).expect("");
    auto server2_guard = waitset.attach_deadline(server2, DEADLINE).expect("");

    auto on_event = [&](WaitSetAttachmentId<ServiceType::Ipc> attachment_id) {
        if (attachment_id.has_event_from(server1_guard)) {
            server1.handle_event();
        } 
        if (attachment_id.has_event_from(server2_guard)) {
            server2.handle_event();
        } 
   
        return CallbackProgression::Continue;
    };

    waitset.wait_and_process(on_event).expect("");
    std::cout << "exit" << std::endl;

    return 0;
}

---

[elfenpiff]

Hey @TE-2

Whether separating the notification channels (/Request and /Response) is a good design choice.

Yes, this is fine. To give you some insights into our behind-the-scenes discussions on this topic. We played around with the idea that the user is able to provide a name to every port and that you can send data or notifications to a port with a specific name, or receive data and notifications only from a port with a specific name.

It would allow you to have an event and a request response service with the same service name, and the client and server port would have a name that you can freely pick. Then you could create two listeners (one with the name of the client port and one with the name server port) and could send a request to a specific server and notify just this one specific server.

Since we are still here in the brainstorming phase, I would like your opinion on whether this would make life easier for you.

For example, what happens if clientA fills the server’s request queue but never sends a corresponding notification?

You cannot prevent this. The worst case is that the client sends the request, and then the process dies. But iceoryx2 is made for mission critical system and we must follow the freedom of interference principle what roughly states that a bug in one part of the system shall not affect another part.
This is why every client - server pair has its own internal connection buffer and another connected client to the same server cannot fill the buffer of the client.

To be also more robust on the server side, you can adjust the request-handling-algorithm so that, when you are triggered by a request-has-been-sent-notification you collect all requests with Server::receive() until only an empty optional is returned.

When a clientB sends both a request and a notification, from my understanding the server could processes a mismatched request, since the two are not directly linked.

This can never happen and is ensured on multiple layers, also on the API. As soon as you receive a request, you get an ActiveRequest object which allows you to send responses only to the corresponding client and the client must keep the PendingResponse object to receive the responses to its request. So with iceoryx2 you shouldn't need to care about request or response ids. Yes they are there but behind the scenes. Also, every ActiveRequest - PendingResponse pair has again its own separate connection so that they are unaffected from any other kind of communication. With this architecture it should be impossible to send a response to the wrong request.

[TE-2]

Hi @elfenpiff,
thanks for your detailed feedback!

An additional hierarchy level such as named ports would actually be very helpful in our use case, as it would allow us to make a service somewhat clearer and more intuitive (a service is represented by only one service name).

With regard to pending enquiries, processing all existing enquiries makes sense.
What also comes to my mind is the order in which the request queues of the individual clients are processed on the server side with Server::receive(), or how they are arranged.
The background is that I would like to ensure that when the system wakes up, the most recent request, which should be associated with the trigger, is processed first. If several clients have filled their queues and never wake up the server, it is theoretically possible that the server will process older entries before the actually relevant request is handled. Depending on the load, this could lead to delays that are difficult to predict.