Merge pull request ClickHouse#84193 from ClickHouse/revert-81595-partial

Revert 81595 partial

Author: clickhouse-gh[bot]

src/Common/ProfileEvents.cpp

@@ -977,24 +977,17 @@ The server successfully detected this situation and will download merged part fr
     M(DiskConnectionsErrors, "Number of cases when creation of a connection for disk is failed", ValueType::Number) \
     M(DiskConnectionsElapsedMicroseconds, "Total time spend on creating connections for disk", ValueType::Microseconds) \
     \
-    M(HTTPConnectionsCreated, "Number of created client HTTP connections", ValueType::Number) \
-    M(HTTPConnectionsReused, "Number of reused client HTTP connections", ValueType::Number) \
-    M(HTTPConnectionsReset, "Number of reset client HTTP connections", ValueType::Number) \
-    M(HTTPConnectionsPreserved, "Number of preserved client HTTP connections", ValueType::Number) \
-    M(HTTPConnectionsExpired, "Number of expired client HTTP connections", ValueType::Number) \
-    M(HTTPConnectionsErrors, "Number of cases when creation of a client HTTP connection failed", ValueType::Number) \
-    M(HTTPConnectionsElapsedMicroseconds, "Total time spend on creating client HTTP connections", ValueType::Microseconds) \
-    \
-    M(HTTPServerConnectionsCreated, "Number of created server HTTP connections", ValueType::Number) \
-    M(HTTPServerConnectionsReused, "Number of reused server HTTP connections", ValueType::Number) \
-    M(HTTPServerConnectionsPreserved, "Number of preserved server HTTP connections. Connection kept alive successfully", ValueType::Number) \
-    M(HTTPServerConnectionsExpired, "Number of expired server HTTP connections.", ValueType::Number) \
-    M(HTTPServerConnectionsClosed, "Number of closed server HTTP connections. Keep alive has not been negotiated", ValueType::Number) \
-    M(HTTPServerConnectionsReset, "Number of reset server HTTP connections. Server closes connection", ValueType::Number) \
-    \
-    M(AddressesDiscovered, "Total count of new addresses in DNS resolve results for HTTP connections", ValueType::Number) \
-    M(AddressesExpired, "Total count of expired addresses which is no longer presented in DNS resolve results for HTTP connections", ValueType::Number) \
-    M(AddressesMarkedAsFailed, "Total count of addresses which have been marked as faulty due to connection errors for HTTP connections", ValueType::Number) \
+    M(HTTPConnectionsCreated, "Number of created http connections", ValueType::Number) \
+    M(HTTPConnectionsReused, "Number of reused http connections", ValueType::Number) \
+    M(HTTPConnectionsReset, "Number of reset http connections", ValueType::Number) \
+    M(HTTPConnectionsPreserved, "Number of preserved http connections", ValueType::Number) \
+    M(HTTPConnectionsExpired, "Number of expired http connections", ValueType::Number) \
+    M(HTTPConnectionsErrors, "Number of cases when creation of a http connection failed", ValueType::Number) \
+    M(HTTPConnectionsElapsedMicroseconds, "Total time spend on creating http connections", ValueType::Microseconds) \
+    \
+    M(AddressesDiscovered, "Total count of new addresses in dns resolve results for http connections", ValueType::Number) \
+    M(AddressesExpired, "Total count of expired addresses which is no longer presented in dns resolve results for http connections", ValueType::Number) \
+    M(AddressesMarkedAsFailed, "Total count of addresses which has been marked as faulty due to connection errors for http connections", ValueType::Number) \
     \
     M(ReadWriteBufferFromHTTPRequestsSent, "Number of HTTP requests sent by ReadWriteBufferFromHTTP", ValueType::Number) \
     M(ReadWriteBufferFromHTTPBytes, "Total size of payload bytes received and sent by ReadWriteBufferFromHTTP. Doesn't include HTTP headers.", ValueType::Bytes) \

src/IO/HTTPChunkedReadBuffer.cpp

@@ -56,38 +56,30 @@ bool HTTPChunkedReadBuffer::nextImpl()
     if (!in)
         return false;
 
-    try
+    /// The footer of previous chunk.
+    if (count())
+        readChunkFooter();
+
+    size_t chunk_size = readChunkHeader();
+    if (0 == chunk_size)
+    {
+        readChunkFooter();
+        in.reset();  // prevent double-eof situation.
+        return false;
+    }
+
+    if (in->available() >= chunk_size)
     {
-        /// The footer of previous chunk.
-        if (count())
-            readChunkFooter();
-
-        size_t chunk_size = readChunkHeader();
-        if (0 == chunk_size)
-        {
-            readChunkFooter();
-            in.reset();  // prevent double-eof situation.
-            return false;
-        }
-
-        if (in->available() >= chunk_size)
-        {
-            /// Zero-copy read from input.
-            working_buffer = Buffer(in->position(), in->position() + chunk_size);
-            in->position() += chunk_size;
-        }
-        else
-        {
-            /// Chunk is not completely in buffer, copy it to scratch space.
-            memory.resize(chunk_size);
-            in->readStrict(memory.data(), chunk_size);
-            working_buffer = Buffer(memory.data(), memory.data() + chunk_size);
-        }
+        /// Zero-copy read from input.
+        working_buffer = Buffer(in->position(), in->position() + chunk_size);
+        in->position() += chunk_size;
     }
-    catch (...)
+    else
     {
-        in.reset();
-        throw;
+        /// Chunk is not completely in buffer, copy it to scratch space.
+        memory.resize(chunk_size);
+        in->readStrict(memory.data(), chunk_size);
+        working_buffer = Buffer(memory.data(), memory.data() + chunk_size);
     }
 
     /// NOTE: We postpone reading the footer to the next iteration, because it may not be completely in buffer,

src/Server/HTTP/HTTPServerConnection.cpp

@@ -2,21 +2,8 @@
 #include <Server/TCPServer.h>
 
 #include <Poco/Net/NetException.h>
-#include <Common/ProfileEvents.h>
 #include <Common/logger_useful.h>
 
-
-namespace ProfileEvents
-{
-    extern const Event HTTPServerConnectionsCreated;
-    extern const Event HTTPServerConnectionsReused;
-    extern const Event HTTPServerConnectionsPreserved;
-    extern const Event HTTPServerConnectionsExpired;
-    extern const Event HTTPServerConnectionsClosed;
-    extern const Event HTTPServerConnectionsReset;
-}
-
-
 namespace DB
 {
 
@@ -38,28 +25,8 @@ void HTTPServerConnection::run()
     std::string server = params->getSoftwareVersion();
     Poco::Net::HTTPServerSession session(socket(), params);
 
-    ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsCreated);
-
-    while (!stopped && tcp_server.isOpen() && session.connected())
+    while (!stopped && tcp_server.isOpen() && session.hasMoreRequests() && session.connected())
     {
-        const bool is_first_request = params->getMaxKeepAliveRequests() == session.getMaxKeepAliveRequests();
-
-        if (!session.hasMoreRequests())
-        {
-            if (is_first_request)
-                // it is strange to have a connection being opened but no request has been sent, account it as an error case
-                ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsReset);
-            else
-                ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsExpired);
-
-            return;
-        }
-        else
-        {
-            if (!is_first_request)
-                ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsReused);
-        }
-
         try
         {
             std::lock_guard lock(mutex);
@@ -102,38 +69,10 @@ void HTTPServerConnection::run()
                             response.sendContinue();
 
                         handler->handleRequest(request, response, write_event);
-
-                        bool keep_alive = false;
-                        if (!params->getKeepAlive() || !request.canKeepAlive())
-                        {
-                            /// Either server is not configured to keep connections alive or client did not ask it
-                            ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsClosed);
-                        }
-                        else if (session.getMaxKeepAliveRequests() == 0 || !session.canKeepAlive())
-                        {
-                            /// connection is expired by max request count
-                            ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsExpired);
-                        }
-                        else if (!response.getKeepAlive())
-                        {
-                            /// server decided to close connection
-                            /// usually it is related to the cases:
-                            /// - the request or response stream is not bounded or
-                            /// - not all data is read from them
-                            ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsReset);
-                        }
-                        else
-                        {
-                            ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsPreserved);
-                            keep_alive = true;
-                        }
-
-                        session.setKeepAlive(keep_alive);
+                        session.setKeepAlive(params->getKeepAlive() && response.getKeepAlive() && session.canKeepAlive());
                     }
                     else
-                    {
                         sendErrorResponse(session, Poco::Net::HTTPResponse::HTTP_NOT_IMPLEMENTED);
-                    }
                 }
                 catch (Poco::Exception &)
                 {
@@ -195,7 +134,6 @@ void HTTPServerConnection::sendErrorResponse(Poco::Net::HTTPServerSession & sess
     response.setKeepAlive(false);
     response.send();
     session.setKeepAlive(false);
-    ProfileEvents::increment(ProfileEvents::HTTPServerConnectionsReset);
 }
 
 }

src/Server/HTTP/HTTPServerRequest.cpp

@@ -56,15 +56,11 @@ HTTPServerRequest::HTTPServerRequest(HTTPContextPtr context, HTTPServerResponse
     /// and retry with exactly the same (incomplete) set of rows.
     /// That's why we have to check body size if it's provided.
     if (getChunkedTransferEncoding())
-    {
         stream = std::make_unique<HTTPChunkedReadBuffer>(std::move(in), HTTP_MAX_CHUNK_SIZE);
-        stream_is_bounded = true;
-    }
     else if (hasContentLength())
     {
         size_t content_length = getContentLength();
         stream = std::make_unique<LimitReadBuffer>(std::move(in), LimitReadBuffer::Settings{.read_no_less = content_length, .read_no_more = content_length, .expect_eof = true});
-        stream_is_bounded = true;
     }
     else if (getMethod() != HTTPRequest::HTTP_GET && getMethod() != HTTPRequest::HTTP_HEAD && getMethod() != HTTPRequest::HTTP_DELETE)
     {
@@ -74,11 +70,8 @@ HTTPServerRequest::HTTPServerRequest(HTTPContextPtr context, HTTPServerResponse
                 "and no chunked/multipart encoding, it may be impossible to distinguish graceful EOF from abnormal connection loss");
     }
     else
-    {
         /// We have to distinguish empty buffer and nullptr.
         stream = std::make_unique<EmptyReadBuffer>();
-        stream_is_bounded = true;
-    }
 }
 
 bool HTTPServerRequest::checkPeerConnected() const

src/Server/HTTP/HTTPServerRequest.h

@@ -47,14 +47,6 @@ class HTTPServerRequest : public HTTPRequest
     X509Certificate peerCertificate() const;
 #endif
 
-    bool canKeepAlive() const
-    {
-        if (stream && stream_is_bounded)
-            return stream->eof();
-
-        return false;
-    }
-
 private:
     /// Limits for basic sanity checks when reading a header
     enum Limits
@@ -73,7 +65,6 @@ class HTTPServerRequest : public HTTPRequest
     Poco::Net::SocketAddress client_address;
     Poco::Net::SocketAddress server_address;
 
-    bool stream_is_bounded = false;
     bool secure;
 
     void readRequest(ReadBuffer & in);

src/Server/HTTP/HTTPServerResponse.cpp

@@ -10,6 +10,7 @@
 #include <Poco/Net/HTTPHeaderStream.h>
 #include <Poco/Net/HTTPStream.h>
 #include <IO/NullWriteBuffer.h>
+#include <sstream>
 
 
 namespace DB
@@ -39,6 +40,9 @@ std::shared_ptr<WriteBuffer> HTTPServerResponse::send()
         // the connection would be poisoned.
         // Next request over that connection reads previously unreaded message as a HTTP status line
 
+        // Send header
+        Poco::Net::HTTPHeaderOutputStream hs(session);
+        write(hs);
         // make sure that nothing is sent to the client if it was HTTP_HEAD request
         stream = std::make_shared<NullWriteBuffer>(write_event);
 
@@ -49,36 +53,75 @@ std::shared_ptr<WriteBuffer> HTTPServerResponse::send()
         // but if we do, then it is safer to close the connection at the end
         setKeepAlive(false);
 
+        // Send header
+        Poco::Net::HTTPHeaderOutputStream hs(session);
+        write(hs);
         stream = std::make_shared<AutoFinalizedWriteBuffer<WriteBufferFromPocoSocket>>(session.socket(), write_event);
     }
     else if (getChunkedTransferEncoding())
     {
+        // Send header
+        Poco::Net::HTTPHeaderOutputStream hs(session);
+        write(hs);
         stream = std::make_shared<AutoFinalizedWriteBuffer<HTTPWriteBufferChunked>>(session.socket(), write_event);
     }
     else if (hasContentLength())
     {
+        // Send header
+        Poco::Net::HTTPHeaderOutputStream hs(session);
+        write(hs);
         stream = std::make_shared<AutoFinalizedWriteBuffer<HTTPWriteBufferFixedLength>>(session.socket(), getContentLength(), write_event);
     }
     else
     {
         setKeepAlive(false);
-
+        // Send header
+        Poco::Net::HTTPHeaderOutputStream hs(session);
+        write(hs);
         stream = std::make_shared<AutoFinalizedWriteBuffer<WriteBufferFromPocoSocket>>(session.socket(), write_event);
     }
 
-    Poco::Net::HTTPHeaderOutputStream hs(session);
-    beginWrite(hs);
-    hs << "\r\n";
-    hs.flush();
-
+    send_started = true;
     return stream;
 }
 
-/// Only this method is called inside WriteBufferFromHTTPServerResponse
-void HTTPServerResponse::beginWrite(std::ostream & ostr)
+std::pair<std::shared_ptr<WriteBuffer>, std::shared_ptr<WriteBuffer>> HTTPServerResponse::beginSend()
 {
-    allowKeepAliveIFFRequestIsFullyRead();
+    poco_assert(!stream);
+    poco_assert(!header_stream);
+
+    /// NOTE: Code is not exception safe.
 
+    if ((request && request->getMethod() == HTTPRequest::HTTP_HEAD) || getStatus() < 200 || getStatus() == HTTPResponse::HTTP_NO_CONTENT
+        || getStatus() == HTTPResponse::HTTP_NOT_MODIFIED)
+    {
+        throw Poco::Exception("HTTPServerResponse::beginSend is invalid for HEAD request");
+    }
+
+    if (hasContentLength())
+    {
+        throw Poco::Exception("HTTPServerResponse::beginSend is invalid for response with Content-Length header");
+    }
+
+    // Write header to buffer
+    std::stringstream header; //STYLE_CHECK_ALLOW_STD_STRING_STREAM
+    beginWrite(header);
+    // Send header
+    auto str = header.str();
+    header_stream = std::make_shared<AutoFinalizedWriteBuffer<WriteBufferFromPocoSocket>>(session.socket(), write_event, str.size());
+    header_stream->write(str.data(), str.size());
+
+    if (getChunkedTransferEncoding())
+        stream = std::make_shared<AutoFinalizedWriteBuffer<HTTPWriteBufferChunked>>(session.socket(), write_event);
+    else
+        stream = std::make_shared<AutoFinalizedWriteBuffer<WriteBufferFromPocoSocket>>(session.socket(), write_event);
+
+    send_started = true;
+    return std::make_pair(header_stream, stream);
+}
+
+void HTTPServerResponse::beginWrite(std::ostream & ostr) const
+{
     HTTPResponse::beginWrite(ostr);
     send_started = true;
 }
@@ -87,11 +130,9 @@ void HTTPServerResponse::sendBuffer(const void * buffer, std::size_t length)
 {
     setContentLength(static_cast<int>(length));
     setChunkedTransferEncoding(false);
-
     // Send header
     Poco::Net::HTTPHeaderOutputStream hs(session);
-    beginWrite(hs);
-    hs << "\r\n";
+    write(hs);
     hs.flush();
 
     if (request && request->getMethod() != HTTPRequest::HTTP_HEAD)
@@ -125,17 +166,8 @@ void HTTPServerResponse::redirect(const std::string & uri, HTTPStatus status)
 
     // Send header
     Poco::Net::HTTPHeaderOutputStream hs(session);
-    beginWrite(hs);
-    hs << "\r\n";
+    write(hs);
     hs.flush();
 }
 
-void HTTPServerResponse::allowKeepAliveIFFRequestIsFullyRead()
-{
-    /// Connection can only be reused if we've fully read the previous request and all its POST data.
-    /// Otherwise we'd misinterpret the leftover data as part of the next request's header.
-    /// HTTPServerRequest::canKeepAlive() checks that request stream is bounded and is fully read.
-    if (!request || !request->canKeepAlive())
-        setKeepAlive(false);
-}
 }