fix(Net): reactor-based http server not response when requests with body (#5227)

* fix: reactor-based http server, send error when exp,  recv req body

* add POST interface in reactor server

* fix: exit stuck

* fix: reactor-based http server redefination

* fix: tsan about socket reactor testConcurrentHandlerRemoval

* address review, add unit tests

* initialize _stopped, and coding style

Author: sky92zwq

Net/include/Poco/Net/HTTPReactorServer.h

@@ -1,3 +1,6 @@
+#ifndef Net_HTTPReactorServer_INCLUDED
+#define Net_HTTPReactorServer_INCLUDED
+
 #include "Poco/Logger.h"
 #include "Poco/Net/HTTPRequestHandlerFactory.h"
 #include "Poco/Net/HTTPServerParams.h"
@@ -17,6 +20,7 @@ class Net_API HTTPReactorServer
 	~HTTPReactorServer();
 	void start();
 	void stop();
+	int port() const { return _tcpReactorServer.port(); }
 	void onMessage(const TcpReactorConnectionPtr& conn);
 	void onError(const Poco::Exception& ex);
 	void sendErrorResponse(HTTPSession& session, HTTPResponse::HTTPStatus status);
@@ -31,3 +35,4 @@ class Net_API HTTPReactorServer
 
 }} // namespace Poco::Net
 
+#endif // Net_HTTPReactorServer_INCLUDED

Net/include/Poco/Net/HTTPReactorServerSession.h

@@ -43,6 +43,8 @@ class Net_API HTTPReactorServerSession : public HTTPSession
 
 	int peek() override;
 
+	int read(char* buffer, std::streamsize length) override;
+
 	int write(const char* buffer, std::streamsize length) override;
 
 	bool parseHeaders(std::size_t pos, std::size_t& bodyStart, std::size_t& contentLength, bool& isChunked);

Net/include/Poco/Net/TCPReactorAcceptor.h

@@ -22,6 +22,7 @@ class Net_API TCPReactorAcceptor : public Poco::Net::SocketAcceptor<TCPReactorSe
 
 	~TCPReactorAcceptor();
 	SocketReactor& reactor();
+	void stop();
 
 	void setRecvMessageCallback(const RecvMessageCallback& cb)
 	{
@@ -40,6 +41,7 @@ class Net_API TCPReactorAcceptor : public Poco::Net::SocketAcceptor<TCPReactorSe
 	std::shared_ptr<ThreadPool>                 _threadPool;
 	RecvMessageCallback                         _recvMessageCallback;
 	TCPServerParams::Ptr                        _pParams;
+	std::atomic<bool>                           _stopped{false};
 };
 
 }} // namespace Poco::Net

Net/include/Poco/Net/TCPReactorServer.h

@@ -6,6 +6,7 @@
 #include "Poco/Net/TCPReactorAcceptor.h"
 #include "Poco/Net/TCPServerParams.h"
 #include "Poco/ThreadPool.h"
+#include <atomic>
 #include <vector>
 
 namespace Poco { namespace Net {
@@ -32,6 +33,8 @@ class Net_API TCPReactorServer
 
 	void stop();
 
+	int port() const { return _port; }
+
 	void setRecvMessageCallback(const RecvMessageCallback& cb);
 
 private:
@@ -41,6 +44,7 @@ class Net_API TCPReactorServer
 	std::vector<ServerSocket>                        _sockets;
 	TCPServerParams::Ptr                             _pParams;
 	int                                              _port;
+	std::atomic<bool>                                _stopped;
 };
 
 }} // namespace Poco::Net

Net/samples/HTTPReactorTimeServer/src/HTTPReactorTimeServer.cpp

@@ -24,12 +24,14 @@
 #include "Poco/DateTimeFormat.h"
 #include "Poco/Exception.h"
 #include "Poco/ThreadPool.h"
+#include "Poco/StreamCopier.h"
 #include "Poco/Util/ServerApplication.h"
 #include "Poco/Util/Option.h"
 #include "Poco/Util/OptionSet.h"
 #include "Poco/Util/HelpFormatter.h"
 #include <cstdio>
 #include <iostream>
+#include <sstream>
 
 #include "Poco/Net/HTTPReactorServer.h"
 
@@ -89,6 +91,36 @@ class TimeRequestHandler: public HTTPRequestHandler
 };
 
 
+class BodyEchoRequestHandler: public HTTPRequestHandler
+	/// Echo the request body back to the client.
+{
+public:
+	BodyEchoRequestHandler() = default;
+
+	void handleRequest(HTTPServerRequest& request, HTTPServerResponse& response)
+	{
+		Application& app = Application::instance();
+		app.logger().information("Request from " + request.clientAddress().toString());
+
+		app.logger().information("Method: " + request.getMethod() + ", URI: " + request.getURI());
+
+		std::string contentType = request.get("Content-Type", "text/plain");
+		app.logger().information("Content-Type: " + contentType);
+
+		std::ostringstream ostrBody;
+		Poco::StreamCopier::copyStream(request.stream(), ostrBody);
+		std::string body = ostrBody.str();
+		app.logger().information("Request body length: " + std::to_string(body.length()));
+
+		response.setChunkedTransferEncoding(true);
+		response.setContentType(contentType);
+
+		std::ostream& ostr = response.send();
+		ostr << body;
+	}
+};
+
+
 class TimeRequestHandlerFactory: public HTTPRequestHandlerFactory
 {
 public:
@@ -100,6 +132,8 @@ class TimeRequestHandlerFactory: public HTTPRequestHandlerFactory
 	{
 		if (request.getURI() == "/")
 			return new TimeRequestHandler(_format, _delay);
+		else if (request.getURI() == "/echo" || request.getURI() == "/body")
+			return new BodyEchoRequestHandler();
 		else
 			return nullptr;
 	}

Net/src/HTTPReactorServer.cpp

@@ -4,17 +4,19 @@
 #include "Poco/Net/HTTPSession.h"
 #include <cstring>
 
-namespace Poco { namespace Net {
+namespace Poco {
+namespace Net {
 
 HTTPReactorServer::HTTPReactorServer(int port, HTTPServerParams::Ptr pParams, HTTPRequestHandlerFactory::Ptr pFactory)
 	: _tcpReactorServer(port, pParams)
 {
 	_pParams = pParams;
 	_pFactory = pFactory;
-	_tcpReactorServer.setRecvMessageCallback([this](const TcpReactorConnectionPtr& conn) {
-		// Handle incoming message
-		this->onMessage(conn);
-	});
+	_tcpReactorServer.setRecvMessageCallback(
+		[this](const TcpReactorConnectionPtr& conn)
+		{
+			this->onMessage(conn);
+		});
 }
 
 HTTPReactorServer::~HTTPReactorServer()
@@ -35,16 +37,15 @@ void HTTPReactorServer::onMessage(const TcpReactorConnectionPtr& conn)
 {
 	try
 	{
-		// Handle read event
 		HTTPReactorServerSession session(conn->socket(), conn->buffer(), _pParams);
 		if (!session.checkRequestComplete())
 		{
 			return;
 		}
-		// Create request and response objects
+
 		HTTPServerResponseImpl response(session);
-		HTTPServerRequestImpl  request(response, session, _pParams);
-		// Process request and generate response
+		HTTPServerRequestImpl request(response, session, _pParams);
+
 		Poco::Timestamp now;
 		response.setDate(now);
 		response.setVersion(request.getVersion());
@@ -54,6 +55,7 @@ void HTTPReactorServer::onMessage(const TcpReactorConnectionPtr& conn)
 		{
 			response.set("Server", server);
 		}
+
 		try
 		{
 			session.requestTrailer().clear();
@@ -66,19 +68,23 @@ void HTTPReactorServer::onMessage(const TcpReactorConnectionPtr& conn)
 
 				pHandler->handleRequest(request, response);
 				session.setKeepAlive(_pParams->getKeepAlive() && response.getKeepAlive());
+				session.popCompletedRequest();
 			}
 			else
+			{
 				sendErrorResponse(session, HTTPResponse::HTTP_NOT_IMPLEMENTED);
+				session.popCompletedRequest();
+			}
 		}
 		catch (Poco::Exception& e)
 		{
 			if (!response.sent())
 			{
 				try
 				{
-					sendErrorResponse(
-						session,
-						e.code() == 0 ? HTTPResponse::HTTP_INTERNAL_SERVER_ERROR : HTTPResponse::HTTPStatus(e.code()));
+					sendErrorResponse(session, e.code() == 0 ? HTTPResponse::HTTP_INTERNAL_SERVER_ERROR
+															 : HTTPResponse::HTTPStatus(e.code()));
+					session.popCompletedRequest();
 				}
 				catch (...)
 				{
@@ -99,13 +105,14 @@ void HTTPReactorServer::sendErrorResponse(HTTPSession& session, HTTPResponse::HT
 	response.setVersion(HTTPMessage::HTTP_1_1);
 	response.setStatusAndReason(status);
 	response.setKeepAlive(false);
+	response.setContentLength(0);
+	response.send();
 
 	session.setKeepAlive(false);
 }
 
 void HTTPReactorServer::onError(const Poco::Exception& ex)
 {
-	// Handle error
 	throw ex;
 }
 

Net/src/HTTPReactorServerSession.cpp

@@ -209,6 +209,19 @@ int HTTPReactorServerSession::peek()
 	}
 }
 
+int HTTPReactorServerSession::read(char* buffer, std::streamsize length)
+{
+	if (_idx < _complete)
+	{
+		int n = static_cast<int>(_complete - _idx);
+		if (n > static_cast<int>(length)) n = static_cast<int>(length);
+		std::memcpy(buffer, _buf.data() + _idx, n);
+		_idx += n;
+		return n;
+	}
+	return 0;
+}
+
 int HTTPReactorServerSession::write(const char* buffer, std::streamsize length)
 {
 	try

Net/src/TCPReactorAcceptor.cpp

@@ -6,8 +6,8 @@ namespace Poco {
 namespace Net {
 
 
-TCPReactorAcceptor::TCPReactorAcceptor(
-	Poco::Net::ServerSocket& socket, Poco::Net::SocketReactor& reactor, TCPServerParams::Ptr pParams)
+TCPReactorAcceptor::TCPReactorAcceptor(Poco::Net::ServerSocket& socket, Poco::Net::SocketReactor& reactor,
+									   TCPServerParams::Ptr pParams)
 	: Poco::Net::SocketAcceptor<TCPReactorServerConnection>(socket, reactor),
 	  _selfReactor(reactor),
 	  _useSelfReactor(pParams->getUseSelfReactor()),
@@ -16,18 +16,39 @@ TCPReactorAcceptor::TCPReactorAcceptor(
 	int workerThreads = _useSelfReactor ? 0 : _pParams->getMaxThreads();
 	if (workerThreads > 0)
 	{
-		_threadPool = std::make_shared<Poco::ThreadPool>("TCPRA", workerThreads, workerThreads);
+		_threadPool = std::make_shared<Poco::ThreadPool>("TCPRW", workerThreads, workerThreads);
+		for (int i = 0; i < workerThreads; i++)
+		{
+			std::shared_ptr<SocketReactor> workerReactor(std::make_shared<SocketReactor>());
+			_wokerReactors.push_back(workerReactor);
+			_threadPool->start(*workerReactor);
+		}
 	}
-	for (int i = 0; i < workerThreads; i++)
+	else
 	{
-		std::shared_ptr<SocketReactor> workerReactor(std::make_shared<SocketReactor>());
-		_wokerReactors.push_back(workerReactor);
-		_threadPool->start(*workerReactor);
+		_useSelfReactor = true;
 	}
 }
 
 TCPReactorAcceptor::~TCPReactorAcceptor()
 {
+	stop();
+}
+
+void TCPReactorAcceptor::stop()
+{
+	if (_stopped.exchange(true))
+	{
+		return;
+	}
+	for (auto& worker : _wokerReactors)
+	{
+		worker->stop();
+	}
+	if (_threadPool)
+	{
+		_threadPool->joinAll();
+	}
 }
 
 SocketReactor& TCPReactorAcceptor::reactor()

Net/src/TCPReactorServer.cpp

@@ -9,15 +9,20 @@ namespace Net {
 
 
 TCPReactorServer::TCPReactorServer(int port, TCPServerParams::Ptr pParams)
-	: _threadPool("TCPR", pParams->getAcceptorNum()),
+	: _threadPool("TCPRA", pParams->getAcceptorNum()),
 	  _reactors(pParams->getAcceptorNum()),
 	  _pParams(pParams),
-	  _port(port)
+	  _port(port),
+	  _stopped(false)
 {
 	for (auto& reactor : _reactors)
 	{
 		ServerSocket socket(_port);
 		_sockets.push_back(socket);
+		if (_sockets.size() == 1)
+		{
+			_port = socket.address().port();
+		}
 		auto acceptor = std::make_shared<TCPReactorAcceptor>(socket, reactor, _pParams);
 		_acceptors.push_back(acceptor);
 	}
@@ -46,6 +51,14 @@ void TCPReactorServer::setRecvMessageCallback(const RecvMessageCallback& cb)
 
 void TCPReactorServer::stop()
 {
+	if (_stopped.exchange(true))
+	{
+		return;
+	}
+	for (auto& acceptor : _acceptors)
+	{
+		acceptor->stop();
+	}
 	for (auto& reactor : _reactors)
 	{
 		reactor.stop();

Net/testsuite/Makefile

@@ -15,7 +15,7 @@ objects = \
 	HTTPRequestTest MessageHeaderTest NetTestSuite UDPEchoServer \
 	HTTPResponseTest MessagesTestSuite NetworkInterfaceTest \
 	HTTPServerTest MulticastEchoServer SocketAddressTest \
-	HTTPReactorServerSessionTest HTTPReactorServerTestSuite \
+	HTTPReactorServerSessionTest HTTPReactorServerTest HTTPReactorServerTestSuite \
 	HTTPCookieTest HTTPCredentialsTest HTMLFormTest HTMLTestSuite \
 	MediaTypeTest QuotedPrintableTest DialogSocketTest \
 	HTTPClientTestSuite FTPClientTestSuite FTPClientSessionTest \