Let the result readers read the response body directly

HttpResponse was a std::istream only to keep the reader interfaces from
the Poco days, and AmortizedIStreamReader was the only consumer, reading
through rdbuf() into a buffer of its own. Both ends are driver code now,
so HttpResponse provides the three operations the format parsers need,
eof(), get() and read(), with inline fast paths over the buffered data,
and the parsers take an HttpResponse reference.

This removes a copy of every byte of the body and AmortizedIStreamReader,
as well as the std::ios specifics this needed: the exception mask to get
errors out of std::istream, the ambiguous traits_type of the multiple
stream bases, and the construction order of the virtual base. Errors are
thrown from eof(), get() and read(), and again from every later call.

Author: slabko

driver/CMakeLists.txt

@@ -78,7 +78,6 @@ add_library (${libname}-impl STATIC
 
     utils/utils.h
     utils/iostream_debug_helpers.h
-    utils/amortized_istream_reader.h
     utils/object_pool.h
     utils/string_pool.h
     utils/unicode_converter.h

driver/format/ODBCDriver2.cpp

@@ -2,7 +2,7 @@
 
 ODBCDriver2ResultSet::ODBCDriver2ResultSet(
     const std::string & timezone,
-    AmortizedIStreamReader & stream,
+    HttpResponse & stream,
     std::unique_ptr<ResultMutator> && mutator
 )
     : ResultSet(stream, std::move(mutator))
@@ -263,10 +263,10 @@ void ODBCDriver2ResultSet::readValue(std::string & src, DataSourceType<DataSourc
 
 ODBCDriver2ResultReader::ODBCDriver2ResultReader(
     const std::string & timezone_,
-    std::istream & raw_stream,
+    HttpResponse & stream_,
     std::unique_ptr<ResultMutator> && mutator
 )
-    : ResultReader(timezone_, raw_stream, std::move(mutator))
+    : ResultReader(timezone_, stream_, std::move(mutator))
 {
     if (stream.eof())
         return;

driver/format/ODBCDriver2.h

@@ -10,7 +10,7 @@ class ODBCDriver2ResultSet
 public:
     explicit ODBCDriver2ResultSet(
         const std::string & timezone,
-        AmortizedIStreamReader & stream,
+        HttpResponse & stream,
         std::unique_ptr<ResultMutator> && mutator
     );
     virtual ~ODBCDriver2ResultSet() override = default;
@@ -69,7 +69,7 @@ class ODBCDriver2ResultReader
 public:
     explicit ODBCDriver2ResultReader(
         const std::string & timezone_,
-        std::istream & raw_stream,
+        HttpResponse & stream_,
         std::unique_ptr<ResultMutator> && mutator
     );
     virtual ~ODBCDriver2ResultReader() override = default;

driver/format/RowBinaryWithNamesAndTypes.cpp

@@ -2,7 +2,7 @@
 
 #include <ctime>
 
-RowBinaryWithNamesAndTypesResultSet::RowBinaryWithNamesAndTypesResultSet(const std::string & timezone, AmortizedIStreamReader & stream, std::unique_ptr<ResultMutator> && mutator)
+RowBinaryWithNamesAndTypesResultSet::RowBinaryWithNamesAndTypesResultSet(const std::string & timezone, HttpResponse & stream, std::unique_ptr<ResultMutator> && mutator)
     : ResultSet(stream, std::move(mutator))
 {
     std::uint64_t num_columns = 0;
@@ -306,10 +306,10 @@ void RowBinaryWithNamesAndTypesResultSet::readValue(DataSourceType<DataSourceTyp
 
 RowBinaryWithNamesAndTypesResultReader::RowBinaryWithNamesAndTypesResultReader(
     const std::string & timezone_,
-    std::istream & raw_stream,
+    HttpResponse & stream_,
     std::unique_ptr<ResultMutator> && mutator
 )
-    : ResultReader(timezone_, raw_stream, std::move(mutator))
+    : ResultReader(timezone_, stream_, std::move(mutator))
 {
     if (stream.eof())
         return;

driver/format/RowBinaryWithNamesAndTypes.h

@@ -8,7 +8,7 @@ class RowBinaryWithNamesAndTypesResultSet
     : public ResultSet
 {
 public:
-    explicit RowBinaryWithNamesAndTypesResultSet(const std::string & timezone, AmortizedIStreamReader & stream, std::unique_ptr<ResultMutator> && mutator);
+    explicit RowBinaryWithNamesAndTypesResultSet(const std::string & timezone, HttpResponse & stream, std::unique_ptr<ResultMutator> && mutator);
     virtual ~RowBinaryWithNamesAndTypesResultSet() override = default;
 
 protected:
@@ -77,7 +77,7 @@ class RowBinaryWithNamesAndTypesResultReader
 public:
     explicit RowBinaryWithNamesAndTypesResultReader(
         const std::string & timezone,
-        std::istream & raw_stream,
+        HttpResponse & stream_,
         std::unique_ptr<ResultMutator> && mutator
     );
     virtual ~RowBinaryWithNamesAndTypesResultReader() override = default;

driver/http_client.cpp

@@ -4,6 +4,7 @@
 #include <curl/curl.h>
 #include <zstd.h>
 
+#include <algorithm>
 #include <cctype>
 #include <charconv>
 #include <cstdio>
@@ -274,13 +275,10 @@ struct HttpResponse::Zstd {
 };
 
 HttpResponse::HttpResponse(std::shared_ptr<HttpSession> session_, HttpRequest request_)
-    : std::istream(static_cast<std::streambuf *>(this))
-    , session(std::move(session_))
+    : session(std::move(session_))
     , request(std::move(request_))
     , handles(std::make_unique<Handles>())
 {
-    exceptions(std::ios::badbit); // failures thrown from underflow() reach the reader
-
     const auto & settings = session->settings;
 
     handles->multi = curl_multi_init();
@@ -391,12 +389,10 @@ std::string HttpResponse::getRedirectUrl() const {
 std::string HttpResponse::readAll(std::size_t limit) {
     std::string res;
     try {
-        char buffer[4096];
-        while (res.size() < limit) {
-            const auto n = sgetn(buffer, sizeof(buffer));
-            if (n <= 0)
-                break;
-            res.append(buffer, static_cast<std::size_t>(n));
+        while (res.size() < limit && (pos < end || fill())) {
+            const auto n = std::min(static_cast<std::size_t>(end - pos), limit - res.size());
+            res.append(pos, n);
+            pos += n;
         }
     }
     catch (...) {
@@ -501,11 +497,24 @@ void HttpResponse::pump() {
         curl_multi_poll(handles->multi, nullptr, 0, 200, nullptr);
 }
 
-HttpResponse::int_type HttpResponse::underflow() {
-    if (gptr() < egptr())
-        return traits_type::to_int_type(*gptr());
+void HttpResponse::readSlow(char * dest, std::size_t count) {
+    const auto requested = count;
+    while (count > 0) {
+        if (pos == end && !fill())
+            throw std::runtime_error("Incomplete input stream, expected at least " + std::to_string(requested) + " more bytes");
+        const auto n = std::min(static_cast<std::size_t>(end - pos), count);
+        std::memcpy(dest, pos, n);
+        dest += n;
+        pos += n;
+        count -= n;
+    }
+}
 
-    setg(nullptr, nullptr, nullptr);
+bool HttpResponse::fill() {
+    if (pos < end)
+        return true;
+
+    pos = end = nullptr;
 
     if (!error && !unsupported_encoding.empty())
         error = std::make_exception_ptr(std::runtime_error("Unsupported Content-Encoding: " + unsupported_encoding));
@@ -530,10 +539,10 @@ HttpResponse::int_type HttpResponse::underflow() {
         if (!done || result == CURLE_OK) {
             if (!zstd) {
                 if (allowed > 0) {
-                    char * begin = raw.data() + raw_pos;
-                    setg(begin, begin, begin + allowed);
+                    pos = raw.data() + raw_pos;
+                    end = pos + allowed;
                     raw_pos += allowed;
-                    return traits_type::to_int_type(*gptr());
+                    return true;
                 }
             }
             else if (allowed > 0 || done) {
@@ -549,8 +558,9 @@ HttpResponse::int_type HttpResponse::underflow() {
                 if (in.pos > 0 || out.pos > 0) // a call without progress returns a size hint, not 0
                     zstd_frame_complete = (rc == 0);
                 if (out.pos > 0) {
-                    setg(decoded.data(), decoded.data(), decoded.data() + out.pos);
-                    return traits_type::to_int_type(*gptr());
+                    pos = decoded.data();
+                    end = pos + out.pos;
+                    return true;
                 }
                 if (in.pos > 0)
                     continue; // input consumed without output yet
@@ -561,7 +571,7 @@ HttpResponse::int_type HttpResponse::underflow() {
             finish();
             if (error)
                 std::rethrow_exception(error);
-            return traits_type::eof();
+            return false;
         }
 
         pump();

driver/http_client.h

@@ -2,19 +2,18 @@
 
 #include <chrono>
 #include <cstdint>
+#include <cstring>
 #include <exception>
-#include <istream>
 #include <map>
 #include <memory>
 #include <stdexcept>
-#include <streambuf>
 #include <string>
 #include <vector>
 
 // HTTP(S) client of the driver, built on libcurl.
 //
 // HttpSession is created per ODBC connection; HttpResponse per request. The response body is
-// read through HttpResponse, which is a std::istream that pulls data from the network lazily.
+// read directly from HttpResponse (eof/get/read), which pulls data from the network lazily.
 //
 // ClickHouse may report an error after it has already started sending a successful response
 // (a "mid-stream exception"): the exception text is appended to the body and the connection is
@@ -71,18 +70,16 @@ struct HttpRequest {
 };
 
 /// A single POST request and its response. The constructor sends the request and blocks until
-/// the response headers are received, throwing HttpTransportError if that fails. The body is
-/// then read through the std::istream interface or directly through rdbuf(). Any failure while
-/// reading the body (transport error, broken compressed data, ClickHouse exception) is thrown from
-/// the read operation itself, and again from every following read. (The stream has std::ios::badbit
-/// set in exceptions(), so std::istream operations rethrow it too, but only once.)
-class HttpResponse
-    : private std::streambuf
-    , public std::istream
-{
+/// the response headers are received, throwing HttpTransportError if that fails. The body is then
+/// read with eof(), get() and read(). Any failure while reading the body (transport error, broken
+/// compressed data, ClickHouse exception) is thrown from these, and again from every later call.
+class HttpResponse {
 public:
     HttpResponse(std::shared_ptr<HttpSession> session, HttpRequest request);
-    ~HttpResponse() override;
+    ~HttpResponse();
+
+    HttpResponse(const HttpResponse &) = delete;
+    HttpResponse & operator=(const HttpResponse &) = delete;
 
     long getStatus() const { return status; }
     bool hasHeader(const std::string & name) const;
@@ -91,15 +88,37 @@ class HttpResponse
     /// The target of a redirect, resolved against the request URL. Empty if there is none.
     std::string getRedirectUrl() const;
 
+    /// True if the body has no more data. Waits for data, if necessary.
+    bool eof() {
+        return (pos == end && !fill());
+    }
+
+    /// Reads one byte of the body; throws if there is none.
+    char get() {
+        if (pos == end && !fill())
+            throw std::runtime_error("Incomplete input stream, expected at least 1 more byte");
+        return *pos++;
+    }
+
+    /// Reads exactly `count` bytes of the body; throws if there are fewer.
+    void read(char * dest, std::size_t count) {
+        if (static_cast<std::size_t>(end - pos) >= count) { // the common case
+            std::memcpy(dest, pos, count);
+            pos += count;
+            return;
+        }
+        readSlow(dest, count);
+    }
+
     /// Reads the rest of the body (at most `limit` bytes). Never throws because of a failed transfer.
     std::string readAll(std::size_t limit);
 
 private:
-    // std::streambuf, std::istream and std::ios all define these.
-    using traits_type = std::streambuf::traits_type;
-    using int_type = std::streambuf::int_type;
-
-    int_type underflow() override;
+    /// Makes the next part of the body available in [pos, end). Returns false at the end of the
+    /// body, throws if the body cannot be read. Must only be called when [pos, end) is empty: it
+    /// may move or overwrite the data that the window pointed to.
+    bool fill();
+    void readSlow(char * dest, std::size_t count);
 
     void pump();
     void finish();
@@ -126,6 +145,9 @@ class HttpResponse
     std::string failed_tail; // undelivered (decoded) body bytes, if the transfer failed
     std::string unsupported_encoding;
 
+    const char * pos = nullptr; // the part of the body that can be read now, without waiting
+    const char * end = nullptr;
+
     std::string raw;         // received body bytes, possibly compressed
     std::size_t raw_pos = 0; // first byte of `raw` not yet handed over to the decoder/reader
     std::size_t raw_total = 0;

driver/result_set.cpp

@@ -122,7 +122,7 @@ void ColumnInfo::updateTypeInfo() {
     }
 }
 
-ResultSet::ResultSet(AmortizedIStreamReader & str, std::unique_ptr<ResultMutator> && mutator)
+ResultSet::ResultSet(HttpResponse & str, std::unique_ptr<ResultMutator> && mutator)
     : stream(str)
     , result_mutator(std::move(mutator))
     , string_pool(1000000)
@@ -303,11 +303,11 @@ void ResultSet::retireRow(Row && row) {
 
 ResultReader::ResultReader(
     const std::string & timezone_,
-    std::istream & raw_stream,
+    HttpResponse & stream_,
     std::unique_ptr<ResultMutator> && mutator
 )
     : timezone(timezone_)
-    , stream(raw_stream)
+    , stream(stream_)
     , result_mutator(std::move(mutator))
 {
 }
@@ -330,17 +330,17 @@ std::unique_ptr<ResultMutator> ResultReader::releaseMutator() {
 std::unique_ptr<ResultReader> make_result_reader(
     const std::string & format,
     const std::string & timezone,
-    std::istream & raw_stream,
+    HttpResponse & stream,
     std::unique_ptr<ResultMutator> && mutator
 ) {
     if (format == "ODBCDriver2") {
-        return std::make_unique<ODBCDriver2ResultReader>(timezone, raw_stream, std::move(mutator));
+        return std::make_unique<ODBCDriver2ResultReader>(timezone, stream, std::move(mutator));
     }
     else if (format == "RowBinaryWithNamesAndTypes") {
         if (!isLittleEndian())
             throw std::runtime_error("'" + format + "' format is supported only on little-endian platforms");
 
-        return std::make_unique<RowBinaryWithNamesAndTypesResultReader>(timezone, raw_stream, std::move(mutator));
+        return std::make_unique<RowBinaryWithNamesAndTypesResultReader>(timezone, stream, std::move(mutator));
     }
 
     throw std::runtime_error("'" + format + "' format is not supported");

driver/result_set.h

@@ -1,9 +1,9 @@
 #pragma once
 
 #include "driver/platform/platform.h"
+#include "driver/http_client.h"
 #include "driver/utils/utils.h"
 #include "driver/utils/object_pool.h"
-#include "driver/utils/amortized_istream_reader.h"
 #include "driver/utils/type_parser.h"
 #include "driver/utils/type_info.h"
 
@@ -93,7 +93,7 @@ class ResultMutator {
 
 class ResultSet {
 public:
-    explicit ResultSet(AmortizedIStreamReader & str, std::unique_ptr<ResultMutator> && mutator);
+    explicit ResultSet(HttpResponse & str, std::unique_ptr<ResultMutator> && mutator);
 
     virtual ~ResultSet();
 
@@ -119,7 +119,7 @@ class ResultSet {
     virtual bool readNextRow(Row & row) = 0;
 
 protected:
-    AmortizedIStreamReader & stream;
+    HttpResponse & stream;
     std::unique_ptr<ResultMutator> result_mutator;
     DefaultConversionContext conversion_context;
     std::vector<ColumnInfo> columns_info;
@@ -137,7 +137,7 @@ class ResultReader {
 protected:
     explicit ResultReader(
         const std::string & timezone_,
-        std::istream & stream,
+        HttpResponse & stream,
         std::unique_ptr<ResultMutator> && mutator
     );
 
@@ -153,15 +153,15 @@ class ResultReader {
 
 protected:
     const std::string timezone;
-    AmortizedIStreamReader stream;
+    HttpResponse & stream;
     std::unique_ptr<ResultMutator> result_mutator;
     std::unique_ptr<ResultSet> result_set;
 };
 
 std::unique_ptr<ResultReader> make_result_reader(
     const std::string & format,
     const std::string & timezone,
-    std::istream & raw_stream,
+    HttpResponse & stream,
     std::unique_ptr<ResultMutator> && mutator
 );