Allow general safe use of ctx.spawn() and ability to inject messages via unbounded channel
#60
Conversation
Signed-off-by: Andrei Sandu <[email protected]>
Signed-off-by: Andrei Sandu <[email protected]>
vstakhov
left a comment
vstakhov
left a comment
There was a problem hiding this comment.
We need to bump the middle version with this change, as we are going to change the existing API.
| /// Send sink for `Message`s to be sent to a subsystem. | ||
| pub tx_bounded: crate::metered::MeteredSender<MessagePacket<Message>>, | ||
| /// Unbounded send sink for `Message`s to be sent to a subsystem. | ||
| pub tx_unbounded: crate::metered::UnboundedMeteredSender<MessagePacket<Message>>, |
|
|
||
| /// Gather running subsystems' outbound streams into one. | ||
| to_orchestra_rx: #support_crate ::stream::Fuse< | ||
| to_orchestra_rx: Option<#support_crate ::stream::Fuse< |
There was a problem hiding this comment.
why did this become an Option?
There was a problem hiding this comment.
I have been using this here: paritytech/polkadot-sdk@291eefe This seemed to be the path of least resistance.
There was a problem hiding this comment.
So the reason is you need to move the receiver to another future task, and therefore - since stop and all the other associated functions require self (non referential) as arguments, you have to be able to retain Self as type (no unpacking) due to the above as well.
I am not a particular fan of using Option<_>, it obfuscates the rationale. Note that with the Option<_> the previous implementation of spawning tasks becomes rather awkward, since one always has to check if rx_* is Some before polling. I'd say either we add a new type, or we create an <OrchestraName>Inner type that is only the part that is being spawned and one then can unpack the <OrchestraName> into the inner type plus receivers.
Not sure there is a better alternative from the top of my head.
This I do understand.
I am missing context here. Whatever long-blocking operation in the subsystems message processing loop is going to be an issue. I don't quite understand how spawn is specific here? I don't quite get how the change enables that. Could you elaborate or add an example under |
|
Ping :) |
There are 2 changes:
to_orchestra_rxto be handled in aOverseertask.Reasoning: the Polkadot overseer handles both events and job spawners in the same select loop. This can create deadlocks in subsystems that use ctx.spawn() to create a job and wait for it, only if the same subsystem has a full channel and an external messages needs to be passed to it from the overseer select loop.
Not sure this is the best API to expose this, happy to hear alternatives