Note: Moved from GitHub discussion to new iceoryx2 community
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.
- Running the event_loop and the client_fn in separate processes works, also with a
while True:loop - Simply replacing the
wait_and_processfunction withwait_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")