Add Join system combinator, allow for multiple multiple input piping

[Trashtalk217]

Objective

• Make it possible to pipe multiple systems into a handler system.

Solution

• Introduce a Cartesian product operator that takes two systems with outputs and combines them into one system with output (systemA(input), systemB(input). This makes the following construction possible:

add_systems(
    ...
    output_systemA.product(output_systemB).pipe(handler_system)
    ...
    output_systemA.product(output_systemB).product(output_systemC).pipe(handler_system2)
    ...
)

---

Changelog

Added

• CartesianProduct with a combine implementation in combinators.rs.
• product method in the mod.rs

Some notes

The main drawback of this approach is the excessive number of parems that turn up in the handler system (I do like a good Lisp, but I digress). Maybe the optimal API would look like this,

(systemA, systemB, systemC, systemD).pipe(handler_system)

// where
fn handler_system(In(a, b, c, d): In<(A, B, C, D)>, ...)

which would be cleaner, but also requires some macro work I'm not sure I can do. There's also a small question of running all the I/O systems in parallel or sequentially. My first guess would be to use abilities Bevy already has with its scheduling, but since I don't know how that works, I can't speak on it.

[github-actions]

Welcome, new contributor!

Please make sure you've read our contributing guide and we look forward to reviewing your pull request shortly ✨

[alice-i-cecile]

This bound feels incorrect: if we're joining the input we should just be able to pass it by value all the way through.

[JoJoJet]

The value gets passed to the first system but not returned, so in order to pass it to the second system we have to be able to copy or clone it.

[Trashtalk217]

Yes, I've tried to remove the Copy, but wasn't able to

[alice-i-cecile]

We should probably try an make these example a little more useful / representative. I'll chew on simple demos.

[Trashtalk217]

My original use case was for a puzzle game where a couple of individual systems queried the puzzle to see if the puzzle was solved (think of checking rows and columns for a sudoku) and then return a boolean if the requirement was met.

The puzzle is solved if all the systems returned true. Technically this can also work with the and combinators, but if you also wanted to display to the user which condition wasn't met, those combinators wouldn't work and you'd need a specific handler system.

[alice-i-cecile]

I'm in favor of this: it greatly extends the expressiveness of this piping API in intuitive ways. I'd ultimately like to see this paired as a join/split API, but there's no need to handle that here.

The main drawback of this approach is the excessive number of parems that turn up in the handler system (I do like a good Lisp, but I digress). Maybe the optimal API would look like this,

Agreed that this is the optimal API: that would lean on the all_tuples macro and will be a bit more advanced. The current API is passable but like you said, extremely lispy.

There's also a small question of running all the I/O systems in parallel or sequentially. My first guess would be to use abilities Bevy already has with its scheduling, but since I don't know how that works, I can't speak on it.

These will end up running sequentially, although I'm reasonably sure you won't be able to write multiple input systems with incompatible data accesses. System piping basically works by "gluing" the systems together, creating a single atomic chunk for the purposes of the scheduler.

There's probably a good argument to make this "atomic" behavior independent of the system piping mechanism (see bevyengine/rfcs#46 for more context). Definitely far outside the scope of an initial PR here.

[SpecificProtagonist]

(systemA, systemB, systemC, systemD).pipe(handler_system)

That would be ideal. If the goal is to pipe multiple systems into one, this is the most direct way to express it, rather than expressing that you want to build the product of the outputs one after the other and feed that into another system.

but also requires some macro work I'm not sure I can do

I'm taking a look at it and the macro part is relatively easy, but I'm stuck at figuring out the types. If you can manually write the 2-tuple and 3-tuple version, I'll gladly turn it into a macro invocation for you.

This feels like it would be more helpfully called join to me.

Whether join or product, I'd say this joins/builds the product of the system outputs, not of the systems themselves.

[JoJoJet]

This looks correct to me.

My one concern is that I'm not sure how useful this would be in general. And given how easy it is to write custom system combinators, this doesn't strictly need to be built in to bevy.
I'm not opposed to adding this, though.

[JoJoJet]

The value gets passed to the first system but not returned, so in order to pass it to the second system we have to be able to copy or clone it.

[Trashtalk217]

Okay, I will likely revert this commit, but I think it's an interesting idea.

In order to try and get the (SystemA, SystemB, SystemC).pipe(handler_system) to work, I've implemented System for (System, System).

This works(ish), but has several problems.

• It takes a lot of code to do it for all tuples we care about, even if we used macro magic, the resulting code would be annoying.
• It's not clear to me how I would implement component.acces() and archetype_component_access(). The current implementation is wrong, but the best I could do.
• You currently still have to turn functions into systems by hand (this can probably be circumvented, but since I've started to reject the idea, I didn't bother).
• It's also not necessarily the 'canonical' way a tuple of systems should be interpreted. It already has a different interpretation inside add_systems(...) where a tuple of systems signifies an anonymous system set.

My conclusion is that a better notation would probably be along the lines of

join(SystemA, SystemB, SystemC, SystemD).pipe(handler_system)

[JoJoJet]

If you end up going the tuple route, it'd be better to implement IntoSystem for (impl IntoSystem, impl IntoSystem). Rather than having the tuple itself be a system, it would convert into a custom system combinator. (Not quite sure if tuples are the best direction, but it's good to explore at least.)

[ItsDoot]

If you end up going the tuple route, it'd be better to implement IntoSystem for (impl IntoSystem, impl IntoSystem).

I ran into wanting this today for easily composable one-shot systems.

[JoJoJet]

I had another thought. Will the (system_a, system_b).pipe(..) syntax conflict with system scheduling?

If I call schedule.add_systems((system_a, system_b)), will these be added as two separate systems or as a compound system? I think not, since CI is passing, but what is preventing this from happening? Is this something that could regress in the future?

This PR is starting to grow on me, but I'd like to make sure we don't ship something that horribly breaks scheduling.

[JoJoJet]

Digging a little deeper, seems like it's because (system_a, system_b) implements IntoSystem<Out = ((), ())>, while IntoSystemConfigs is only implemented for IntoSystem<Out = ()>.

This is incredibly subtle, I'd like to make this more explicit somehow.

[JoJoJet]

Here's another solution for join.

pub fn join<A, B, AMarker, BMarker>(
    mut system_a: A,
    mut system_b: B,
) -> impl FnMut(In<A::In>, ParamSet<(A::Param, B::Param)>) -> (A::Out, B::Out)
where
    A: SystemParamFunction<AMarker>,
    B: SystemParamFunction<BMarker, In = A::In>,
    A::In: Copy,
{
    move |In(input), mut params| {
        let out_a = system_a.run(input, params.p0());
        let out_b = system_b.run(input, params.p1());
        (out_a, out_b)
    }
}

Instead of a.join(b).pipe(..) or (a, b).pipe, you'd call join(a, b).pipe(..). This doesn't even require system combinators, since all of the combination occurs within the body of the join function. Also, anyone can just copy-paste this code and use it right now. I don't think it has to be in bevy proper to work.

[ActuallyHappening]

Here's another solution for join.

pub fn join<A, B, AMarker, BMarker>(
    mut system_a: A,
    mut system_b: B,
) -> impl FnMut(In<A::In>, ParamSet<(A::Param, B::Param)>) -> (A::Out, B::Out)
where
    A: SystemParamFunction<AMarker>,
    B: SystemParamFunction<BMarker, In = A::In>,
    A::In: Copy,
{
    move |In(input), mut params| {
        let out_a = system_a.run(input, params.p0());
        let out_b = system_b.run(input, params.p1());
        (out_a, out_b)
    }
}

Instead of a.join(b).pipe(..) or (a, b).pipe, you'd call join(a, b).pipe(..). This doesn't even require system combinators, since all of the combination occurs within the body of the join function. Also, anyone can just copy-paste this code and use it right now. I don't think it has to be in bevy proper to work.

This works wonders thank you!
This function should be found in the documentation or linked from somewhere because it makes piping far more expressive. However, calling anything of the sort:

join(system1.pipe(system2), ...)

fails. Is there a workaround for this? My logic looks something like this:

.add_systems(
    Update,
    join(
        gather_player_movement.pipe(vectorise_input_flags),
        get_max_power_vectors.pipe(get_relative_strengths),
    )
    .pipe(apply_thrust_stages),
)

[JoJoJet]

In order to pass a system into this style of combinator, the system needs to use this style of combinator all the way down, which .pipe() does not do. You'd have to make your own version of pipe, like this:

pub fn pipe<A, B, AMarker, BMarker>(
    mut system_a: A,
    mut system_b: B,
) -> impl FnMut(In<A::In>, ParamSet<(A::Param, B::Param)>) -> B::Out
where
    A: SystemParamFunction<AMarker>,
    B: SystemParamFunction<BMarker, In = A::Out>,
{
    move |In(input), mut params| {
        let value = system_a.run(input, params.p0());
        system_b.run(value, params.p1())
    }
}

Which you could then use like

.add_systems(
    Update,
    join(
        pipe(gather_player_movement, vectorise_input_flags),
        pipe(get_max_power_vectors, get_relative_strengths),
    )
    .pipe(apply_thrust_stages),
)

[ActuallyHappening]

In order to pass a system into this style of combinator, the system needs to use this style of combinator all the way down, which .pipe() does not do. You'd have to make your own version of pipe, like this:

pub fn pipe<A, B, AMarker, BMarker>(
    mut system_a: A,
    mut system_b: B,
) -> impl FnMut(In<A::In>, ParamSet<(A::Param, B::Param)>) -> B::Out
where
    A: SystemParamFunction<AMarker>,
    B: SystemParamFunction<BMarker, In = A::Out>,
{
    move |In(input), mut params| {
        let value = system_a.run(input, params.p0());
        system_b.run(value, params.p1())
    }
}

Which you could then use like

.add_systems(
    Update,
    join(
        pipe(gather_player_movement, vectorise_input_flags),
        pipe(get_max_power_vectors, get_relative_strengths),
    )
    .pipe(apply_thrust_stages),
)

Again thank you very much this worked like a charm!
My logic has grown considerably but by defining a join2 and join3 variant, I can express (rather complicatedly) my final player movement / thrust logic:

pipe(
	join3(
		pipe(
			join2(
				pipe(
					join2(gather_input_flags, get_base_normal_vectors),
					flag_normal_vectors,
				),//.pipe(info),
				max_velocity_magnitudes,
			),//.pipe(info),
			get_relative_strengths,
		),//.pipe(info),
		get_base_normal_vectors,
		force_factors,
	),//.pipe(info),
	save_thrust_stages,
).pipe(apply_thrust).pipe(ignore),

This takes a long time to compile though, even on latest nightly, currently rendering piping as a suboptimal for current bevy projects at the type complexity I am working at. I can probably simplify the types a lot and compress into one system (which I was using before), but I need to thread through a bunch of system parameters manually which is tedious. The manual solution is still the only way to achieve my goal currently (without waiting 30 seconds for incremental compiles, using older only ~5 seconds)

[JoJoJet]

Another option is to turn each of those systems into a regular function which accepts a custom #[derive(SystemParam)] struct, then have one big system that collects all of the system params and uses them to call each function in order.

[ActuallyHappening]

Right now manually threading systems is not to bad, with only around 8 different system params threaded through.
When I start optimising for performance I see only one feasible route, however, which is using resources to store the results of non-trivial computations. I would consider a 'trivial' computation one that I only do once per Update and only consume once per Update, so standard piping like trivial_creator.pipe(trivial_consumer) would be perfectly acceptable.

But my current system has a few systems whose computations are used multiple times in different contexts, so by piping them using .pipe(...) I would be calling them twice per Update since I need to .pipe them twice, e.g. .add_systems(Update, (collect.pipe(consumer_1), collect.pipe(consumer_2)). Depending on how important / performant the system collect is, this may or may not be an acceptable performance cost. Assuming its not, you could store the computation in an 'anonymous' resource and 'anonymous' system set, anonymous in quotations because without macros this isn't strictly true and because the resource / system set types might eventually be used for more than merely piping data:

use bevy::prelude::*;

// example types

#[derive(Resource, Clone)] // Clone for simplicity, there may be a way to relax this requirement
pub struct ExampleCollectReturn;

/// do some computation, maybe read from a [Query]
pub fn collect() -> ExampleCollectReturn {
  ExampleCollectReturn
}

/// time for demonstration purposes, you can put any system params you like
pub fn consumer_1(In(_collect_return): In<ExampleCollectReturn>, time: Res<Time>) {
  // do something with the result
}

/// query for demonstration purposes, you can put any system params you like
pub fn consumer_2(In(_collect_return): In<ExampleCollectReturn>, query: Query<Entity>) {
  // do something with the result
}

// optimization specific types

#[derive(Resource)]
pub struct _PipingResource {
  collect_return: ExampleCollectReturn,
}

#[derive(Hash, PartialEq, Eq, Debug, Clone, SystemSet)]
pub struct _PipingSet;

fn _store_collect_return(In(collect_return): In<ExampleCollectReturn>, mut commands: Commands) {
  commands.insert_resource(_PipingResource { collect_return });
}

fn _get_collect_return(_data: Res<_PipingResource>) -> ExampleCollectReturn {
  // There is probably way to avoid cloning and use references / Box, but its fine for my use cases
  _data.collect_return.clone()
}

// example usage (plugin)

pub struct ExamplePlugin;
impl Plugin for ExamplePlugin {
  fn build(&self, app: &mut App) {
    // not optimized way
    app.add_systems(Update, (collect.pipe(consumer_1), collect.pipe(consumer_2)));

    // optimized way
    app.add_systems(
      Update,
      (
        collect.pipe(_store_collect_return).in_set(_PipingSet),
        _get_collect_return.pipe(consumer_1).after(_PipingSet),
        _get_collect_return.pipe(consumer_2).after(_PipingSet),
      ),
    );
  }
}

If this pattern is desirable, it is possible to design a macro-free API for this using a new derive macro, assuming you can break the collectors and consumers into distinct stages. I can draft a proposal in this issue if I begin using this pattern.