Support for rkyv

[chikko80]

Hi everyone,

First off, thank you for your incredible work on this project—I’m genuinely amazed by what you’ve built!

I’ve been searching for a low-latency IPC solution for my app for quite some time, and I could never quite find the right fit—until now

For my use case, I noticed that support for non-SHM types is currently missing. However, I see that it’s already mentioned on your roadmap. Would it be possible to get an approximate ETA for this feature?

Also, if someone has a bit of time, it would be really helpful if the roadmap could be updated (it seems to have not been refreshed for about four months). This would make it easier for us to follow what’s coming next and plan accordingly.

Thanks again for all your hard work!

[elfenpiff]

@chikko80 Thanks for the hint. I updated the roadmap and also put this as a point in the release document.

I will take a look into rkyv to provide some rough estimation. I think it is unrealistic for v0.6, but if it is not too much of an effort to implement we can prioritize this for v0.7 - which would be beginning of Q2 2025.

[chikko80]

I built a small wrapper around your library to use the dynamic data feature described in the examples to kinda do what i initially asked for. It uses rkyv to serialize data and perform zero-copy deserialization upon receiving (from what i am understanding).

My implementation might not be the most performant or the way you would implement it, but it works for my current use case (non-prod environment) — as far as I can tell. Maybe, it provides some quick insights for you, how the library could be used.

use iceoryx2::{
    port::{publisher::Publisher, subscriber::Subscriber},
    prelude::*,
};

pub struct IceoryxDynamicPayloadPublisher {
    publisher: Publisher<ipc::Service, [u8], u128>,
}

impl IceoryxDynamicPayloadPublisher {
    pub fn new(server_name: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let node = NodeBuilder::new().create::<iceoryx2::service::ipc::Service>()?;

        let service = node
            .service_builder(&server_name.try_into()?)
            .publish_subscribe::<[u8]>()
            .user_header::<u128>()
            .open_or_create()?;

        let publisher = service
            .publisher_builder()
            .initial_max_slice_len(65536)
            .allocation_strategy(AllocationStrategy::PowerOfTwo)
            .create()?;

        Ok(Self { publisher })
    }

    pub fn send_payload(
        &self,
        payload: impl std::fmt::Debug
            + for<'a> rkyv::Serialize<
                rkyv::rancor::Strategy<
                    rkyv::ser::Serializer<
                        rkyv::util::AlignedVec,
                        rkyz::ser::allocator::ArenaHandle<'a>,
                        rkyv::ser::sharing::Share,
                    >,
                    rkyv::rancor::Error,
                >,
            >,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let bytes = &rkyv::to_bytes::<rkyv::rancor::Error>(&payload).unwrap()[..];
        let test = self.publisher.loan_slice_uninit(bytes.len()).unwrap();
        let mut sample = test.write_from_slice(bytes);
        *sample.user_header_mut() = Self::get_current_time();
        sample.send()?;
        Ok(())
    }

    fn get_current_time() -> u128 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_micros()
    }
}

pub struct IceoryxDynamicPayloadSubscriber {
    pub subscriber: Subscriber<ipc::Service, [u8], u128>,
}

impl IceoryxDynamicPayloadSubscriber {
    pub fn new(server_name: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let node = NodeBuilder::new().create::<ipc::Service>()?;

        let service = node
            .service_builder(&server_name.try_into()?)
            .publish_subscribe::<[u8]>()
            .user_header::<u128>()
            .open_or_create()?;

        let subscriber = service.subscriber_builder().create()?;

        Ok(Self { subscriber })
    }
}

Upon receiving data, I interpret the bytes as follows:

let payload = sample.payload();

// Or you can use the unsafe API for maximum performance
#[allow(unsafe_code)]
let archived = unsafe { rkyv::access_unchecked::<ArchivedIpcResponse>(payload) };

Here, archived will reflect the data you published via iceorxy.

[elfenpiff]

@chikko80 this is really helpful! Thank you for the code snippet.