Da

✏️ Editing: Server.h

/*
 *  Phusion Passenger - https://www.phusionpassenger.com/
 *  Copyright (c) 2014-2025 Asynchronous B.V.
 *
 *  "Passenger", "Phusion Passenger" and "Union Station" are registered
 *  trademarks of Asynchronous B.V.
 *
 *  Permission is hereby granted, free of charge, to any person obtaining a copy
 *  of this software and associated documentation files (the "Software"), to deal
 *  in the Software without restriction, including without limitation the rights
 *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *  copies of the Software, and to permit persons to whom the Software is
 *  furnished to do so, subject to the following conditions:
 *
 *  The above copyright notice and this permission notice shall be included in
 *  all copies or substantial portions of the Software.
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 *  THE SOFTWARE.
 */
#ifndef _PASSENGER_SERVER_KIT_SERVER_H_
#define _PASSENGER_SERVER_KIT_SERVER_H_

#include <psg_sysqueue.h>

#include <boost/cstdint.hpp>
#include <boost/config.hpp>
#include <boost/scoped_ptr.hpp>
#include <oxt/system_calls.hpp>
#include <oxt/backtrace.hpp>
#include <oxt/macros.hpp>
#include <vector>
#include <new>
#include <ev++.h>

#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <cstdio>
#include <jsoncpp/json.h>

#include <LoggingKit/LoggingKit.h>
#include <SafeLibev.h>
#include <Constants.h>
#include <ServerKit/Context.h>
#include <ServerKit/Errors.h>
#include <ServerKit/Hooks.h>
#include <ServerKit/Client.h>
#include <ServerKit/ClientRef.h>
#include <ConfigKit/ConfigKit.h>
#include <Algorithms/MovingAverage.h>
#include <Utils.h>
#include <Utils/ScopeGuard.h>
#include <StrIntTools/StrIntUtils.h>
#include <IOTools/IOUtils.h>
#include <SystemTools/SystemTime.h>

namespace Passenger {
namespace ServerKit {

using namespace std;
using namespace boost;
using namespace oxt;

// We use 'this' so that the macros work in derived template classes like HttpServer<>.
#define SKS_LOG(level, file, line, expr)  P_LOG(Passenger::LoggingKit::context, level, file, line, "[" << this->getServerName() << "] " << expr)
#define SKS_ERROR(expr)  P_ERROR("[" << this->getServerName() << "] " << expr)
#define SKS_WARN(expr)   P_WARN("[" << this->getServerName() << "] " << expr)
#define SKS_INFO(expr)   P_INFO("[" << this->getServerName() << "] " << expr)
#define SKS_NOTICE(expr) P_NOTICE("[" << this->getServerName() << "] " << expr)
#define SKS_DEBUG(expr)  P_DEBUG("[" << this->getServerName() << "] " << expr)
#define SKS_TRACE(level, expr) P_TRACE(level, "[" << this->getServerName() << "] " << expr)

#define SKS_NOTICE_FROM_STATIC(server, expr) P_NOTICE("[" << server->getServerName() << "] " << expr)

#define SKC_LOG(client, level, expr) SKC_LOG_FROM_STATIC(this, client, level, expr)
#define SKC_ERROR(client, expr) SKC_ERROR_FROM_STATIC(this, client, expr)
#define SKC_WARN(client, expr) SKC_WARN_FROM_STATIC(this, client, expr)
#define SKC_NOTICE(client, expr) SKC_NOTICE_FROM_STATIC(this, client, expr)
#define SKC_INFO(client, expr) SKC_INFO_FROM_STATIC(this, client, expr)
#define SKC_DEBUG(client, expr) SKC_DEBUG_FROM_STATIC(this, client, expr)
#define SKC_DEBUG_WITH_POS(client, file, line, expr) \
	SKC_DEBUG_FROM_STATIC_WITH_POS(this, client, file, line, expr)
#define SKC_TRACE(client, level, expr) SKC_TRACE_FROM_STATIC(this, client, level, expr)
#define SKC_TRACE_WITH_POS(client, level, file, line, expr) \
	SKC_TRACE_FROM_STATIC_WITH_POS(this, client, level, file, line, expr)

#define SKC_LOG_FROM_STATIC(server, client, level, expr) \
	do { \
		if (Passenger::LoggingKit::getLevel() >= level) { \
			char _clientName[16]; \
			int _clientNameSize = server->getClientName((client), _clientName, sizeof(_clientName)); \
			P_LOG(LoggingKit::context, level, __FILE__, __LINE__, \
				"[Client " << StaticString(_clientName, _clientNameSize) << "] " << expr); \
		} \
	} while (0)
#define SKC_ERROR_FROM_STATIC(server, client, expr) \
	SKC_LOG_FROM_STATIC(server, client, Passenger::LoggingKit::ERROR, expr)
#define SKC_WARN_FROM_STATIC(server, client, expr) \
	SKC_LOG_FROM_STATIC(server, client, Passenger::LoggingKit::WARN, expr)
#define SKC_NOTICE_FROM_STATIC(server, client, expr) \
	SKC_LOG_FROM_STATIC(server, client, Passenger::LoggingKit::NOTICE, expr)
#define SKC_INFO_FROM_STATIC(server, client, expr) \
	SKC_LOG_FROM_STATIC(server, client, Passenger::LoggingKit::INFO, expr)
#define SKC_DEBUG_FROM_STATIC(server, client, expr) \
	SKC_LOG_FROM_STATIC(server, client, Passenger::LoggingKit::DEBUG, expr)
#define SKC_DEBUG_FROM_STATIC_WITH_POS(server, client, file, line, expr) \
	do { \
		if (OXT_UNLIKELY(Passenger::LoggingKit::getLevel() >= Passenger::LoggingKit::DEBUG)) { \
			char _clientName[16]; \
			int _clientNameSize = server->getClientName((client), _clientName, sizeof(_clientName)); \
			P_DEBUG_WITH_POS(file, line, \
				"[Client " << StaticString(_clientName, _clientNameSize) << "] " << expr); \
		} \
	} while (0)
#define SKC_TRACE_FROM_STATIC(server, client, level, expr) \
	SKC_TRACE_FROM_STATIC_WITH_POS(server, client, level, __FILE__, __LINE__, expr)
#define SKC_TRACE_FROM_STATIC_WITH_POS(server, client, level, file, line, expr) \
	do { \
		if (OXT_UNLIKELY(Passenger::LoggingKit::getLevel() >= Passenger::LoggingKit::INFO + level)) { \
			char _clientName[16]; \
			int _clientNameSize = server->getClientName((client), _clientName, sizeof(_clientName)); \
			P_TRACE_WITH_POS(level, file, line, \
				"[Client " << StaticString(_clientName, _clientNameSize) << "] " << expr); \
		} \
	} while (0)

#define SKC_LOG_EVENT(klass, client, eventName) SKC_LOG_EVENT_FROM_STATIC(this, klass, client, eventName)
#define SKC_LOG_EVENT_FROM_STATIC(server, klass, client, eventName) \
	TRACE_POINT_WITH_DATA_FUNCTION(klass::_getClientNameFromTracePoint, client); \
	SKC_TRACE_FROM_STATIC(server, client, 3, "Event: " eventName)

/*
 * BEGIN ConfigKit schema: Passenger::ServerKit::BaseServerSchema
 * (do not edit: following text is automatically generated
 * by 'rake configkit_schemas_inline_comments')
 *
 *   accept_burst_count           unsigned integer   -   default(32)
 *   client_freelist_limit        unsigned integer   -   default(0)
 *   min_spare_clients            unsigned integer   -   default(0)
 *   start_reading_after_accept   boolean            -   default(true)
 *
 * END
 */
class BaseServerSchema: public ConfigKit::Schema {
private:
	void initialize() {
		using namespace ConfigKit;

add("accept_burst_count", UINT_TYPE, OPTIONAL, 32);
		add("start_reading_after_accept", BOOL_TYPE, OPTIONAL, true);
		add("min_spare_clients", UINT_TYPE, OPTIONAL, 0);
		add("client_freelist_limit", UINT_TYPE, OPTIONAL, 0);
	}

public:
	BaseServerSchema() {
		initialize();
		finalize();
	}

BaseServerSchema(bool _subclassing) {
		initialize();
	}
};

struct BaseServerConfigRealization {
	unsigned int acceptBurstCount: 7;
	bool startReadingAfterAccept: 1;
	unsigned int minSpareClients: 12;
	unsigned int clientFreelistLimit: 12;

BaseServerConfigRealization(const ConfigKit::Store &config)
		: acceptBurstCount(config["accept_burst_count"].asUInt()),
		  startReadingAfterAccept(config["start_reading_after_accept"].asBool()),
		  minSpareClients(config["min_spare_clients"].asUInt()),
		  clientFreelistLimit(config["client_freelist_limit"].asUInt())
		{ }

void swap(BaseServerConfigRealization &other) BOOST_NOEXCEPT_OR_NOTHROW {
		#define SWAP_BITFIELD(Type, name) \
			do { \
				Type tmp = name; \
				name = other.name; \
				other.name = tmp; \
			} while (false)

SWAP_BITFIELD(unsigned int, acceptBurstCount);
		SWAP_BITFIELD(bool, startReadingAfterAccept);
		SWAP_BITFIELD(unsigned int, minSpareClients);
		SWAP_BITFIELD(unsigned int, clientFreelistLimit);

#undef SWAP_BITFIELD
	}
};

struct BaseServerConfigChangeRequest {
	boost::scoped_ptr<ConfigKit::Store> config;
	boost::scoped_ptr<BaseServerConfigRealization> configRlz;
};

/**
 * A highly optimized generic base class for evented socket servers, implementing basic,
 * low-level connection management.
 *
 * ## Features
 *
 * ### Client objects
 *
 * Every connected client is represented by a client object, which inherits from
 * ServerKit::BaseClient. The client object provides input and output, and you can
 * extend with your own fields.
 *
 * Client objects are reference counted, for easy memory management.
 *
 * Creation and destruction is very efficient, because client objects are put on a
 * freelist upon destruction, so that no malloc calls are necessary next time.
 *
 * ### Zero-copy buffers
 *
 * All input is handled in a zero-copy manner, by using the mbuf system.
 *
 * ### Channel I/O abstraction
 *
 * All input is handled through the Channel abstraction, and all output is
 * handled through the FileBufferedFdSinkChannel abstraction.. This makes writing
 * evented servers very easy.
 *
 * ### Multiple listen endpoints
 *
 * The server can listen on multiple server endpoints at the same time (e.g. TCP and
 * Unix domain sockets), up to SERVER_KIT_MAX_SERVER_ENDPOINTS.
 *
 * ### Automatic backoff when too many file descriptors are active
 *
 * If ENFILES or EMFILES is encountered when accepting new clients, Server will stop
 * accepting new clients for a few seconds so that doesn't keep triggering the error
 * in a busy loop.
 *
 * ### Logging
 *
 * Provides basic logging macros that also log the client name.
 */
template<typename DerivedServer, typename Client = Client>
class BaseServer: public HooksImpl {
public:
	/***** Types *****/
	typedef BaseServer BaseClass;
	typedef Client ClientType;
	typedef ClientRef<DerivedServer, Client> ClientRefType;
	STAILQ_HEAD(FreeClientList, Client);
	TAILQ_HEAD(ClientList, Client);

enum State {
		ACTIVE,
		TOO_MANY_FDS,
		SHUTTING_DOWN,
		FINISHED_SHUTDOWN
	};

static const unsigned int MAX_ACCEPT_BURST_COUNT = 127;

typedef void (*Callback)(DerivedServer *server);
	typedef BaseServerConfigChangeRequest ConfigChangeRequest;

/***** Configuration *****/
	ConfigKit::Store config;
	BaseServerConfigRealization configRlz;
	Callback shutdownFinishCallback;

/***** Working state and statistics (do not modify) *****/
	State serverState;
	FreeClientList freeClients;
	ClientList activeClients, disconnectedClients;
	unsigned int freeClientCount, activeClientCount, disconnectedClientCount;
	unsigned int peakActiveClientCount;
	unsigned long totalClientsAccepted, lastTotalClientsAccepted;
	unsigned long long totalBytesConsumed;
	ev_tstamp lastStatisticsUpdateTime;
	double clientAcceptSpeed1m, clientAcceptSpeed1h;

private:
	Context *ctx;
	unsigned int nextClientNumber: 28;
	uint8_t nEndpoints: 3;
	bool accept4Available: 1;
	ev::timer acceptResumptionWatcher;
	ev::timer statisticsUpdateWatcher;
	ev::io endpoints[SERVER_KIT_MAX_SERVER_ENDPOINTS];

/***** Private methods *****/

void preinitialize(Context *context) {
		STAILQ_INIT(&freeClients);
		TAILQ_INIT(&activeClients);
		TAILQ_INIT(&disconnectedClients);

acceptResumptionWatcher.set(context->libev->getLoop());
		acceptResumptionWatcher.set<
			BaseServer<DerivedServer, Client>,
			&BaseServer<DerivedServer, Client>::onAcceptResumeTimeout>(this);

statisticsUpdateWatcher.set(context->libev->getLoop());
		statisticsUpdateWatcher.set<
			BaseServer<DerivedServer, Client>,
			&BaseServer<DerivedServer, Client>::onStatisticsUpdateTimeout>(this);
	}

static void _onAcceptable(EV_P_ ev_io *io, int revents) {
		static_cast<BaseServer *>(io->data)->onAcceptable(io, revents);
	}

void onAcceptable(ev_io *io, int revents) {
		TRACE_POINT();
		unsigned int acceptCount = 0;
		bool error = false;
		int fd, errcode = 0;
		Client *client;
		Client *acceptedClients[MAX_ACCEPT_BURST_COUNT];

P_ASSERT_EQ(serverState, ACTIVE);
		SKS_DEBUG("New clients can be accepted on a server socket");

for (unsigned int i = 0; i < configRlz.acceptBurstCount; i++) {
			fd = acceptNonBlockingSocket(io->fd);
			if (fd == -1) {
				error = true;
				errcode = errno;
				break;
			}

FdGuard guard(fd, NULL, 0);
			client = checkoutClientObject();
			TAILQ_INSERT_HEAD(&activeClients, client, nextClient.activeOrDisconnectedClient);
			acceptedClients[acceptCount] = client;
			activeClientCount++;
			acceptCount++;
			totalClientsAccepted++;
			client->number = getNextClientNumber();
			reinitializeClient(client, fd);
			P_LOG_FILE_DESCRIPTOR_PURPOSE(fd, "Server " << getServerName()
				<< ", client " << getClientName(client));
			guard.clear();
		}

if (acceptCount > 0) {
			SKS_DEBUG(acceptCount << " new client(s) accepted; there are now " <<
				activeClientCount << " active client(s)");
		}
		if (error && errcode != EAGAIN && errcode != EWOULDBLOCK) {
			SKS_ERROR("Cannot accept client: " << getErrorDesc(errcode) <<
				" (errno=" << errcode << "). " <<
				"Stop accepting clients for 3 seconds. " <<
				"Current client count: " << activeClientCount);
			serverState = TOO_MANY_FDS;
			acceptResumptionWatcher.set(3, 0);
			acceptResumptionWatcher.start();
			for (uint8_t i = 0; i < nEndpoints; i++) {
				ev_io_stop(ctx->libev->getLoop(), &endpoints[i]);
			}
		}

onClientsAccepted(acceptedClients, acceptCount);
	}

void onAcceptResumeTimeout(ev::timer &timer, int revents) {
		TRACE_POINT();
		P_ASSERT_EQ(serverState, TOO_MANY_FDS);
		SKS_NOTICE("Resuming accepting new clients");
		serverState = ACTIVE;
		for (uint8_t i = 0; i < nEndpoints; i++) {
			ev_io_start(ctx->libev->getLoop(), &endpoints[i]);
		}
	}

int acceptNonBlockingSocket(int serverFd) {
		union {
			struct sockaddr_in inaddr;
			struct sockaddr_in6 inaddr6;
			struct sockaddr_un unaddr;
		} u;
		socklen_t addrlen = sizeof(u);

if (accept4Available) {
			int fd = callAccept4(serverFd,
				(struct sockaddr *) &u,
				&addrlen,
				O_NONBLOCK);
			// FreeBSD returns EINVAL if accept4() is called with invalid flags.
			if (fd == -1 && (errno == ENOSYS || errno == EINVAL)) {
				accept4Available = false;
				return acceptNonBlockingSocket(serverFd);
			} else {
				return fd;
			}
		} else {
			int fd = syscalls::accept(serverFd,
				(struct sockaddr *) &u,
				&addrlen);
			FdGuard guard(fd, __FILE__, __LINE__);
			if (fd == -1) {
				return -1;
			} else {
				try {
					setNonBlocking(fd);
				} catch (const SystemException &e) {
					SKS_DEBUG("Unable to set non-blocking flag on accepted client socket: " <<
						e.what() << " (errno=" << e.code() << ")");
					errno = e.code();
					return -1;
				}
				guard.clear();
				return fd;
			}
		}
	}

void onStatisticsUpdateTimeout(ev::timer &timer, int revents) {
		TRACE_POINT();

this->onUpdateStatistics();
		this->onFinalizeStatisticsUpdate();

timer.repeat = timeToNextMultipleD(5, ev_now(this->getLoop()));
		timer.again();
	}

unsigned int getNextClientNumber() {
		return nextClientNumber++;
	}

Client *checkoutClientObject() {
		// Try to obtain client object from freelist.
		if (!STAILQ_EMPTY(&freeClients)) {
			return checkoutClientObjectFromFreelist();
		} else {
			return createNewClientObject();
		}
	}

Client *checkoutClientObjectFromFreelist() {
		assert(freeClientCount > 0);
		SKS_TRACE(3, "Checking out client object from freelist (" <<
			freeClientCount << " -> " << (freeClientCount - 1) << ")");
		Client *client = STAILQ_FIRST(&freeClients);
		P_ASSERT_EQ(client->getConnState(), Client::IN_FREELIST);
		client->refcount.store(2, boost::memory_order_relaxed);
		freeClientCount--;
		STAILQ_REMOVE_HEAD(&freeClients, nextClient.freeClient);
		return client;
	}

Client *createNewClientObject() {
		Client *client;
		SKS_TRACE(3, "Creating new client object");
		try {
			client = new Client(this);
		} catch (const std::bad_alloc &) {
			return NULL;
		}
		onClientObjectCreated(client);
		return client;
	}

void clientReachedZeroRefcount(Client *client) {
		TRACE_POINT();
		assert(disconnectedClientCount > 0);
		assert(!TAILQ_EMPTY(&disconnectedClients));

SKC_TRACE(client, 3, "Client object reached a reference count of 0");
		TAILQ_REMOVE(&disconnectedClients, client, nextClient.activeOrDisconnectedClient);
		disconnectedClientCount--;

if (addClientToFreelist(client)) {
			SKC_TRACE(client, 3, "Client object added to freelist (" <<
				(freeClientCount - 1) << " -> " << freeClientCount << ")");
		} else {
			SKC_TRACE(client, 3, "Client object destroyed; not added to freelist " <<
				"because it's full (" << freeClientCount << ")");
			delete client;
		}

if (serverState == SHUTTING_DOWN
		 && activeClientCount == 0
		 && disconnectedClientCount == 0)
		{
			finishShutdown();
		}
	}

bool addClientToFreelist(Client *client) {
		if (freeClientCount < configRlz.clientFreelistLimit) {
			STAILQ_INSERT_HEAD(&freeClients, client, nextClient.freeClient);
			freeClientCount++;
			client->refcount.store(2, boost::memory_order_relaxed);
			client->setConnState(Client::IN_FREELIST);
			return true;
		} else {
			return false;
		}
	}

void passClientToEventLoopThread(Client *client) {
		// The shutdown procedure waits until all ACTIVE and DISCONNECTED
		// clients are gone before destroying a Server, so we know for sure
		// that this async callback outlives the Server.
		ctx->libev->runLater(boost::bind(&BaseServer::passClientToEventLoopThreadCallback,
			this, ClientRefType(client, __FILE__, __LINE__)));
	}

void passClientToEventLoopThreadCallback(ClientRefType clientRef) {
		// Do nothing. Once this method returns, the reference count of the
		// client drops to 0, and clientReachedZeroRefcount() is called.
	}

const char *getServerStateString() const {
		switch (serverState) {
		case ACTIVE:
			return "ACTIVE";
		case TOO_MANY_FDS:
			return "TOO_MANY_FDS";
		case SHUTTING_DOWN:
			return "SHUTTING_DOWN";
		case FINISHED_SHUTDOWN:
			return "FINISHED_SHUTDOWN";
		default:
			return "UNKNOWN";
		}
	}

void finishShutdown() {
		TRACE_POINT();
		compact(LoggingKit::INFO);

acceptResumptionWatcher.stop();
		statisticsUpdateWatcher.stop();

SKS_NOTICE("Shutdown finished");
		serverState = FINISHED_SHUTDOWN;
		if (shutdownFinishCallback) {
			shutdownFinishCallback(static_cast<DerivedServer *>(this));
		}
	}

void logClientDataReceived(Client *client, const MemoryKit::mbuf &buffer, int errcode) {
		if (buffer.size() > 0) {
			SKC_TRACE(client, 3, "Processing " << buffer.size() << " bytes of client data");
		} else if (errcode == 0) {
			SKC_TRACE(client, 2, "Client sent EOF");
		} else {
			SKC_TRACE(client, 2, "Error reading from client socket: " <<
				getErrorDesc(errcode) << " (errno=" << errcode << ")");
		}
	}

static Channel::Result _onClientDataReceived(Channel *_channel,
		const MemoryKit::mbuf &buffer, int errcode)
	{
		FdSourceChannel *channel = reinterpret_cast<FdSourceChannel *>(_channel);
		Client *client = static_cast<Client *>(static_cast<BaseClient *>(
			channel->getHooks()->userData));
		BaseServer *server = getServerFromClient(client);
		const size_t bufferSize = buffer.size();

server->logClientDataReceived(client, buffer, errcode);
		Channel::Result result = server->onClientDataReceived(client, buffer, errcode);

// This counter is mostly useful for unit tests, so it's too much hassle to
		// support cases where result.consumed < 1.
		size_t consumed = std::max<size_t>(0,
			std::min<size_t>(result.consumed, bufferSize));
		server->totalBytesConsumed += consumed;
		SKC_TRACE_FROM_STATIC(server, client, 2,
			consumed << " bytes of client data consumed in this callback");

return result;
	}

static void _onClientOutputError(FileBufferedFdSinkChannel *_channel, int errcode) {
		FileBufferedFdSinkChannel *channel =
			reinterpret_cast<FileBufferedFdSinkChannel *>(_channel);
		Client *client = static_cast<Client *>(static_cast<BaseClient *>(
			channel->getHooks()->userData));
		BaseServer *server = getServerFromClient(client);
		server->onClientOutputError(client, errcode);
	}

protected:
	/***** Hooks *****/

virtual void onClientObjectCreated(Client *client) {
		TRACE_POINT();

client->hooks.impl        = this;
		client->hooks.userData    = static_cast<BaseClient *>(client);

client->input.setContext(ctx);
		client->input.setHooks(&client->hooks);
		client->input.setDataCallback(_onClientDataReceived);

client->output.setContext(ctx);
		client->output.setHooks(&client->hooks);
		client->output.errorCallback = _onClientOutputError;
	}

virtual void onClientsAccepted(Client **clients, unsigned int size) {
		unsigned int i;

peakActiveClientCount = std::max(peakActiveClientCount, activeClientCount);

for (i = 0; i < size; i++) {
			Client *client = clients[i];

onClientAccepted(client);
			if (client->connected()) {
				if (configRlz.startReadingAfterAccept) {
					client->input.startReading();
				} else {
					client->input.startReadingInNextTick();
				}
			}
			// A Client object starts with a refcount of 2 so that we can
			// be sure it won't be destroyted while we're looping inside this
			// function. But we also need an extra unref here.
			unrefClient(client, __FILE__, __LINE__);
		}
	}

virtual void onClientAccepted(Client *client) {
		// Do nothing.
	}

virtual void onClientDisconnecting(Client *client) {
		// Do nothing.
	}

virtual void onClientDisconnected(Client *client) {
		// Do nothing.
	}

virtual bool shouldDisconnectClientOnShutdown(Client *client) {
		return false;
	}

virtual Channel::Result onClientDataReceived(Client *client, const MemoryKit::mbuf &buffer,
		int errcode)
	{
		if (buffer.empty()) {
			disconnect(&client);
		}
		return Channel::Result(0, true);
	}

virtual void onClientOutputError(Client *client, int errcode) {
		SKC_LOG_EVENT(DerivedServer, client, "onClientOutputError");
		char message[1024];
		int ret = snprintf(message, sizeof(message),
			"client socket write error: %s (errno=%d)",
			getErrorDesc(errcode), errcode);
		disconnectWithError(&client, StaticString(message, ret),
			getClientOutputErrorDisconnectionLogLevel(client, errcode));
	}

virtual LoggingKit::Level getClientOutputErrorDisconnectionLogLevel(
		Client *client, int errcode) const
	{
		return LoggingKit::WARN;
	}

virtual void onUpdateStatistics() {
		SKS_DEBUG("Updating statistics");
		ev_tstamp now = ev_now(this->getLoop());
		ev_tstamp duration = now - lastStatisticsUpdateTime;

// Statistics are updated about every 5 seconds, so about 12 updates
		// per minute. We want the old average to decay to 5% after 1 minute
		// and 1 hour, respectively, so:
		// 1 minute: 1 - exp(ln(0.05) / 12) = 0.22092219194555585
		// 1 hour  : 1 - exp(ln(0.05) / (60 * 12)) = 0.0041520953856636345
		clientAcceptSpeed1m = expMovingAverage(clientAcceptSpeed1m,
			(totalClientsAccepted - lastTotalClientsAccepted) / duration,
			0.22092219194555585);
		clientAcceptSpeed1h = expMovingAverage(clientAcceptSpeed1h,
			(totalClientsAccepted - lastTotalClientsAccepted) / duration,
			0.0041520953856636345);
	}

virtual void onFinalizeStatisticsUpdate() {
		lastTotalClientsAccepted = totalClientsAccepted;
		lastStatisticsUpdateTime = ev_now(this->getLoop());
	}

virtual void reinitializeClient(Client *client, int fd) {
		client->setConnState(Client::ACTIVE);
		SKC_TRACE(client, 2, "Client associated with file descriptor: " << fd);
		client->input.reinitialize(fd);
		client->output.reinitialize(fd);
	}

virtual void deinitializeClient(Client *client) {
		client->input.deinitialize();
		client->output.deinitialize();
	}

virtual void onShutdown(bool forceDisconnect) {
		// Do nothing.
	}

public:
	/***** Public methods *****/

BaseServer(Context *context, const BaseServerSchema &schema,
		const Json::Value &initialConfig = Json::Value(),
		const ConfigKit::Translator &translator = ConfigKit::DummyTranslator())
		: config(schema, initialConfig, translator),
		  configRlz(config),
		  shutdownFinishCallback(NULL),
		  serverState(ACTIVE),
		  freeClientCount(0),
		  activeClientCount(0),
		  disconnectedClientCount(0),
		  peakActiveClientCount(0),
		  totalClientsAccepted(0),
		  lastTotalClientsAccepted(0),
		  totalBytesConsumed(0),
		  lastStatisticsUpdateTime(ev_time()),
		  clientAcceptSpeed1m(-1),
		  clientAcceptSpeed1h(-1),
		  ctx(context),
		  nextClientNumber(1),
		  nEndpoints(0),
		  accept4Available(true)
	{
		preinitialize(context);
	}

virtual ~BaseServer() {
		P_ASSERT_EQ(serverState, FINISHED_SHUTDOWN);
	}

/***** Initialization, listening and shutdown *****/

virtual void initialize() {
		statisticsUpdateWatcher.set(5, 5);
		statisticsUpdateWatcher.start();
	}

// Pre-create multiple client objects so that they get allocated
	// near each other in memory. Hopefully increases CPU cache locality.
	void createSpareClients() {
		for (unsigned int i = 0; i < configRlz.minSpareClients; i++) {
			Client *client = createNewClientObject();
			client->setConnState(Client::IN_FREELIST);
			STAILQ_INSERT_HEAD(&freeClients, client, nextClient.freeClient);
			freeClientCount++;
		}
	}

void listen(int fd) {
		#ifdef EOPNOTSUPP
			#define EXTENSION_EOPNOTSUPP EOPNOTSUPP
		#else
			#define EXTENSION_EOPNOTSUPP ENOTSUP
		#endif

TRACE_POINT();
		assert(nEndpoints < SERVER_KIT_MAX_SERVER_ENDPOINTS);
		int flag = 1;
		setNonBlocking(fd);
		if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int)) == -1
		 && errno != ENOPROTOOPT
		 && errno != ENOTSUP
		 && errno != EXTENSION_EOPNOTSUPP)
		{
			int e = errno;
			SKS_WARN("Cannot disable Nagle's algorithm on a TCP socket: " <<
				strerror(e) << " (errno=" << e << ")");
		}
		ev_io_init(&endpoints[nEndpoints], _onAcceptable, fd, EV_READ);
		endpoints[nEndpoints].data = this;
		ev_io_start(ctx->libev->getLoop(), &endpoints[nEndpoints]);
		nEndpoints++;

#undef EXTENSION_EOPNOTSUPP
	}

void shutdown(bool forceDisconnect = false) {
		if (serverState != ACTIVE) {
			return;
		}

vector<Client *> clients;
		Client *client;
		typename vector<Client *>::iterator v_it, v_end;

SKS_DEBUG("Shutting down");
		serverState = SHUTTING_DOWN;
		onShutdown(forceDisconnect);

// Stop listening on all endpoints.
		acceptResumptionWatcher.stop();
		for (uint8_t i = 0; i < nEndpoints; i++) {
			ev_io_stop(ctx->libev->getLoop(), &endpoints[i]);
		}

if (activeClientCount == 0 && disconnectedClientCount == 0) {
			finishShutdown();
			return;
		}

// Once we've set serverState to SHUTTING_DOWN, `activeClientCount` will no
		// longer grow, but may change due to hooks and callbacks.
		// So we make a copy of the client list here and operate on that.
		clients.reserve(activeClientCount);
		TAILQ_FOREACH (client, &activeClients, nextClient.activeOrDisconnectedClient) {
			P_ASSERT_EQ(client->getConnState(), Client::ACTIVE);
			refClient(client, __FILE__, __LINE__);
			clients.push_back(client);
		}

// Disconnect each active client.
		v_end = clients.end();
		for (v_it = clients.begin(); v_it != v_end; v_it++) {
			client = (Client *) *v_it;
			Client *c = client;
			if (forceDisconnect || shouldDisconnectClientOnShutdown(client)) {
				disconnectWithError(&client, "server is shutting down");
			}
			unrefClient(c, __FILE__, __LINE__);
		}

// When all active and disconnected clients are gone,
		// finishShutdown() will be called to set state to FINISHED_SHUTDOWN.
	}

void feedNewClients(const int *fds, unsigned int size) {
		Client *client;
		Client *acceptedClients[MAX_ACCEPT_BURST_COUNT];

assert(size > 0);
		assert(size <= MAX_ACCEPT_BURST_COUNT);
		P_ASSERT_EQ(serverState, ACTIVE);

activeClientCount += size;
		totalClientsAccepted += size;

for (unsigned int i = 0; i < size; i++) {
			client = checkoutClientObject();
			TAILQ_INSERT_HEAD(&activeClients, client, nextClient.activeOrDisconnectedClient);
			acceptedClients[i] = client;
			client->number = getNextClientNumber();
			reinitializeClient(client, fds[i]);
			P_LOG_FILE_DESCRIPTOR_PURPOSE(fds[i], "Server " << getServerName()
				<< ", client " << getClientName(client));
		}

SKS_DEBUG(size << " new client(s) accepted; there are now " <<
			activeClientCount << " active client(s)");

onClientsAccepted(acceptedClients, size);
	}

/***** Server management *****/

virtual void compact(LoggingKit::Level logLevel = LoggingKit::NOTICE) {
		unsigned int count = freeClientCount;

while (!STAILQ_EMPTY(&freeClients)) {
			Client *client = STAILQ_FIRST(&freeClients);
			P_ASSERT_EQ(client->getConnState(), Client::IN_FREELIST);
			client->refcount.store(2, boost::memory_order_relaxed);
			freeClientCount--;
			STAILQ_REMOVE_HEAD(&freeClients, nextClient.freeClient);
			delete client;
		}
		assert(freeClientCount == 0);

SKS_LOG(logLevel, __FILE__, __LINE__,
			"Freed " << count << " spare client objects");
	}

/***** Client management *****/

// Ensures that the buffer is NULL-terminated.
	virtual unsigned int getClientName(const Client *client, char *buf, size_t size) const {
		unsigned int ret = uintToString(client->number, buf, size - 1);
		buf[ret] = '\0';
		return ret;
	}

string getClientName(const Client *client) {
		char buf[128];
		unsigned int size = getClientName(client, buf, sizeof(buf));
		return string(buf, size);
	}

vector<ClientRefType> getActiveClients() {
		vector<ClientRefType> result;
		Client *client;

TAILQ_FOREACH (client, &activeClients, nextClient.activeOrDisconnectedClient) {
			P_ASSERT_EQ(client->getConnState(), Client::ACTIVE);
			result.push_back(ClientRefType(client, __FILE__, __LINE__));
		}
		return result;
	}

Client *lookupClient(int fd) {
		Client *client;

TAILQ_FOREACH (client, &activeClients, nextClient.activeOrDisconnectedClient) {
			P_ASSERT_EQ(client->getConnState(), Client::ACTIVE);
			if (client->fd == fd) {
				return client;
			}
		}
		return NULL;
	}

Client *lookupClient(const StaticString &clientName) {
		Client *client;

TAILQ_FOREACH (client, &activeClients, nextClient.activeOrDisconnectedClient) {
			P_ASSERT_EQ(client->getConnState(), Client::ACTIVE);
			char buf[512];
			unsigned int size;

size = getClientName(client, buf, sizeof(buf));
			if (StaticString(buf, size) == clientName) {
				return client;
			}
		}
		return NULL;
	}

bool clientOnUnixDomainSocket(Client *client) {
		union {
			struct sockaddr genericAddress;
			struct sockaddr_un unixAddress;
			struct sockaddr_in inetAddress;
		} addr;
		socklen_t len = sizeof(addr);
		int ret;

do {
			ret = getsockname(client->getFd(), &addr.genericAddress, &len);
		} while (ret == -1 && errno == EINTR);
		if (ret == -1) {
			int e = errno;
			throw SystemException("Unable to autodetect socket type (getsockname() failed)", e);
		} else {
			#ifdef AF_UNIX
				return addr.genericAddress.sa_family == AF_UNIX;
			#else
				return addr.genericAddress.sa_family == AF_LOCAL;
			#endif
		}
	}

/** Increase client reference count. */
	void refClient(Client *client, const char *file, unsigned int line) {
		int oldRefcount = client->refcount.fetch_add(1, boost::memory_order_relaxed);
		SKC_TRACE_WITH_POS(client, 3, file, line,
			"Refcount increased; it is now " << (oldRefcount + 1));
	}

/** Decrease client reference count. Adds client to the
	 * freelist if reference count drops to 0.
	 */
	void unrefClient(Client *client, const char *file, unsigned int line) {
		int oldRefcount = client->refcount.fetch_sub(1, boost::memory_order_release);
		assert(oldRefcount >= 1);

SKC_TRACE_WITH_POS(client, 3, file, line,
			"Refcount decreased; it is now " << (oldRefcount - 1));
		if (oldRefcount == 1) {
			boost::atomic_thread_fence(boost::memory_order_acquire);

if (ctx->libev->onEventLoopThread()) {
				assert(client->getConnState() != Client::IN_FREELIST);
				/* As long as the client is still in the ACTIVE state, it has at least
				 * one reference, namely from the Server itself. Therefore it's impossible
				 * to get to a zero reference count without having disconnected a client. */
				P_ASSERT_EQ(client->getConnState(), Client::DISCONNECTED);
				clientReachedZeroRefcount(client);
			} else {
				// Let the event loop handle the client reaching the 0 refcount.
				SKC_TRACE(client, 3, "Passing client object to event loop thread");
				passClientToEventLoopThread(client);
			}
		}
	}

bool disconnect(int fd) {
		assert(serverState != FINISHED_SHUTDOWN);
		Client *client = lookupClient(fd);
		if (client != NULL) {
			return disconnect(&client);
		} else {
			return false;
		}
	}

bool disconnect(const StaticString &clientName) {
		assert(serverState != FINISHED_SHUTDOWN);
		Client *client = lookupClient(clientName);
		if (client != NULL) {
			return disconnect(&client);
		} else {
			return false;
		}
	}

bool disconnect(Client **client) {
		Client *c = *client;
		if (c->getConnState() != Client::ACTIVE) {
			return false;
		}

int fdnum = c->getFd();
		SKC_TRACE(c, 2, "Disconnecting; there are now " << (activeClientCount - 1) <<
			" active clients");
		onClientDisconnecting(c);

c->setConnState(ClientType::DISCONNECTED);
		TAILQ_REMOVE(&activeClients, c, nextClient.activeOrDisconnectedClient);
		activeClientCount--;
		TAILQ_INSERT_HEAD(&disconnectedClients, c, nextClient.activeOrDisconnectedClient);
		disconnectedClientCount++;

deinitializeClient(c);
		SKC_TRACE(c, 2, "Closing client file descriptor: " << fdnum);
		try {
			safelyClose(fdnum);
			P_LOG_FILE_DESCRIPTOR_CLOSE(fdnum);
		} catch (const SystemException &e) {
			SKC_WARN(c, "An error occurred while closing the client file descriptor: " <<
				e.what() << " (errno=" << e.code() << ")");
		}

*client = NULL;
		onClientDisconnected(c);
		unrefClient(c, __FILE__, __LINE__);
		return true;
	}

void disconnectWithWarning(Client **client, const StaticString &message) {
		SKC_WARN(*client, "Disconnecting client with warning: " << message);
		disconnect(client);
	}

void disconnectWithError(Client **client, const StaticString &message,
		LoggingKit::Level logLevel = LoggingKit::WARN)
	{
		SKC_LOG(*client, logLevel, "Disconnecting client with error: " << message);
		disconnect(client);
	}

/***** Introspection *****/

OXT_FORCE_INLINE
	Context *getContext() {
		return ctx;
	}

OXT_FORCE_INLINE
	const Context *getContext() const {
		return ctx;
	}

OXT_FORCE_INLINE
	struct ev_loop *getLoop() const {
		return ctx->libev->getLoop();
	}

virtual StaticString getServerName() const {
		return P_STATIC_STRING("Server");
	}

bool prepareConfigChange(const Json::Value &updates,
		vector<ConfigKit::Error> &errors, BaseServerConfigChangeRequest &req)
	{
		req.config.reset(new ConfigKit::Store(config, updates, errors));
		if (errors.empty()) {
			req.configRlz.reset(new BaseServerConfigRealization(*req.config));
		}
		return errors.empty();
	}

void commitConfigChange(BaseServerConfigChangeRequest &req)
		BOOST_NOEXCEPT_OR_NOTHROW
	{
		config.swap(*req.config);
		configRlz.swap(*req.configRlz);
	}

virtual Json::Value inspectConfig() const {
		return config.inspect();
	}

virtual Json::Value inspectStateAsJson() const {
		Json::Value doc = ctx->inspectStateAsJson();
		const Client *client;

doc["pid"] = (unsigned int) getpid();
		doc["server_state"] = getServerStateString();
		doc["free_client_count"] = freeClientCount;
		Json::Value &activeClientsDoc = doc["active_clients"] = Json::Value(Json::objectValue);
		doc["active_client_count"] = activeClientCount;
		Json::Value &disconnectedClientsDoc = doc["disconnected_clients"] = Json::Value(Json::objectValue);
		doc["disconnected_client_count"] = disconnectedClientCount;
		doc["peak_active_client_count"] = peakActiveClientCount;
		doc["client_accept_speed"]["1m"] = averageSpeedToJson(
			capFloatPrecision(clientAcceptSpeed1m * 60),
			"minute", "1 minute", -1);
		doc["client_accept_speed"]["1h"] = averageSpeedToJson(
			capFloatPrecision(clientAcceptSpeed1h * 60),
			"minute", "1 hour", -1);
		doc["total_clients_accepted"] = (Json::UInt64) totalClientsAccepted;
		doc["total_bytes_consumed"] = (Json::UInt64) totalBytesConsumed;

TAILQ_FOREACH (client, &activeClients, nextClient.activeOrDisconnectedClient) {
			Json::Value subdoc;
			char clientName[16];

getClientName(client, clientName, sizeof(clientName));
			activeClientsDoc[clientName] = inspectClientStateAsJson(client);
		}

TAILQ_FOREACH (client, &disconnectedClients, nextClient.activeOrDisconnectedClient) {
			Json::Value subdoc;
			char clientName[16];

getClientName(client, clientName, sizeof(clientName));
			disconnectedClientsDoc[clientName] = inspectClientStateAsJson(client);
		}

return doc;
	}

virtual Json::Value inspectClientStateAsJson(const Client *client) const {
		Json::Value doc;
		char clientName[16];

assert(client->getConnState() != Client::IN_FREELIST);
		getClientName(client, clientName, sizeof(clientName));
		doc["connection_state"] = client->getConnStateString();
		doc["name"] = clientName;
		doc["number"] = client->number;
		doc["refcount"] = client->refcount.load(boost::memory_order_relaxed);
		doc["output_channel_state"] = client->output.inspectAsJson();

return doc;
	}

/***** Miscellaneous *****/

/** Get a thread-safe reference to the client. As long as the client
	 * has a reference, it will never be added to the freelist.
	 */
	ClientRefType getClientRef(Client *client, const char *file, unsigned int line) {
		return ClientRefType(client, file, line);
	}

/**
	 * Returns a pointer to the BaseServer that created the given Client object.
	 * Unlike the void pointer that Client::getServerBaseClassPointer() returns,
	 * this method's typed return value allows safe recasting of the result pointer.
	 */
	static const BaseServer *getConstServerFromClient(const Client *client) {
		return static_cast<BaseServer *>(client->getServerBaseClassPointer());
	}

static BaseServer *getServerFromClient(Client *client) {
		return static_cast<BaseServer *>(client->getServerBaseClassPointer());
	}

/***** Friend-public methods and hook implementations *****/

void _refClient(Client *client, const char *file, unsigned int line) {
		refClient(client, file, line);
	}

void _unrefClient(Client *client, const char *file, unsigned int line) {
		unrefClient(client, __FILE__, __LINE__);
	}

static bool _getClientNameFromTracePoint(char *output, unsigned int size, void *userData) {
		Client *client = static_cast<Client *>(userData);
		BaseServer *server = getServerFromClient(client);
		char *pos = output;
		const char *end = output + size;

pos = appendData(pos, end, "Client ");
		server->getClientName(client, pos, end - pos);
		return true;
	}

virtual bool hook_isConnected(Hooks *hooks, void *source) {
		Client *client = static_cast<Client *>(static_cast<BaseClient *>(hooks->userData));
		return client->connected();
	}

virtual void hook_ref(Hooks *hooks, void *source, const char *file, unsigned int line) {
		Client *client = static_cast<Client *>(static_cast<BaseClient *>(hooks->userData));
		refClient(client, file, line);
	}

virtual void hook_unref(Hooks *hooks, void *source, const char *file, unsigned int line) {
		Client *client = static_cast<Client *>(static_cast<BaseClient *>(hooks->userData));
		unrefClient(client, file, line);
	}
};

template <typename Client = Client>
class Server: public BaseServer<Server<Client>, Client> {
public:
	Server(Context *context, const BaseServerSchema &schema,
		const Json::Value &initialConfig = Json::Value())
		: BaseServer<Server, Client>(context, schema, initialConfig)
		{ }
};

} // namespace ServerKit
} // namespace Passenger

#endif /* _PASSENGER_SERVER_KIT_SERVER_H_ */

📁 Sid Gifari File Manager