Invoker #88

slinkydeveloper · 2023-02-14T10:10:46Z

Depends on #80, Fix #82.

This PR implements the basic logic of the invoker:

Each invocation has an InvocationStateMachine assigned to it, handling the different states of the invocation:

To execute the invocation, an InvocationTask is created, running in a separate tokio task. This is where we implement the protocol network logic:
- The InvocationTask is composed by 3 loops: write only, when replaying the journal and waiting for response headers, read and write when both completions are sent from invoker main loop and when new entries are available from the service endpoint, and read only when the completions channel is closed. In the request response mode, we will be able to reuse the same code simply disabling the completions channel.
- Every encoding/decoding logic is implemented by Encoder/Decoder from Introduce invoker::message module #80
- InvocationTask ends either after the http stream(s) are closed gracefully, or after a network failure, sending a signal back to the invoker, including last known journal revision and journal entry.
The invoker has a main loop which orchestrates signals coming from the partition processor and from InvocationTasks, eventually applying state changes to InvocationStateMachine, sending completions to single InvocationTasks, or cleaning up internal state.
- The invoker is the component responsible for deciding when the invocation is suspended, or ended. In future InvocationTask might do that as well, when we'll reintroduce the SuspensionMessage.

This PR also includes:

The port of test_utils from the PoC runtime.
A manual test to run against the sdk-java Counter example

tillrohrmann

Thanks for creating this PR @slinkydeveloper. This looks like a really good piece of work :-)

My main question is around the unbounded channels within the invoker. I think that this is something we should have an answer for (not necessarily addressing it right away) because it might affect stability.

Another question is around the handling of stored entry acks. I assume that this part is not fully implemented yet, right?

I noticed that you expect an empty Completion for a Custom journal entry? So far I thought that the partition processor would send for every journal entry an stored entry ack which could be used to signal that a Custom entry has been persisted (e.g. a side effect entry). This is probably something we need to align on.

deny.toml

tillrohrmann · 2023-02-14T13:05:31Z

src/invoker/src/lib.rs

+}
+
+pub trait ServiceEndpointRegistry {
+    fn resolve_endpoint(&self, service_name: &str) -> Option<EndpointMetadata>;


Wondering whether this needs to become eventually asynchronous because in the distributed setup we will probably have to talk to the meta component.

For the time being I would keep it synchronous, as we still need to unfold this discussion: #91

Making this async opens up another can of worms, such as do you block the whole invoker waiting on the meta to give you the endpoint metadata? Also, in that case, would it make sense for the partition processor to directly resolve the EndpointMetadata and pass them when creating invoke messages for the invoker?

src/invoker/src/lib.rs

src/invoker/tests/counter_test.rs

src/invoker/tests/mocks.rs

src/invoker/src/invocation_task.rs

tillrohrmann · 2023-02-14T16:08:11Z

src/invoker/src/invocation_task.rs

+                response_res = &mut req_fut, if response.is_none() => {
+                    response = Some(
+                        response_res??
+                    );
+                },
+                Some(je) = journal_stream.next() => {
+                    self.write(http_stream_tx, ProtocolMessage::UnparsedEntry(je)).await?;
+                    self.next_journal_index += 1;


Nice, this parallelizes the request initiation and data sending if the underlying connection supports buffering, right?

This is not the only reason. In one of the iterations of the protocol I figured out that I could send the response header frame as soon as I get the request, without waiting for the journal to be sent over. This could provide faster feedback to the invoker, for example when connection cannot be established, or when a service is not known. For the latter, we should move the status check from the run_internal to the match arm of response_res.

Because it is not mandated in the protocol spec when the sdk writes the response headers, we simply don't make any assumption here.

Perhaps another point here is that hyper doesn't split the "request buffer ready" and "response header frame received" in two separate futures, which requires a bit of playing around with spawning a new tokio task. Hopefully this will get better with hyper 1.0, where most likely we'll have more control of http frames.

src/invoker/src/invoker.rs

src/invoker/src/invocation_task.rs

slinkydeveloper · 2023-02-15T14:00:08Z

@tillrohrmann For the discussion about bounded/unbounded channels, i've opened an issue here #92

tillrohrmann

Thanks for the update @slinkydeveloper. I think there are two bigger problems we have to tackle at some point.

Make sure that InvocationTask does not leak resources
Rethink propagation of journal revision to the service endpoints in order to make it work (I guess this would require a protocol change)

tillrohrmann · 2023-02-16T11:27:04Z

src/invoker/src/invoker.rs

+        // TODO we don't need this field right?
+        #[allow(dead_code)]
        journal_revision: JournalRevision,


I think we don't need this field. I have removed it in #89.

src/invoker/src/invoker.rs

tillrohrmann · 2023-02-16T13:55:44Z

src/invoker/src/invocation_task.rs

+};
+use super::{EndpointMetadata, InvokeInputJournal, JournalMetadata, JournalReader, ProtocolType};
+
+#[allow(clippy::declare_interior_mutable_const)]


Why is it a false positive? Is it because Bytes is classified wrongly and does not expose interior mutability?

src/invoker/src/invocation_task.rs

tillrohrmann · 2023-02-16T15:30:01Z

src/invoker/src/invocation_task.rs

+        let buf = self.encoder.encode(msg);
+
+        if let Err(hyper_err) = http_stream_tx.send_data(buf).await {
+            // is_closed() can happen only with sender's channel


What is sender's channel?

https://docs.rs/hyper/latest/hyper/struct.Error.html#method.is_closed It's the body sender channel. Essentially is_closed() means the request channel is closed. This is fine to ignore, as it simply means we cannot write the request anymore, because perhaps the sdk wants to suspend.

Ok, thanks for the clarification. Can you update the comment to reflect this? E.g. "is_closed is only true iff the request channel has been closed. This can happen if the service endpoint wants to suspend."

src/invoker/src/invocation_task.rs

slinkydeveloper · 2023-02-17T11:46:42Z

@tillrohrmann I rebased on main, and addressed all your comments. This is ready for another pass. I also included a commit to remove EntryType, which shouldn't be useful anymore, let me know if you want to keep this commit or not.

tillrohrmann

Thanks for updating this PR @slinkydeveloper. The changes look good to me. +1 for merging after resolving my last minor comments.

src/invoker/src/invoker.rs

tillrohrmann · 2023-02-17T11:07:09Z

src/invoker/src/invocation_task.rs

+};
+use super::{EndpointMetadata, InvokeInputJournal, JournalMetadata, JournalReader, ProtocolType};
+
+#[allow(clippy::declare_interior_mutable_const)]


Then let's add this to the comment that we suspect that it is because of the Bytes.

tillrohrmann · 2023-02-17T12:54:42Z

src/invoker/src/invocation_task.rs

+        let buf = self.encoder.encode(msg);
+
+        if let Err(hyper_err) = http_stream_tx.send_data(buf).await {
+            // is_closed() can happen only with sender's channel


Ok, thanks for the clarification. Can you update the comment to reflect this? E.g. "is_closed is only true iff the request channel has been closed. This can happen if the service endpoint wants to suspend."

tillrohrmann · 2023-02-17T12:57:59Z

src/invoker/src/invoker.rs

                    break;
                }
            }
        }
+
+        // Wait for all the tasks to shutdown
+        if tokio::time::timeout(shutdown_timeout, invocation_tasks.shutdown())


Why are we pushing the timeout into the invoker? Wouldn't it be enough that someone who is waiting on the task to complete adds a timeout (like the owner of the invoker)?

tillrohrmann · 2023-02-17T13:05:58Z

src/worker/src/lib.rs

@@ -27,6 +29,8 @@ type PartitionProcessor = partition::PartitionProcessor<
 >;
 type TargetedFsmCommand = Targeted<partition::Command>;

+const INVOKER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);


The restate binary will add a timeout for the shutdown https://github.com/restatedev/restate/blob/main/src/restate/src/main.rs#L47. That's why I believe that we don't need this timeout here.

tillrohrmann · 2023-02-17T13:06:25Z

src/service_protocol/src/codec.rs

@@ -27,12 +27,12 @@ pub enum ErrorKind {
 macro_rules! match_decode {
    ($ty:expr, $buf:expr, { $($variant:ident),* }) => {
        match $ty {
-              $(EntryType::$variant => paste::paste! {
+              $(RawEntryHeader::$variant { .. } => paste::paste! {


That's a good simplification. Thanks for fixing it :-)

Define the Invoker interface with InvocationTask (tasks reading and writing single invocation streams).

…r retrying

…9437c..552361ef 552361ef handlerType -> ty f38562fd Rename entries and manifest to have names closer to the concept names we use everywhere else (restatedev#88) git-subtree-dir: crates/service-protocol/service-protocol git-subtree-split: 552361ef45340173e8beb361e35fb7b6e5d6effe

…9437c..552361ef 552361ef handlerType -> ty f38562fd Rename entries and manifest to have names closer to the concept names we use everywhere else (#88) git-subtree-dir: crates/service-protocol/service-protocol git-subtree-split: 552361ef45340173e8beb361e35fb7b6e5d6effe

slinkydeveloper force-pushed the invoker_wip branch from a7fad7b to 53bd2d4 Compare February 14, 2023 10:26

slinkydeveloper changed the title ~~[WIP] Invoker~~ Invoker Feb 14, 2023

slinkydeveloper requested a review from tillrohrmann February 14, 2023 10:34

tillrohrmann reviewed Feb 14, 2023

View reviewed changes

tillrohrmann reviewed Feb 16, 2023

View reviewed changes

slinkydeveloper force-pushed the invoker_wip branch 2 times, most recently from 843bfde to 05ff67d Compare February 17, 2023 11:44

slinkydeveloper requested a review from tillrohrmann February 17, 2023 11:46

tillrohrmann approved these changes Feb 17, 2023

View reviewed changes

slinkydeveloper added 17 commits February 17, 2023 14:18

Lay in some foundations of the Invoker state machines.

47b458b

Define the Invoker interface with InvocationTask (tasks reading and writing single invocation streams).

Implement abort partition

87a1081

Implement Invoker

5527759

Implemented invocation task

c84c654

Add test_utils from the PoC

0d845f0

Added manual test

9e2d113

Remove allow dead_code

b2dfd96

Trivial feedback

ca9b129

Feedback

7312461

Feedback

da9ae08

Abort on drop

b79b996

Move shutdown timeout definition out of the invoker

d4ca9a6

Introduce JournalTracker to improve the logic to track stored acks fo…

95cb018

…r retrying

Rebase fixes

c77ef38

Remove EntryType

a939ded

Feedback

52fe2df

Rebase

1e01c52

slinkydeveloper force-pushed the invoker_wip branch from 05ff67d to 1e01c52 Compare February 17, 2023 13:19

slinkydeveloper merged commit ff38ea3 into restatedev:main Feb 17, 2023

slinkydeveloper deleted the invoker_wip branch February 17, 2023 14:14

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Invoker #88

Invoker #88

slinkydeveloper commented Feb 14, 2023 •

edited

Loading

tillrohrmann left a comment

tillrohrmann Feb 14, 2023

slinkydeveloper Feb 15, 2023

tillrohrmann Feb 14, 2023

slinkydeveloper Feb 15, 2023

slinkydeveloper commented Feb 15, 2023

tillrohrmann left a comment

tillrohrmann Feb 16, 2023

tillrohrmann Feb 16, 2023

tillrohrmann Feb 16, 2023

slinkydeveloper Feb 17, 2023

tillrohrmann Feb 17, 2023

slinkydeveloper commented Feb 17, 2023

tillrohrmann left a comment

tillrohrmann Feb 17, 2023

tillrohrmann Feb 17, 2023

tillrohrmann Feb 17, 2023

tillrohrmann Feb 17, 2023

tillrohrmann Feb 17, 2023

Invoker #88

Invoker #88

Conversation

slinkydeveloper commented Feb 14, 2023 • edited Loading

tillrohrmann 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

slinkydeveloper commented Feb 15, 2023

tillrohrmann 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

slinkydeveloper commented Feb 17, 2023

tillrohrmann 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

slinkydeveloper commented Feb 14, 2023 •

edited

Loading