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

Workarounds and Failing tests

test_timeout fails in assert int(info["timeout_disconnects"]) >= 1. It was non critical and this test has also failed upstream a couple of times.
Added DflyInstance::has_arg to avoid propagating io_loop_v2 flag on tests that use older versions of dragonfly
Calls ResetOnRecvHook inside Migrate() instead of OnPreMigrateThread. The reason for that is that OnPreMigrateThread check fails if the connection is closing and because deregistering the hook is preemptive -- this invariant might break. To mitigate this, we deregister inside Migrate() instead and if after it the connection is closing we exit instead of starting the migration.
added migration_allowed_to_register_ to flag when a migration should register a callback. A connection might migrate before it reaches its asynchronous flow (io_loop_v2) which can lead to double registration errors (one from the migration and one once it reaches io_loop_v2). To respect the synchronous parts of the connection, we use a flag to indicate when a migration should also register a callback.
corner case in ConsumeInput() read comment in: https://github.com/dragonflydb/dragonfly/pull/6069/files#diff-b88cbca1b2c9337ab2a37262f9d516b117f4996da25e2ae173e0e129f2b82b20R1207

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.

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

kostasrim · 2025-12-10T09:39:32Z

augment review

kostasrim · 2025-12-10T09:39:50Z

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

augmentcode

Review completed. 2 suggestions posted.

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

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

augmentcode · 2025-12-10T09:42:09Z

src/facade/dragonfly_connection.cc

+}
+
+variant<error_code, Connection::ParserStatus> Connection::IoLoopV2() {
+  error_code ec;


error_code ec; is declared but never used in IoLoopV2(), which can trigger an unused-variable warning; consider removing it.

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

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

romange · 2025-12-10T14:02:48Z

src/facade/dragonfly_connection.cc

  VLOG(1) << "Connection::OnShutdown";

  BreakOnce(POLLHUP);
+  io_ec_ = make_error_code(errc::connection_aborted);


I would expect that it's not needed as Connection::Shutdown() should generate a socket notification

it doesn't for some some reason -- test_no_tls_on_admin_port

I want to get this in a green state and then I can follow up

romange · 2025-12-10T14:03:26Z

src/facade/dragonfly_connection.cc


 void Connection::OnPostMigrateThread() {
-  DVLOG(1) << "[" << id_ << "] OnPostMigrateThread";
+  DVLOG(1) << "[" << id_ << "] OnPostMigrateThread " << GetClientId();


it's not needed. GetClientId(); is id_ which is already printed in DVLOG

romange · 2025-12-10T14:04:12Z

src/facade/dragonfly_connection.cc

+  const bool io_loop_v2 = GetFlag(FLAGS_experimental_io_loop_v2);
+  if (io_loop_v2 && !is_tls_ && socket_ && socket_->IsOpen() && allowed_to_register_) {
+    socket_->RegisterOnRecv([this](const FiberSocketBase::RecvNotification& n) {
+      CHECK(this);


please remove check(this) - it's always true.

romange · 2025-12-10T15:10:46Z

src/facade/dragonfly_connection.cc

 }

-void Connection::HandleMigrateRequest() {
+void Connection::HandleMigrateRequest(bool unregister) {


you do not use unregister

romange · 2025-12-10T15:14:43Z

src/facade/dragonfly_connection.cc

+      // Migrations should call RegisterRecv if the connection has reached here once.
+      // Otherwise, a migration won't register and instead wait for the connection state
+      // to first reach here and then call RegisterRecv inside IoLoopV2.
+      allowed_to_register_ = true;


I think allowed_to_register_ is redundant - I did not see how you use it and it's always true inside RegisterRecv

It's needed here as we might end up calling RegisterOnRecv twice. Everything above the call to IoLoopV2 is synchronous and fiber blocking. So, the connection might migrate before it reaches IoLoopV2, call RegisteOnRecv() as part of the PostMigration(). Eventually, it will reach IoLoopV2 body which will try to register and check fail. I edited the comment.

romange

several minor comments

romange · 2025-12-11T08:22:32Z

src/facade/dragonfly_connection.cc

+    if (io_loop_v2 && !is_tls_) {
+      // Everything above the IoLoopV2 is fiber blocking. A connection can migrate before
+      // it reaches here and will cause a double RegisterOnRecv check fail. To avoid this,
+      // a migration shall only call RegisterOnRev if it reached the main IoLoopV2 below.


I think it's not enough. Similarly to CancelOnErrorCb - you need to unregister from the polling in OnPreMigrateThread function

WIll take a look

I see. So, I do have it, but inside Migrate() and not in OnPreMigrateThread. The reason is that ResetOnRecvHook() is preemptive and by the time we wake up the check:

658 CHECK(!cc_->conn_closing);

Can fail. To avoid this, I moved it to Migrate() such that if after reseting the hook the connection is closing, we exit instead of moving on with the migration. It's a subtle corner case.

Logically, we should deregister in OnPreMigrate just like we register on OnPostMigrate -- however I don't see a clean way to do this

we could call ResetOnRecvHook after the check(), after socket_->CancelOnErrorCb();,
but ok - we can check it later.

I do not want you to do more changes in this PR. Can you please summarize
TODOs - failing/skipped tests, unresolved issues/workarounds like this one, pubsub failure etc in the PR description so it won't get lost completely after we merge?

we could call ResetOnRecvHook after the check(), after socket_->CancelOnErrorCb();,
but ok - we can check it later.

Yes but then we would move on with a migration on a closing connection. Doin that seems at least redundant 🤷

I do not want you to do more changes in this PR. Can you please summarize TODOs - failing/skipped tests, unresolved issues/workarounds like this one, pubsub failure etc in the PR description so it won't get lost completely after we merge?

sure! one sec

@romange done!

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

kostasrim added 9 commits December 9, 2025 15:00

CI

3e4727d

Merge branch 'main' into kpr31

05181be

skip

ee632c5

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

Merge branch 'main' into kpr31

0b2b127

helio

8cc279a

pop

d8f3f03

manual trigger

af32a6e

helio

fbf0266

one to rull them all

6210e17

kostasrim requested a review from romange December 10, 2025 09:39

augmentcode bot reviewed Dec 10, 2025

View reviewed changes

kostasrim added 2 commits December 10, 2025 13:18

pollhup on migration path

b657fe8

preemption point

bb4e230

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

romange reviewed Dec 10, 2025

View reviewed changes

reset for all conn

ba35ecb

romange reviewed Dec 10, 2025

View reviewed changes

kostasrim added 4 commits December 10, 2025 17:33

comments

10d96d0

remove

21a3fab

revert

146ed0d

revert

7221f9b

romange approved these changes Dec 10, 2025

View reviewed changes

romange reviewed Dec 11, 2025

View reviewed changes

kostasrim merged commit 128df73 into main Dec 11, 2025
14 of 15 checks passed

kostasrim deleted the kpr31 branch December 11, 2025 09:40

chore: asynchronous IO for connection fiber #6069

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!

Workarounds and Failing tests

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!

kostasrim commented Dec 10, 2025

Uh oh!

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

Uh oh!

Uh oh!

kostasrim commented Nov 17, 2025 •

edited

Loading

romange Dec 3, 2025 •

edited

Loading

kostasrim commented Dec 10, 2025 •

edited

Loading

kostasrim Dec 10, 2025 •

edited

Loading