test

void libhttppp::HttpEvent::DisconnectEvent(HttpRequest &/*curreq*/,const int /*tid*/,ULONG_PTR /*args*/){

}

void libhttppp::HttpEvent::_dispatchH2Stream(HttpRequest &cureq,

bool libhttppp::HttpEvent::_dispatchH2Stream(HttpRequest &cureq,

                                             std::string &out,

                                             uint32_t sid,

                                             const std::vector<hpack::HeaderField> &decoded,

                    if (cureq2.SendData.pos < cureq2.SendData.size())

                        netplus::requestWritablePoll(cureq2);

                });

            return;

            return true;

        }

        // Connection lookup failed (shouldn't happen while we're still

        // holding event_mutex for this call) — fall through to the

    // Dispatch to user's virtual RequestEvent

    RequestEvent(tempreq, tid, args);

    _finishH2Dispatch(cureq, out, sid, std::move(tempreq_ptr), tid, args);

    return false;

}

void libhttppp::HttpEvent::_finishH2Dispatch(HttpRequest &cureq,

    out.reserve(4096);  // Pre-allocate to avoid repeated reallocations

    size_t off = 0;

    int reprocess_count = 0;

    // Set by _dispatchH2Stream whenever it offloads a stream to the

    // background pool instead of finishing it synchronously. An offloaded

    // dispatch leaves `out` empty and `off` already reset to 0 (the frame

    // was still consumed from RecvData before the call) — without this

    // flag, the reprocess condition below would see "no progress" and stop

    // looking at any further buffered frames, permanently stranding every

    // other multiplexed stream's HEADERS frame in RecvData with nothing

    // left to ever re-trigger processing it.

    bool anyOffloaded = false;

    // Consume HTTP/2 client connection preface if present

    if (cureq.RecvData.size() >= H2_CLIENT_PREFACE_LEN &&

                    cureq.RecvData.erase(cureq.RecvData.begin(),

                                         cureq.RecvData.begin() + off);

                    off = 0;

                    _dispatchH2Stream(cureq, out, sid, decoded, "", tid, args);

                    anyOffloaded |= _dispatchH2Stream(cureq, out, sid, decoded, "", tid, args);

                    goto done;

                } else {

                    // Body expected via DATA frames (POST, PUT, etc.)

                        cureq.RecvData.erase(cureq.RecvData.begin(),

                                             cureq.RecvData.begin() + off);

                        off = 0;

                        _dispatchH2Stream(cureq, out, sid,

                        anyOffloaded |= _dispatchH2Stream(cureq, out, sid,

                                          it->second.headers, "", tid, args);

                        cureq.h2state().pendingIncoming.erase(it);

                        goto done;

                            cureq.RecvData.erase(cureq.RecvData.begin(),

                                                 cureq.RecvData.begin() + off);

                            off = 0;

                            _dispatchH2Stream(cureq, out, sid,

                            anyOffloaded |= _dispatchH2Stream(cureq, out, sid,

                                              it->second.headers,

                                              it->second.body,

                                              tid, args);

    // If frames remain in RecvData (e.g. HEADERS that arrived during

    // streaming), process them now instead of waiting for the next

    // EPOLLIN which may never fire.

    // Only reprocess if we actually made progress (consumed frames or

    // generated output). Otherwise we'd spin forever on an incomplete

    // frame that needs more network data to arrive.

    // Only reprocess if we actually made progress (consumed frames,

    // generated output, or offloaded a stream). Otherwise we'd spin

    // forever on an incomplete frame that needs more network data to

    // arrive. anyOffloaded matters here: an offloaded dispatch consumes

    // its frame from RecvData just like a synchronous one, but leaves

    // `out` empty and `off` already reset — without checking it, any

    // further multiplexed streams' HEADERS frames buffered in this same

    // read would never get looked at (nothing else will ever re-trigger

    // processing them, since the client is done sending and is only

    // waiting on responses).

    // Guard: limit reprocessing iterations to prevent infinite spin when

    // frames continuously generate output (e.g. ACKs) without consuming data.

    if (reprocess_count < 16 && (!out.empty() || off > 0) && cureq.RecvData.size() >= H2_FRAME_HEADER_LEN) {

    if (reprocess_count < 16 && (!out.empty() || off > 0 || anyOffloaded) && cureq.RecvData.size() >= H2_FRAME_HEADER_LEN) {

        ++reprocess_count;

        out.clear();

        out.reserve(4096);

        // constructor. See shouldOffloadH2Dispatch / _dispatchH2Stream.

        std::unique_ptr<netplus::ThreadPool>  _h2DispatchPool;

        void _dispatchH2Stream(HttpRequest &cureq, std::string &out,

        // Returns true if the stream was handed off to the background

        // offload pool instead of being finished synchronously — callers

        // must treat that as "progress" for reprocess-loop purposes even

        // though it leaves `out` unchanged, since the frame was still

        // consumed from RecvData either way.

        bool _dispatchH2Stream(HttpRequest &cureq, std::string &out,

                               uint32_t sid,

                               const std::vector<hpack::HeaderField> &decoded,

                               const std::string &reqBody,

add_test(connection_test connection_test)

add_executable(httpd_h2_concurrency_test

    httpd_h2_concurrency_test.cpp

)

if(WIN32)

    target_link_libraries (httpd_h2_concurrency_test httppp-static ws2_32)

else()

    target_link_libraries (httpd_h2_concurrency_test httppp-static)

endif()

add_test(httpd_h2_concurrency_test httpd_h2_concurrency_test)

add_executable(url_test

    url.cpp

)

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

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.

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

// Regression test for the HTTP/2 stream-dispatch staircase bug: multiple

// streams multiplexed on one connection must not serialize behind each

// other's RequestEvent. Runs the SAME synthetic slow server twice — once

// with h2OffloadThreads=0 (must reproduce the staircase, proving the

// synchronous fallback path is unchanged) and once with offload enabled

// (must show the streams complete with substantially overlapping wall

// clocks instead).

#include <algorithm>

#include <chrono>

#include <cstring>

#include <iostream>

#include <thread>

#include <vector>

#include <arpa/inet.h>

#include <netinet/in.h>

#include <sys/socket.h>

#include <unistd.h>

#include <netplus/socket.h>

#include "hpack.h"

#include "http.h"

#include "httpd.h"

namespace {

constexpr uint8_t FRAME_DATA = 0x00;

constexpr uint8_t FRAME_HEADERS = 0x01;

constexpr uint8_t FRAME_SETTINGS = 0x04;

constexpr uint8_t FLAG_END_STREAM = 0x01;

constexpr uint8_t FLAG_END_HEADERS = 0x04;

constexpr size_t FRAME_HEADER_LEN = 9;

constexpr char CLIENT_PREFACE[] = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";

void appendFrameHeader(std::vector<char> &out, uint32_t len, uint8_t type,

                        uint8_t flags, uint32_t streamId) {

    out.push_back(static_cast<char>((len >> 16) & 0xFF));

    out.push_back(static_cast<char>((len >> 8) & 0xFF));

    out.push_back(static_cast<char>(len & 0xFF));

    out.push_back(static_cast<char>(type));

    out.push_back(static_cast<char>(flags));

    out.push_back(static_cast<char>((streamId >> 24) & 0x7F));

    out.push_back(static_cast<char>((streamId >> 16) & 0xFF));

    out.push_back(static_cast<char>((streamId >> 8) & 0xFF));

    out.push_back(static_cast<char>(streamId & 0xFF));

}

// A slow synthetic backend stand-in: sleeps for a caller-supplied number of

// milliseconds (parsed out of the request path, e.g. "/slow/50") before

// responding, standing in for a real blocking mediadb round-trip.

class SlowServer : public libhttppp::HttpEvent {

public:

    SlowServer(std::vector<netplus::socket*> serversocket, size_t h2OffloadThreads)

        : HttpEvent(serversocket, 1000, h2OffloadThreads) {}

    bool shouldOffloadH2Dispatch(libhttppp::HttpRequest&, uint32_t) const override {

        // Gated only by whether the pool exists (h2OffloadThreads), so one

        // class serves both the synchronous and offloaded test runs.

        return true;

    }

    void RequestEvent(libhttppp::HttpRequest &req, const int, ULONG_PTR) override {

        const std::string &path = req.getRequestURL();

        int delayMs = 0;

        auto pos = path.rfind('/');

        if (pos != std::string::npos) {

            try { delayMs = std::stoi(path.substr(pos + 1)); } catch (...) {}

        }

        if (delayMs > 0)

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

        libhttppp::HttpResponse res;

        res.setContentType("text/plain");

        res.send(req, std::string("ok"), 2);

    }

};

// Builds one HEADERS(+END_HEADERS+END_STREAM) frame requesting

// "/slow/<delayMs>" on the given stream id.

std::vector<char> buildSlowRequest(uint32_t streamId, int delayMs) {

    std::string path = "/slow/" + std::to_string(delayMs);

    std::string hpack_out = libhttppp::hpack::Encoder::encodeRequestHeaders(

        "GET", path, "http", "127.0.0.1");

    std::vector<char> out;

    appendFrameHeader(out, static_cast<uint32_t>(hpack_out.size()), FRAME_HEADERS,

                       FLAG_END_HEADERS | FLAG_END_STREAM, streamId);

    out.insert(out.end(), hpack_out.begin(), hpack_out.end());

    return out;

}

// Connects, sends the H2 preface + an empty SETTINGS frame + one HEADERS

// frame per stream (all in one burst, before reading anything back — this

// is what makes it a genuine multiplexing test rather than N sequential

// requests), then reads frames until every stream has seen END_STREAM.

// Returns each stream's completion timestamp.

std::vector<std::chrono::steady_clock::time_point> runConcurrentRequest(

    int port, const std::vector<uint32_t> &streamIds, int delayMs) {

    int fd = ::socket(AF_INET, SOCK_STREAM, 0);

    sockaddr_in addr{};

    addr.sin_family = AF_INET;

    addr.sin_port = htons(static_cast<uint16_t>(port));

    inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr);

    if (::connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) < 0) {

        ::close(fd);

        throw std::runtime_error("connect failed");

    }

    std::vector<char> req(CLIENT_PREFACE, CLIENT_PREFACE + sizeof(CLIENT_PREFACE) - 1);

    appendFrameHeader(req, 0, FRAME_SETTINGS, 0, 0); // empty client SETTINGS

    for (uint32_t sid : streamIds) {

        auto hdrs = buildSlowRequest(sid, delayMs);

        req.insert(req.end(), hdrs.begin(), hdrs.end());

    }

    size_t sent = 0;

    while (sent < req.size()) {

        ssize_t n = ::send(fd, req.data() + sent, req.size() - sent, 0);

        if (n <= 0) { ::close(fd); throw std::runtime_error("send failed"); }

        sent += static_cast<size_t>(n);

    }

    std::map<uint32_t, std::chrono::steady_clock::time_point> completed;

    std::vector<char> buf;

    char tmp[4096];

    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);

    while (completed.size() < streamIds.size() &&

           std::chrono::steady_clock::now() < deadline) {

        ssize_t n = ::recv(fd, tmp, sizeof(tmp), 0);

        if (n <= 0) break;

        buf.insert(buf.end(), tmp, tmp + n);

        size_t off = 0;

        while (off + FRAME_HEADER_LEN <= buf.size()) {

            uint32_t len = (static_cast<uint8_t>(buf[off]) << 16) |

                           (static_cast<uint8_t>(buf[off + 1]) << 8) |

                           static_cast<uint8_t>(buf[off + 2]);

            uint8_t type = static_cast<uint8_t>(buf[off + 3]);

            uint8_t flags = static_cast<uint8_t>(buf[off + 4]);

            uint32_t sid = ((static_cast<uint8_t>(buf[off + 5]) & 0x7F) << 24) |

                           (static_cast<uint8_t>(buf[off + 6]) << 16) |

                           (static_cast<uint8_t>(buf[off + 7]) << 8) |

                           static_cast<uint8_t>(buf[off + 8]);

            if (off + FRAME_HEADER_LEN + len > buf.size())

                break; // incomplete frame, wait for more bytes

            if ((type == FRAME_HEADERS || type == FRAME_DATA) &&

                (flags & FLAG_END_STREAM) && sid != 0) {

                completed.emplace(sid, std::chrono::steady_clock::now());

            }

            off += FRAME_HEADER_LEN + len;

        }

        buf.erase(buf.begin(), buf.begin() + off);

    }

    ::close(fd);

    if (completed.size() != streamIds.size())

        throw std::runtime_error("not all streams completed in time");

    std::vector<std::chrono::steady_clock::time_point> times;

    for (uint32_t sid : streamIds) times.push_back(completed[sid]);

    return times;

}

// Binds an ephemeral-port TCP listener, starts a SlowServer on it in a

// background thread, and returns the bound port. The server thread is

// intentionally never joined — the test process exits via std::exit once

// all assertions are done, which is fine for a short-lived test binary and

// avoids netplus::event::Running (a process-wide static) making a second,

// sequential runEventloop() call impossible after the first is stopped.

int startServer(std::unique_ptr<netplus::tcp> &sockOwner, size_t h2OffloadThreads) {

    sockOwner = std::make_unique<netplus::tcp>("127.0.0.1", 0, 16, -1);

    sockOwner->bind();

    sockaddr_in bound{};

    socklen_t blen = sizeof(bound);

    ::getsockname(sockOwner->fd(), reinterpret_cast<sockaddr*>(&bound), &blen);

    int port = ntohs(bound.sin_port);

    sockOwner->listen();

    auto *server = new SlowServer({sockOwner.get()}, h2OffloadThreads);

    std::thread([server] { server->runEventloop(); }).detach();

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

    return port;

}

} // namespace

int main() {

    const std::vector<uint32_t> streamIds = {1, 3, 5, 7, 9};

    const int delayMs = 50;

    int failures = 0;

    // Run 1: h2OffloadThreads = 0 — must reproduce the staircase, proving

    // the untouched synchronous fallback path still behaves as before.

    {

        std::unique_ptr<netplus::tcp> sock;

        int port = startServer(sock, /*h2OffloadThreads=*/0);

        auto times = runConcurrentRequest(port, streamIds, delayMs);

        auto totalMs = std::chrono::duration_cast<std::chrono::milliseconds>(

            times.back() - times.front()).count();

        std::cout << "[sync]    spread=" << totalMs << "ms (expect >= ~"

                  << (streamIds.size() - 1) * delayMs / 2 << "ms, staircase)"

                  << std::endl;

        // Staircase signature: each stream finishes roughly delayMs after

        // the previous one, so the spread across all of them should be a

        // sizeable fraction of (N-1)*delayMs, not near zero.

        if (totalMs < static_cast<long long>((streamIds.size() - 1) * delayMs) / 2) {

            std::cerr << "FAIL: synchronous path did not show the expected "

                          "staircase (spread=" << totalMs << "ms) — either the "

                          "fallback path regressed, or this environment "

                          "schedules unrealistically fast" << std::endl;

            ++failures;

        }

    }

    // Run 2: h2OffloadThreads > 0 with shouldOffloadH2Dispatch() true —

    // must show the streams completing with overlapping wall clocks.

    {

        std::unique_ptr<netplus::tcp> sock;

        int port = startServer(sock, /*h2OffloadThreads=*/4);

        auto times = runConcurrentRequest(port, streamIds, delayMs);

        auto totalMs = std::chrono::duration_cast<std::chrono::milliseconds>(

            times.back() - times.front()).count();

        std::cout << "[offload] spread=" << totalMs << "ms (expect < ~"

                  << 3 * delayMs << "ms, overlap)" << std::endl;

        if (totalMs >= 3 * delayMs) {

            std::cerr << "FAIL: offloaded streams did not overlap "

                          "(spread=" << totalMs << "ms, expected < "

                       << 3 * delayMs << "ms)" << std::endl;

            ++failures;

        }

    }

    std::cout << (failures == 0 ? "PASS" : "FAIL") << std::endl;

    std::exit(failures == 0 ? 0 : 1);

}