epoll `test_replication_all` always hangs when killing processes at the end

[abhijat]

At the end of the test_replication_all (which has several combinations and several instances of dragonfly) the running dragonfly instances (master and replica) are stopped.

For some reason now the stop never proceeds smoothly (in python, the OS pid seems to exit fine), when we call proc.terminate() then the following proc.communicate() hangs until timeout (2 minutes).

Every shutdown takes over 2 minutes to complete, the test becomes much slower as a result.

[abhijat]

The dragonfly process (replica) are actually stuck. Printing the fibers, the following stands out:

I20251023 18:25:14.173482 213279 scheduler.cc:534] ------------ Fiber shard_stable_sync_read (suspended:83036ms) ------------
0x1369911  std::function<>::operator()()
0x1368731  util::fb2::detail::FiberInterface::ExecuteOnFiberStack()::{lambda()#1}::operator()()
0x136969d  boost::context::detail::fiber_ontop<>()
0x1369140  boost::context::fiber::resume_with<>()
0x136819b  util::fb2::detail::FiberInterface::SwitchTo()
0x135de53  util::fb2::detail::Scheduler::Preempt()
0x41a98e  util::fb2::detail::FiberInterface::Suspend()
0x13f2020  util::fb2::EpollSocket::PendingReq::Suspend()
0x13f4a16  util::fb2::EpollSocket::RecvMsg()
0x13f5b5c  util::fb2::EpollSocket::Recv()
0x134529d  util::FiberSocketBase::ReadSome()
0x13fc88f  io::PrefixSource::ReadSome()
0x13fe1af  io::Source::ReadSome()
0x13fc478  io::Source::ReadAtLeast()
0x1007900  dfly::JournalReader::EnsureRead()

This is after the Replica::Stop() method gets stuck for over 20 seconds, while waiting for the sync_fb_ to join.

[kostasrim]

I think the epoll reactor does not wake up the blocked/pending read during the shutdown. Will give it a check when I get a small free window

[romange]

@abhijat does it reproduce consistently?
we should indeed verify that Shutdown was called on the socket and it did not break RecvMsg call.
Could be done with logging if this scenario is easily reproducing.

[abhijat]

I can reproduce it locally consistently with

pytest tests/dragonfly -k test_replication_all --df force_epoll=true

The following parameters reproduce this when added on the test

@pytest.mark.parametrize(
    "t_master, t_replicas, seeder_config, stream_target",
    [
        pytest.param(8, 6 * [1], dict(key_target=5_000), 2_000, marks=M_SLOW),
    ],
)
@pytest.mark.parametrize("mode", [{}])
@pytest.mark.parametrize("background_snapshotting", [False])

Will add some logs, it looks like the DflyShardReplica does not join.

[abhijat]

From the log, the error handler is triggered, which calls dfly::DflyShardReplica::Cancel, which shuts down the socket. But dfly::DflyShardReplica::sync_fb_ does not join, it is still stuck on RecvMsg.

@kostasrim I can also see the test stops hanging and starts to work fine, if I move back the socket close call to ProtocolClient::~ProtocolClient from ProtocolClient::CloseSocket (as it was before #5927) but it doesn't make sense why this would work since the socket is closed correctly by the error handler.

[abhijat]

I think the epoll reactor does not wake up the blocked/pending read during the shutdown. Will give it a check when I get a small free window

We have the following sequence of calls:

• replica goes into stable sync, waits on Replica::JoinDflyFlows
• the flows themselves are blocked on RecvMsg inside tx_reader.NextTxData(&reader, cntx)
• Replica::Stop is called, which triggers the error handler, which calls DflyShardReplica::Cancel on each flow
• this calls ProtocolClient::CloseSocket which in turn shuts down and then closes the socket
• closing the socket fd will remove it from the epoll interest list, but we also do that explicitly in EpollProactor::Disarm() through EpollDel

After the last point, where the fd is removed from epoll interest list, it will not receive any further notifications, so the flows blocked on step 2 will never get unblocked, and step 1 is also blocked forever.

This should not be a problem because in step 4 (ProtocolClient::CloseSocket) we first shutdown the socket before closing it, which should be enough to trigger the reader on step 2 to see that the connection is broken.

I think what happens is that sometimes (very often) the socket close and shutdown happen so close together such that the reader does not get an event and by the end of close, the fd is removed from the interest list of epoll so it will never get an event after that.

To test this theory I added a sleep between shutdown and close so that epoll might be able to recognize the shutdown:

diff --git a/src/server/protocol_client.cc b/src/server/protocol_client.cc
index 166d439d..b913d53b 100644
--- a/src/server/protocol_client.cc
+++ b/src/server/protocol_client.cc
@@ -228,6 +228,7 @@ void ProtocolClient::CloseSocket() {
         auto ec = sock_->Shutdown(SHUT_RDWR);
         LOG_IF(ERROR, ec) << "Could not shutdown socket " << ec;
       }
+      ThisFiber::SleepFor(100ms);

       auto ec = sock_->Close();  // Quietly close.
       LOG_IF(WARNING, ec) << "Error closing socket " << ec << "/" << ec.message();

With this sleep added the tests seem to run as normal. Without this the shutdown always hangs.

[abhijat]

I think what happens is that sometimes (very often) the socket close and shutdown happen so close together such that the reader does not get an event and by the end of close

I don't have any proof of this, its just a speculation. My theory is that shutdown and close in quick succession make the reader miss the shutdown event. Is there anything in epoll that guarantees this will not happen?

[kostasrim]

@abhijat yes exactly and I agree it seems that we do miss the notification/event. From the body of RecvMsg

  404   do {                                                                                                                                                                                                      
  405     if (fd_ & IS_SHUTDOWN) {                                                                                                                                                                                  406       ec = make_error_code(errc::connection_aborted);                                                                                                                                                         407       break;                                                                                                                                                                                                  408     }                                                                                                                                                                                                         409                                                                                                                                                                                                               410     ssize_t res = recvmsg(fd, const_cast<msghdr*>(&msg), flags);                                                                                                                                            
  411     if (res > 0) {  // if res is 0, that means a peer closed the socket.                                                                                                                                      412       return res;                                                                                                                                                                                             413     }                                                                                                                                                                                                         414                                                                                                                                                                                                             
  415     if (res == 0) {                                                                                                                                                                                         
  416       ec = make_error_code(errc::connection_aborted);                                                                                                                                                       
  417       break;                                                                                                                                                                                                
  418     }                                                                                                                                                                                                       
  419                                                                                                                                                                                                             
  420     if (errno != EAGAIN) {                                                                                                                                                                                  
  421       ec = from_errno();                                                                                                                                                                                    
  422       break;                                                                                                                                                                                                
  423     }                                                                                                                                                                                                       
  424                                                                                                                                                                                                             
  425     PendingReq req(&read_req_);                                                                                                                                                                             
  426     ec = req.Suspend(timeout());                                                                                                                                                                            
  427   } while (!ec);  

We suspend with timeout() = uint32_max which is indefinite. Therefore, the request will never timeout neither will it get notified and as a side effect it will never loop to fd_ & IS_SHUTDOWN on line 405. This seems like a deadlock because:

1. Operation never times out
2. It can reach to a state where it never receives an event

Worth try lowering the timeout to a second or something to test this :)

[kostasrim]

Ok I think I fully understand what happens:

I don't have any proof of this, its just a speculation. My theory is that shutdown and close in quick succession make the reader miss the shutdown event. Is there anything in epoll that guarantees this will not happen?

Kinda. It should always happen right ? It gets fixed because sleep yields the fiber which allows the event loop to observe the Shutdown. The question is why ? Let me explain, you almost answered this yourself:

1. When we create an epoll socket we register a callback (see EpollSocket:Wakey). The epoll is not a proactor, so it works a little differently from iouring. Specifically, when we suspend Recv because of EAGAIN we wait for the fd to become readable. The fiber is never placed on the active queue of the scheduler (the active queue is the fibers that the scheduler is allowed to run). Instead, the epoll reactor main loop will notify about the readiness of a fd via a registered cb: Wakey(). Once called, it will place the fiber back to the scheduler's active queue which will eventually wake the fiber back for execution.

2. Now if you do Shutdown() and then Close() you do this:

 >>369 void EpollProactor::Disarm(int fd, unsigned arm_index) {                                                                                                                                                    
  370   DCHECK(pthread_self() == thread_id_);                                                                                                                                                                     
  371                                                                                                                                                                                                             
  372   DVLOG(2) << "Disarming " << fd << " on " << arm_index;                                                                                                                                                    
  373   CHECK_LT(arm_index, centries_.size());                                                                                                                                                                    
  374                                                                                                                                                                                                             
  375   centries_[arm_index].cb = nullptr;                   <------------------------------- Bye Bye handler                                                                                                                                                     
  376   centries_[arm_index].index = next_free_ce_;                                                                                                                                                               
  377                                                                                                                                                                                                             
  378   next_free_ce_ = arm_index;                                                                                                                                                                                
  379   EpollDel(epoll_fd_, fd);                                                                                                                                                                                  
  380 }  

and because timeout() is indefinite you never wake up. You just removed the cb that calls Wakey(). We will never get any event and therefore the fiber will never be placed to the active queue of the scheduler.

That's how we deadlock.

Ths fix is simple, add a Yield between the calls (not a sleep) such that we allow the event to complete. (maybe it's a little more complicated because we might not gurantee that after the yield the loop exited -- so double check this)

IMO these are subtle implementation differences of iouring/epoll that leak to the API users (us) unnoticed