feat(server): Buffered streamer + use on stable state #639

dranikpg · 2023-01-03T21:55:12Z

This PR introduces:

BufferedStreamerBase for managing wakeups/blocks to allow creating double-buffered streamers easily
JournalStreamer that uses BufferedStreamerBase

Why BufferedStreamerBase separately? In the future, We might want to replace the snapshot channel with it in case of single producer - single consumer to remove allocations, synchronization, fragmentation, etc...

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg · 2023-01-03T21:56:30Z

src/server/io_utils.h

+  // Write stream in buffered parts until producer finishes or error occurs.
+  std::error_code WriteAll(io::Sink* dest);


The BufferedStreamBase doesn't run the writer fiber itself because in the case of, for example, a snapshot the writing fiber is managed externally

dranikpg · 2023-01-03T21:57:09Z

src/server/io_utils.cc

+void BufferedStreamerBase::ReportWritten() {
+  buffered_++;
+  // Check and possibly wait for consumer to keep up.
+  waker_.await([this]() { return !IsStalled() || IsStopped(); });
+  // Notify consumer we have new data.
+  waker_.notify();
+}
+
+std::error_code BufferedStreamerBase::WriteAll(io::Sink* dest) {
+  CHECK_NE(buf_ptr_, nullptr);
+
+  while (!IsStopped()) {
+    // Wait for more data or stop signal.
+    waker_.await([this]() { return buffered_ > 0 || IsStopped(); });
+    if (IsStopped())
+      break;
+
+    // Steal-swap producer buffer.
+    std::swap(buf_stash_, *buf_ptr_);
+    buffered_ = 0;
+
+    // If producer stalled, notify we consumed data and it can unblock.
+    waker_.notify();
+    RETURN_ON_ERR(dest->Write(buf_stash_.InputBuffer()));


This looks much simpler than last time

dranikpg · 2023-01-03T21:58:12Z

src/server/dflycmd.cc

+
+  // Self referential because buf reference pointr to writer.


TODO: fix comment

src/server/io_utils.h

romange

Looks good. I am not sure about "wait/notify" sequence, commented about it.

romange · 2023-01-04T12:09:50Z

src/server/dflycmd.cc

+  write_fb_ = Fiber(&JournalStreamer::WriterFb, this, dest);
+  journal_cb_id_ = journal_->RegisterOnChange([this](const journal::Entry& entry) {
+    writer_.Write(entry);
+    ReportWritten();


I think it's a bit confusing to have both the backpressure and notification bundled together.
Suppose you are stalled. You want to notify your consumer before you block.

You block because the consumer does not keep up, there is no point in notifying it before - its already busy. Once its no longer stalled it will have consumed your latest portions

romange · 2023-01-04T12:11:39Z

src/server/io_utils.cc

+  waker_.notify();
+}
+
+std::error_code BufferedStreamerBase::WriteAll(io::Sink* dest) {


Write part is a confusing since both the consumer and producer parts are "writing".

I needed to look into the function to understand what it does. Maybe call it ConsumeIntoSink or similar?

src/server/io_utils.cc

adiholden · 2023-01-04T12:40:36Z

src/server/io_utils.cc

+      break;
+
+    // Steal-swap producer buffer.
+    std::swap(buf_stash_, *buf_ptr_);


I think that the flow of switching the buffer of journal writer without it knowing is not very intuitive.
How about creating journal writer with buffer not a member but one that is set from outside, with this approach you will have SetBuffer func to replace the buf and you can use it here.

The only counter-intuitive thing to stealing a buffer is that it can't cause a data race because the writer and consumer have to be on the same thread. Making it use the journal writer internally won't make it re-usable.

What could be done instead is:

making the journal writer unbuffered, i.e. write directly to a sink

the sink will be some buffer owned by the buffered stream (so it will own both)

Otherwise, we can keep the internal buffer inside the journal writer, but now flash to a sink (which will be the IoBuf) after each operations. The downside is that this introduced an intermediate copy - but its probably very cheap?

@romange What do you think?

it's localized and does not affect lots of places in the code.
and we can make it more clear by adding more comments.

adiholden · 2023-01-04T12:41:03Z

The flush function of journal writer can be removed right?

src/server/io_utils.cc

Signed-off-by: Vladislav Oleshko <[email protected]>

adiholden · 2023-01-08T07:18:21Z

src/server/dflycmd.cc

-    if (cb_id)
-      sf_->journal()->UnregisterOnChange(cb_id);
+  // Register cleanup.
+  flow->cleanup = [this, streamer, flow]() {


Is it possible that flow cleanup will be triggered between the start above and the update of the cleanup lambda?

No, because unlike the replica it has a per-session mutex that is locked on state transitions

And it doesn't contain any await points

adiholden · 2023-01-08T07:22:22Z

src/server/io_utils.cc

+  while (!IsStopped()) {
+    // Wait for more data or stop signal.
+    waker_.await([this]() { return buffered_ > 0 || IsStopped(); });
+    if (IsStopped())


If producer is done doesnt it make more sense to finish write all the buffered journal data and then return?

Indeed.
It doesn't matter currently because its used as cancellation, but the general behavior should be this

Signed-off-by: Vladislav Oleshko <[email protected]>

romange · 2023-01-09T18:55:17Z

Maybe rebase it to let tests run?

dranikpg · 2023-01-09T19:01:34Z

Maybe rebase it to let tests run?

I will update the branch when pytests are merged (plz re-approve them), it will trigger a new ci flow

romange · 2023-01-10T13:58:11Z

src/server/io_utils.cc

+    waker_.notify();
+    RETURN_ON_ERR(dest->Write(consumer_buf_.InputBuffer()));
+
+    if (IsStopped()) {


instead of checking here, you can unconditionally call waker_.notifyAll(); after the loop.

romange · 2023-01-10T14:01:43Z

src/server/io_utils.cc

+
+    // If producer stalled, notify we consumed data and it can unblock.
+    waker_.notify();
+    RETURN_ON_ERR(dest->Write(consumer_buf_.InputBuffer()));


what happens if we return an error but the producer is blocked ?

This is the whole issue I was solving and what the comment in dflycmd is about. The function that receives the error cancels the context and finalizes the streamer without waiting for explicit cleanup

src/server/io_utils.h

romange · 2023-01-10T14:05:22Z

src/server/io_utils.h

+  // Write some data into the internal buffer.
+  // Consumer needs to be woken up manually with NotifyWritten to avoid waking it up for small
+  // writes.
+  io::Result<size_t> WriteSome(const iovec* vec, uint32_t len) override;


I would write here in the comments how your ideal produce flow looks like in terms. for example,

while (should_write()) { bsb->WriteSome(...); bsb->NotifyWritten(); } bsb->Finalize();

if this flow is correct, why you do not call NotifyWritten inside WriteSome?

Because notify issues a wakeup which I don't always want (I've written this in the comment)

Now that the writer is unbuffered (notify is essentially instead of flush), we don't want a wakeup after writing every small part. In the future we even might not want to wake after ever entry - for example when we mass evict entries.

Signed-off-by: Vladislav Oleshko <[email protected]>

feat(server): Buffered streamer + use on stable state

b02294b

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg commented Jan 3, 2023

View reviewed changes

dranikpg requested review from romange and adiholden January 4, 2023 08:55

romange reviewed Jan 4, 2023

View reviewed changes

adiholden reviewed Jan 4, 2023

View reviewed changes

src/server/io_utils.cc Show resolved Hide resolved

dranikpg added 3 commits January 5, 2023 16:35

fix(server): Make buffered streamer own both buffers

695ff4b

Signed-off-by: Vladislav Oleshko <[email protected]>

Merge branch 'main' into async-streamer

2b4ad82

fix(server): Fix journal test

cde1874

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg requested review from romange and adiholden January 5, 2023 13:49

adiholden reviewed Jan 8, 2023

View reviewed changes

adiholden previously approved these changes Jan 8, 2023

View reviewed changes

fix(server): Small buffered streamer fix

bacdac6

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg dismissed adiholden’s stale review via bacdac6 January 8, 2023 08:00

adiholden previously approved these changes Jan 8, 2023

View reviewed changes

romange previously approved these changes Jan 8, 2023

View reviewed changes

Merge branch 'main' into async-streamer

59aa85b

dranikpg marked this pull request as draft January 9, 2023 21:37

dranikpg dismissed stale reviews from romange and adiholden via 7bd0654 January 10, 2023 12:40

dranikpg marked this pull request as ready for review January 10, 2023 12:41

dranikpg enabled auto-merge (squash) January 10, 2023 12:41

dranikpg requested review from romange and adiholden January 10, 2023 12:41

romange reviewed Jan 10, 2023

View reviewed changes

fix(server): Fix cancellation deadlock

ff1d573

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg force-pushed the async-streamer branch from 7bd0654 to ff1d573 Compare January 10, 2023 19:51

dranikpg marked this pull request as draft January 10, 2023 20:02

auto-merge was automatically disabled January 10, 2023 20:02
Pull request was converted to draft

dranikpg added 2 commits January 13, 2023 12:02

Merge branch 'main' into async-streamer

7fcdb95

fix(server): Fix bug

1a75abc

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg marked this pull request as ready for review January 13, 2023 10:54

dranikpg requested a review from romange January 13, 2023 10:54

dranikpg mentioned this pull request Jan 13, 2023

feat(server): Improve stable state throughput (Experimental) #596

Closed

adiholden approved these changes Jan 15, 2023

View reviewed changes

Merge branch 'main' into async-streamer

ce95c3f

dranikpg merged commit 7eff61c into dragonflydb:main Jan 15, 2023

dranikpg deleted the async-streamer branch February 27, 2023 16:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(server): Buffered streamer + use on stable state #639

feat(server): Buffered streamer + use on stable state #639

dranikpg commented Jan 3, 2023 •

edited

Loading

dranikpg Jan 3, 2023

dranikpg Jan 3, 2023

dranikpg Jan 3, 2023

romange left a comment

romange Jan 4, 2023

dranikpg Jan 5, 2023

romange Jan 4, 2023

adiholden Jan 4, 2023

dranikpg Jan 5, 2023 •

edited

Loading

dranikpg Jan 5, 2023 •

edited

Loading

dranikpg Jan 5, 2023

romange Jan 5, 2023

adiholden commented Jan 4, 2023

adiholden Jan 8, 2023

dranikpg Jan 8, 2023

dranikpg Jan 8, 2023

adiholden Jan 8, 2023

dranikpg Jan 8, 2023

romange commented Jan 9, 2023

dranikpg commented Jan 9, 2023

romange Jan 10, 2023

romange Jan 10, 2023

dranikpg Jan 10, 2023

romange Jan 10, 2023

dranikpg Jan 10, 2023 •

edited

Loading

		// Write stream in buffered parts until producer finishes or error occurs.
		std::error_code WriteAll(io::Sink* dest);

feat(server): Buffered streamer + use on stable state #639

feat(server): Buffered streamer + use on stable state #639

Conversation

dranikpg commented Jan 3, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

romange 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

dranikpg Jan 5, 2023 • edited Loading

Choose a reason for hiding this comment

dranikpg Jan 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

adiholden commented Jan 4, 2023

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

romange commented Jan 9, 2023

dranikpg commented Jan 9, 2023

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

dranikpg Jan 10, 2023 • edited Loading

Choose a reason for hiding this comment

dranikpg commented Jan 3, 2023 •

edited

Loading

dranikpg Jan 5, 2023 •

edited

Loading

dranikpg Jan 5, 2023 •

edited

Loading

dranikpg Jan 10, 2023 •

edited

Loading