[Proposal] Separating callback chain building and actual invocation

[NobodyXu]

Currently, the callback chain building and the actual invocation is done at the same time.

This means that invocation of Generator::run needs to check for initialization and that under debug mode, there might be 2n stack frames where n is the number of pipelines.

Here, I propose a new API for Generator:

trait Callback {
    fn call(&mut self) -> GeneratorResult;
}

trait Generator {
    type Output: Callback;

    fn run(&mut self, output: impl FnMut(Self::Output) -> crate::ValueResult) -> Self::Output;
}

Call to Generator::run will initialize the pipeline (e.g. for Dedup, it would be getting the first element) and then return a callback to a closure that have the entire pipelines already built.

As a result, even under the debug build where no inline is performed, the stack frames used is guaranteed to be equal to the number of pipelines.

The function returned will be called inside for_each, try_for_each and IteratorAdaptor.

This would also make it clear to new contributors on which part is responsible for initialization and which part is accounted for the "loop body" and possibly remove some uses of Option.

[AndWass]

Hi

I don't think you can do a trait like that, you can't have impl FnMut() -> GeneratorResult as a function return value :(

I agree it would be nice to seperate the pipeline creation, and pipeline execution into seperate steps. However I am not sure it can be truly done. In Dedup for instance, if it fails to get the first element at build-time, it will still have to try to get it at run-time.

Say the upstream Generator is trying to fetch data from a channel, if it fails (no element to read at this time) it reports stop, if the channel is closed it reports complete, otherwise it sends each available value through the pipeline. With such a source-generator you can't rely on the first value being available when you build the pipeline. And so you must be able to do the initialization step even when you do later runs on the pipeline.

If we had a marker trait trait NeverStoppingGenerator: Generator we could move the initialization into Dedup::new for any source that is a NeverStoppingGenerator and remove that code entirely from Dedup::run when Src implements NeverStoppingGenerator, but that requires specialization I think?

[NobodyXu]

I don't think you can do a trait like that, you can't have impl FnMut() -> GeneratorResult as a function return value :(

IMHO it should be possible, as I recalled from my memory rust's async function is just a sugar for fn f(...) -> impl Future

Say the upstream Generator is trying to fetch data from a channel, if it fails (no element to read at this time) it reports stop, if the channel is closed it reports complete, otherwise it sends each available value through the pipeline. With such a source-generator you can't rely on the first value being available when you build the pipeline. And so you must be able to do the initialization step even when you do later runs on the pipeline.

If so, then I guess the terminating for_each will just simply return I guess?

And IMO the initialization process can be restarted by calling for_each again,

It is an interesting problem though: I don't think pushgen will work well with channel, which is async by nature.

While pushgen employs callback to avoid branching, it isn't async, which I think program which uses channel would expect.

[AndWass]

IMHO it should be possible, as I recalled from my memory rust's async function is just a sugar for fn f(...) -> impl Future

An async fn some_async() is just sugar for fn some_async() -> impl Future yes, but the Future trait itself, is alot more basic. And a quick test on the playground confirms that you can't have impl Trait as return-type in traits.

If so, then I guess the terminating for_each will just simply return I guess?

for_each will stop and return GeneratorResult::Stopped yes.

And IMO the initialization process can be restarted by calling for_each again,

True

It is an interesting problem though: I don't think pushgen will work well with channel, which is async by nature.

While pushgen employs callback to avoid branching, it isn't async, which I think program which uses channel would expect.

I was more thinking if you have a generator that uses try_recv to generate data. It might not be a common use-case but I think it can prove valuable

[NobodyXu]

An async fn some_async() is just sugar for fn some_async() -> impl Future yes, but the Future trait itself, is alot more basic. And a quick test on the playground confirms that you can't have impl Trait as return-type in traits.

Yep, turns out only way to do this is to add an item to the Test, which implements another trait Callback since we cannot infer the type of closure.

playground

I was more thinking if you have a generator that uses try_recv to generate data. It might not be a common use-case but I think it can prove valuable

Yep, though I think my proposal will not forbid this API.

[AndWass]

Yah I definitely think it's an interesting idea to split the cold and hot paths. Some experimentation and testing is necessary before committing to it though

[NobodyXu]

What kind of experiments do you think is necessary for this?

[NobodyXu]

I was just trying to define a clear example for the new trait, however I immediately discover that without impl FnMut() -> GeneratorResult as the return value, the code failed apart because the output is a generic parameter and it is necessary for run to return different types based on the generic parameter.

[AndWass]

Ah ok. I will try and think of something to try as well, but right now I don't have any clear idea.

[NobodyXu]

I just realized that all pipelines either takes &mut to itself (like Dedup) or takes nothing (Cloned).

So what I now proposed instead is to use an erased interface:

use core::ffi::c_void;
use core::mem::transmute;
use core::ptr::null_mut;
use core::marker::PhantomData;

/// ErasedFnPointer can either points to a free function or associated one that
/// `&mut self`
struct ErasedFnPointer<'a, T, Ret> {
    struct_pointer: *mut c_void,
    wrapper: fn(&ErasedFnPointer<'a, T, Ret>, T) -> Ret, 
    fp: *const (),
    phantom: PhantomData<&'a ()>
}
impl<'a, T, Ret> ErasedFnPointer<'a, T, Ret> {
    fn associated_wrapper<S>(&self, param: T) -> Ret {
        let sp = self.struct_pointer as *mut S;
        let s = unsafe { &mut (*sp) };
        let fp = unsafe { transmute::<_, fn(&mut S, T) -> Ret>(self.fp) };
        fp(s, param)
    }
    
    pub fn from_associated<S>(struct_pointer: &'a mut S, fp: fn(&mut S, T) -> Ret)
        -> ErasedFnPointer<'a, T, Ret>
    {
        ErasedFnPointer {
            struct_pointer: struct_pointer as *mut _ as *mut c_void,
            fp: fp as *const (),
            wrapper: ErasedFnPointer::associated_wrapper::<S>,
            phantom: PhantomData
        }
    }
    
    fn free_wrapper(&self, param: T) -> Ret {
        let fp = unsafe { transmute::<_, fn(T) -> Ret>(self.fp) };
        fp(param)
    }
    
    pub fn from_free(fp: fn(T) -> Ret) -> ErasedFnPointer<'static, T, Ret> {
        ErasedFnPointer {
            struct_pointer: null_mut(),
            fp: fp as *const(),
            wrapper: ErasedFnPointer::free_wrapper,
            phantom: PhantomData
        }
    }
    
    pub fn call(&self, param: T) -> Ret {
        (self.wrapper)(self, param)
    }
}

trait Callback {
    fn call(&mut self) -> GeneratorResult;
}

trait Generator {
    type Output: Callback;

    fn run(&mut self, output: ErasedFnPointer<Self::Output, crate::ValueResult>) -> Self::Output;
}

The compiler can easily inline the *_wrapper and the underlying function pointer, while Generator can have a stable interface without any generic parameter, so that it can return a single type only.

This makes the separation I described before possible without losing performance, but it also might help with compilation time a bit since there is no need for monomorphization and no duplicate code is generated when one pipeline uses the other more than once (e.x. Dedup).

playground

Edit:

One problem with ErasedFnPointer is that it doesn’t have PhantomData<&S> field, so it might actually caused a problem for enforcing the mutability rule.

I think I got it fixed now with lifetime annotation and PhantomData.

[AndWass]

Hi

I'm not sold at the moment to be honest. This is IMO starting to get more complicated than the system already in place. Have you implemented any of the more involved adaptors (like Dedup or Flatten) using this pattern? I think before any change is decided on the basic benchmarks must be ported to this kind of system and checked against current system.

It is also quite easy to run into UB with the pointer solution; playground

[NobodyXu]

Hi

I'm not sold at the moment to be honest. This is IMO starting to get more complicated than the system already in place. Have you implemented any of the more involved adaptors (like Dedup or Flatten) using this pattern? I think before any change is decided on the basic benchmarks must be ported to this kind of system and checked against current system.

It is also quite easy to run into UB with the pointer solution; playground

About the UB part, I have actually updated my code yesterday and introduces lifetime annotation inside ErasedFnPointer along with PhantomData to fix the lifetime issues.

With the updated code, it will actually fail to compile.

Edit:

Looks like I forgot to update the link yesterday and now I have updated the link in my original comment.

[NobodyXu]

It’s actually possible to implement take one easily with closure, since in rust, associated functions are just regular functions:

fn main() {
    let mut x = None;
    ErasedFnPointer::from_associated(&mut x, |x, param| {
        *x = Some(param);
        println!("{:#?}", x);
    }).call(1);
}

[NobodyXu]

For the part where deduped values is pushed, I guess the parameter output (the callback) in Generator::run has to be saved inside Dedup so that it can be accessed.

However, it does raise issues with lifetime, unless GeneratorExt::for_each and GeneratorExt::try_for_each takes self instead of &mut self.

Personally, I don’t think of this as a major downside, as users can always wraps the pipelines in a closure to use it multiple times, but it does impact existing uses of pushgen.

[NobodyXu]

Here’s a link to the code inside gist

[AndWass]

Thanks for the writeup! At the moment I won't pursue this split. I don't really see the benefit right now and I don't really have the time to do any benchmarking and testing.