Add on dispatch to extraction

[vertexclique]

Checklist

• [ ] tests are passing with cargo test.
• [ ] tests and/or benchmarks are included
• [ ] documentation is changed or added
• [ ] commit message is clear

[vertexclique]

@scrabsha Idea is removing redundant code like:

                loop {
                    MessageHandler::new(ctx.recv().await?)
                        .on_tell(|group_bcast: Arc<SignedMessage>, _sender_addr| {
                            let mut group_bcast = group_bcast;
                            MessageHandler::new(extract(group_bcast))
                                .on_broadcast(|broadcast: &DataModel, _sender_addr| {

                                 ...... DO SOMETHING WITH DISPATCHED MESSAGE

[scrabsha]

Thanks for the example. Took me a while, but i think i get it now!

[scrabsha]

Hi there!

I made a compiling version of what you wrote in this pr (code). As you can see, it is very similar to what you wrote here. I have the following questions:

• what should we do when the message contained in the Arc is not a broadcast message? My understanding so far is that a dispatch corresponds to a message that is sent to several actors. As such, it is always a broadcast. Am i right?
• I am not sure that the extract function works as expected. If more than one process starts waiting for its exclusive access to the inner data, it may result in a deadlock. Is there anything in the code that prevent it?

(tldr: I think I lack experience in how bastion works to properly implement on_dispatch)

[vertexclique]

@scrabsha

what should we do when the message contained in the Arc is not a broadcast message? My understanding so far is that a dispatch corresponds to a message that is sent to several actors. As such, it is always a broadcast. Am i right?

yes they are always broadcast

I am not sure that the extract function works as expected. If more than one process starts waiting for its exclusive access to the inner data, it may result in a deadlock. Is there anything in the code that prevent it?

No solution atm as of yet. It needs to be verified that someone doesn't hold a strong reference for whole programs' time. Good way of getting rid of it is that having a memory reclamation for the messages. I am writing a Mr scheme for it. Until then we can use crossbeam-epoch.

[scrabsha]

I am not sure that the extract function works as expected. If more than one process starts waiting for its exclusive access to the inner data, it may result in a deadlock. Is there anything in the code that prevent it?

No solution atm as of yet. It needs to be verified that someone doesn't hold a strong reference for whole programs' time. Good way of getting rid of it is that having a memory reclamation for the messages. I am writing a Mr scheme for it. Until then we can use crossbeam-epoch.

@vertexclique toghether with o0ignition0o, we found a simpler solution. As we always are dealing with a broadcast, we can clone the SignedMessage itself (using Msg::try_clone internally). This is rather inexpensive since we are cloning an Arc, and fixes any possible deadlock.

The changes can be found here. I need to write some tests before opening a PR.

[o0Ignition0o]

+1

I think it will fit well with our dispatcher refactorings, that will replace broadcast* (which aren't broadcasts)

[vertexclique]

Yes, that's better. I will try this with Kafka streamer if you open a PR :)

[o0Ignition0o]

https://github.com/bastion-rs/bastion/pull/319

@vertexclique getting there! I can even register children on the fly and they start receiving messages <3

Documentation and a couple of tests missing, otherwise i m pretty happy with the API

[vertexclique]

Closing in favor of #319