Add a DirectedGossip struct

[tomaka]

Provides the primitive that will make it possible to tackle paritytech/polkadot#1453

Adds a new gossip module in sc-network, containing DirectedGossip.
The way it works is what is described here.

Here's how you would plug it into Polkadot, regarding paritytech/polkadot#1453:

• When the network bridge learns about a new connection, it builds a new DirectedGossipPrototype. This type doesn't have any template parameter.
• The NetworkBridgeEvent::PeerConnected enum variant should contain a new field of type DirectedGossipPrototype (or some abstraction of it). The bridge passes the newly-created object there.
• Whatever subsystem picks up the PeerConnected event turns the prototype into an actual DirectedGossip, passing the protocol name, queue size, and messages type.
• The subsystem can then manipulate the queue of notifications using push_or_discard, push_unbounded, or retain.
• When a PeerDisconnected message is received, the subsystem discards the DirectedGossip for this peer.

[tomaka]

I marked as "in-progress" because tests have to be written.

[twittner]

I suggest to remove this trait. The two type parameters B and H are only necessary in the private spawn_task method and adding those is far less code than this trait plus impl. And your calls to NetworkService::notification_sender are not dynamically dispatched.

[tomaka]

It's necessary for DirectedGossipPrototype to work. See your other comment.

[twittner]

This is pub so I assume this will be used from somewhere external? Could you explain this a little more please? Currently it is not in use and the DirectedGossip and task construction could happen directly in DirectedGossip::new.

[tomaka]

I admit that this is type is specifically targeted for the Polkadot use case.

In Polkadot's code, the so-called network bridge communicates via messages with other subsystems. For example, when we connect, a PeerConnected message is sent.

The network bridge doesn't know what is the type of the networking messages that the various subsystems would manipulate (the M generic in this PR), and the various subsystems don't know what the B and H generics would be.

This DirectedGossipPrototype would easily plug into that scheme. We would send a DirectedGossipPrototype to the subsystems, without the need for the subsystem to know what B and H would be.

[twittner]

Should Polkadot-specific code not be put in Polkadot?

[rphmeier]

The network bridge doesn't know what is the type of the networking messages that the various subsystems would manipulate (the M generic in this PR), and the various subsystems don't know what the B and H generics would be.

I am actually changing this right now, so we may not need this type. Will push a PR shortly

[rphmeier]

paritytech/polkadot#1535 - now the network bridge is aware of the specific information flowing over the network and we may be able to avoid these prototypes. However, different subsystems will still want Senders for different types. However, the sender for a specific variant should also be encoding as |x| ValidationProtocolV1::SomeConcreteVariant(x).encode() which I think is covered by the encode-fn part of the API here

[tomaka]

Should Polkadot-specific code not be put in Polkadot?

To answer specifically this: the way I see it, this code isn't really Polkadot-specific.
This code provides a convenient-to-use high-level API on top of a low-level primitive, and people are free to use either the convenient API if it suits them, or the low-level primitive if it doesn't suit them.

I think that, in general, we want to be as "universal" as possible when designing low-level code, but as long as the lower-level primitives are exposed, higher-level code can be a bit more targeted towards certain use-cases.

[mxinden]

Do I understand correctly that this uses the unstable feature to have futures-timer? If so, should we not do that across the entire crate for all usage of futures-timer?

[tomaka]

No, it is for Convar. It is unfortunate that we have to depend on the unstable feature, but I couldn't find any crate other than async-std that provides an asynchronous Condvar.

[mxinden]

If this is something we want to avoid, one can use a channel as a condvar.

use futures::channel::mpsc::channel;

fn main() {
    let (tx, rx) = channel(0);
    
    let producer = async move {
        // Produce something and put it to place X.
        
        // Signal the consumer that that something can be optained.
        tx.send(()).await;
    }
    
    let consumer = async move {
        // Wait for the signal. Maybe include a timeout like you do today with 10 secs.
        let _ = rx.next().await;
        
        // Access place X.
    }
}

Problem is that this can't be used in the Drop implementations and thus cleanup would depend on the 10 sec timeout.

[twittner]

Problem is that this can't be used in the Drop implementations [...]

It does not always need to. When the Sender is dropped, the receiver will notice and the task can terminate. Instead of let _ = rx.next().await; one would write if rx.next().await.is_some() { ... }. In the Drop impl of QueueGuard one can use Sender::try_send.

[tomaka]

I find it a bit weird to use a channel, which involves an additional Arc, Mutex and Vec, just to wake up a task, rather than a Waker.

But I have now also tried using a Waker, and the implementation is considerably more tricky and difficult to read because of potential race conditions and having to introduce manual polling within an async function and having to implement your own Waker.

Before going on, I'd like to understand what is wrong with the Condvar solution, as a Condvar is exactly the tool that is designed for this specific job.

[mxinden]

Why offer two ways to queue_or_discard? Is writing self.lock_queue().await.push_or_discard(message); as a user not fine as well?

[rphmeier]

This API could return a Result<(), M> or similar as well

[tomaka]

It's a convenient shortcut.

[rphmeier]

Same here. It would be good to get back the message if the queue is full

[tomaka]

But what would you do with the message that is returned? Put it in another queue?
The only sane things you can do when the queue is full is either discard the message or force-close the connection.

Additionally, what if the connection with the remote is closed? Are we supposed to return back the message as well? If so, then it's very problematic because we can't detect this.

[rphmeier]

The only sane things you can do when the queue is full is either discard the message or force-close the connection.

Isn't retain another option? We could decide to do that based on the type of the message we are trying to send.

And is there no sane thing you can do? Why can't you wait for space to appear in the queue or something like that?

[tomaka]

The way I see retain is that you'd call it all the time, even if there is space in the buffer. As far as I can tell, when a message is obsolete, there is no point in leaving it in the queue anyway.

Why can't you wait for space to appear in the queue or something like that?

The entire reason for this API to exist is to remove the need for any waiting. See also this paragraph.

Ultimately there has to be a code somewhere that holds some sort of HashMap<PeerId, DirectedGossip>. If it needs to send a message to one of the peers and its buffer is full, then it shouldn't wait for this peer and instead continue its processing.

[rphmeier]

It would be nice to have a way to wait, although I agree that the API for that should be used sparingly so you don't degrade to the performance of the slowest peer. There are cases where we don't want to drop messages, for instance when responding to a validator's request.

[tomaka]

There are cases where we don't want to drop messages

I believe that everything that would fall in this category should be covered by request-response protocols.

Rather than adding a wait, I could restore the push_unbounded method that I've removed after a review.

[mxinden]

@rphmeier @tomaka is this conversation resolved?

[tomaka]

I have addressed all concerns, except for the QueueSenderPrototype (formerly DirectedGossipPrototype) that might be useless and add a very small overhead.

@rphmeier Do you confirm that the NetworkBridge would directly create the QueueSenders itself? In other words, that it knows what the M generic is, and that there's no need for some "prototype" anymore?
If so I'll remove it.

[rphmeier]

Do you confirm that the NetworkBridge would directly create the QueueSenders itself? In other words, that it knows what the M generic is, and that there's no need for some "prototype" anymore?

Yes, after paritytech/polkadot#1537 is in

[tomaka]

bot rebase

[ghost]

Rebasing.

[twittner]

Any comment on @mxinden's suggestion to use a channel instead of a Condvar which would avoid depending on unstable features from async-std?

[tomaka]

If we do indeed want to not depend on unstable, then I'd suggest having a Waker instead of a channel.
However, the Mutex/Condvar "pattern" is what you're supposed to use for that kind of code, so do we actually want to get rid of the Condvar?

[tomaka]

bot rebase

[ghost]

Rebasing.

[tomaka]

Friendly reviewing ping

[mxinden]

This looks good to me overall.

I am still in favor of using a channel instead of an unstable Condvar (see https://github.com/paritytech/substrate/pull/6803/files#r466192207). As far as I can tell this would also remove the need for the stop_task AtomicBool (see https://github.com/paritytech/substrate/pull/6803/files#r466298691). I wouldn't block merging the pull request for that reason though.

[mxinden]

Suggested change

	pub struct QueueSender<M> {
	pub struct QueuedSender<M> {

Not sure whether this is a typo or intentional. In case it is the former I would prefer QueuedSender.

[mxinden]

👍

[mxinden]

@rphmeier @tomaka is this conversation resolved?

[rphmeier]

@rphmeier @tomaka is this conversation resolved

On discarding, I will defer to Pierre for now. If we find a need, the behavior can be easily changed later.

[tomaka]

bot merge

[ghost]

Trying merge.