Ping-Pong a memory address for repeated reuse

[nosjojo]

I'm working on a tool that needs to move a LOT of data (GB/s) between processes. Since I'm trying to operate as fast as possible, I don't want to reallocate memory for anything, I'd like to do all the processing in-place, but I'm having trouble figuring out how to correctly recycle a memory address.

I can't share my actual code, but I took the complex data example and mutated it into a pseudo recreation of it. In my actual implementation, these sub/pub instances are in different processes, but this example is enough to trigger the error.

use iceoryx2::prelude::*;
use iceoryx2_bb_container::vec::FixedSizeVec;

#[derive(Debug, Default, PlacementDefault)]
#[repr(C)]
pub struct Data {
    buffer: FixedSizeVec<u8, { 1024 * 1024 * 8 }>,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let node = NodeBuilder::new().create::<ipc::Service>()?;

    let service_out = node
        .service_builder(&"DataOut".try_into()?)
        .publish_subscribe::<Data>()
        .max_publishers(16)
        .max_subscribers(16)
        .open_or_create()?;

    let service_in = node
        .service_builder(&"DataReturn".try_into()?)
        .publish_subscribe::<Data>()
        .max_publishers(16)
        .max_subscribers(16)
        .open_or_create()?;
    let publisher_out = service_out.publisher_builder().create()?;
    let publisher_return = service_in.publisher_builder().create()?;
    let subscriber_out = service_out.subscriber_builder().create()?;
    let subscriber_return = service_in.subscriber_builder().create()?;

    let mut alldata = Vec::with_capacity(5);
    for _ in 0..5 {
        alldata.push(Data::default());
    }

    loop {
        let mut sample = publisher_out.loan_uninit()?;
        unsafe { Data::placement_default(sample.payload_mut().as_mut_ptr()) };
        let mut sample = unsafe { sample.assume_init() };

        let payload = sample.payload_mut();
        payload.buffer = alldata.pop().unwrap().buffer;

        sample.send()?;

        while let Some(sample) = subscriber_out.receive()? {
            println!("received data. {}", sample.buffer.len());
        }
        let mut ret_sample = publisher_return.loan_uninit()?;
        unsafe { Data::placement_default(ret_sample.payload_mut().as_mut_ptr()) };
        let mut ret_sample = unsafe { ret_sample.assume_init() };

        ret_sample.send()?;

        if let Some(returned_buf) = subscriber_return.receive()? {
            alldata.push(returned_buf.payload());
        }
    }
}

The crux of my issue comes in when I try to send the buffer back to the originator. I was storing the available buffers in a Vec, but the returned type in the message is &Data, not Data. If I try to dereference it, I get an error because it wants to copy and we (rightfully) don't have a Copy trait.

Did I miss something really obvious? Is there a better/correct way to do this? My main goal is to avoid reallocations, I'm already pushing this system to the edge.

[elfenpiff]

@nosjojo

My main goal is to avoid reallocations, I'm already pushing this system to the edge.

What kind of reallocations? iceoryx2 allocates all the memory it requires at once when you create the publisher - it is the data segment of the publisher and is managed by a pool allocator. This pool allocator is incredibly fast, so you do not have to worry about overhead here.
The placement new trait also ensures that the data is constructed in place of the memory so there are also no reallocations.

I was storing the available buffers in a Vec, but the returned type in the message is &Data, not Data.

I am a bit confused here. Do you mean that Sample::payload() returns a reference to Data which is stored in shared memory?

Just let me rephrase your use case so that I am sure that I understood it.

1. You do not want to reallocate memory.
2. A subscriber receives a huge chunk of memory.
3. You want to forward/send the underlying memory to another subscriber?? So that you do not have to write it again?

[nosjojo]

The answer is yes to all 3 questions. You can toss that code snippet into a valid project and it'll highlight what I mean by data vs. &data, since it's a build error.

The system this will run on has to do a lot of data capture and processing in near real time, so any overhead in the code, like memory allocation, can cause it to fail. So we front load the allocation. And since we move TBs of data through it, we can't just allocate for everything all at once.
The way we did this in C was to create shared memory and pass a semaphore around to control access. I'm essentially trying to recreate that flow.
Send address to capture card -> fill with data -> send address to processing -> return address for reuse.

[elfenpiff]

@nosjojo This will not yet work with iceoryx2.

What you are describing is something we describe as a pipeline. Process A produces data and sends it to process B, it consumes and modifies the data a bit and sends it to process C which again modifies the data. In your case process A and C would be the same and it would be a circular pipeline - which is not a problem.

But a pipeline comes with a restriction - only one subscriber is allowed. Otherwise one subscriber is reading a sample while another one is modifying it and causing data races.

Another challenge you would have here is what happens when, in this pipeline, one process crashes? We must be able to at least reclaim the lost sample since the publisher owns the data segment. When a subscriber dies and the sample cannot be reclaimed then it would cause a memory leak on publisher side.

I assume also that you do not need features like a subscriber buffer or a history for late joining subscribers with such an architecture. So, in the end, you need a pattern where you have a shared storage / memory for multiple processes and a token that is passed around that grants you read/write access to it. If you do not have the token, you are not allowed to read or write data into this shared storage.
Would such a thing solve your problem? I think we can "quickly" implement such a pattern - we already have a design for it but didn't expect that users are already requesting something like this.

[nosjojo]

At a glance, that solution sounds like it would work. If the design is viable, I'm willing to try it out.

I'm going to take a crack at removing the feedback loop and letting it allocate for each buffer, since it's always possible it will work.