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

test

parent 18f83a5d
Loading
Loading
Loading
Loading
+236 −7
Changes for src/http.cpp: 236 added lines, 7 removed lines.
Original line number Diff line number Diff line
@@ -2485,7 +2485,8 @@ static std::string h2cBuildWindowUpdate(uint32_t stream_id, uint32_t increment)
const std::vector<char> libhttppp::HttpClient::_h2Request(
    const std::string &method,
    HttpRequest &nreq,
    const std::vector<char> *postBody)
    const std::vector<char> *postBody,
    const BodySource *bodySource)
{
    std::vector<char> ret;

@@ -2588,7 +2589,8 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(

        uint32_t stream_id = _h2NextStreamId;
        _h2NextStreamId += 2;  // Client streams use odd IDs: 1, 3, 5, ...
        bool end_stream = (postBody == nullptr || postBody->empty());
        const bool hasBufferedBody = postBody && !postBody->empty();
        bool end_stream = !hasBufferedBody && !bodySource;
        uint8_t hdr_flags = H2C_FLAG_END_HEADERS;
        if (end_stream) hdr_flags |= H2C_FLAG_END_STREAM;

@@ -2601,7 +2603,7 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
        std::vector<uint8_t> raw;
        raw.reserve(16384);

        if (postBody && !postBody->empty()) {
        if (hasBufferedBody || bodySource) {
            // Track the peer's (server's) flow control window.
            // Initial value is 65535 until we receive the server's SETTINGS.
            int32_t peer_conn_window = 65535;
@@ -2676,10 +2678,9 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
                }
            }

            size_t offset = 0;
            while (offset < postBody->size()) {
            // Blocks until the peer has granted some window, returns how much.
            auto wait_window = [&]() -> int32_t {
                int32_t allowed = std::min(peer_conn_window, peer_stream_window);

                // If window exhausted, block-read for WINDOW_UPDATE
                while (allowed <= 0) {
                    size_t n = recv_blocking(_recvScratch);
@@ -2694,7 +2695,36 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
                    process_fc_frames();
                    allowed = std::min(peer_conn_window, peer_stream_window);
                }
                return allowed;
            };

            if (bodySource) {
                std::vector<char> chunkBuf(H2C_MAX_FRAME_SIZE);
                for (;;) {
                    size_t have = (*bodySource)(chunkBuf.data(), chunkBuf.size());
                    if (have == 0) {
                        // The source can only say a chunk was its last by returning 0
                        // afterward, so END_STREAM goes on an empty DATA frame (which
                        // needs no flow-control window).
                        send_all(h2cBuildFrame(H2C_FRAME_DATA, H2C_FLAG_END_STREAM,
                                               stream_id, std::string()));
                        break;
                    }
                    size_t hoff = 0;
                    while (hoff < have) {
                        size_t chunk = std::min(have - hoff,
                                                static_cast<size_t>(wait_window()));
                        send_all(h2cBuildFrame(H2C_FRAME_DATA, 0, stream_id,
                                               std::string(chunkBuf.data() + hoff, chunk)));
                        peer_conn_window -= static_cast<int32_t>(chunk);
                        peer_stream_window -= static_cast<int32_t>(chunk);
                        hoff += chunk;
                    }
                }
            } else {
                size_t offset = 0;
                while (offset < postBody->size()) {
                    int32_t allowed = wait_window();
                    size_t chunk = std::min({postBody->size() - offset,
                                             static_cast<size_t>(allowed),
                                             H2C_MAX_FRAME_SIZE});
@@ -2707,6 +2737,7 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
                    offset += chunk;
                }
            }
        }

        // 4) Read response frames
        bool got_end_stream = false;
@@ -2899,7 +2930,8 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
  const std::vector<char> libhttppp::HttpClient::_h3Request(
    const std::string &method,
    HttpRequest &nreq,
    const std::vector<char> *postBody)
    const std::vector<char> *postBody,
    const BodySource *bodySource)
  {
    std::vector<char> ret;

@@ -2953,7 +2985,24 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
      }

      uint64_t stream_id = q->openStream(true);
      if (!bodySource) {
        q->sendStreamData(stream_id, stream_payload, true);
      } else {
        // One chunk of lookahead: QUIC can't carry FIN on an empty write (sendStreamData's
        // vector overload sends nothing for empty data), so the last DATA frame has to
        // carry it -- and a chunk is only known to be last once the next read returns 0.
        std::vector<uint8_t> cur(64 * 1024), next(64 * 1024);
        size_t curLen = (*bodySource)(reinterpret_cast<char *>(cur.data()), cur.size());
        q->sendStreamData(stream_id, stream_payload, curLen == 0);
        while (curLen > 0) {
          size_t nextLen = (*bodySource)(reinterpret_cast<char *>(next.data()), next.size());
          std::vector<uint8_t> frame;
          h3AppendFrame(frame, 0x00, std::vector<uint8_t>(cur.begin(), cur.begin() + curLen));
          q->sendStreamData(stream_id, frame, nextLen == 0);
          cur.swap(next);
          curLen = nextLen;
        }
      }

      std::vector<uint8_t> raw;
      std::vector<char> body;
@@ -3306,6 +3355,96 @@ const std::vector<char> libhttppp::HttpClient::Request(int requestType, libhttpp
    return _doH1Request(method, requestType, nreq, body, maxTries, /*followRedirects=*/false);
}

const std::vector<char> libhttppp::HttpClient::RequestStream(int requestType,
                                                             libhttppp::HttpRequest &nreq,
                                                             const BodySource &source,
                                                             int64_t contentLength)
{
    const char *method = RequestMethodName(requestType);
    if (!*method) {
        HTTPException excep;
        excep[HTTPException::Error] << "HttpClient::RequestStream: unknown request type " << requestType;
        throw excep;
    }

    try {
        _ensureConnected();

        if (!nreq.getHeaderData("host"))
            nreq.setHeaderData("host")->push_back(
                _url.getHost() + ":" + std::to_string(_url.getPort()));
        nreq.setRequestType(requestType);
        if (nreq.getRequestURL().empty())
            nreq.setRequestURL(_url.getPath());

        // Framing is ours to decide, whatever the caller's headers claimed.
        nreq.deldata("content-length");
        nreq.deldata("transfer-encoding");
        if (contentLength >= 0)
            nreq.setHeaderData("content-length")->push_back(std::to_string(contentLength));

        if (_isH3)
            return _h3Request(method, nreq, nullptr, &source);
        if (_isH2)
            return _h2Request(method, nreq, nullptr, &source);

        // HTTP/1.x: Content-Length when known, chunked otherwise (RFC 9112 §7.1).
        const bool chunked = contentLength < 0;
        if (chunked)
            nreq.setHeaderData("transfer-encoding")->push_back("chunked");
        if (nreq.getRequestVersion().empty())
            nreq.setRequestVersion(HTTPVERSION(1.1));
        if (!_cltsock || _cltsock->fd() < 0)
            resetConnection();

        std::string header;
        nreq.printHeader(header);
        _sendAll(header);

        std::vector<char> buf(64 * 1024);
        uint64_t sent = 0;
        for (;;) {
            size_t n = source(buf.data(), buf.size());
            if (n == 0)
                break;
            sent += n;
            if (!chunked && sent > static_cast<uint64_t>(contentLength)) {
                HTTPException ee;
                ee[HTTPException::Error] << "RequestStream: body exceeds Content-Length "
                                         << contentLength;
                throw ee;
            }
            if (chunked) {
                char sizeLine[32];
                int sl = std::snprintf(sizeLine, sizeof(sizeLine), "%zx\r\n", n);
                _sendAll(sizeLine, static_cast<size_t>(sl));
                _sendAll(buf.data(), n);
                _sendAll("\r\n", 2);
            } else {
                _sendAll(buf.data(), n);
            }
        }
        if (chunked) {
            _sendAll("0\r\n\r\n", 5);
        } else if (sent != static_cast<uint64_t>(contentLength)) {
            HTTPException ee;
            ee[HTTPException::Error] << "RequestStream: body ended after " << sent
                                     << " of " << contentLength << " bytes";
            throw ee;
        }
        return _h1ReadResponse(method);
    } catch (netplus::NetException &e) {
        // Mid-body: the connection's framing state is unknown, it can't be reused.
        resetConnection();
        HTTPException ee;
        ee[HTTPException::Error] << e.what();
        throw ee;
    } catch (...) {
        resetConnection();
        throw;
    }
}

const std::vector<char> libhttppp::HttpClient::Get(libhttppp::HttpRequest &nreq, size_t maxTries)
{
    return _doH1Request("GET", GETREQUEST, nreq, nullptr, maxTries, /*followRedirects=*/true);
@@ -4806,6 +4945,96 @@ bool libhttppp::HttpRequest::isChunkedRequest(){
//                (followed by any pipelined bytes) and Content-Length is set
//            0 : incomplete, need more data (RecvData untouched)
//           -1 : malformed chunked framing
// How long a streaming-body read() may wait for the next bytes before giving up -- the
// same order as the H1 offload socket timeout, so a stalled uploader can't pin a worker.
static constexpr auto kRequestBodyIdleTimeout = std::chrono::seconds(60);

bool libhttppp::H2RequestBodyPipe::push(const char *data, size_t len){
  {
    std::lock_guard<std::mutex> lk(_mtx);
    if(_aborted)
      return false;
    if(_buf.size() - _off + len > _capacity)
      return false;
    if(_off > 0 && _off >= _buf.size() / 2){
      _buf.erase(0, _off);
      _off = 0;
    }
    _buf.append(data, len);
  }
  _cv.notify_all();
  return true;
}

void libhttppp::H2RequestBodyPipe::finish(){
  {
    std::lock_guard<std::mutex> lk(_mtx);
    _eof = true;
  }
  _cv.notify_all();
}

void libhttppp::H2RequestBodyPipe::abort(){
  {
    std::lock_guard<std::mutex> lk(_mtx);
    if(_eof)
      return; // whole body already arrived -- nothing left to cut short
    _aborted = true;
  }
  _cv.notify_all();
}

bool libhttppp::H2RequestBodyPipe::aborted() const{
  std::lock_guard<std::mutex> lk(_mtx);
  return _aborted;
}

bool libhttppp::H2RequestBodyPipe::finished() const{
  std::lock_guard<std::mutex> lk(_mtx);
  return _done;
}

size_t libhttppp::H2RequestBodyPipe::read(char *buf, size_t len){
  size_t n = 0;
  bool eof = false;
  {
    std::unique_lock<std::mutex> lk(_mtx);
    if(_done)
      return 0;
    if(!_cv.wait_for(lk, kRequestBodyIdleTimeout,
                     [&]{ return _aborted || _eof || _off < _buf.size(); })){
      _aborted = true;
      HTTPException e;
      e[HTTPException::Error] << "request body: timed out waiting for data";
      throw e;
    }
    if(_aborted){
      HTTPException e;
      e[HTTPException::Error] << "request body: stream reset before end of body";
      throw e;
    }
    n = std::min(len, _buf.size() - _off);
    std::memcpy(buf, _buf.data() + _off, n);
    _off += n;
    if(_off == _buf.size()){
      _buf.clear();
      _off = 0;
      if(_eof)
        _done = (n == 0);
    }
    eof = _eof;
  }
  if(n > 0 && !eof && onConsumed)
    onConsumed(n);
  return n;
}

size_t libhttppp::HttpRequest::readBody(char *buf, size_t len){
  if(!_bodyStream)
    return 0;
  return _bodyStream->read(buf, len);
}

int libhttppp::HttpRequest::decodeChunkedBody(){
  const char *d = RecvData.data();
  const size_t n = RecvData.size();
+91 −2
Changes for src/http.h: 91 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -35,6 +35,9 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <map>
#include <unordered_map>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <mutex>

#include <netplus/socket.h>
#include <netplus/connection.h>
@@ -139,6 +142,18 @@ namespace libhttppp {
      const std::vector<char> Request(int requestType, HttpRequest &nreq,
                                      const std::vector<char> *body = nullptr, size_t maxTries=0);

      // Streaming-upload variant of Request(): the body is pulled from `source` (returns
      // the number of bytes written into buf, 0 at end of body) and sent as it's produced,
      // so relaying a large upload never holds the whole body in memory. contentLength < 0
      // means unknown -- HTTP/1.1 then uses chunked transfer-encoding, H2/H3 frame it
      // natively. The response is returned fully buffered, same as Request(). Never retries
      // and never follows redirects, since the body can't be replayed; on any error the
      // connection is reset (it's mid-body) and the exception propagates.
      using BodySource = std::function<size_t(char *buf, size_t len)>;
      const std::vector<char> RequestStream(int requestType, HttpRequest &nreq,
                                            const BodySource &source,
                                            int64_t contentLength = -1);

      // Streaming API: send request, return parsed response headers only.
      // After this call, use readBodyChunk() to read body data incrementally.
      HttpResponse GetStream(HttpRequest &nreq);
@@ -261,12 +276,16 @@ namespace libhttppp {
      // HTTP/3 client helpers
      uint32_t _h2NextStreamId = 1;     // next client-initiated stream ID (odd)
      std::unique_ptr<hpack::Decoder> _h2Decoder; // persistent HPACK decoder for connection
      // bodySource (RequestStream()) replaces postBody when set -- the body is pulled and
      // framed incrementally instead of sent from one buffer.
      const std::vector<char> _h2Request(const std::string &method,
                                          HttpRequest &nreq,
                                          const std::vector<char> *postBody = nullptr);
                                          const std::vector<char> *postBody = nullptr,
                                          const BodySource *bodySource = nullptr);
      const std::vector<char> _h3Request(const std::string &method,
                  HttpRequest &nreq,
                  const std::vector<char> *postBody = nullptr);
                  const std::vector<char> *postBody = nullptr,
                  const BodySource *bodySource = nullptr);

      // Streaming state
      enum StreamMode { STREAM_NONE, STREAM_CONTENT_LENGTH, STREAM_CHUNKED, STREAM_EOF,
@@ -486,6 +505,56 @@ namespace libhttppp {
  };
  
 
  // Incremental request-body source, attached to a request when
  // HttpEvent::shouldStreamRequestBody() opted it in -- see HttpRequest::readBody().
  class RequestBodyStream {
  public:
    virtual ~RequestBodyStream() = default;
    // Blocks until body data is available. Returns the number of bytes written to buf,
    // 0 once the body is complete. Throws HTTPException if the body can't be completed
    // (peer reset/closed, malformed framing, idle timeout).
    virtual size_t read(char *buf, size_t len) = 0;
    // True once read() has returned 0 (the whole body was consumed).
    virtual bool finished() const = 0;
    // Wake any blocked read() and make it (and every later one) throw -- unless the whole
    // body has already been received, in which case it can still be read to the end.
    virtual void abort() = 0;
  };

  // HTTP/2 request body handed from the connection's event-loop thread (which receives the
  // DATA frames) to the offload worker running RequestEvent. Bounded: the dispatcher only
  // grants the peer stream-level flow-control credit for what read() has consumed (see
  // onConsumed), so a well-behaved peer can never have more than one window in flight.
  class H2RequestBodyPipe : public RequestBodyStream {
  public:
    explicit H2RequestBodyPipe(size_t capacity) : _capacity(capacity) {}

    // Producer side (event loop). False if the pipe was aborted or the peer overran
    // capacity (i.e. ignored flow control) -- the caller resets the stream.
    bool push(const char *data, size_t len);
    void finish();
    bool aborted() const;

    size_t read(char *buf, size_t len) override;
    bool finished() const override;
    void abort() override;

    // Called from read()'s thread, never under the pipe's own lock, with the number of
    // bytes just consumed -- unless the body is already complete, since a peer that has
    // sent END_STREAM needs no more credit.
    std::function<void(size_t)> onConsumed;

  private:
    mutable std::mutex      _mtx;
    std::condition_variable _cv;
    std::string             _buf;
    size_t                  _off = 0;
    size_t                  _capacity;
    bool                    _eof = false;
    bool                    _aborted = false;
    bool                    _done = false;
  };

  class HttpRequest : public HttpHeader, public netplus::con{
  public:
    HttpRequest();
@@ -522,6 +591,13 @@ namespace libhttppp {
    bool                      isChunkedRequest();
    int                       decodeChunkedBody();

    /* Streaming request body (server side). Set only when HttpEvent::shouldStreamRequestBody()
     * returned true for this request: RequestEvent then runs as soon as the headers arrive,
     * RecvData holds no body, and the body is pulled incrementally with readBody() (blocking,
     * 0 = end of body) instead -- so an upload of any size costs bounded memory. */
    bool                      hasStreamingBody() const { return static_cast<bool>(_bodyStream); }
    size_t                    readBody(char *buf, size_t len);

    /*mobilphone switch*/
    bool isMobile();

@@ -559,6 +635,7 @@ namespace libhttppp {

    int               _RequestType = PARSEREQUEST;
    size_t            _MaxUploadSize = DEFAULT_UPLOADSIZE;
    std::shared_ptr<RequestBodyStream> _bodyStream;

    // Cached strings derived from _firstHeaderData pseudo-headers.
    // Populated by parseH1/parseH2/parseH3, read by getters.
@@ -611,6 +688,18 @@ namespace libhttppp {
      bool headersComplete = false;       // true once END_HEADERS received
      bool endStreamOnHeaders = false;    // END_STREAM was on HEADERS frame
      bool streaming = false;             // body handled by onH2DataChunk callback
      // Body handed to an already-dispatched RequestEvent (shouldStreamRequestBody()).
      std::shared_ptr<H2RequestBodyPipe> bodyPipe;

      H2PendingIncoming() = default;
      H2PendingIncoming(H2PendingIncoming &&) = default;
      H2PendingIncoming &operator=(H2PendingIncoming &&) = default;
      H2PendingIncoming(const H2PendingIncoming &) = delete;
      H2PendingIncoming &operator=(const H2PendingIncoming &) = delete;
      // Every way a pending stream goes away before END_STREAM (RST_STREAM, connection
      // teardown) erases it -- aborting here unblocks the worker reading its body instead of
      // leaving it to wait out the pipe's idle timeout. A no-op after END_STREAM.
      ~H2PendingIncoming() { if (bodyPipe) bodyPipe->abort(); }
    };

    struct H2State {
+298 −1

File changed.

Preview size limit exceeded, changes collapsed.

+25 −0
Changes for src/httpd.h: 25 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -111,6 +111,18 @@ namespace libhttppp {
            return false;
        }

        // Return true to have a body-bearing request dispatched as soon as its headers have
        // arrived, with the body read incrementally inside RequestEvent via
        // HttpRequest::readBody() instead of buffered first -- memory per upload stays bounded
        // (H1: one socket read; H2: one stream flow-control window) whatever its size. Called
        // with the request's method/URL/headers already parsed (an H2 call sees a scratch
        // request object, not the one RequestEvent later gets). Only honored where the request
        // is also offloaded (shouldOffloadH1Dispatch/shouldOffloadH2Dispatch true, with a
        // pool), since reading the body blocks; H3 bodies are still buffered. Default false.
        virtual bool shouldStreamRequestBody(HttpRequest &req) const {
            return false;
        }

        virtual void RequestEvent(HttpRequest &curreq,const int tid,ULONG_PTR args);
        virtual void ResponseEvent(HttpRequest &curreq,const int tid,ULONG_PTR args);
        virtual void ConnectEvent(HttpRequest &curreq,const int tid,ULONG_PTR args);
@@ -196,11 +208,24 @@ namespace libhttppp {
        // 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.
        // bodyStream: a streaming body (see _startH2BodyStream) to attach to the per-stream
        // request instead of reqBody.
        bool _dispatchH2Stream(HttpRequest &cureq, std::string &out,
                               uint32_t sid,
                               const std::vector<hpack::HeaderField> &decoded,
                               const std::string &reqBody,
                               const int tid, ULONG_PTR args,
                               std::shared_ptr<H2RequestBodyPipe> bodyStream = nullptr);
        // Dispatches an H2 stream whose body is still to come right away (headers only), when
        // shouldStreamRequestBody() opts it in; DATA frames then feed pending.bodyPipe. Returns
        // false (nothing done) when the stream doesn't qualify.
        bool _startH2BodyStream(HttpRequest &cureq, std::string &out, uint32_t sid,
                                HttpRequest::H2PendingIncoming &pending,
                                const int tid, ULONG_PTR args);
        // H1 counterpart: dispatches a body-bearing request right after its headers, with an
        // incremental reader over the connection's socket attached. False when it doesn't
        // qualify (the caller then buffers the body as before).
        bool _startH1BodyStream(HttpRequest &cureq, const int tid, ULONG_PTR args);
        // Runs RequestEvent(cureq,...) for a fully-parsed HTTP/1.x request, offloading to
        // _h1DispatchPool when shouldOffloadH1Dispatch() opts in. consumeBodyBytes is how many
        // already-fully-buffered request body bytes to erase from RecvData once RequestEvent