Commit 91f071d5 authored by jan.koester's avatar jan.koester
Browse files

test

parent bdf7ed4c

src/quic.cpp

+59 −14
Original line number Diff line number Diff line
@@ -1985,6 +1985,10 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
                    existing->_pending_max_data = false;

                }

                existing->pmtudTick();

                // F20: sendMaxStreamData/sendMaxData above only queued their

                // packets (batch=true) — flush now, before this block's own

                // scope ends, so they never sit unsent.

                existing->flushBatch();

            }



            if (!existing->_pending_dispatches.empty()) {
@@ -2162,6 +2166,8 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
                    existing->_pending_max_data = false;

                }

                existing->pmtudTick();

                // F20: flush what sendMaxStreamData/sendMaxData just queued.

                existing->flushBatch();

            }



            // Dispatch stream callbacks off-thread (see scheduleDispatches)
@@ -2348,6 +2354,8 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
                winner->_pending_max_data = false;

            }

            winner->pmtudTick();

            // F20: flush what sendMaxStreamData/sendMaxData just queued.

            winner->flushBatch();

        }



        if (!winner->_pending_dispatches.empty()) {
@@ -2425,6 +2433,8 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
            child_ptr->_pending_max_data = false;

        }

        child_ptr->pmtudTick();

        // F20: flush what sendMaxStreamData/sendMaxData just queued.

        child_ptr->flushBatch();

    }



    // Dispatch any pending child callbacks off-thread (see scheduleDispatches)
@@ -3163,7 +3173,7 @@ std::vector<uint8_t> quic::buildShortHeaderPacket(const std::vector<uint8_t>& pa 
// DATA_BLOCKED, STREAM_DATA_BLOCKED) — see _ctrl_header_scratch/

// _ctrl_pkt_scratch's declaration in socket.h. Returns sendPacket()'s

// result so callers keep their existing retry-on_failure behavior.

ssize_t quic::sendShortHeaderControlPacket(const std::vector<uint8_t>& frame) {

ssize_t quic::sendShortHeaderControlPacket(const std::vector<uint8_t>& frame, bool batch) {

    auto& header = _ctrl_header_scratch;

    header.clear();
@@ -3181,6 +3191,15 @@ ssize_t quic::sendShortHeaderControlPacket(const std::vector<uint8_t>& frame) { 


    protectPacketInto(header, frame, pn, EncryptionLevel::Application, _app_keys.send, _ctrl_pkt_scratch);

    applyHeaderProtection(_ctrl_pkt_scratch, EncryptionLevel::Application, _app_keys.send);

    if (batch) {

        // F20: queue instead of an immediate syscall — caller guarantees a

        // flushBatch() before it returns (see this parameter's comment in

        // socket.h). Report as fully sent: the batch's own partial/failure

        // handling (flushBatch(), quic.cpp) retries on the next flush

        // rather than needing this call's return value.

        batchPacket(_ctrl_pkt_scratch.data(), _ctrl_pkt_scratch.size());

        return static_cast<ssize_t>(_ctrl_pkt_scratch.size());

    }

    return sendPacket(_ctrl_pkt_scratch.data(), _ctrl_pkt_scratch.size());

}
@@ -3769,6 +3788,9 @@ void quic::pumpNetwork(int flags) { 
                sendMaxData(_pending_max_data_value);

                _pending_max_data = false;

            }

            // F20: flush what sendMaxStreamData/sendMaxData just queued —

            // still inside this block's lock, before it goes out of scope.

            flushBatch();

        }



        // Collect pending dispatches
@@ -5290,12 +5312,10 @@ void quic::sendMaxStreamData(uint64_t stream_id, uint64_t max_data) { 
    bytes = encodeVarInt(max_data, buf);

    frame.insert(frame.end(), buf, buf + bytes);



    ssize_t sent = sendShortHeaderControlPacket(frame);

    if (sent < 0) {

        // Flow control frames are critical — immediate non-blocking retry.

        // Do not sleep here: caller typically holds quic_mtx().

        sendPacket(_ctrl_pkt_scratch.data(), _ctrl_pkt_scratch.size());

    }

    // F20: batched — every call site is one of the repeated "flush pending

    // flow control" blocks, each already followed by a flushBatch() call,

    // so this never sits queued past the caller's own return.

    sendShortHeaderControlPacket(frame, /*batch=*/true);

}



// ============================================================================
@@ -5309,12 +5329,8 @@ void quic::sendMaxData(uint64_t max_data) { 
    size_t bytes = encodeVarInt(max_data, buf);

    frame.insert(frame.end(), buf, buf + bytes);



    ssize_t sent = sendShortHeaderControlPacket(frame);

    if (sent < 0) {

        // Flow control frames are critical — immediate non-blocking retry.

        // Do not sleep here: caller typically holds quic_mtx().

        sendPacket(_ctrl_pkt_scratch.data(), _ctrl_pkt_scratch.size());

    }

    // F20: batched — see sendMaxStreamData()'s comment above.

    sendShortHeaderControlPacket(frame, /*batch=*/true);

}



// ============================================================================
@@ -6221,6 +6237,29 @@ size_t quic::sendData(buffer& data, int flags) { 
// ============================================================================



size_t quic::sendStreamData(uint64_t stream_id, const uint8_t* data, size_t len, bool fin) {

    // F30 (Performance-Report): bewusst NICHT umgesetzt. Der Report schlägt vor,

    // die AES-GCM-Verschlüsselung aus dem quic_mtx()-Lock herauszulösen und nur

    // die PN-Reservierung sowie die _sent_packets/cwnd-Buchführung unter einem

    // kurzen Lock zu halten, um die gemessene Kontention zwischen einem

    // sendenden Application-Thread und dem gleichzeitig ACKs verarbeitenden

    // Netzwerk-Thread derselben Verbindung zu reduzieren.

    //

    // Das würde eine Umstrukturierung erfordern, bei der PN-Reservierung,

    // Verschlüsselung und der eigentliche sendPacket()-Syscall mit

    // unterschiedlicher Lock-Granularität ablaufen, während die PN-Reihenfolge

    // auf der Leitung und die Konsistenz von _sent_packets strikt erhalten

    // bleiben müssen. Ein dabei eingeschlichener Race würde sich auf diesem

    // Loopback-Testsystem (nahezu keine Latenz, kaum reales Queuing) mit hoher

    // Wahrscheinlichkeit NICHT zeigen, aber genau in dem Produktions-Szenario

    // mit echter Latenz/Jitter auftreten können, das dieser Fix eigentlich

    // verbessern soll (Paket-Reordering, korrumpierte Retransmission-Buchführung).

    // Aufwand und Risiko sind laut Report selbst "hoch"/"hoch" — ohne

    // Mehrkern-/Variable-Bandbreite-Validierung außerhalb dieser Sandbox ist

    // das Risiko eines stillen, schwer reproduzierbaren Bugs im kritischen

    // Send-Pfad (den F31 gerade erst durch 3x volle Testsuite + 3x

    // quic_concurrent_test als korrekt verifiziert hat) höher als der Gewinn.

    // Entscheidung: nicht implementiert, transparent dokumentiert statt

    // riskant erzwungen.

    std::unique_lock<std::recursive_mutex> lock(quic_mtx());



    if (_conn_state.load() != ConnectionState::Connected && !_handshake_complete) {
@@ -6899,6 +6938,8 @@ void quic::pumpIncomingLocked() { 
                t->_pending_max_data = false;

            }

            t->pmtudTick();

            // F20: flush what sendMaxStreamData/sendMaxData just queued.

            t->flushBatch();

        }



        // Dispatch stream callbacks for routed (sibling) connections too —
@@ -6952,6 +6993,8 @@ void quic::flushPendingFlowControlAndCheckLoss() { 
        }



        pmtudTick();

        // F20: flush what sendMaxStreamData/sendMaxData just queued.

        flushBatch();

    }



    // After processing incoming packets (which may contain ACKs),
@@ -6994,6 +7037,8 @@ void quic::flush_out() { 
            sendMaxData(_pending_max_data_value);

            _pending_max_data = false;

        }

        // F20: flush what sendMaxStreamData/sendMaxData just queued.

        flushBatch();

    }

}

src/socket.h

+16 −1
Original line number Diff line number Diff line
@@ -1330,7 +1330,22 @@ namespace netplus { 
		// sender (ACK, MAX_DATA, MAX_STREAM_DATA, DATA_BLOCKED,

		// STREAM_DATA_BLOCKED) — mirrors the allocation-free pattern

		// sendStreamData(uint8_t*, ...) already uses for bulk data.

		ssize_t sendShortHeaderControlPacket(const std::vector<uint8_t>& frame);

		//

		// `batch` (F20): when true, queues the packet via batchPacket()

		// instead of sending it immediately via sendPacket(). Only pass

		// true from call sites that are guaranteed to call flushBatch()

		// themselves before returning (currently: sendMaxStreamData() and

		// sendMaxData(), always invoked from one of the repeated "flush

		// pending flow control" blocks that already end in a flushBatch()

		// call — see their call sites). DATA_BLOCKED/STREAM_DATA_BLOCKED/

		// STOP_SENDING stay unbatched (the default): they fire from

		// scattered, not-uniformly-flush-bounded call sites, and batching

		// them without that guarantee risks a frame sitting queued

		// indefinitely — exactly the class of "peer never gets the

		// update, connection stalls" bug this project has repeatedly hit

		// elsewhere (see memory: dispatch staircase, mediadb1/authdb

		// unresponsive incidents).

		ssize_t sendShortHeaderControlPacket(const std::vector<uint8_t>& frame, bool batch = false);

		std::vector<uint8_t> _ctrl_header_scratch;

		std::vector<uint8_t> _ctrl_pkt_scratch;