Commit e196ce60 authored by jan.koester's avatar jan.koester
Browse files

fixed locking

parent a2e5648d

src/CMakeLists.txt

+1 −0
Original line number Diff line number Diff line
@@ -58,6 +58,7 @@ set(headers 
    exception.h

    eventapi.h

    connection.h

    rwlock.h

    random.h

    crypto/base64.h

    crypto/des.h

src/quic.cpp

+10 −10
Original line number Diff line number Diff line
@@ -2119,7 +2119,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
        if (purge_now - _last_registry_purge >= _registry_purge_interval) {

            _last_registry_purge = purge_now;

            std::vector<std::shared_ptr<quic>> to_destroy;

            std::unique_lock<std::shared_mutex> reg_lock(_registry_mutex);

            std::unique_lock<netplus::WriterPreferringSharedMutex> reg_lock(_registry_mutex);

            for (auto it = _child_connections.begin(); it != _child_connections.end(); ) {

                if (it->second && it->second->_conn_state.load() == ConnectionState::Closed) {

                    to_destroy.push_back(std::move(it->second));
@@ -2269,7 +2269,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
        // touching anything about the connection itself.

        std::shared_ptr<quic> existing;

        {

            std::shared_lock<std::shared_mutex> reg_lock(_registry_mutex);

            std::shared_lock<netplus::WriterPreferringSharedMutex> reg_lock(_registry_mutex);

            auto it = _child_connections.find(cidKey(&dgram[1], cid_len));

            if (it != _child_connections.end()) existing = it->second;

        }
@@ -2341,7 +2341,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 


    std::shared_ptr<quic> existing;

    {

        std::shared_lock<std::shared_mutex> reg_lock(_registry_mutex);

        std::shared_lock<netplus::WriterPreferringSharedMutex> reg_lock(_registry_mutex);

        auto it = _child_connections.find(cidKey(dcid));

        if (it != _child_connections.end()) existing = it->second;

    }
@@ -2358,7 +2358,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
            // discipline as the purge step above, and for the same reason.

            {

                std::vector<std::shared_ptr<quic>> to_destroy;

                std::unique_lock<std::shared_mutex> reg_lock2(_registry_mutex);

                std::unique_lock<netplus::WriterPreferringSharedMutex> reg_lock2(_registry_mutex);

                auto it1 = _child_connections.find(cidKey(existing->_local_cid));

                if (it1 != _child_connections.end()) {

                    to_destroy.push_back(std::move(it1->second));
@@ -2483,7 +2483,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
        auto now = std::chrono::steady_clock::now();

        bool already_retried = false;

        {

            std::unique_lock<std::shared_mutex> reg_lock(_registry_mutex);

            std::unique_lock<netplus::WriterPreferringSharedMutex> reg_lock(_registry_mutex);

            for (auto it = _recent_retries.begin(); it != _recent_retries.end(); ) {

                if (now - it->second > kRetryDedupeWindow) it = _recent_retries.erase(it);

                else ++it;
@@ -2562,7 +2562,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
    // done any expensive crypto work or touched the network for it yet.

    std::shared_ptr<quic> winner;

    {

        std::unique_lock<std::shared_mutex> reg_lock(_registry_mutex);

        std::unique_lock<netplus::WriterPreferringSharedMutex> reg_lock(_registry_mutex);

        auto it = _child_connections.find(cidKey(dcid));

        if (it != _child_connections.end()) {

            winner = it->second;
@@ -6579,7 +6579,7 @@ void quic::processClientHello(const std::vector<uint8_t>& msg) { 
                    bool already_used = false;

                    {

                        auto now = std::chrono::steady_clock::now();

                        std::unique_lock<std::shared_mutex> reg_lock(listener->_registry_mutex);

                        std::unique_lock<netplus::WriterPreferringSharedMutex> reg_lock(listener->_registry_mutex);

                        for (auto it = listener->_used_ticket_ids.begin(); it != listener->_used_ticket_ids.end(); ) {

                            if (now > it->second) it = listener->_used_ticket_ids.erase(it);

                            else ++it;
@@ -7815,7 +7815,7 @@ void quic::pumpIncomingLocked() { 
            // Short header — route via DCID

            bool no_children;

            {

                std::shared_lock<std::shared_mutex> reg_lock(parent->_registry_mutex);

                std::shared_lock<netplus::WriterPreferringSharedMutex> reg_lock(parent->_registry_mutex);

                no_children = parent->_child_connections.empty();

            }

            if (!_parent && no_children) {
@@ -7823,7 +7823,7 @@ void quic::pumpIncomingLocked() { 
            } else {

                static constexpr size_t CID_LEN = 8;

                if (len < 1 + CID_LEN) continue;

                std::shared_lock<std::shared_mutex> reg_lock(parent->_registry_mutex);

                std::shared_lock<netplus::WriterPreferringSharedMutex> reg_lock(parent->_registry_mutex);

                auto it = parent->_child_connections.find(cidKey(data + 1, CID_LEN));

                if (it != parent->_child_connections.end()) target = it->second;

            }
@@ -7849,7 +7849,7 @@ void quic::pumpIncomingLocked() { 
            if (len >= 7) {

                size_t dcid_len = static_cast<size_t>(data[5]);

                if (6 + dcid_len <= len) {

                    std::shared_lock<std::shared_mutex> reg_lock(parent->_registry_mutex);

                    std::shared_lock<netplus::WriterPreferringSharedMutex> reg_lock(parent->_registry_mutex);

                    auto it = parent->_child_connections.find(cidKey(data + 6, dcid_len));

                    if (it != parent->_child_connections.end()) target = it->second;

                }

src/rwlock.h

0 → 100644
+81 −0
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.

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



#pragma once



#include <shared_mutex>



#if defined(__GLIBC__)

#include <pthread.h>

#endif



namespace netplus {



#if defined(__GLIBC__)

// std::shared_mutex gives no portable way to request writer-preferring

// fairness, and glibc's default pthread_rwlock policy

// (PTHREAD_RWLOCK_PREFER_READER_NP) lets a steady stream of short-lived

// readers starve out a waiting writer indefinitely — reproduced live on a

// production mediadb node, where a connection's own congestion-wait retry

// loop (quic.cpp's sendStreamData(), calling pumpIncomingLocked() up to

// ~2000x per stalled chunk) kept re-acquiring a registry's shared lock for

// the duration of a large transfer, starving accept()'s one-shot exclusive

// lock for registering a brand-new connection. Same API surface as

// std::shared_mutex so it drops into existing std::shared_lock/

// std::unique_lock call sites unchanged.

class WriterPreferringSharedMutex {

public:

    WriterPreferringSharedMutex() {

        pthread_rwlockattr_t attr;

        pthread_rwlockattr_init(&attr);

        pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);

        pthread_rwlock_init(&_lock, &attr);

        pthread_rwlockattr_destroy(&attr);

    }

    ~WriterPreferringSharedMutex() { pthread_rwlock_destroy(&_lock); }



    WriterPreferringSharedMutex(const WriterPreferringSharedMutex&) = delete;

    WriterPreferringSharedMutex& operator=(const WriterPreferringSharedMutex&) = delete;



    void lock() { pthread_rwlock_wrlock(&_lock); }

    void unlock() { pthread_rwlock_unlock(&_lock); }

    bool try_lock() { return pthread_rwlock_trywrlock(&_lock) == 0; }



    void lock_shared() { pthread_rwlock_rdlock(&_lock); }

    void unlock_shared() { pthread_rwlock_unlock(&_lock); }

    bool try_lock_shared() { return pthread_rwlock_tryrdlock(&_lock) == 0; }



private:

    pthread_rwlock_t _lock;

};

#else

// pthread_rwlockattr_setkind_np is a glibc extension, unavailable on

// musl/BSD/Windows builds — fall back to the portable default there.

using WriterPreferringSharedMutex = std::shared_mutex;

#endif



} // namespace netplus

src/socket.h

+11 −1
Original line number Diff line number Diff line
@@ -43,6 +43,7 @@ 
#include <functional>



#include "config.h"

#include "rwlock.h"

#include "crypto/aes.h"

#include "crypto/x509.h"

#include "crypto/sha.h"
@@ -1622,7 +1623,16 @@ namespace netplus { 
		// unique_lock for the rare registry mutation. Declared before

		// _child_connections for the same destruction-order reason as

		// _quic_mutex above.

		mutable std::shared_mutex _registry_mutex;

		//

		// Writer-preferring, not plain std::shared_mutex: a busy

		// connection's sendStreamData() congestion-wait loop calls

		// pumpIncomingLocked() up to ~2000x per stalled chunk, each taking

		// a shared_lock here — with glibc's default reader-preferring

		// policy that can starve out accept()'s exclusive lock (needed to

		// register a brand-new connection) for the entire transfer.

		// Reproduced live in production (mediadb cluster DEGRADED

		// flapping / "Failed to connect to node" errors).

		mutable netplus::WriterPreferringSharedMutex _registry_mutex;



		// Connection tracking for server mode (maps DCID to child quic*).

		// Keyed by the raw CID bytes as a std::string (see cidKey() in