Commit 5c69dde6 authored by jan.koester's avatar jan.koester
Browse files

epoll: close a connection by identity, not by fd number 

IoEventHandler looked the connection up by fd and then waited for its
event_mutex. If another worker closed it meanwhile (idle reaper, a close
from a re-armed event via requestWritablePoll), the waiting worker still
saw a non-null csock and went on: recv() on the closed fd ("tcp::recvData
failed: Bad file descriptor"), callbacks on a disconnected con, and its
own error path then called CloseEventHandler(fd) -- which, once the
kernel had reused the fd number, closed an unrelated new connection.
The reaper had the same problem: it collected fd numbers and closed by
number later.

- IoEventHandler: after taking event_mutex, go on only if c is still the
  connection registered under fd.
- closeConnection(c, fd): takes event_mutex first, then erases and
  EPOLL_CTL_DELs only if the map still holds c (lock order
  event_mutex -> POLL_HANDLER_MUTEX, as in detachConnection).
- reapIdleConnections: keeps the con, skips it if a worker holds its
  event_mutex (busy, not idle) and re-checks idleness under the lock.

New close_race_test: callbacks after DisconnectEvent in 16 of 20 rounds
before, 0 after.

Co-Authored-By: default avatarClaude Opus 5.5 <noreply@anthropic.com>
parent 8e72ce6e

src/event/epoll.cpp

+34 −22
Changes for src/event/epoll.cpp: 34 added lines, 22 removed lines.
Original line number Diff line number Diff line
@@ -318,7 +318,11 @@ namespace netplus { 
            bool doClose = false;

            {

                std::lock_guard<std::recursive_mutex> lk(c->event_mutex);

                if (!c->slots[0].csock) return;

                // Another thread (the idle reaper, a close from another event) may have

                // closed this connection while we waited for the lock -- and the kernel

                // may already have handed the fd number to a new connection. Only go on

                // if c is still the connection registered under fd.

                if (!c->slots[0].csock || getConByFd(fd) != c) return;



                EpollArmGuard rearm(fd, c, this);
@@ -567,43 +571,44 @@ namespace netplus { 
                }

            } // unlock event_mutex



            // Drop local reference under the same lock to serialize refcount updates

            c.reset();

            

            if (doClose) {

                CloseEventHandler(fd, tid, args);

                closeConnection(c, fd, tid, args);

            }

            c.reset();

        }



        // ------------------------------------------------------------

        // Close handler (safe + no deadlock)

        // ------------------------------------------------------------

        void CloseEventHandler(int fd, const int tid, ULONG_PTR args) override {

            std::shared_ptr<con> ccon;



            epoll_ctl(_pollFD, EPOLL_CTL_DEL, fd, nullptr);

            std::shared_ptr<con> ccon = getConByFd(fd);

            if (ccon) closeConnection(ccon, fd, tid, args);

        }



        // Closes exactly ccon -- not whatever connection is registered under fd by the

        // time we get here: once ccon's socket is closed the kernel can reuse the fd

        // number for a new connection, and closing by number alone would then take

        // down that unrelated connection. Lock order event_mutex -> POLL_HANDLER_MUTEX,

        // the same as IoEventHandler and detachConnection.

        void closeConnection(std::shared_ptr<con> ccon, int fd, const int tid, ULONG_PTR args) {

            {

                std::lock_guard<std::recursive_mutex> lk(ccon->event_mutex);

                {

                    std::unique_lock<std::shared_mutex> lock(POLL_HANDLER_MUTEX);

                    auto it = CONNECTIONS.find(fd);

                if (it == CONNECTIONS.end())

                    return;

                ccon = std::move(it->second);

                    if (it == CONNECTIONS.end() || it->second != ccon)

                        return; // already closed by another thread

                    CONNECTIONS.erase(it);

                }



            if (!ccon) return;



            // 3. Lock per connection and close the socket[cite: 1]

            {

                std::lock_guard<std::recursive_mutex> lk(ccon->event_mutex);

                // still ours: the socket isn't closed yet, so fd can't have been reused

                epoll_ctl(_pollFD, EPOLL_CTL_DEL, fd, nullptr);

                safeCloseSocket(ccon);

            }



            try {

                evconnection->DisconnectEvent(*ccon, tid, args);

            } catch (...) {

                // ignore user callback failures[cite: 1]

                // ignore user callback failures

            }

        }
@@ -629,19 +634,26 @@ namespace netplus { 
            if (now - _lastSweepAttempt < sweepIntervalSeconds()) return;

            _lastSweepAttempt = now;



            std::vector<int> idleFds;

            std::vector<std::pair<int, std::shared_ptr<con>>> idle;

            {

                std::shared_lock<std::shared_mutex> lk(POLL_HANDLER_MUTEX);

                for (auto& kv : CONNECTIONS) {

                    if (kv.second->ownerPollFd != _pollFD) continue; // not ours to close

                    if (now - kv.second->lasteventime < idleTimeoutSeconds) continue;

                    if (kv.second->hasPendingOps()) continue; // background dispatch task still using this connection

                    idleFds.push_back(kv.first);

                    idle.emplace_back(kv.first, kv.second);

                }

            }

            // Close outside POLL_HANDLER_MUTEX -- CloseEventHandler takes it again itself

            // Close outside POLL_HANDLER_MUTEX -- closeConnection takes it again itself

            // (unique_lock), and it's not recursive.

            for (int fd : idleFds) CloseEventHandler(fd, tid, args);

            for (auto& [fd, c] : idle) {

                // A worker holding event_mutex is handling this connection right now, so

                // it isn't idle -- skip it rather than block the sweep behind it.

                std::unique_lock<std::recursive_mutex> lk(c->event_mutex, std::try_to_lock);

                if (!lk.owns_lock()) continue;

                if (time(nullptr) - c->lasteventime < idleTimeoutSeconds) continue;

                closeConnection(c, fd, tid, args);

            }

        }



    private:

test/CMakeLists.txt

+8 −0
Changes for test/CMakeLists.txt: 8 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -303,3 +303,11 @@ if(UNIX) 
    target_link_libraries(socketwait_disconnect_test netplus-static)

    add_test(NAME socketwait_disconnect_test COMMAND socketwait_disconnect_test)

endif()



add_executable(close_race_test close_race_test.cpp)

if(WIN32)

    target_link_libraries(close_race_test netplus-static ws2_32)

else()

    target_link_libraries(close_race_test netplus-static)

endif()

add_test(NAME close_race_test COMMAND close_race_test)

test/close_race_test.cpp

0 → 100644
+132 −0
Changes for test/close_race_test.cpp: 132 added lines, 0 removed lines.
Original line number Diff line number Diff line 
/*******************************************************************************

 * Copyright (c) 2026, Jan Koester jan.koester@gmx.net

 * All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or without

 * modification, are permitted provided that the following conditions are met:

 * Redistributions of source code must retain the above copyright

 *      notice, this list of conditions and the following disclaimer.

 * Redistributions in binary form must reproduce the above copyright

 *      notice, this list of conditions and the following disclaimer in the

 *      documentation and/or other materials provided with the distribution.

 * Neither the name of the <organization> nor the

 *      names of its contributors may be used to endorse or promote products

 *      derived from this software without specific prior written permission.

 *

 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND

 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

 * DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY

 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES

 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;

 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND

 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 *******************************************************************************/



// A connection closed by one worker while a second worker is already waiting on its

// event_mutex must not be handled again by that second worker: with the socket closed

// it used to recv() on the dead fd ("tcp::recvData failed: Bad file descriptor") or,

// once the kernel reused the fd number, on somebody else's connection.

//

// requestWritablePoll() re-arms the fd while RequestEvent still runs, so a second

// worker picks the connection up and blocks on event_mutex; RequestEvent then throws,

// which closes the connection. No callback may reach the connection after its

// DisconnectEvent.



#include <atomic>

#include <chrono>

#include <iostream>

#include <mutex>

#include <set>

#include <stdexcept>

#include <thread>



#include "connection.h"

#include "eventapi.h"

#include "exception.h"



using namespace netplus;



namespace {



class CloseRaceServer : public event {

  public:

    std::atomic<int> lateCallbacks{0};

    std::atomic<int> disconnects{0};



    CloseRaceServer(std::vector<netplus::socket*> serversockets, int timeout)

        : event(serversockets, timeout, 0) {}



    void RequestEvent(con &curcon, const int, ULONG_PTR) override {

        checkAlive(curcon);

        curcon.RecvData.clear();

        requestWritablePoll(curcon);

        // let another worker take the re-armed event and queue up on event_mutex

        std::this_thread::sleep_for(std::chrono::milliseconds(100));

        throw std::runtime_error("close this connection");

    }



    void ResponseEvent(con &curcon, const int, ULONG_PTR) override { checkAlive(curcon); }

    void ConnectEvent(con &, const int, ULONG_PTR) override {}



    void DisconnectEvent(con &curcon, const int, ULONG_PTR) override {

        std::lock_guard<std::mutex> l(_m);

        _closed.insert(&curcon);

        ++disconnects;

    }



    void CreateConnection(std::shared_ptr<con> &res) override {

        res = std::make_shared<con>(this);

        // a new connection may get the address of a freed one

        std::lock_guard<std::mutex> l(_m);

        _closed.erase(res.get());

    }



  private:

    void checkAlive(con &c) {

        std::lock_guard<std::mutex> l(_m);

        if (_closed.count(&c)) ++lateCallbacks;

    }



    std::mutex _m;

    std::set<con*> _closed;

};



} // namespace



int main() {

    const int rounds = 20;

    tcp serverSocket("127.0.0.1", 18086, 1024, -1);

    CloseRaceServer srv({&serverSocket}, 200);



    event::Running = true;

    std::thread serverThread([&srv]() { srv.runEventloop(); });

    std::this_thread::sleep_for(std::chrono::milliseconds(500));



    for (int i = 0; i < rounds; ++i) {

        tcp client;

        client.connect("127.0.0.1", 18086);

        buffer first((char*)"A", 1);

        client.sendData(first);

        // more data waiting, so the second worker has something to read

        std::this_thread::sleep_for(std::chrono::milliseconds(20));

        buffer second((char*)"B", 1);

        try { client.sendData(second); } catch (...) {}

        std::this_thread::sleep_for(std::chrono::milliseconds(250));

    }



    event::Running = false;

    serverThread.join();



    std::cout << "[close_race_test] rounds=" << rounds

              << " disconnects=" << srv.disconnects

              << " callbacks after DisconnectEvent=" << srv.lateCallbacks << std::endl;

    if (srv.disconnects < rounds || srv.lateCallbacks != 0) {

        std::cerr << "  FAIL" << std::endl;

        return 1;

    }

    std::cout << "  OK" << std::endl;

    return 0;

}