Compatibility of WaitSet and wait operations with threads in python

[TDietz21]

Hello together,

I am currently using iceoryx2 with the python language bindings and aim to develop a simple server/client exchange via the request/response pattern.

To this end, I normally use a Server class that sets up communication with iceoryx2 and then starts a thread to run the event-based communication. However, I ran into the issue that after setting up this thread, the Server class does not initialize properly.
I have isolated all of this in, what I think, is the most minimal example for this behaviour.

The server event_loop sets up the iox2 objects and then performs wait_and_process() via the waitset to listen to any events.
The client function also sets up the iox2 objects and directly sends the payload, notifies the request event service and waits for the response from the eventloop.

However, executing this code does not terminate. To rule out that this is due wait_and_process() being activated, thus needing an event to wake_up and terminate, I replaced the 'while True' loop with a 'running' variable that ensures the loop is only run once.
Still, just by looking at the terminal, execution seems to freeze once the event_loop ist started.

Some more things I already considered.

1. Running the event_loop and the client_fn in separate processes works, also with a while True: loop
2. Simply replacing the wait_and_process function with wait_and_process_with_timeout() results in a timeout on the client side

I am a bit clueless on what to do now. I also suspected whether the GIL is handled correctly, but the iceoryx2 documentation specifically states that for blocking_waits in python the GIL is released.

My question is therefore, does anyone know what else could be the issue with this code and can confirm that this form of communication properly aligns with the iceoryx2 communication patterns/examples?

minimal_issue_example.py

import logging
import time
import threading
import iceoryx2 as iox2
import ctypes

main_logger = logging.getLogger(__name__)
cycle_time = iox2.Duration.from_millis(1)

SERVICE_NAME = "Test"

class TransmissionData(ctypes.Structure):
    """The strongly typed payload type."""

    _fields_ = [("x", ctypes.c_int32)]

    def __str__(self) -> str:
        """Returns human-readable string of the contents."""
        return f"TransmissionData {{ x: {self.x} }}"

    @staticmethod
    def type_name() -> str:
        """Returns the system-wide unique type name required for communication."""
        return "TransmissionData"


def handle_event(server, listener, notifier, _logger):
    """
    Process all pending events and requests.
    """
    for event in listener.try_wait_all():
        _logger.info(event)
        if not event:
            break

        active_request = server.receive()
        if not active_request:
            _logger.warning("Event signaled but no request available in queue")
            continue
        _logger.info("Request Received")

        payload: TransmissionData = active_request.payload().contents
        old_x = payload.x

        response_payload = TransmissionData()
        response_payload.x = 2 * old_x

        # Send response
        active_request.send_copy(response_payload)
        notifier.notify()
        _logger.info("Response Sent")

        active_request.delete()


def event_loop():
    # Setup of Iceoryx2 objects
    event_logger = logging.getLogger(__name__ + "_event_loop")
    node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)
    service_rr = (
        node.service_builder(iox2.ServiceName.new(SERVICE_NAME))
        .request_response(TransmissionData, TransmissionData)
        .open_or_create()
    )

    request_event = node.service_builder(iox2.ServiceName.new(SERVICE_NAME + "/Request")).event().open_or_create()
    response_event = node.service_builder(iox2.ServiceName.new(SERVICE_NAME + "/Response")).event().open_or_create()

    iox_server = service_rr.server_builder().create()
    listener = request_event.listener_builder().create()
    notifier = response_event.notifier_builder().create()

    waitset = iox2.WaitSetBuilder.new().create(iox2.ServiceType.Ipc)
    guard = waitset.attach_notification(listener)

    # Handling eventloop via iceoryx2
    running = True
    while running:
        attachments, result = waitset.wait_and_process()
        # attachments, result = waitset.wait_and_process_with_timeout(cycle_time)
        if result in (
            iox2.WaitSetRunResult.TerminationRequest,
            iox2.WaitSetRunResult.Interrupt,
        ):
            break

        for attachment in attachments:
            if attachment == iox2.WaitSetAttachmentId.from_guard(guard):
                event_logger.info("Event Handle Triggered")
                handle_event(iox_server, listener, notifier, event_logger)
        running = False

    return


def client_fn():
    # Setup of Iceoryx2 objects
    client_logger = logging.getLogger(__name__ + "_client_fn")

    node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)
    service_rr = (
        node.service_builder(iox2.ServiceName.new(SERVICE_NAME))
        .request_response(TransmissionData, TransmissionData)
        .open_or_create()
    )
    request_event = node.service_builder(iox2.ServiceName.new(SERVICE_NAME + "/Request")).event().open_or_create()
    response_event = node.service_builder(iox2.ServiceName.new(SERVICE_NAME + "/Response")).event().open_or_create()

    client = service_rr.client_builder().create()
    notifier = request_event.notifier_builder().create()
    listener = response_event.listener_builder().create()

    # Sending data via iceoryx2 request/response
    payload = TransmissionData()
    payload.x = 5

    iox_request = client.loan_uninit()
    iox_request = iox_request.write_payload(payload)
    pending_response = iox_request.send()
    notifier.notify()
    client_logger.info(f"Request sent.")

    wait_result = listener.timed_wait_one(iox2.Duration.from_secs(1))
    if not wait_result:
        client_logger.error(f"Response timeout for service: {SERVICE_NAME}")

    iox_response = pending_response.receive()
    if iox_response:
        response: TransmissionData = iox_response.payload().contents
        client_logger.info(f"Response received with value {str(response)}")

    return


if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)

    event_thread = threading.Thread(
        target=event_loop,
        name="EventLoop",
        daemon=False,
    )
    client_thread = threading.Thread(
        target=client_fn,
        name="Client",
        daemon=False,
    )

    event_thread.start()
    main_logger.info("Event Thread Started")

    client_thread.start()
    main_logger.info("Client Thread Started")

[TDietz21]

I will close this as I moved the discussion as a topic to the iceoryx2 community