paritytech / orchestra

A partial actor pattern with a global orchestrator.

orchestra

The orchestra pattern is a partial actor pattern, with a global orchestrator regarding relevant work items.

proc-macro

The proc macro provides a convenience generator with a builder pattern, where at it's core it creates and spawns a set of subsystems, which are purely declarative.

    #[orchestra(signal=SigSigSig, event=Event, gen=AllMessages, error=OrchestraError)]
    pub struct Opera {
        #[subsystem(MsgA, sends: [MsgB])]
        sub_a: AwesomeSubSysA,

        #[cfg(any(feature = "feature1", feature = "feature2"))]
        #[subsystem(MsgB, sends: [MsgA])]
        sub_b: AwesomeSubSysB,
    }

• Each subsystem is annotated with #[subsystem(_)] where MsgA respectively MsgB are the messages being consumed by that particular subsystem. Each of those subsystems is required to implement the subsystem trait with the correct trait bounds. Commonly this is achieved by using #[subsystem] and #[contextbounds] macro.
  • #[contextbounds(Foo, error=Yikes, prefix=wherethetraitsat)] can applied to impl-blocks and fn-blocks. It will add additional trait bounds for the generic Context with Context: FooContextTrait for <Context as FooContextTrait>::Sender: FooSenderTrait besides a few more. Note that Foo here references the name of the subsystem as declared in #[orchestra(..)] macro.
  • #[subsystem(Foo, error=Yikes, prefix=wherethetraitsat)] is a extension to the above, implementing trait Subsystem<Context, Yikes>.
• error= tells the orchestra to use the user provided error type, if not provided a builtin one is used. Note that this is the one error type used throughout all calls, so make sure it does impl From<E> for all other error types E that are relevant to your application.
• event= declares an external event type, that injects certain events into the orchestra, without participating in the subsystem pattern.
• signal= defines a signal type to be used for the orchestra. This is a shared "tick" or "clock" for all subsystems.
• gen= defines a wrapping enum type that is used to wrap all messages that can be consumed by any subsystem.
• Features can be feature gated by #[cfg(feature = "feature")] attribute macro expressions. Currently supported are any, all, not and feature.

    /// Execution context, always required.
    pub struct DummyCtx;

    /// Task spawner, always required
    /// and must implement `trait orchestra::Spawner`.
    pub struct DummySpawner;

    fn main() {
        let _orchestra = Opera::builder()
            .sub_a(AwesomeSubSysA::default())
            .sub_b(AwesomeSubSysB::default())
            .spawner(DummySpawner)
            .build();
    }

In the shown main, the orchestra is created by means of a generated, compile time erroring builder pattern.

The builder requires all subsystems, baggage fields (additional struct data) and spawner to be set via the according setter method before build method could even be called. Failure to do such an initialization will lead to a compile error. This is implemented by encoding each builder field in a set of so called state generics, meaning that each field can be either Init<T> or Missing<T>, so each setter translates a state from Missing to Init state for the specific struct field. Therefore, if you see a compile time error that blames about Missing where Init is expected it usually means that some subsystems or baggage fields were not set prior to the build call.

To exclude subsystems from such a check, one can set wip attribute on some subsystem that is not ready to be included in the orchestra:

    #[orchestra(signal=SigSigSig, event=Event, gen=AllMessages, error=OrchestraError)]
    pub struct Opera {
        #[subsystem(MsgA, sends: MsgB)]
        sub_a: AwesomeSubSysA,

        #[subsystem(MsgB, sends: MsgA), wip]
        sub_b: AwesomeSubSysB, // This subsystem will not be required nor allowed to be set
    }

Baggage fields can be initialized more than one time, however, it is not true for subsystems: subsystems must be initialized only once (another compile time check) or be replaced by a special setter like method replace_<subsystem>.

A task spawner and subsystem context are required to be defined with Spawner and respectively SubsystemContext implemented.

Debugging

As always, debugging is notoriously annoying with bugged proc-macros, see feature "expand".

Features

feature "expand"

expander is employed to yield better error messages. Enable with --features=orchestra/expand.

feature "dotgraph"

Generate a directed graph which shows the connectivity according to the declared messages to be send and consumed. Enable with --features=orchestra/dotgraph. The path to the generated file will be displayed and is of the form ${OUT_DIR}/${orchestra|lowercase}-subsystem-messaging.dot. Use dot -Tpng ${OUT_DIR}/${orchestra|lowercase}-subsystem-messaging.dot > connectivity.dot to convert to i.e. a png image or use your favorite dot file viewer. It also creates a .svg alongside the .dot graph, derived from the .dot graph for direct usage.

Caveats

No tool is without caveats, and orchestra is no exception.

Large Message Types

It is not recommended to have large messages that are sent via channels, just like for other implementations of channels. If you need to transfer data that is larger than a few dozend bytes, use Box<_> around it or use a global identifier to access persisted state such as a database, depending on the use case.

Response Channels

It seems very appealing to have response channels as part of messages, and for many cases, these are a very convenient way of maintaining a strucutured data flow, yet they are ready to shoot you in the foot when not used diligently. The diligence required is regarding three topics:

1. Circular message dependencies leading to a dead-lock for single threaded subsystems
2. Too deep message dependencies across many subsystems
3. Delays due to response channels

Each of them has a variety of solutions, such as local caching to remedy frequent lookups of the same information or splitting up subsystem into multiple to avoid circular dependencies or merging tiny topologically closely related to one subsystem in some exceptional cases, but strongly depend on the individual context in which orchestra is used.

To find these, the feature dotgraph is providing a visualization of all interactions of the subsystem to subsystem level (not on the message level, yet) to investigate cycles. Keep an eye on warnings during the generation phase.

License

Licensed under either of

• Apache License, Version 2.0, (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0)
• MIT license (LICENSE-MIT or http://opensource.org/licenses/MIT)

at your option.

orchestra's Issues

move unbounded or bounded sends to declaration, rathern than impl locality by default

Unify the sender API and only provide one API async fn send_message(msg: ..). bounded or unbounded is an impl detail that should be provided at the subsystem declaration, i.e.#[subsystem(DisputeDistributionMessage, sends: [ RuntimeApiMessage, DisputeCoordinatorMessage as unbounded, NetworkBridgeMessage,])] and thesend_messageimpl will take care of using the correct channels.ChannelsOut will also contain less entries.

   #[subsystem(DisputeDistributionMessage, sends: [
              RuntimeApiMessage,
              DisputeCoordinatorMessage as unbounded,
              NetworkBridgeMessage,
])]

and the send_message impl will take care of using the correct channels. ChannelsOut will also contain less entries.

Create a virtual manifest

https://doc.rust-lang.org/cargo/reference/workspaces.html#virtual-workspace

dotgraph build errors

paritytech/polkadot-sdk#2836

[orchestra-proc-macro-0.3.3/src/graph.rs:207] err_msg = "Expected value after assignment"
Hetscher/Hiccup: Failed to parse dotfile
caused by: Expected value after assignment
expander: writing orchestra-expansion-c5de4377c228.rs

`CHANNEL_CAPACITY` is unused

Since 0.0.5 I get the warning message CHANNEL_CAPACITY from generated code.

Crates.io ownership

Hey @drahnr 👋

could you please

cargo owner --add parity-crate-owner --add github:paritytech:core-devs

for orchestra, orchestra-proc-macro and prioritized-metered-channel? 🙏

Thank you in advance :)

Implement prioritized channels

Quoting from Polkadot issue paritytech/polkadot-sdk#824:

Extend the metered::* types to have send::<Priority = Standard> generic argument which can either be Droppable, Low, Standard, High (not an enum!, all marker types).

Pros:

• allow a more nuanced handling of messages
• keep the semantics simple

Cons:

• Complexity:
  • Will require more logic around sending
  • Requires multiple channels per subsystem

Disable signed commits

Signed commits do not allow me to rebase, even if my branch sits exactly on-top of main. So everything has to be squashed.

Implement `try_send_message`

This would allow implementing some custom logic to handle blocking sends.

Reuse subsystems in different orchestra

Some time ago, we introduced a minimal Polkadot node in Cumulus. It runs a subset of subsystems that is usually present in the polkadot node. For the initial implementation, I replaced unneeded subsystems with DummySubsystem.

This works quite well, but I would like to clean this up and define a new orchestra that only accepts the required subsystem set.

Naive definition of a new orchestra and usage with polkadot subsystems gives this error:

error[E0271]: type mismatch resolving `<CollatorOverseerSubsystemContext<AvailabilityRecoveryMessage> as SubsystemContext>::OutgoingMessages == AvailabilityRecoveryOutgoingMessages`
   --> client/relay-chain-minimal-node/src/collator_overseer.rs:167:4
    |
167 |         .availability_recovery(AvailabilityRecoverySubsystem::with_chunks_only(
    |          ^^^^^^^^^^^^^^^^^^^^^ expected enum `polkadot_overseer::AvailabilityRecoveryOutgoingMessages`, found enum `collator_overseer::AvailabilityRecoveryOutgoingMessages`
    |
    = note: enum `collator_overseer::AvailabilityRecoveryOutgoingMessages` and enum `polkadot_overseer::AvailabilityRecoveryOutgoingMessages` have similar names, but are actually distinct types

I guess this is expected since we define the trait location via prefix manually. So it seems like there is a 1:1 mapping between subsystem <-> orchestra?

Before diving deeper into this, I would like to have an opinion from someone more experienced with the orchestra (maybe @drahnr?). Is this even supported currently? How would I go about this?

stage 2 enhancements

After paritytech/polkadot#3427 is implemented, there is another suite of improvments in the pipeline:

• #10
• #9
• #11

Provide separate sinks for signals and messages

As it was discussed, a subsystem must be able to process signals and messages separately so it can prioritize signal processing if message processing takes a long time. Two separate async methods, recv_signal() and recv_msg(), should be introduced to separate the flows.

Implement per subsystem message_capacity override.

Something like #[subsystem(message_capacity=32768, SomeMessage, sends: [])].

This is needed to properly size the channels for subsystems that need to handle occasionally large bursts of work without blocking any potential sender.

attempt to get rid of `contextbounds`-proc-macro

since it's required to have the Context: SubsystemContext around to spawn a task, #[overseer::contextbounds(..)] was introduced. This could be avoided by allowing every Sender: ${Subsystem}SenderTrait to also implement a new trait SpawnTask or even trait SpawnNamed, since all that is required is sending a future to the overseer. Common practice for i.e. the job based subsystems and some subsystems. This could alleviate the need for that second proc-macro entirely and simplify the code based and improve compilation speed (hypothesis).

Handle where clauses for baggage generics

Left over from paritytech/polkadot#3427

broadcast_signal: parallelize sending and still use a timeout for completion

If a subsystem is having a hard time and it's signal queue becomes full, all the subsequent subsystems will receive the signal much later on. We could have a timeout for completion of all futures, but this would still back pressure on the source of the signal. We could instead track individual timeouts for each send with a FuturesUnordered.

For example, if in the below snippet of generated code sub2 blocks (up to timeout - 10s), then updates to sub3..6 will also be deleayed, leading to a general slowdown, as well as missing on some messages sent between unaffected systems and sub2 which can only be received after the signals are processed.

    /// Broadcast a signal to all subsystems.
    pub async fn broadcast_signal(
        &mut self,
        signal: OverseerSignal,
    ) -> ::std::result::Result<(), SubsystemError> {
        self.sub1.send_signal(signal.clone()).await?;
        self.sub2.send_signal(signal.clone()).await?;
        self.sub3.send_signal(signal.clone()).await?;
        self.sub4.send_signal(signal.clone()).await?;
        self.sub5.send_signal(signal.clone()).await?;
        self.sub6.send_signal(signal.clone()).await?;

        let _ = signal;
        Ok(())
    }

Memory bounded channels

We are using bounded channels to allow the consumer to backpressure the producer when it doesn't keep up. This is useful, but currently we can only do it based on message count. When the sender side is forwarding messages(notifications) from a network stack, (like distribution subsystems in Polkadot), messages are not created equal wrt how much memory they consume. Most of the messages are < 512b, but there are also big messages of up to MAX_NOTIFICATION_SIZE (64kb).

Larger channels are very useful as they allow to buffer bursts of messages at the cost of memory. Implementing the bounds on memory rather than message count helps us be very accurate about this tradeoff and dynamically allowing very large queues of smaller messages.

Consider exposing an `UnboundedMeteredSender` as orchestra handle

This is useful to avoid spawning a task when sending a high ToF message. See paritytech/polkadot#7038

Possible extractions from the polkadot repo

• subystem-test-helpers as orchestra-test-helpers
• malus extension/replacement framework as orchestra-malus

Allow subsets of the graph to be printed

          For `Polkadot` the output of the graph is quite hard to reason about, so one improvement I was thinking about is to add some parameter such that only selected subsystems are present in the graph. WDYT ?

Originally posted by @sandreim in #43 (comment)

Avoid dependency on consumed message type, generate a marker type

The generated code strongly relies on the consumed message type, which complicates the generated code significantly. Using a marker type and additional trait to annotate sent and consumed types, would likely be advantageous to reduce complexity and have one defining information anchor.

Include network protocol utilities into this repo?

Ideally Orchestra is a 'one-stop-shop' for all the tools you need to extend Substrate services with custom node logic. It would be useful to provide the network-bridge or a factory for network bridges, as well as some of the tooling around req/res that exists in the Polkadot repo

Confusing error for unconsumed message

#[orchestra(gen=AllMessages, ..)]
struct X {
#[subsystem(MsgA, sends: [Unconsumed])
foo: Foo,
#[subsystem(MsgB, sends: [Unconsumed])
bar: Bar,
}

will raise an error about missing enum variant AllMessages::Unconsumed which is non-obvious how to resolve.

Will be obsoleted once #11 is implemented, would possibly make sense to add a sense check to the proc-macro for the time being.
CC @MathisWellmann