chore: asynchronous IO for connection fiber #6069

kostasrim · 2025-11-17T13:50:51Z

This PR adds IoLoopV2 that removes fiber blocking read calls from the flow. This allows the connection fiber to handle other events while recv calls are handled asynchronously from readiness events triggered by a registered multishot poll.

add IoLoopV2 that uses non fiber blocking async reads via poll registration
add tls label in pytests
add a manual trigger workflow to run all the regression tests (both epoll/uring) without tls

Follow up on a separate PR:

Allow also EnableRecvMultishot()

Resolves #6028

kostasrim · 2025-11-28T09:49:25Z

src/facade/dragonfly_connection.cc

+    }
+
+    phase_ = PROCESS;
+    bool is_iobuf_full = io_buf_.AppendLen() == 0;


No functional change from here and onwards in comparison IoLoop

kostasrim · 2025-11-28T09:56:03Z

tests/dragonfly/connection_test.py

    await check_stats()


+@pytest.mark.tls


no changes to the tests. I added the tls label to disable tls tests on the manual workflow I introduced (because tls is broken for now).

yep, so instead - avoid enabling v2 for tls sockets. we have is_tls_ flag for that

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim · 2025-12-01T10:51:00Z

https://github.com/dragonflydb/dragonfly/actions/runs/19820043215

augmentcode

Review completed. 5 suggestions posted.

Comment augment review to trigger a new review at any time.

augmentcode · 2025-12-01T10:52:40Z

.github/actions/regression-tests/action.yml

        export ROOT_DIR="${GITHUB_WORKSPACE}/tests/dragonfly/valkey_search"
        export UBSAN_OPTIONS=print_stacktrace=1:halt_on_error=1 # to crash on errors

+        if [[ "${{inputs.df-arg}}" == 'epoll' ]]; then


DF_ARG is only set when inputs.df-arg equals 'epoll', which prevents other values (e.g., the workflow’s 'expiremental_io_loop_v2') from being propagated to pytest.

_{🤖 Was this useful? React with 👍 or 👎}

we already have epoll argument why do you need this?

augmentcode · 2025-12-01T10:52:40Z

.github/actions/regression-tests/action.yml

          export FILTER="${{inputs.filter}} and not exclude_epoll"
          # Run only replication tests with epoll
-          timeout 80m pytest -m "$FILTER" --durations=10 --timeout=300 --color=yes --json-report --json-report-file=report.json dragonfly --df force_epoll=true --log-cli-level=INFO || code=$?
+          timeout 80m pytest -m "$FILTER" $DF_ARG --durations=10 --timeout=300 --color=yes --json-report --json-report-file=report.json dragonfly --df force_epoll=true --df --log-cli-level=INFO || code=$?


There is an extra --df with no value after force_epoll=true in the pytest command, which is likely to break argument parsing.

_{🤖 Was this useful? React with 👍 or 👎}

augmentcode · 2025-12-01T10:52:40Z

src/facade/dragonfly_connection.cc

+        io_ec_ = make_error_code(errc::connection_aborted);
+      }
+
+      LOG_IF(FATAL, !io_ec_) << "Recv error: " << strerror(-res) << " errno " << errno;


strerror(-res) uses the negated recv return value instead of errno, so the logged error message text will be incorrect.

_{🤖 Was this useful? React with 👍 or 👎}

augmentcode · 2025-12-01T10:52:40Z

src/facade/dragonfly_connection.cc

+        [this]() { return io_buf_.InputLen() > 0 || io_ec_ || io_buf_.AppendLen() == 0; });
+
+    if (io_ec_) {
+      LOG_IF(WARNING, cntx()->replica_conn) << "async io error: " << ec;


The log line prints ec, which is uninitialized here, instead of the actual I/O error io_ec_.

_{🤖 Was this useful? React with 👍 or 👎}

tests/dragonfly/pymemcached_test.py

kostasrim · 2025-12-01T10:53:43Z

src/facade/dragonfly_connection.cc

 #include <absl/strings/str_cat.h>
 #include <absl/time/time.h>

+#include <condition_variable>


TODO remove the includes. plugin is too aggressive

yes, please

kostasrim · 2025-12-01T10:56:11Z

@romange you can start taking a look.

I added a workflow + flag to run ioloopv2. The former is to avoid running the tests on both paths every time. Once we slowly migrate I will remove the workflow file -- it's just a temporary solution
I only used epoll based notifications -- will follow up with enable_multi_shot_=true as well
I introduced ioloopv2 to provide clear serparation between the two implementations

romange · 2025-12-01T19:30:26Z

.github/actions/regression-tests/action.yml

        export ROOT_DIR="${GITHUB_WORKSPACE}/tests/dragonfly/valkey_search"
        export UBSAN_OPTIONS=print_stacktrace=1:halt_on_error=1 # to crash on errors

+        if [[ "${{inputs.df-arg}}" == 'epoll' ]]; then


we already have epoll argument why do you need this?

romange · 2025-12-01T19:30:45Z

.github/actions/regression-tests/action.yml

          export FILTER="${{inputs.filter}} and not exclude_epoll"
          # Run only replication tests with epoll
-          timeout 80m pytest -m "$FILTER" --durations=10 --timeout=300 --color=yes --json-report --json-report-file=report.json dragonfly --df force_epoll=true --log-cli-level=INFO || code=$?
+          timeout 80m pytest -m "$FILTER" $DF_ARG --durations=10 --timeout=300 --color=yes --json-report --json-report-file=report.json dragonfly --df force_epoll=true --df --log-cli-level=INFO || code=$?


.github/workflows/ioloop-v2-regtests.yml

romange · 2025-12-01T19:33:44Z

src/facade/dragonfly_connection.cc

 #include <absl/strings/str_cat.h>
 #include <absl/time/time.h>

+#include <condition_variable>


yes, please

src/facade/dragonfly_connection.cc

romange · 2025-12-01T19:43:16Z

src/facade/dragonfly_connection.cc

+    }
+
+    io::MutableBytes buf = io_buf_.AppendBuffer();
+    int res = recv(socket_->native_handle(), buf.data(), buf.size(), 0);


Please use TryRecv interface

romange · 2025-12-01T19:49:32Z

src/facade/dragonfly_connection.cc

+  do {
+    HandleMigrateRequest();
+
+    // We *must* poll again for readiness. The event handler we registered above


it is possible to improve for sure. 'We must' is a too strong statement as you can introduce a state that tracks whether the read is needed and when it's not needed.
Either change to TODO or fix it in this PR.

Well, is it a regression to poll unconditionally ? I imagine, the overhead for EAGAIN is minimal so we don't really need a variable for this. I wrote must because I found poll to be the simplest way. I reworded the comment to be less strong. My only question is, what is the reason to prefer a flag over an unconditional poll assuming EAGAIN is not eating CPU time?

it's a system call and we add another system-call per each loop. It's fine for now but I think we will have to add a state variable to track when this call is needed. I do not know how it effects performance, in fact, we will need to do some benchmarking tests on this PR. I will do it next week

romange · 2025-12-01T19:52:38Z

src/facade/dragonfly_connection.cc

+        [this]() { return io_buf_.InputLen() > 0 || io_ec_ || io_buf_.AppendLen() == 0; });
+
+    if (io_ec_) {
+      LOG_IF(WARNING, cntx()->replica_conn) << "async io error: " << ec;


romange · 2025-12-01T19:54:58Z

src/facade/dragonfly_connection.h

  util::fb2::Fiber async_fb_;             // async fiber (if started)

+  std::error_code io_ec_;
+  util::fb2::EventCount io_event_;


do not use EventCount for thread-local synchronization - use CondVarAny instead

romange · 2025-12-01T19:56:03Z

tests/dragonfly/connection_test.py

    await check_stats()


+@pytest.mark.tls


yep, so instead - avoid enabling v2 for tls sockets. we have is_tls_ flag for that

Signed-off-by: Kostas Kyrimis <[email protected]>

romange · 2025-12-03T11:16:44Z

src/facade/dragonfly_connection.cc

+        return;
+      }
+
+      LOG_EVERY_T(ERROR, 10) << "Recv error: " << ec;


romange · 2025-12-03T11:17:58Z

src/facade/dragonfly_connection.cc

+    io::Result<size_t> res = socket_->TryRecv(buf);
+
+    // error path
+    if (!res) {


for readability, I would reverse this condition and handle the happy case first. aka
if (res && *res) {

romange · 2025-12-03T11:19:45Z

tests/dragonfly/pymemcached_test.py


 from . import dfly_args
 from .instance import DflyInstance
+from .utility import *


not needed?

romange · 2025-12-03T11:22:08Z

src/facade/dragonfly_connection.cc

    UpdateIoBufCapacity(io_buf_, stats_, [&]() { io_buf_.EnsureCapacity(64); });
-    auto res = IoLoop();
+    variant<error_code, Connection::ParserStatus> res;
+    if (GetFlag(FLAGS_expiremental_io_loop_v2) && !is_tls_) {


introduce a boolean variable instead of duplicating condition below.

romange · 2025-12-03T11:23:57Z

src/facade/dragonfly_connection.cc

+      return;
+    }
+    // A recv call can return fewer bytes than requested even if the
+    // socket buffer actually contains enough data to satisfy the full request.


why do you say this? yes, there are some corner cases but in general, Recv will try to unload as much as possible and fill the destination buffer

recv can return less data than available on the socket in non-blocking mode. so even if the read buffers contain 1k bytes and even if you try to read 1k, recv might return less.

I don't want to sound too confident here, that's my current understanding of the behaviour in non blocking mode.

Either way, I want to describe the reason why poll is important: a) because it could be that we read less data than it is available or b) the capacity of io_buf_ is less than what recv can satisfy in one call

I think this exact example can not happen. i,e, with 1KB you will always read 1KB. There are some limits of how much it will read in one call, but besides that, my understanding is that a) can happen only because of b)

also, define "available". how can you observe "available" size without reading the data?

actually, you can: ioctl(socket_fd, FIONREAD, &bytes_available)

Hmm, gemini confirms your statement. Good to know.

Yeah perplexity confirmed this for me as well but I do not trust it that much -- If I recall correctly I think I read this somewhere in a conversation on the kernel mailing list but I didn't verify it myself (and hence wrote I am not sure)

actually, you can: ioctl(socket_fd, FIONREAD, &bytes_available)

yes the receive buffers

romange · 2025-12-03T11:24:55Z

src/facade/dragonfly_connection.cc

+
+  // TODO EnableRecvMultishot
+
+  // Breaks with TLS. RegisterOnRecv is unimplemented.


this comment should move where you check is_tls_

romange · 2025-12-03T13:04:36Z

LGTM in general, let's try submitting it this week.

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim · 2025-12-03T13:15:46Z

LGTM in general, let's try submitting it this week.

I addressed your comments and run the tests again. Plz rereview

romange · 2025-12-03T15:55:32Z

src/facade/dragonfly_connection.cc

          << "Redis parser error: " << result << " during parse: " << ToSV(read_buffer);
    }
    read_buffer.remove_prefix(consumed);
+    total += consumed;


sorry I did not catch earlier - I suggest removing total counter and instead calling here
io_buf_.ConsumeInput(consumed);

I think it's better because as you said - we preempt here, during which time we can read more data.
this will lead to much smoother "pipeline-like" I/O.

Currently: you read, say 4KB, and your io_buf does not have more space. you start executing commands.
your read notifications won't fill the buffer until you finish running all the commands and empty your io_buf. only then you will call the recv and copy the next chunk.

Copilot

Pull request overview

This PR introduces IoLoopV2, a new asynchronous I/O implementation for Dragonfly's connection handling that eliminates fiber-blocking read calls. The new implementation uses multishot poll registration for non-blocking socket reads, allowing the connection fiber to handle other events while recv operations are processed asynchronously. This is currently experimental and only enabled for non-TLS connections when the expiremental_io_loop_v2 flag is set.

Key changes:

Adds IoLoopV2() method that uses asynchronous, non-blocking reads via poll registration instead of fiber-blocking recv calls
Fixes a bug in ParseRedis() where buffer consumption was incorrectly calculated when preemption occurred mid-parsing
Adds GitHub workflow for regression testing with IoLoopV2 enabled, excluding TLS tests

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 6 comments.

File	Description
src/facade/dragonfly_connection.h	Adds declarations for `DoReadOnRecv()`, `IoLoopV2()`, and synchronization primitives (`io_ec_`, `io_event_`) needed for asynchronous I/O
src/facade/dragonfly_connection.cc	Implements the new asynchronous I/O loop with poll-based readiness notifications, fixes buffer consumption tracking in `ParseRedis()`, and integrates the new loop as an experimental opt-in feature
.github/workflows/ioloop-v2-regtests.yml	Adds manual workflow for regression testing with IoLoopV2 enabled, excluding TLS tests
.github/actions/regression-tests/action.yml	Updates regression test action to support passing experimental flags to dragonfly binary

Copilot · 2025-12-03T17:49:15Z

.github/workflows/ioloop-v2-regtests.yml

+          aws-access-key-id: ${{ secrets.AWS_S3_ACCESS_KEY }}
+          aws-secret-access-key: ${{ secrets.AWS_S3_ACCESS_SECRET }}
+          s3-bucket: ${{ secrets.S3_REGTEST_BUCKET }}
+          df-arg: "expiremental_io_loop_v2"


Inconsistent spelling: "expiremental_io_loop_v2" should be "experimental_io_loop_v2" to match the spelling in the action.yml file (line 61).

Suggested change

df-arg: "expiremental_io_loop_v2"

df-arg: "experimental_io_loop_v2"

Copilot · 2025-12-03T17:49:15Z

src/facade/dragonfly_connection.cc

+    }
+
+    if (io_buf_.AppendLen() == 0) {
+      // We will regrow in IoLoop


Comment says "We will regrow in IoLoop" but this method is called from IoLoopV2, not IoLoop. The comment should be updated to say "We will regrow in IoLoopV2" for clarity.

Suggested change

// We will regrow in IoLoop

// We will regrow in IoLoopV2

Copilot · 2025-12-03T17:49:15Z

src/facade/dragonfly_connection.cc

+  size_t max_iobfuf_len = GetFlag(FLAGS_max_client_iobuf_len);
+
+  auto* peer = socket_.get();
+  recv_buf_.res_len = 0;
+
+  // TODO EnableRecvMultishot
+
+  peer->RegisterOnRecv([this](const FiberSocketBase::RecvNotification& n) {
+    DoReadOnRecv(n);
+    io_event_.notify_one();
+  });
+
+  do {
+    HandleMigrateRequest();
+
+    // Poll again for readiness. The event handler registered above is edge triggered
+    // (called once per socket readiness event). So, for example, it could be that the
+    // cb read less data than it is available because of io_buf_ capacity. If after
+    // an iteration the fiber does not poll the socket for more data it might deadlock.
+    // TODO maybe use a flag instead of a poll
+    DoReadOnRecv(FiberSocketBase::RecvNotification{true});
+    fb2::NoOpLock noop;
+    io_event_.wait(
+        noop, [this]() { return io_buf_.InputLen() > 0 || io_ec_ || io_buf_.AppendLen() == 0; });
+
+    if (io_ec_) {
+      LOG_IF(WARNING, cntx()->replica_conn) << "async io error: " << io_ec_;
+      return std::exchange(io_ec_, {});
+    }
+
+    phase_ = PROCESS;
+    bool is_iobuf_full = io_buf_.AppendLen() == 0;
+
+    if (io_buf_.InputLen() > 0) {
+      if (redis_parser_) {
+        parse_status = ParseRedis(max_busy_read_cycles_cached);
+      } else {
+        DCHECK(memcache_parser_);
+        parse_status = ParseMemcache();
+      }
+    } else {
+      parse_status = NEED_MORE;
+      DCHECK(io_buf_.AppendLen() == 0);
+    }
+
+    if (reply_builder_->GetError()) {
+      return reply_builder_->GetError();
+    }
+
+    if (parse_status == NEED_MORE) {
+      parse_status = OK;
+
+      size_t capacity = io_buf_.Capacity();
+      if (capacity < max_iobfuf_len) {
+        size_t parser_hint = 0;
+        if (redis_parser_)
+          parser_hint = redis_parser_->parselen_hint();  // Could be done for MC as well.
+
+        // If we got a partial request and we managed to parse its
+        // length, make sure we have space to store it instead of
+        // increasing space incrementally.
+        // (Note: The buffer object is only working in power-of-2 sizes,
+        // so there's no danger of accidental O(n^2) behavior.)
+        if (parser_hint > capacity) {
+          UpdateIoBufCapacity(io_buf_, stats_,
+                              [&]() { io_buf_.Reserve(std::min(max_iobfuf_len, parser_hint)); });
+        }
+
+        // If we got a partial request because iobuf was full, grow it up to
+        // a reasonable limit to save on Recv() calls.
+        if (is_iobuf_full && capacity < max_iobfuf_len / 2) {


Typo in variable name: "max_iobfuf_len" should be "max_iobuf_len" (missing 'u'). This typo is used consistently in lines 2241, 2294, 2306, and 2311.

Copilot · 2025-12-03T17:49:16Z

src/facade/dragonfly_connection.cc

+  // if (std::holds_alternative<io::MutableBytes>(n.read_result))
+  using RecvNot = util::FiberSocketBase::RecvNotification::RecvCompletion;
+  if (std::holds_alternative<RecvNot>(n.read_result)) {
+    if (!std::get<bool>(n.read_result)) {


Inconsistent type handling: Line 2193 defines RecvNot as an alias for RecvNotification::RecvCompletion and line 2194 checks for this type, but line 2195 uses std::get<bool> instead of std::get<RecvNot>. Either use RecvNot consistently or remove the unnecessary typedef if RecvCompletion is indeed just bool.

Suggested change

if (!std::get<bool>(n.read_result)) {

if (!std::get<RecvNot>(n.read_result)) {

Copilot · 2025-12-03T17:49:16Z

src/facade/dragonfly_connection.cc

+    return;
+  }
+
+  DCHECK(false) << "Sould not reach here";


Typo in error message: "Sould" should be "Should".

Suggested change

DCHECK(false) << "Sould not reach here";

DCHECK(false) << "Should not reach here";

Copilot · 2025-12-03T17:49:16Z

src/facade/dragonfly_connection.cc

          "If non-zero, waits for this time for more I/O "
          " events to come for the connection in case there is only one command in the pipeline. ");

+ABSL_FLAG(bool, expiremental_io_loop_v2, false, "new io loop");


Inconsistent spelling: flag is named "expiremental_io_loop_v2" (should be "experimental_io_loop_v2").

Suggested change

ABSL_FLAG(bool, expiremental_io_loop_v2, false, "new io loop");

ABSL_FLAG(bool, experimental_io_loop_v2, false, "new io loop");

kostasrim self-assigned this Nov 17, 2025

kostasrim force-pushed the kpr31 branch from e62d5df to 9ab2832 Compare November 28, 2025 09:48

kostasrim commented Nov 28, 2025

View reviewed changes

kostasrim force-pushed the kpr31 branch from 0c382f8 to 654c8f3 Compare November 28, 2025 09:54

kostasrim commented Nov 28, 2025

View reviewed changes

kostasrim added 3 commits November 28, 2025 11:58

chore: asynchronous IO for connection fiber

8d28c55

Signed-off-by: Kostas Kyrimis <[email protected]>

simple workflows + testing infra

ec18a18

Signed-off-by: Kostas Kyrimis <[email protected]>

revert logs

d587691

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim force-pushed the kpr31 branch from 654c8f3 to d587691 Compare November 28, 2025 09:58

kostasrim added 2 commits November 28, 2025 12:14

helio

334ade2

try to manually trigger v2 regtest workflow

f5b2787

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim force-pushed the kpr31 branch from db1ce16 to f5b2787 Compare November 28, 2025 10:31

kostasrim added 3 commits November 28, 2025 12:36

revert

74428b5

Signed-off-by: Kostas Kyrimis <[email protected]>

try

ae7188d

fixes

1a5c383

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim marked this pull request as ready for review December 1, 2025 10:49

kostasrim changed the title ~~[wip] chore: asynchronous IO for connection fiber~~ chore: asynchronous IO for connection fiber Dec 1, 2025

augmentcode bot reviewed Dec 1, 2025

View reviewed changes

kostasrim commented Dec 1, 2025

View reviewed changes

kostasrim requested a review from romange December 1, 2025 10:54

romange requested changes Dec 1, 2025

View reviewed changes

kostasrim added 2 commits December 2, 2025 15:05

comments

380cfde

Signed-off-by: Kostas Kyrimis <[email protected]>

comments

5eb6dc9

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim requested a review from romange December 3, 2025 10:44

romange reviewed Dec 3, 2025

View reviewed changes

comments

4664e4b

Signed-off-by: Kostas Kyrimis <[email protected]>

kostasrim requested a review from romange December 3, 2025 13:15

romange reviewed Dec 3, 2025

View reviewed changes

romange requested a review from Copilot December 3, 2025 17:44

Copilot started reviewing on behalf of romange December 3, 2025 17:45 View session

Copilot finished reviewing on behalf of romange December 3, 2025 17:48

Copilot AI reviewed Dec 3, 2025

View reviewed changes

kostasrim added 2 commits December 4, 2025 11:18

comments

53b70fa

order

dcb1594


		// TODO EnableRecvMultishot

		// Breaks with TLS. RegisterOnRecv is unimplemented.

	df-arg: "expiremental_io_loop_v2"
	df-arg: "experimental_io_loop_v2"

	if (!std::get<bool>(n.read_result)) {
	if (!std::get<RecvNot>(n.read_result)) {

	DCHECK(false) << "Sould not reach here";
	DCHECK(false) << "Should not reach here";

	ABSL_FLAG(bool, expiremental_io_loop_v2, false, "new io loop");
	ABSL_FLAG(bool, experimental_io_loop_v2, false, "new io loop");

chore: asynchronous IO for connection fiber #6069

Are you sure you want to change the base?

chore: asynchronous IO for connection fiber #6069

Conversation

kostasrim commented Nov 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kostasrim commented Dec 1, 2025

Uh oh!

augmentcode bot left a comment

Choose a reason for hiding this comment

Uh oh!

augmentcode bot Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

augmentcode bot Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

augmentcode bot Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

augmentcode bot Dec 1, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kostasrim commented Dec 1, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

romange Dec 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

romange Dec 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

kostasrim commented Nov 17, 2025 •

edited

Loading

romange Dec 3, 2025 •

edited

Loading

romange Dec 3, 2025 •

edited

Loading