Awaiting a Listener in tokio

General
1

Hi,

Our application is async, built on tokio, and we want to receive iceoryx2 samples in async code (publish-subscribe): wake up when a sample arrives, without polling and without dedicating a thread to a blocking wait.

Current solution
publisher calls notify() on an event service after every send(), and subscriber registers the listener’s fd with tokio:

type Service = ipc_threadsafe::Service;

let event = node.service_builder(&"my_service/notify".try_into()?).event().open_or_create()?;

// Publisher
let notifier = event.notifier_builder().create()?;
sample.send()?;
notifier.notify()?;

// Subscriber
let listener = event.listener_builder().create()?;
let fd = unsafe { BorrowedFd::borrow_raw(listener.file_descriptor().native_handle()) }
      .try_clone_to_owned()?;
let ready = AsyncFd::with_interest(ListenerFd(fd), Interest::READABLE)?;

loop {
    if subscriber.has_samples()? { return Ok(()); }
    let mut guard = ready.readable().await?;
    guard.clear_ready();
    listener.try_wait_all(|_| {})?;
}

Please notice the unsafe.
Is this the correct approach, or is there a better way?

2

This is currently the right approach. We also wanted to provide an async API for iceoryx2, but have not yet found the time to implement it.

The idea is to provide more high-level ports and a more convenient API that might not be certifiable but would be perfect for such use cases.

It would roughly look like this:

pub struct AsyncListener {
  listener: Arc<Listener>,
  ready: AsyncFd<ListenerFd>,
}

impl AsyncListener {
  pub async fn wait(&self) -> io::Result<Vec<EventId>>;
}

The idea would be that you could use it like that:

let listener = event.async_listener_builder().create()?;

loop {
  for event_id in listener.wait().await? {
    // add logic depending on what event id you have received
  }
}

Maybe it makes even sense to go one step further and combine events always with pubsub on such a high level API. So that you have a high-level AsyncPublisher/AsyncSubscriber in the iceoryx2-userland crate which automatically sends events for:

  • sample has been sent
  • subscriber connected/disconnected
  • publisher connected/disconnected
  • sample consumed
  • …

What would be valuable would be an API draft or pseudo-example that illustrates for us how you would like to use iceoryx2 in an async environment.

Based on such a draft we could plan the next steps on our side.

3

Thank you!

This is a new project, once we settle I will try to provide more information about the use-case.