Make Core Controller application socket connections non-blocking (#2593)

`ApplicationPool::Socket::checkoutConnection()` initiates blocking
connects, which could block the event loop if the application process's
socket backlog is full.

---------

Co-authored-by: Hongli Lai <hongli@phusion.nl>

Author: CamJN

CHANGELOG

@@ -4,6 +4,7 @@ Release 6.0.27 (Not yet released)
  * Fix compilation on FreeBSD.
  * [Ruby] Fix compatibility with Rack 2 While maintaining compatibility with Rack 3. Closes GH-2595.
  * [Ruby] Use non-deprecated functions in native extensions.
+ * Fix an issue where Passenger could freeze while connecting to application processes (event loop blocking).
  * Updated various library versions used in precompiled binaries (used for e.g. gem installs):
    - ccache: 4.10.2 -> 4.11.2
    - cmake: 3.31.3 -> 3.31.6
@@ -19,6 +20,7 @@ Release 6.0.27 (Not yet released)
      - 3.4.1 -> 3.4.2
    - rubygems: 3.6.2 -> 3.6.6
    - zstd: 1.5.6 -> 1.5.7
+ * 
 
 
 Release 6.0.26

build/support/cxx_dependency_map.rb

@@ -3022,6 +3022,7 @@
    "src/cxx_supportlib/DataStructures/HashedStaticString.h",
    "src/cxx_supportlib/Exceptions.h",
    "src/cxx_supportlib/FileDescriptor.h",
+   "src/cxx_supportlib/FileTools/FileManip.h",
    "src/cxx_supportlib/IOTools/BufferedIO.h",
    "src/cxx_supportlib/IOTools/IOUtils.h",
    "src/cxx_supportlib/LoggingKit/Assert.h",
@@ -3030,7 +3031,9 @@
    "src/cxx_supportlib/LoggingKit/LoggingKit.h",
    "src/cxx_supportlib/StaticString.h",
    "src/cxx_supportlib/StrIntTools/StrIntUtils.h",
+   "src/cxx_supportlib/Utils.h",
    "src/cxx_supportlib/Utils/FastStringStream.h",
+   "src/cxx_supportlib/Utils/ScopeGuard.h",
    "src/cxx_supportlib/oxt/backtrace.hpp",
    "src/cxx_supportlib/oxt/detail/backtrace_disabled.hpp",
    "src/cxx_supportlib/oxt/detail/backtrace_enabled.hpp",

packaging/debian

@@ -42,6 +42,7 @@ passenger_enabled on;
 <%= nginx_option(app, :app_type) %>
 <%= nginx_option(app, :startup_file) %>
 <%= nginx_option(app, :app_start_command) %>
+<%= nginx_option(app, :app_connect_timeout) %>
 <%= nginx_option(app, :start_timeout) %>
 <%= nginx_option(app, :min_instances) %>
 <%= nginx_option(app, :max_request_queue_size) %>

resources/templates/standalone/server.erb

@@ -52,10 +52,19 @@ class AbstractSession {
 	virtual StaticString getProtocol() const = 0;
 	virtual unsigned int getStickySessionId() const = 0;
 	virtual const ApiKey &getApiKey() const = 0;
+	/** The connection fd that was established by \c initiate(), or -1 if not initiated yet. Note that this fd is non-blocking. */
 	virtual int fd() const = 0;
 	virtual bool isClosed() const = 0;
 
-	virtual void initiate(bool blocking = true) = 0;
+	/**
+	 * Initiates a non-blocking connect to the application socket, or reuse an existing connection.
+	 *
+	 * @return Whether the non-blocking connect finished (true) or whether you need to wait for it (false).
+	 * @throws SystemException Something went wrong.
+	 * @throws RuntimeException Something went wrong.
+	 * @throws boost::thread_interrupted A system call has been interrupted.
+	 */
+	virtual bool initiate() = 0;
 
 	virtual void requestOOBW() { /* Do nothing */ }
 

src/agent/Core/ApplicationPool/AbstractSession.h

@@ -185,20 +185,15 @@ class Session: public AbstractSession {
 		return getSocket()->protocol;
 	}
 
-
-	virtual void initiate(bool blocking = true) override {
+	// See AbstractSession.h for docs
+	virtual bool initiate() override {
 		assert(!closed);
 		ScopeGuard g(boost::bind(&Session::callOnInitiateFailure, this));
 		Connection connection = socket->checkoutConnection();
 		connection.fail = true;
-		if (connection.blocking && !blocking) {
-			FdGuard g2(connection.fd, NULL, 0);
-			setNonBlocking(connection.fd);
-			g2.clear();
-			connection.blocking = false;
-		}
 		g.clear();
 		this->connection = connection;
+		return connection.ready;
 	}
 
 	bool initiated() const {

src/agent/Core/ApplicationPool/Session.h

@@ -51,13 +51,13 @@ struct Connection {
 	int fd;
 	bool wantKeepAlive: 1;
 	bool fail: 1;
-	bool blocking: 1;
+	bool ready: 1;
 
 	Connection()
 		: fd(-1),
 		  wantKeepAlive(false),
 		  fail(false),
-		  blocking(true)
+		  ready(false)
 		{ }
 
 	void close() {
@@ -85,13 +85,15 @@ class Socket {
 		return concurrency;
 	}
 
+	/** Initiates a non-blocking connect. */
 	Connection connect() const {
 		Connection connection;
 		P_TRACE(3, "Connecting to " << address);
-		connection.fd = connectToServer(address, __FILE__, __LINE__);
+		NConnect_State state(address, __FILE__, __LINE__);
+		connection.ready = state.connectToServer();
 		connection.fail = true;
+		connection.fd = state.getFd().detach();
 		connection.wantKeepAlive = false;
-		connection.blocking = true;
 		P_LOG_FILE_DESCRIPTOR_PURPOSE(connection.fd, "App " << pid << " connection");
 		return connection;
 	}
@@ -164,15 +166,22 @@ class Socket {
 	}
 
 	/**
-	 * Connect to this socket or reuse an existing connection.
+	 * Initiates a non-blocking connection to this socket or reuse an existing connection.
+	 * Use `result.ready` to check whether the connect is finished or whether you need
+	 * to wait for it to finish.
 	 *
 	 * One MUST call checkinConnection() when one's done using the Connection.
 	 * Failure to do so will result in a resource leak.
+	 *
+	 * @throws SystemException Something went wrong.
+	 * @throws RuntimeException Something went wrong.
+	 * @throws boost::thread_interrupted A system call has been interrupted.
 	 */
 	Connection checkoutConnection() {
 		boost::unique_lock<boost::mutex> l(connectionPoolLock);
 
 		if (!idleConnections.empty()) {
+			TRACE_POINT();
 			P_TRACE(3, "Socket " << address << ": checking out connection from connection pool (" <<
 				idleConnections.size() << " -> " << (idleConnections.size() - 1) <<
 				" items). Current total number of connections: " << totalConnections);
@@ -181,6 +190,7 @@ class Socket {
 			totalIdleConnections--;
 			return connection;
 		} else {
+			TRACE_POINT();
 			Connection connection = connect();
 			totalConnections++;
 			P_TRACE(3, "Socket " << address << ": there are now " <<

src/agent/Core/ApplicationPool/Socket.h

@@ -28,10 +28,22 @@
 
 #include <boost/thread.hpp>
 #include <string>
+#include <utility>
 #include <cassert>
+#include <cstring>
+#include <unistd.h>
+#include <LoggingKit/Logging.h>
 #include <IOTools/IOUtils.h>
 #include <IOTools/BufferedIO.h>
+#include <StrIntTools/StrIntUtils.h>
 #include <Core/ApplicationPool/AbstractSession.h>
+#include <Exceptions.h>
+#include <StaticString.h>
+#include <FileDescriptor.h>
+#include <Utils.h>
+#include <Utils/ScopeGuard.h>
+#include <FileTools/FileManip.h>
+#include <oxt/system_calls.hpp>
 
 namespace Passenger {
 namespace ApplicationPool2 {
@@ -57,6 +69,7 @@ class TestSession: public AbstractSession {
 	mutable bool closed;
 	mutable bool success;
 	mutable bool wantKeepAlive;
+	mutable bool forcingNonInstantConnect;
 
 public:
 	TestSession()
@@ -67,7 +80,8 @@ class TestSession: public AbstractSession {
 		  stickySessionId(0),
 		  closed(false),
 		  success(false),
-		  wantKeepAlive(false)
+		  wantKeepAlive(false),
+		  forcingNonInstantConnect(false)
 		{ }
 
 	virtual void ref() const override {
@@ -163,12 +177,57 @@ class TestSession: public AbstractSession {
 		return wantKeepAlive;
 	}
 
-	virtual void initiate(bool blocking = true) override {
+	void forceNonInstantConnect() {
 		boost::lock_guard<boost::mutex> l(syncher);
-		connection = createUnixSocketPair(__FILE__, __LINE__);
+		forcingNonInstantConnect = true;
+	}
+
+	virtual bool initiate() override {
+		boost::lock_guard<boost::mutex> l(syncher);
+
+		// Create a unique temporary directory for the socket
+		string tempDirPathTemplate = StaticString(getSystemTempDir()) + "/passenger.session.XXXXXX";
+		DynamicBuffer tempDirPath(tempDirPathTemplate.size() + 1);
+		memcpy(tempDirPath.data, tempDirPathTemplate.c_str(), tempDirPathTemplate.size() + 1);
+		if (mkdtemp(tempDirPath.data) == NULL) {
+			int e = errno;
+			throw SystemException("Cannot create temporary directory", e);
+		}
+		ScopeGuard g([&]() {
+			try {
+				removeDirTree(tempDirPath.data);
+			} catch (const std::exception &e) {
+				P_ERROR("Error deleting temporary directory " << tempDirPath.data << ": " << e.what());
+			}
+		});
+
+		// Create server socket
+		string socketPath = StaticString(tempDirPath.data) + "/socket";
+		FileDescriptor serverFd(createUnixServer(socketPath.c_str(), 0, true, __FILE__, __LINE__),
+								__FILE__, __LINE__);
+
+		// Create client socket (non-blocking)
+		NUnix_State clientState;
+		setupNonBlockingUnixSocket(clientState, socketPath, __FILE__, __LINE__);
+		bool immediatelyConnected = connectToUnixServer(clientState);
+		connection.first = std::move(clientState.fd);
+
+		// Accept connection (blocking)
+		FileDescriptor serverSideFd(oxt::syscalls::accept(serverFd, NULL, NULL),
+			__FILE__, __LINE__);
+		if (serverSideFd == -1) {
+			int e = errno;
+			throw SystemException("Cannot accept connection", e);
+		}
+
+		// Store the server-side fd.
+		connection.second = std::move(serverSideFd);
 		peerBufferedIO = BufferedIO(connection.second);
-		if (!blocking) {
-			setNonBlocking(connection.first);
+
+		if (forcingNonInstantConnect) {
+			return false;
+		} else {
+			return immediatelyConnected;
 		}
 	}
 

src/agent/Core/ApplicationPool/TestSession.h

@@ -48,6 +48,7 @@
 #include <sys/types.h>
 #include <sys/uio.h>
 #include <utility>
+#include <exception>
 #include <cstdio>
 #include <cstdlib>
 #include <cstddef>
@@ -100,10 +101,6 @@ class Controller: public ServerKit::HttpServer<Controller, Client> {
 	typedef ServerKit::FileBufferedChannel FileBufferedChannel;
 	typedef ServerKit::FileBufferedFdSinkChannel FileBufferedFdSinkChannel;
 
-	// If you change this value, make sure that Request::sessionCheckoutTry
-	// has enough bits.
-	static const unsigned int MAX_SESSION_CHECKOUT_TRY = 10;
-
 	ControllerMainConfig mainConfig;
 	ControllerRequestConfigPtr requestConfig;
 	StringKeyTable< boost::shared_ptr<Options> > poolOptionsCache;
@@ -195,9 +192,13 @@ class Controller: public ServerKit::HttpServer<Controller, Client> {
 		const AbstractSessionPtr &session, const ExceptionPtr &e);
 	void maybeSend100Continue(Client *client, Request *req);
 	void initiateSession(Client *client, Request *req);
+	void finishInitiatingSession(Client *client, Request *req);
+	static void onSessionSocketConnected(EV_P_ ev_io *io, int revents);
+	static void onSessionSocketConnectTimeout(EV_P_ ev_timer *io, int flag);
 	static void checkoutSessionLater(Request *req);
 	void reportSessionCheckoutError(Client *client, Request *req,
 		const ExceptionPtr &e);
+	void handleSessionInitiationError(Client *client, Request *req, const std::exception &e);
 	int lookupCodeFromHeader(Request *req, const char* header, int statusCode);
 	void writeRequestQueueFullExceptionErrorResponse(Client *client,
 		Request *req, const boost::shared_ptr<RequestQueueFullException> &e);
@@ -311,6 +312,7 @@ class Controller: public ServerKit::HttpServer<Controller, Client> {
 	void endRequestWithSimpleResponse(Client **c, Request **r,
 		const StaticString &body, int code = 200);
 	void endRequestAsBadGateway(Client **client, Request **req);
+	void endRequestAsGatewayTimeout(Client **client, Request **req);
 	void writeBenchmarkResponse(Client **client, Request **req,
 		bool end = true);
 	bool getBoolOption(Request *req, const HashedStaticString &name,
@@ -355,9 +357,15 @@ class Controller: public ServerKit::HttpServer<Controller, Client> {
 
 	virtual void asyncGetFromApplicationPool(Request *req,
 		ApplicationPool2::GetCallback callback);
+	virtual int getSessionSocketConnectIoWatchConditions() const;
+	virtual double getSessionSocketEffectiveConnectTimeout(Request *req) const;
 
 
 public:
+	// If you change this value, make sure that Request::sessionCheckoutTry
+	// has enough bits.
+	static constexpr unsigned int MAX_SESSION_CHECKOUT_TRY = 10;
+
 	typedef ControllerConfigChangeRequest ConfigChangeRequest;
 
 	// Dependencies