Introduce segmented queue in the invoker #482

slinkydeveloper · 2023-06-02T14:56:47Z

tillrohrmann · 2023-06-05T09:12:19Z

src/queue/src/segmented_queue.rs

@@ -97,10 +125,15 @@ impl<T: Serialize + DeserializeOwned + Send + 'static> SegmentQueue<T> {
        }
    }

+    pub async fn dequeue(&mut self) -> Option<T> {
+        self.wait_for_non_emptiness().await;


If I understand the code correctly, then this call will prevent us from entering a busy loop where the outer select loop will always complete on dequeue if the queue is empty. Is this correct? If yes, then I am wondering whether this shouldn't rather be a concern of the caller (the outer select loop introducing a guard for the dequeue branch (if None was returned, then disable, when enqueue something then enabling this branch)).

You have understood correctly @tillrohrmann. I didn't know that it is possible to disable branches.
But with this approach, we have now two variants try_dequeue and dequeue where the former will complete right away, and the later will be awaiting.

Seems useful to have, and it is convenient and cheap for the caller.

If yes, then I am wondering whether this shouldn't rather be a concern of the caller

I don't agree with this. To me the changes introduced in this PR make the interface look more like a channel interface, and this is IMO more clear and idiomatic, in particular given that the interface has async in the dequeue signature. Overall, this reads nicer.

Also the fact that this future, in the location where we use it, depends on what the outer loop is just a detail of how the SegmentedQueue works now and in future, hopefully, this will look more like a real channel (as @igalshilman and I discussed), where the sender part is owned directly by partition processors.

I am not 100% certain, but I think that the NonEmptyFuture does not fully work as intended. I think that the registration of the waker does not have any effect and it would be enough to return Poll::Pending if queue.is_empty(). The reason is that the returned future from this method captures &mut self. Hence, in order to call enqueue one will have to drop this future and thereby rendering the waker useless.

Hmm I think you're right, in another scenario this most likely won't compile because of the double borrowing of &mut self...

Small follow-up question: If dequeue waits until the queue is non-empty, then I assume that we don't have to return a Option<T> but can return a T.

@tillrohrmann you right, there is no need to capture the waker in this case. Other then this, I'd leave this as-is,
And will also leave any followup queue design discussions outside of this scope :-) that might or might not happen in the future (no pun indented)

then I assume that we don't have to return a Option but can return a T.

Makes sense, channels use Option<> because they can be closed, but that's a concept this queue doesn't have.

So what you guys think I should do here? Just rollback the last two commits and we use the queue in a busy loop?

Adding a guard on the select arm should prevent the busy loop, I guess.

Ok so I'm gonna keep igal's suggested changes for the time being, and just modify the usage here in the invoker

igalshilman

Thanks @slinkydeveloper, from my side the changes looks good!
After Till's comment is resolved either way, +1 to merge.

igalshilman · 2023-06-05T09:31:19Z

src/invoker/src/invoker.rs

-                PollNext::Right
-            });
+        // Prepare the segmented queue
+        let mut segmented_input_queue = restate_queue::SegmentQueue::init(tmp_dir, 1_056_784)


should this value come from the configuration?

If you set it up, it does, otherwise defaults to a temp dir.

sorry I should have been more specific. I mean the queue size.

Ah, sure we can do that if we want to, although I'm not sure this is a knob someone will ever touch.

Also fix typo.

slinkydeveloper · 2023-06-19T09:29:37Z

Tested with 1 in_memory_threshold and it works :) @tillrohrmann perhaps do a last pass and we can merge it.

tillrohrmann

Thanks for updating this PR @slinkydeveloper. LGTM. +1 for merging.

tillrohrmann · 2023-06-20T15:50:36Z

src/invoker/src/invoker.rs

+    /// # Temporary directory
+    ///
+    /// Temporary directory to use for the invoker temporary files.
+    /// If empty, the system temporary directory will be used instead.
+    #[cfg_attr(
+        feature = "options_schema",
+        schemars(with = "String", default = "Options::default_tmp_dir")
+    )]
+    tmp_dir: PathBuf,


At some point we should create by default a Restate working directory under which we assemble all the different directories so that people are aware of what Restate writes to disk. Not for now, though.

tillrohrmann · 2023-06-20T15:55:42Z

src/invoker/src/invoker.rs

-                PollNext::Right
-            });
+        // Prepare the segmented queue
+        let mut segmented_input_queue = SegmentQueue::init(tmp_dir, 1_056_784)


Not needed right now but we might wanna the in memory threshold configurable in order to control Restate's memory footprint a tad bit better. At the moment we still have the problem that invocations can have arbitrary size, though.

For the time being, let's keep this hardcoded value. I feel this config option will probably remain forever untouched, so perhaps let's make it configurable the day someone really finds out this queue to be problematic to avoid polluting now our configs of very complicated options to understand.

slinkydeveloper requested a review from igalshilman June 2, 2023 14:56

tillrohrmann reviewed Jun 5, 2023

View reviewed changes

igalshilman approved these changes Jun 5, 2023

View reviewed changes

slinkydeveloper force-pushed the issues/285 branch from 1e79055 to 7122d3a Compare June 5, 2023 09:53

slinkydeveloper changed the title ~~[WIP] Introduce segmented queue in the invoker~~ Introduce segmented queue in the invoker Jun 8, 2023

slinkydeveloper force-pushed the issues/285 branch from 84de828 to e332214 Compare June 8, 2023 14:50

slinkydeveloper added 6 commits June 19, 2023 10:19

Add fs_util and move some fs functions there.

be85dde

Add init method to initialize the temp dir.

b4e04c1

Also fix typo.

Integrate the segmented queue into the invoker

4455729

Apply changes as suggested from Igal

d040b25

Rebase

0b46735

Remove future we don't need anymore, and revert dequeue/try_dequeue

c0ddb64

slinkydeveloper force-pushed the issues/285 branch from e332214 to 7ceeed4 Compare June 19, 2023 08:19

slinkydeveloper requested a review from tillrohrmann June 19, 2023 09:29

slinkydeveloper force-pushed the issues/285 branch from 7ceeed4 to c0ddb64 Compare June 19, 2023 09:30

tillrohrmann approved these changes Jun 20, 2023

View reviewed changes

slinkydeveloper merged commit 76d1b55 into restatedev:main Jun 20, 2023

slinkydeveloper deleted the issues/285 branch June 20, 2023 16:04

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce segmented queue in the invoker #482

Introduce segmented queue in the invoker #482

slinkydeveloper commented Jun 2, 2023 •

edited

Loading

tillrohrmann Jun 5, 2023

igalshilman Jun 5, 2023

slinkydeveloper Jun 5, 2023

tillrohrmann Jun 5, 2023 •

edited

Loading

slinkydeveloper Jun 5, 2023

tillrohrmann Jun 5, 2023

igalshilman Jun 5, 2023

slinkydeveloper Jun 8, 2023

tillrohrmann Jun 8, 2023

slinkydeveloper Jun 8, 2023

igalshilman left a comment

igalshilman Jun 5, 2023

slinkydeveloper Jun 5, 2023

igalshilman Jun 5, 2023

slinkydeveloper Jun 5, 2023

slinkydeveloper commented Jun 19, 2023

tillrohrmann left a comment

tillrohrmann Jun 20, 2023

slinkydeveloper Jun 20, 2023

tillrohrmann Jun 20, 2023

slinkydeveloper Jun 20, 2023

Introduce segmented queue in the invoker #482

Introduce segmented queue in the invoker #482

Conversation

slinkydeveloper commented Jun 2, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann Jun 5, 2023 • edited Loading

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

igalshilman 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 Jun 19, 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

slinkydeveloper commented Jun 2, 2023 •

edited

Loading

tillrohrmann Jun 5, 2023 •

edited

Loading