chore: Streamer is rewritten with async interface #3108

romange · 2024-05-31T09:10:22Z

No description provided.

dranikpg · 2024-05-31T10:30:57Z

What's missing is backpressure 🙂

romange · 2024-05-31T11:45:06Z

What's missing is backpressure 🙂

AwaitIfWritten is backpressure

romange · 2024-06-01T04:12:03Z

src/server/journal/streamer.cc

          // No record to write, just await if data was written so consumer will read the data.
-          return AwaitIfWritten();


I do not completely understand the semantics of this call.
based on the code I think that waker_.await([this]() { return !IsStalled() || IsStopped(); }); does not wait that everything is completely flushed. Am I wrong @adiholden ?

src/server/journal/streamer.cc

dranikpg · 2024-06-02T20:13:19Z

src/server/journal/streamer.cc

+    if (pending_buf_.empty()) {
+      dest_->AsyncWrite(src, [buf, this, len = str.size()](std::error_code ec) {
+        delete[] buf;
+        OnCompletion(ec, len);
+      });
+    } else {
+      iovec v[2];
+      v[0] = IoVec(pending_buf_);
+      v[1] = IoVec(src);
+
+      dest_->AsyncWrite(
+          v, 2,
+          [buf0 = std::move(pending_buf_), buf, this, len = total_pending](std::error_code ec) {
+            delete[] buf;
+            OnCompletion(ec, len);
+          });


nit: if pending buf is empty, does it hurt to have it in the iovec? Otherwise we can conditionally build the iovec. Either way I imagine this can be simpler

dranikpg · 2024-06-02T20:14:50Z

src/server/journal/streamer.cc

+    is_stopped_ = true;
    cntx_->ReportError(ec);


report error should lead to cntx->cancelled() being true, so is_stopped is redundant?

yeah, I was contemplating over it. I will remove

dranikpg · 2024-06-02T20:17:37Z

src/server/journal/streamer.cc

+  auto next = chrono::steady_clock::now() + kSinkThrottleTime;
+  std::cv_status status =
+      waker_.await_until([this]() { return !IsStalled() || IsStopped(); }, next);
+  if (status == std::cv_status::timeout) {
+    LOG(WARNING) << "Stream timed out, inflight bytes " << in_flight_bytes_;
+    is_stopped_ = true;
+    cntx_->ReportError(make_error_code(errc::stream_timeout));
  }


so a replica lag of 0.5 second is "impossible"? There won't be any way to disable this later 🙂

I will make it a flag. It's not just a lag. It means we have in_flight_bytes_ more than a threshold AND we wait on it to decrease for more than 0.5s. But I agree that if we say serialize 4GB value, we won't update in_flight_bytes_ until the whole request has not been sent and it may take quite a while

romange · 2024-06-05T13:57:48Z

src/server/journal/streamer.cc

-               // UnregisterOnChange will try to lock the mutex therefor calling UnregisterOnChange
-               // before Finalize may cause deadlock.
+  VLOG(1) << "JournalStreamer::Cancel";
+  cntx_->Cancel();


should I call cntx_->Cancel(); here @dranikpg ?

I assume not, becauase it cancel is already called when cleaning up. I'd remove it just to be sure, but the error should be ignored if the context already has one set

dranikpg · 2024-06-05T20:27:13Z

src/server/journal/streamer.cc

-               // UnregisterOnChange will try to lock the mutex therefor calling UnregisterOnChange
-               // before Finalize may cause deadlock.
+  VLOG(1) << "JournalStreamer::Cancel";
+  cntx_->Cancel();


I assume not, becauase it cancel is already called when cleaning up. I'd remove it just to be sure, but the error should be ignored if the context already has one set

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

romange force-pushed the Pr3 branch from 5b26e02 to cb070c6 Compare May 31, 2024 09:21

romange requested a review from dranikpg May 31, 2024 09:22

romange force-pushed the Pr3 branch 2 times, most recently from 97f5747 to 5bfce7b Compare May 31, 2024 17:49

romange requested a review from adiholden May 31, 2024 18:42

romange force-pushed the Pr3 branch from 5bfce7b to 77f0eef Compare May 31, 2024 20:32

romange marked this pull request as ready for review May 31, 2024 21:34

romange force-pushed the Pr3 branch from 77f0eef to f59b943 Compare May 31, 2024 21:36

romange commented Jun 1, 2024

View reviewed changes

romange force-pushed the Pr3 branch 2 times, most recently from eded400 to 8c4020d Compare June 1, 2024 09:22

romange commented Jun 1, 2024

View reviewed changes

src/server/journal/streamer.cc Outdated Show resolved Hide resolved

romange force-pushed the Pr3 branch 7 times, most recently from 2089db8 to 5445de1 Compare June 2, 2024 17:07

dranikpg previously approved these changes Jun 2, 2024

View reviewed changes

romange dismissed dranikpg’s stale review via 49cf37b June 5, 2024 05:55

romange force-pushed the Pr3 branch 2 times, most recently from 49cf37b to ad5a77d Compare June 5, 2024 13:56

romange commented Jun 5, 2024

View reviewed changes

romange requested a review from dranikpg June 5, 2024 13:58

dranikpg approved these changes Jun 5, 2024

View reviewed changes

romange force-pushed the Pr3 branch from ad5a77d to 1e8be8f Compare June 10, 2024 09:56

dranikpg previously approved these changes Jun 10, 2024

View reviewed changes

romange added 4 commits June 13, 2024 10:43

chore: Streamer is rewritten with async interface

daeba05

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

chore: better error reporting

4c4c161

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

chore: comments

d8c04ef

chore: more fixes

5576033

romange dismissed dranikpg’s stale review via 5576033 June 13, 2024 07:56

romange force-pushed the Pr3 branch from 1e8be8f to 5576033 Compare June 13, 2024 07:56

dranikpg approved these changes Jun 13, 2024

View reviewed changes

romange merged commit a800631 into main Jun 13, 2024
10 checks passed

romange deleted the Pr3 branch June 13, 2024 09:29

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

chore: Streamer is rewritten with async interface #3108

chore: Streamer is rewritten with async interface #3108

romange commented May 31, 2024

dranikpg commented May 31, 2024

romange commented May 31, 2024

romange Jun 1, 2024 •

edited

Loading

dranikpg Jun 2, 2024

dranikpg Jun 2, 2024

romange Jun 3, 2024

dranikpg Jun 2, 2024

romange Jun 3, 2024

romange Jun 5, 2024

dranikpg Jun 5, 2024

dranikpg Jun 5, 2024

		// No record to write, just await if data was written so consumer will read the data.
		return AwaitIfWritten();

chore: Streamer is rewritten with async interface #3108

chore: Streamer is rewritten with async interface #3108

Conversation

romange commented May 31, 2024

dranikpg commented May 31, 2024

romange commented May 31, 2024

romange Jun 1, 2024 • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

romange Jun 1, 2024 •

edited

Loading