Connected LivenessService to comms pub/sub #686

sdbondi · 2019-08-21T05:00:02Z

Description

LivenessHandler connected up to a stream of PingPong messages
Refactored the way services are initialized, keeping it super simple
and to allow more flexibility

Tari Services are one or more (usually) long-running asynchronous tasks which furnish request via a provided interface (called a Handle). Dropping the handle must cancel the task. This is (usually) implemented using an mpsc channel and/or subscription stream. That is, once all senders have disconnected, the receiver stream will close and so will the task. Services usually subscribe to one or many comms messages and update state or perform operations as necessary.

Services are set up by implementing the ServiceInitializer trait - this trait can be passed to the StackBuilder which is responsible for initializing the services and signalling when all services are initialized.

Services can access other service handles allowing them to make requests on other services. This is done by retrieving the appropriate handle from the ServiceHandles collection which is provided to every ServiceInitializer.

Handles can easily (but not necessarily) be implemented by using the tari_p2p::executor::transport::channel(S) function. This function takes in a service S and returns a Requester/Responder pair - the requester can be used to make asynchronous requests to the given service S. A handle, however, can be implemented in whatever way the service implementer sees fit.

Motivation and Context

Closes #676
Ref #621

How Has This Been Tested?

Unit tests (integration tests to follow)

Types of changes

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to change)
Feature refactor (No new feature or functional changes, but performance or technical debt improvements)
New Tests
Documentation

Checklist:

I'm merging against the development branch
I ran cargo-fmt --all before pushing
My change requires a change to the documentation.
I have updated the documentation accordingly.
I have added tests to cover my changes.

CjS77 · 2019-08-21T05:25:00Z

base_layer/p2p/src/executor/handles/and_then_service.rs

@@ -0,0 +1,140 @@
+// Copyright 2019 The Tari Project


You should submit this as a PR to the Tower repo, if it it's generally useful

Was thinking that it could potentially be a FutureExt combinator for tower services, will look into it

base_layer/p2p/src/executor/handles/and_then_service.rs

base_layer/p2p/src/executor/handles/mod.rs

philipr-za

This is some crazy stuff, when we get it polished it's going to be really slick.

So I have to agree with Cayle that the ServiceHandle, AndThenService and service composition stuff is extremely difficult to grok. I think it will be valuable to spend some time writing an extensive overview doc about how it all fits together because the few docstrings that there are do not actually describe how the full system goes together and how it resolves and, to be honest, I still don't 100% understand the full process of resolving the handles into a composed multi service thing.

Secondly, I think the Executor service stuff should move out of the P2P crate as I think it is going to be the framework for all domain services not just those used for the P2P stuff so lets talk about where it goes? This is essentially the framework we are going to use to compose our full applications.

base_layer/p2p/src/executor/handles/future.rs

base_layer/p2p/src/executor/handles/and_then_service.rs

base_layer/p2p/src/executor/handles/future.rs

base_layer/p2p/src/executor/stack.rs

base_layer/p2p/src/services/liveness/handler.rs

philipr-za · 2019-08-21T09:55:45Z

base_layer/p2p/src/services/liveness/mod.rs

+
+/// Initializer for the Liveness service handle and service future.
+pub struct LivenessInitializer {
+    inbound_subscriber: Arc<InboundTopicSubscriber<TariMessageType>>,


This feels a little weird. Perhaps we should find a way for a service to require that on creation it is passed an already constructed TopicSubscription with the correct topic and then make it the problem of the thing that constructs the services?

Ye this is service dependent - so you can do it in any way you like - however I'd like to keep service specific code in the initializer for that service rather than passing in LivenessInitializer::new(comms.inbound_subcriber().subscription(TariMessageType::new(NetMessage::PingPong))

You can do LivenessInitializer::new(comms.inbound_subscriber()) - might even just pass in comms and leave the linking of comms to the service completely up to the service initializer implementation.

I think pass in the Comms

Done - although the downside is that this requires a full comms stack to test. I've added a function to create the LivenessInitializer from a subscriber to keep tests easy. In the near future, we probably want to make it easy to setup a full (development) comms stack in integration tests

Making a mock Comms stack that just has the Subscriber and Future based interface should be pretty easy.

philipr-za · 2019-08-21T10:01:20Z

base_layer/p2p/src/services/domain_deserializer.rs

+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let msg = self.message.take().expect("poll called twice on Deserializer");
+        blocking(|| {


This seems overkill to need blocking here? It's just a memory mapping? This is definitely a discussion to have to decide where the threshold is for an operation to be put into Blocking but a straight up memory based read seems like its plenty fast enough to be done directly?

Yup - maybe - not sure either - it's a memory mapping because we happen to be using bincode - however that could change and is up to the MessageFormat impl. Tasks are super cheap to create according to tokio. Definitely up for debate and I'm sure we'll come across many of these cases.

philipr-za · 2019-08-21T10:02:48Z

base_layer/p2p/src/services/domain_deserializer.rs

+
+impl<T: MessageFormat> Future for DomainMessageDeserializer<T> {
+    type Error = BlockingError;
+    type Item = Result<(MessageInfo, T), MessageError>;


If we are going as far as to make a future for this then I think its time to have a proper struct to hold the fields of the (MessageInfo, T) tuple.

No strong opinions on that - will do in the next iteration :)

It is more apparent that this tuple is janky when you look at the code that uses it. When you have matches and code that are using result.0 and result.1 to access the two fields it doesn't look good.

I've started with using a DomainMessage<T> struct in another branch which I'll PR soon.

philipr-za · 2019-08-21T12:07:24Z

So I just read what you wrote in the PR and I think that goes a long way to clarifying how the Service framework should be used. The academic in me also thinks that the whole structure could be neatly conveyed/summarized with a simple diagram.

CjS77 · 2019-08-22T10:19:51Z

Could you squash all your commits together please?

- LivenessHandler connected up to a stream of PingPong messages - Refactored the way services are initialized, keeping it super simple and to allow more flexibility.

Merge pull request #686 Description LivenessHandler connected up to a stream of PingPong messages Refactored the way services are initialized, keeping it super simple and to allow more flexibility Tari Services are one or more (usually) long-running asynchronous tasks which furnish request via a provided interface (called a Handle). Dropping the handle must cancel the task. This is (usually) implemented using an mpsc channel and/or subscription stream. That is, once all senders have disconnected, the receiver stream will close and so will the task. Services usually subscribe to one or many comms messages and update state or perform operations as necessary. Services are set up by implementing the ServiceInitializer trait - this trait can be passed to the StackBuilder which is responsible for initializing the services and signalling when all services are initialized. Services can access other service handles allowing them to make requests on other services. This is done by retrieving the appropriate handle from the ServiceHandles collection which is provided to every ServiceInitializer. Handles can easily (but not necessarily) be implemented by using the tari_p2p::executor::transport::channel(S) function. This function takes in a service S and returns a Requester/Responder pair - the requester can be used to make asynchronous requests to the given service S. A handle, however, can be implemented in whatever way the service implementer sees fit. Motivation and Context Closes #676 Ref #621 How Has This Been Tested? Unit tests (integration tests to follow)

This was referenced Aug 21, 2019

Integration tests for liveness service #687

Merged

Added async cursive PingPong example app #688

Merged

CjS77 reviewed Aug 21, 2019

View reviewed changes

CjS77 assigned philipr-za Aug 21, 2019

sdbondi force-pushed the sb-p2p-liveness-recv-pingpong branch from ba5efe2 to 968e122 Compare August 21, 2019 07:06

philipr-za reviewed Aug 21, 2019

View reviewed changes

sdbondi force-pushed the sb-p2p-liveness-recv-pingpong branch from 074a9d4 to 9cbd1a9 Compare August 21, 2019 17:28

philipr-za previously approved these changes Aug 22, 2019

View reviewed changes

sdbondi dismissed philipr-za’s stale review via 4cf5d76 August 22, 2019 10:51

sdbondi force-pushed the sb-p2p-liveness-recv-pingpong branch from aa6fc28 to 4cf5d76 Compare August 22, 2019 10:51

Connected LivenessService to comms pub/sub

f7408fc

- LivenessHandler connected up to a stream of PingPong messages - Refactored the way services are initialized, keeping it super simple and to allow more flexibility.

sdbondi force-pushed the sb-p2p-liveness-recv-pingpong branch from 4cf5d76 to f7408fc Compare August 22, 2019 12:28

CjS77 approved these changes Aug 22, 2019

View reviewed changes

CjS77 merged commit f7408fc into development Aug 22, 2019

CjS77 deleted the sb-p2p-liveness-recv-pingpong branch August 22, 2019 12:36

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Connected LivenessService to comms pub/sub #686

Connected LivenessService to comms pub/sub #686

sdbondi commented Aug 21, 2019 •

edited

Loading

CjS77 Aug 21, 2019

sdbondi Aug 21, 2019

philipr-za left a comment

philipr-za Aug 21, 2019

sdbondi Aug 21, 2019

philipr-za Aug 21, 2019

sdbondi Aug 22, 2019 •

edited

Loading

philipr-za Aug 22, 2019

philipr-za Aug 21, 2019

sdbondi Aug 21, 2019 •

edited

Loading

philipr-za Aug 21, 2019 •

edited

Loading

sdbondi Aug 21, 2019

philipr-za Aug 21, 2019

sdbondi Aug 22, 2019 •

edited

Loading

philipr-za commented Aug 21, 2019

CjS77 commented Aug 22, 2019

Connected LivenessService to comms pub/sub #686

Connected LivenessService to comms pub/sub #686

Conversation

sdbondi commented Aug 21, 2019 • edited Loading

Description

Motivation and Context

How Has This Been Tested?

Types of changes

Checklist:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipr-za 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

sdbondi Aug 22, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sdbondi Aug 21, 2019 • edited Loading

Choose a reason for hiding this comment

philipr-za Aug 21, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sdbondi Aug 22, 2019 • edited Loading

Choose a reason for hiding this comment

philipr-za commented Aug 21, 2019

CjS77 commented Aug 22, 2019

sdbondi commented Aug 21, 2019 •

edited

Loading

sdbondi Aug 22, 2019 •

edited

Loading

sdbondi Aug 21, 2019 •

edited

Loading

philipr-za Aug 21, 2019 •

edited

Loading

sdbondi Aug 22, 2019 •

edited

Loading