RFC LibraChannel Data Structure #1462

ankushagarwal · 2019-10-22T02:16:36Z

Summary

This PR is intended to foster discussion for issue #1323.

We introduce a new mpsc data structure called LibraChannel which has two ends libra_channel::Receiver, libra_channel::Sender similar to existing mpsc data structures.
This data structure is parametrized on
- The max number of messages to store per validator
- The per-validator queue style: LIFO, FIFO
- A generic message type T which implements a ValidatorMessage trait
The libra_channel::Sender<T> implements a fn put(&mut self, data: T) method which inserts the message into the internal data structure. This is a non-blocking synchronous call
The libra_channel::Receiver<T> implements the Stream trait so that the consumer of this receiver can easily use this inside a select! block
Added an example on how to use libra_channel in libra_channel_example.rs

Notes:

The validator fairness is not implemented properly right now. Right now all the messages are stored in a struct called PerValidatorQueue. Validator fairness can easily be implemented by improving the algorithms in add_message and get_message methods of PerValidatorQueue
Since we are implementing our own poll_next(), it gives us complete control of the order in which we want the consumer of this channel (FIFO, LIFO, etc)
We can create one libra_channel per message type and the consumer can do a select! over these as illustrated in the fn recv_loop

bmwill · 2019-10-22T18:10:05Z

common/channel/src/libra_channel.rs

+
+/// The sending end of LibraChannel
+#[derive(Clone)]
+pub struct LibraChannelSender<T: ValidatorMessage> {


Since the crate is named "libra-channel" you don't need to prefix every type in this crate with "LibraChannel". I would recommend dropping the prefix throughout the crate.

dmitri-perelman

This is going to be awesome! Some comments inline.

dmitri-perelman · 2019-10-22T22:22:10Z

common/channel/src/libra_channel.rs

+pub trait ValidatorMessage {
+    /// Extract the Validator from which this message arrived. This
+    /// will be used for ensuring fairness among validators.
+    fn get_validator(&self) -> &str;


minor: I don't think we want a string here.
Today we're using a struct AccountAddress for peer ids. So we could either use a template parameter, which implements Eq, Hash (in this case the channel is going to provide fairness based on an abstract Key), or a specific struct that we can use for validators. @phlip9, what's the ultimate struct in libra types for the validator? Is it AccountAddress?

It is AccountAddress

Updated to use AccountAddress. Let me know if you want me to update it to a generic parameter.

dmitri-perelman · 2019-10-22T22:27:40Z

common/channel/src/libra_channel.rs

+    fn get_validator(&self) -> &str;
+}
+
+/// Type is an enum which can be used as a configuration option for


nit: Could you extend the comments and explain the policy for treating the messages when the queue is full depending on the policy? (drop the oldest message in the queue if it's LIFO / drop the new message if it's FIFO).

dmitri-perelman · 2019-10-22T22:29:54Z

common/channel/src/libra_channel.rs

+}
+
+impl<T: ValidatorMessage> PerValidatorQueue<T> {
+    fn new(queue_type: Type, max_queue_size: usize) -> Self {


Could you please add a TODO for GC of old validator entries?

I gave this some thought and I was not completely convinced that a GC mechanism is needed. The event processor should consume the queues for validators and if a validator dies, we will just have a couple of empty queues in a HashMap, which I think is fine.

Maybe GC makes sense when the validators are highly dynamic and new validators keep coming and old validators keep dying.

dmitri-perelman · 2019-10-22T22:39:06Z

common/channel/src/libra_channel.rs

+        for q in self.per_validator_queue.values_mut() {
+            if !q.is_empty() {
+                return match self.queue_type {
+                    Type::FIFO => q.pop_back(),


I think it should be a reverse. The insert is using push_back(), so LIFO is pop_back() and FIFO is pop_front(), isn't it?

Yup. Fixed.

dmitri-perelman · 2019-10-22T22:41:39Z

common/channel/src/libra_channel.rs

+        }
+    }
+
+    fn add_message(&mut self, message: T) {


nit: Could you please add a high-level comment describing the the overall structure and the semantics? (just a couple of lines)

dmitri-perelman · 2019-10-22T22:51:07Z

common/channel/src/libra_channel.rs

+        }
+        if let Some(q) = self.per_validator_queue.get_mut(validator) {
+            q.push_back(message);
+            if q.len() == self.max_queue_size + 1 {


Imho a more explicit approach would be to have VecDeque of size max_queue_size, and then upon insert:

if q.len() == self.max_queue_size { match self.queue_type { // drop the new message in FIFO ordering Type::FIFO => (), // drop the oldest message in LIFO ordering Type::LIFO => { q.pop_front(); q.push_back(message); }, } } else { q.push_back(message); }

Obviously, it's a matter of taste, just saying :).

This one makes more sense. Updated.

dmitri-perelman · 2019-10-22T22:56:11Z

common/channel/src/libra_channel.rs

+}
+
+impl<T: ValidatorMessage> Sender<T> {
+    /// TODO: We can have this return the old data if the new data is replacing some existing data


Yup, although it'd then depend on FIFO / LIFO ordering (in FIFO ordering the returned value would be the message itself?)

True. Thinking some more about it, the actual message would not be of any use if we using it for back-pressure. Only a boolean value is enough to indicate that the queue is full.

dmitri-perelman · 2019-10-22T22:56:40Z

common/channel/src/libra_channel.rs

+    /// This puts the data into the internal data structure. This is a non-blocking synchronous call
+    pub fn put(&mut self, data: T) {
+        let mut shared_state = self.shared_state.lock().unwrap();
+        // TODO: Enforce maximum size


nit: this comment is unnecessary, the data structure is taking care of it.

dmitri-perelman · 2019-10-22T22:58:33Z

common/channel/src/libra_channel.rs

+    }
+}
+
+impl<T: ValidatorMessage> FusedStream for Receiver<T> {


Could you elaborate on this one a little bit more? Why is it needed and why is it always false? :)

Sorry I missed this. The FusedStream is a trait extension for the Stream trait.

Stream trait's specification has undefined behaviour when calling poll_next() after a Stream has ended (by sending a Poll::Ready(None) in its last response).

FusedStream removes this undefined behaviour by providing a is_terminated function to allow a caller (The select!() macro in this case) to decide whether to call poll_next() on the underlying Stream.

is_terminated may also return true if a Stream has become inactive and can no longer make progress and should be ignored or dropped rather than being polled again.

In the case of LibraChannel, I don't see this channel ending anytime, so I have set is_terminated to false.

If we want to provide a provision to close the channel from the sender side, we can do so by providing a close_channel() method on the Sender struct which sets a boolean value in the SharedState struct, but I don't see a case for that now.

ankushagarwal · 2019-10-23T00:05:10Z

cc @bothra90

dmitri-perelman · 2019-10-23T17:01:44Z

Hey I'm sorry for saying that: could we move the "round-robin" logic to a separate PR? This PR is already loaded with important business logic. I'd love to land the first piece with the async stuff and the basics of the per-validator queue and discuss the round robin in a separate PR / commit.

ankushagarwal · 2019-10-23T17:39:53Z

Happy to do that. I plan to close this PR and open two separate PRs for the libra channel and the round-robin data structure.

Let me know if you have any other comments or concerns in this prototype before I start formalizing this.

ankushagarwal · 2019-10-23T21:03:23Z

Opened other PRs which have formal implementations.

Summary libra_channel provides an mpsc channel which has two ends libra_channel::Receiver and libra_channel::Sender similar to existing mpsc data structures. What makes it different from existing mpsc channels is that we have full control over how the internal queueing in the channel happens and how we schedule messages to be sent out from this channel. Note that libra_channel does not provide an implementation of this internal queueing mechanism, but provides a trait (MessageQueue) which the consumers of this channel can use to create libra_channels' according to their needs. Related issue #1323 Related RFC PR : #1462 Test Plan Added unit tests Closes: #1483 Approved by: dmitri-perelman

ankushagarwal requested a review from dmitri-perelman October 22, 2019 02:16

ankushagarwal mentioned this pull request Oct 22, 2019

Smarter incoming channels for networking and consensus #1323

Closed

bmwill reviewed Oct 22, 2019

View reviewed changes

ankushagarwal requested a review from phlip9 October 22, 2019 18:22

ankushagarwal marked this pull request as ready for review October 22, 2019 18:45

dmitri-perelman reviewed Oct 22, 2019

View reviewed changes

RFC LibraChannel Data Structure

9b9de9b

ankushagarwal mentioned this pull request Oct 23, 2019

[enhancement] Create libra_channel module which provides an mpsc channel #1483

Closed

ankushagarwal closed this Oct 23, 2019

ankushagarwal deleted the libra-channel branch October 24, 2019 19:13

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

RFC LibraChannel Data Structure #1462

RFC LibraChannel Data Structure #1462

ankushagarwal commented Oct 22, 2019 •

edited

Loading

bmwill Oct 22, 2019

ankushagarwal Oct 22, 2019

dmitri-perelman left a comment

dmitri-perelman Oct 22, 2019

bothra90 Oct 23, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

dmitri-perelman Oct 22, 2019

ankushagarwal Oct 23, 2019

ankushagarwal commented Oct 23, 2019

dmitri-perelman commented Oct 23, 2019

ankushagarwal commented Oct 23, 2019 •

edited

Loading

ankushagarwal commented Oct 23, 2019

RFC LibraChannel Data Structure #1462

RFC LibraChannel Data Structure #1462

Conversation

ankushagarwal commented Oct 22, 2019 • edited Loading

Summary

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dmitri-perelman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ankushagarwal commented Oct 23, 2019

dmitri-perelman commented Oct 23, 2019

ankushagarwal commented Oct 23, 2019 • edited Loading

ankushagarwal commented Oct 23, 2019

ankushagarwal commented Oct 22, 2019 •

edited

Loading

ankushagarwal commented Oct 23, 2019 •

edited

Loading