/*
* Phusion Passenger - https://www.phusionpassenger.com/
* Copyright (c) 2011-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_APPLICATION_POOL_GROUP_PROCESS_LIST_MANAGEMENT_CPP_
#define _PASSENGER_APPLICATION_POOL_GROUP_PROCESS_LIST_MANAGEMENT_CPP_
#ifdef INTELLISENSE
#include
#endif
#include
/*************************************************************************
*
* Process list management functions for ApplicationPool2::Group
*
*************************************************************************/
namespace Passenger {
namespace ApplicationPool2 {
using namespace std;
using namespace boost;
/****************************
*
* Private methods
*
****************************/
Process *
Group::findProcessWithStickySessionId(unsigned int id) const {
ProcessList::const_iterator it, end = enabledProcesses.end();
for (it = enabledProcesses.begin(); it != end; it++) {
Process *process = it->get();
if (process->getStickySessionId() == id) {
return process;
}
}
return NULL;
}
Process *
Group::findProcessWithStickySessionIdOrLowestBusyness(unsigned int id) const {
int leastBusyProcessIndex = -1;
int lowestBusyness = 0;
unsigned int i, size = enabledProcessBusynessLevels.size();
const int *enabledProcessBusynessLevels = &this->enabledProcessBusynessLevels[0];
for (i = 0; i < size; i++) {
Process *process = enabledProcesses[i].get();
if (process->getStickySessionId() == id) {
return process;
} else if (leastBusyProcessIndex == -1 || enabledProcessBusynessLevels[i] < lowestBusyness) {
leastBusyProcessIndex = i;
lowestBusyness = enabledProcessBusynessLevels[i];
}
}
if (leastBusyProcessIndex == -1) {
return NULL;
} else {
return enabledProcesses[leastBusyProcessIndex].get();
}
}
Process *
Group::findProcessWithLowestBusyness(const ProcessList &processes) const {
if (processes.empty()) {
return NULL;
}
int lowestBusyness = -1;
Process *leastBusyProcess = NULL;
ProcessList::const_iterator it;
ProcessList::const_iterator end = processes.end();
for (it = processes.begin(); it != end; it++) {
Process *process = (*it).get();
int busyness = process->busyness();
if (lowestBusyness == -1 || lowestBusyness > busyness) {
lowestBusyness = busyness;
leastBusyProcess = process;
}
}
return leastBusyProcess;
}
/**
* Cache-optimized version of findProcessWithLowestBusyness() for the common case.
*/
Process *
Group::findEnabledProcessWithLowestBusyness() const {
if (enabledProcesses.empty()) {
return NULL;
}
int leastBusyProcessIndex = -1;
int lowestBusyness = 0;
unsigned int i, size = enabledProcessBusynessLevels.size();
const int *enabledProcessBusynessLevels = &this->enabledProcessBusynessLevels[0];
for (i = 0; i < size; i++) {
if (leastBusyProcessIndex == -1 || enabledProcessBusynessLevels[i] < lowestBusyness) {
leastBusyProcessIndex = i;
lowestBusyness = enabledProcessBusynessLevels[i];
}
}
return enabledProcesses[leastBusyProcessIndex].get();
}
/**
* Return the process with the given sticky session ID if it exists.
* If not, then find the "best" enabled process to route a request to,
* according to the same criteria documented for findBestProcess().
*
* - If the process with the given sticky session ID exists, then always
* returns that process. Meaning that this process could be `!canBeRoutedTo()`.
* - If there is no process that can be routed to, then returns nullptr.
*/
Process *
Group::findBestProcessPreferringStickySessionId(unsigned int id) const {
Process *bestProcess = nullptr;
ProcessList::const_iterator it;
ProcessList::const_iterator end = enabledProcesses.end();
for (it = enabledProcesses.begin(); it != end; it++) {
Process *process = it->get();
if (process->getStickySessionId() == id) {
return process;
} else if (!process->isTotallyBusy()
&& (
bestProcess == nullptr
|| process->generation > bestProcess->generation
|| (process->generation == bestProcess->generation && process->spawnStartTime < bestProcess->spawnStartTime)
|| (process->generation == bestProcess->generation && process->spawnStartTime == bestProcess->spawnStartTime && process->busyness() < bestProcess->busyness())
)
) {
bestProcess = process;
}
}
return bestProcess;
}
/**
* Given a ProcessList, find the "best" process to route a request to.
* At the moment, "best" is defined as the process with the highest generation,
* lowest start time, and lowest busyness, in that order of priority.
*
* If there is no process that can be routed to, then returns nullptr.
*
* @post result == nullptr || result.canBeRoutedTo()
*/
Process *
Group::findBestProcess(const ProcessList &processes) const {
if (processes.empty()) {
return nullptr;
}
Process *bestProcess = nullptr;
ProcessList::const_iterator it;
ProcessList::const_iterator end = processes.end();
for (it = processes.begin(); it != end; it++) {
Process *process = it->get();
if (!process->isTotallyBusy()
&& (
bestProcess == nullptr
|| process->generation > bestProcess->generation
|| (process->generation == bestProcess->generation && process->spawnStartTime < bestProcess->spawnStartTime)
|| (process->generation == bestProcess->generation && process->spawnStartTime == bestProcess->spawnStartTime && process->busyness() < bestProcess->busyness())
)
) {
bestProcess = process;
}
}
return bestProcess;
}
/**
* Adds a process to the given list (enabledProcess, disablingProcesses, disabledProcesses)
* and sets the process->enabled flag accordingly.
* The process must currently not be in any list. This function does not fix
* getWaitlist invariants or other stuff.
*/
void
Group::addProcessToList(const ProcessPtr &process, ProcessList &destination) {
destination.push_back(process);
process->setIndex(destination.size() - 1);
if (&destination == &enabledProcesses) {
process->enabled = Process::ENABLED;
enabledCount++;
enabledProcessBusynessLevels.push_back(process->busyness());
if (process->isTotallyBusy()) {
nEnabledProcessesTotallyBusy++;
}
} else if (&destination == &disablingProcesses) {
process->enabled = Process::DISABLING;
disablingCount++;
} else if (&destination == &disabledProcesses) {
assert(process->sessions == 0);
process->enabled = Process::DISABLED;
disabledCount++;
} else if (&destination == &detachedProcesses) {
assert(process->isAlive());
process->enabled = Process::DETACHED;
if (!this->options.abortWebsocketsOnProcessShutdown && this->options.appType == P_STATIC_STRING("nodejs")) {
// When Passenger is not allowed to abort websockets the application needs a way to know graceful shutdown
// is in progress. The solution for the most common use (Node.js) is to send a SIGINT. This is the general
// termination signal for Node; later versions of pm2 also use it (with a 1.6 sec grace period, Passenger just waits)
kill(process->getPid(), SIGINT);
}
callAbortLongRunningConnectionsCallback(process);
} else {
P_BUG("Unknown destination list");
}
}
/**
* Removes a process to the given list (enabledProcess, disablingProcesses, disabledProcesses).
* This function does not fix getWaitlist invariants or other stuff.
*/
void
Group::removeProcessFromList(const ProcessPtr &process, ProcessList &source) {
ProcessPtr p = process; // Keep an extra reference count just in case.
source.erase(source.begin() + process->getIndex());
process->setIndex(-1);
switch (process->enabled) {
case Process::ENABLED:
assert(&source == &enabledProcesses);
enabledCount--;
if (process->isTotallyBusy()) {
nEnabledProcessesTotallyBusy--;
}
break;
case Process::DISABLING:
assert(&source == &disablingProcesses);
disablingCount--;
break;
case Process::DISABLED:
assert(&source == &disabledProcesses);
disabledCount--;
break;
case Process::DETACHED:
assert(&source == &detachedProcesses);
break;
default:
P_BUG("Unknown 'enabled' state " << (int) process->enabled);
}
// Rebuild indices
ProcessList::iterator it, end = source.end();
unsigned int i = 0;
for (it = source.begin(); it != end; it++, i++) {
const ProcessPtr &process = *it;
process->setIndex(i);
}
// Rebuild enabledProcessBusynessLevels
if (&source == &enabledProcesses) {
enabledProcessBusynessLevels.clear();
for (it = source.begin(); it != end; it++, i++) {
const ProcessPtr &process = *it;
enabledProcessBusynessLevels.push_back(process->busyness());
}
enabledProcessBusynessLevels.shrink_to_fit();
}
}
void
Group::removeFromDisableWaitlist(const ProcessPtr &p, DisableResult result,
boost::container::vector &postLockActions)
{
const deque &disableWaitlist = this->disableWaitlist;
deque::const_iterator it, end = disableWaitlist.end();
deque newList;
for (it = disableWaitlist.begin(); it != end; it++) {
const DisableWaiter &waiter = *it;
const ProcessPtr process = waiter.process;
if (process == p) {
postLockActions.push_back(boost::bind(waiter.callback, p, result));
} else {
newList.push_back(waiter);
}
}
this->disableWaitlist = newList;
}
void
Group::clearDisableWaitlist(DisableResult result,
boost::container::vector &postLockActions)
{
// This function may be called after processes in the disableWaitlist
// have been disabled or enabled, so do not assume any value for
// waiter.process->enabled in this function.
postLockActions.reserve(postLockActions.size() + disableWaitlist.size());
while (!disableWaitlist.empty()) {
const DisableWaiter &waiter = disableWaitlist.front();
postLockActions.push_back(boost::bind(waiter.callback, waiter.process, result));
disableWaitlist.pop_front();
}
}
void
Group::enableAllDisablingProcesses(boost::container::vector &postLockActions) {
P_DEBUG("Enabling all DISABLING processes with result DR_ERROR");
deque::iterator it, end = disableWaitlist.end();
for (it = disableWaitlist.begin(); it != end; it++) {
const DisableWaiter &waiter = *it;
const ProcessPtr process = waiter.process;
// A process can appear multiple times in disableWaitlist.
assert(process->enabled == Process::DISABLING
|| process->enabled == Process::ENABLED);
if (process->enabled == Process::DISABLING) {
removeProcessFromList(process, disablingProcesses);
addProcessToList(process, enabledProcesses);
P_DEBUG("Enabled process " << process->inspect());
}
}
clearDisableWaitlist(DR_ERROR, postLockActions);
}
/**
* The `immediately` parameter only has effect if the detached processes checker
* thread is active. It means that, if the thread is currently sleeping, it should
* wake up immediately and perform work.
*/
void
Group::startCheckingDetachedProcesses(bool immediately) {
if (!detachedProcessesCheckerActive) {
P_DEBUG("Starting detached processes checker");
getPool()->nonInterruptableThreads.create_thread(
boost::bind(&Group::detachedProcessesCheckerMain, this, shared_from_this()),
"Detached processes checker: " + getName(),
POOL_HELPER_THREAD_STACK_SIZE
);
detachedProcessesCheckerActive = true;
} else if (detachedProcessesCheckerActive && immediately) {
detachedProcessesCheckerCond.notify_all();
}
}
void
Group::detachedProcessesCheckerMain(GroupPtr self) {
TRACE_POINT();
Pool *pool = getPool();
Pool::DebugSupportPtr debug = pool->debugSupport;
if (debug != NULL && debug->detachedProcessesChecker) {
debug->debugger->send("About to start detached processes checker");
debug->messages->recv("Proceed with starting detached processes checker");
}
boost::unique_lock lock(pool->syncher);
while (true) {
assert(detachedProcessesCheckerActive);
if (getLifeStatus() == SHUT_DOWN || boost::this_thread::interruption_requested()) {
UPDATE_TRACE_POINT();
P_DEBUG("Stopping detached processes checker");
detachedProcessesCheckerActive = false;
break;
}
UPDATE_TRACE_POINT();
if (!detachedProcesses.empty()) {
P_TRACE(2, "Checking whether any of the " << detachedProcesses.size() <<
" detached processes have exited...");
ProcessList::iterator it, end = detachedProcesses.end();
ProcessList processesToRemove;
for (it = detachedProcesses.begin(); it != end; it++) {
const ProcessPtr process = *it;
switch (process->getLifeStatus()) {
case Process::ALIVE:
if (process->canTriggerShutdown()) {
P_DEBUG("Detached process " << process->inspect() <<
" has 0 active sessions now. Triggering shutdown.");
process->triggerShutdown();
assert(process->getLifeStatus() == Process::SHUTDOWN_TRIGGERED);
}
break;
case Process::SHUTDOWN_TRIGGERED:
if (process->canCleanup()) {
P_DEBUG("Detached process " << process->inspect() << " has shut down. Cleaning up associated resources.");
process->cleanup();
assert(process->getLifeStatus() == Process::DEAD);
processesToRemove.push_back(process);
} else if (process->shutdownTimeoutExpired()) {
P_WARN("Detached process " << process->inspect() <<
" didn't shut down within " PROCESS_SHUTDOWN_TIMEOUT_DISPLAY
". Forcefully killing it with SIGKILL.");
kill(process->getPid(), SIGKILL);
}
break;
default:
P_BUG("Unknown 'lifeStatus' state " << (int) process->getLifeStatus());
}
}
UPDATE_TRACE_POINT();
end = processesToRemove.end();
for (it = processesToRemove.begin(); it != end; it++) {
removeProcessFromList(*it, detachedProcesses);
}
}
UPDATE_TRACE_POINT();
if (detachedProcesses.empty()) {
UPDATE_TRACE_POINT();
P_DEBUG("Stopping detached processes checker");
detachedProcessesCheckerActive = false;
boost::container::vector actions;
if (shutdownCanFinish()) {
UPDATE_TRACE_POINT();
finishShutdown(actions);
}
verifyInvariants();
verifyExpensiveInvariants();
lock.unlock();
UPDATE_TRACE_POINT();
runAllActions(actions);
break;
} else {
UPDATE_TRACE_POINT();
verifyInvariants();
verifyExpensiveInvariants();
}
// Not all processes can be shut down yet. Sleep for a while unless
// someone wakes us up.
UPDATE_TRACE_POINT();
detachedProcessesCheckerCond.timed_wait(lock,
posix_time::milliseconds(100));
}
}
/****************************
*
* Public methods
*
****************************/
/**
* Attaches the given process to this Group and mark it as enabled. This
* function doesn't touch `getWaitlist` so be sure to fix its invariants
* afterwards if necessary, e.g. by calling `assignSessionsToGetWaiters()`.
*/
AttachResult
Group::attach(const ProcessPtr &process,
boost::container::vector &postLockActions)
{
TRACE_POINT();
assert(process->getGroup() == NULL || process->getGroup() == this);
assert(process->isAlive());
assert(isAlive());
if (processUpperLimitsReached()) {
return AR_GROUP_UPPER_LIMITS_REACHED;
} else if (poolAtFullCapacity()) {
return AR_POOL_AT_FULL_CAPACITY;
} else if (!isWaitingForCapacity() && anotherGroupIsWaitingForCapacity()) {
return AR_ANOTHER_GROUP_IS_WAITING_FOR_CAPACITY;
}
process->initializeStickySessionId(generateStickySessionId());
if (options.forceMaxConcurrentRequestsPerProcess != -1) {
process->forceMaxConcurrency(options.forceMaxConcurrentRequestsPerProcess);
}
P_DEBUG("Attaching process " << process->inspect());
addProcessToList(process, enabledProcesses);
/* Now that there are enough resources, relevant processes in
* 'disableWaitlist' can be disabled.
*/
const deque &disableWaitlist = this->disableWaitlist;
deque::const_iterator it, end = disableWaitlist.end();
deque newDisableWaitlist;
for (it = disableWaitlist.begin(); it != end; it++) {
const DisableWaiter &waiter = *it;
const ProcessPtr process2 = waiter.process;
// The same process can appear multiple times in disableWaitlist.
assert(process2->enabled == Process::DISABLING
|| process2->enabled == Process::DISABLED);
if (process2->sessions == 0) {
if (process2->enabled == Process::DISABLING) {
P_DEBUG("Disabling DISABLING process " << process2->inspect() <<
"; disable command succeeded immediately");
removeProcessFromList(process2, disablingProcesses);
addProcessToList(process2, disabledProcesses);
} else {
P_DEBUG("Disabling (already disabled) DISABLING process " <<
process2->inspect() << "; disable command succeeded immediately");
}
postLockActions.push_back(boost::bind(waiter.callback, process2, DR_SUCCESS));
} else {
newDisableWaitlist.push_back(waiter);
}
}
this->disableWaitlist = newDisableWaitlist;
// Update GC sleep timer.
wakeUpGarbageCollector();
postLockActions.push_back(boost::bind(&Group::runAttachHooks, this, process));
return AR_OK;
}
/**
* Detaches the given process from this Group. This function doesn't touch
* getWaitlist so be sure to fix its invariants afterwards if necessary.
* `pool->detachProcessUnlocked()` does that so you should usually use
* that method over this one.
*/
void
Group::detach(const ProcessPtr &process, boost::container::vector &postLockActions) {
TRACE_POINT();
assert(process->getGroup() == this);
assert(process->isAlive());
assert(isAlive());
if (process->enabled == Process::DETACHED) {
P_DEBUG("Detaching process " << process->inspect() << ", which was already being detached");
return;
}
const ProcessPtr p = process; // Keep an extra reference just in case.
P_DEBUG("Detaching process " << process->inspect());
if (process->enabled == Process::ENABLED || process->enabled == Process::DISABLING) {
assert(enabledCount > 0 || disablingCount > 0);
if (process->enabled == Process::ENABLED) {
removeProcessFromList(process, enabledProcesses);
} else {
removeProcessFromList(process, disablingProcesses);
removeFromDisableWaitlist(process, DR_NOOP, postLockActions);
}
} else {
assert(process->enabled == Process::DISABLED);
assert(!disabledProcesses.empty());
removeProcessFromList(process, disabledProcesses);
}
addProcessToList(process, detachedProcesses);
startCheckingDetachedProcesses(false);
postLockActions.push_back(boost::bind(&Group::runDetachHooks, this, process));
}
/**
* Detaches all processes from this Group. This function doesn't touch
* getWaitlist so be sure to fix its invariants afterwards if necessary.
*/
void
Group::detachAll(boost::container::vector &postLockActions) {
assert(isAlive());
P_DEBUG("Detaching all processes in group " << info.name);
foreach (ProcessPtr process, enabledProcesses) {
addProcessToList(process, detachedProcesses);
}
foreach (ProcessPtr process, disablingProcesses) {
addProcessToList(process, detachedProcesses);
}
foreach (ProcessPtr process, disabledProcesses) {
addProcessToList(process, detachedProcesses);
}
enabledProcesses.clear();
disablingProcesses.clear();
disabledProcesses.clear();
enabledProcessBusynessLevels.clear();
enabledCount = 0;
disablingCount = 0;
disabledCount = 0;
nEnabledProcessesTotallyBusy = 0;
clearDisableWaitlist(DR_NOOP, postLockActions);
startCheckingDetachedProcesses(false);
}
/**
* Marks the given process as enabled. This function doesn't touch getWaitlist
* so be sure to fix its invariants afterwards if necessary.
*/
void
Group::enable(const ProcessPtr &process, boost::container::vector &postLockActions) {
assert(process->getGroup() == this);
assert(process->isAlive());
assert(isAlive());
if (process->enabled == Process::DISABLING) {
P_DEBUG("Enabling DISABLING process " << process->inspect());
removeProcessFromList(process, disablingProcesses);
addProcessToList(process, enabledProcesses);
removeFromDisableWaitlist(process, DR_CANCELED, postLockActions);
} else if (process->enabled == Process::DISABLED) {
P_DEBUG("Enabling DISABLED process " << process->inspect());
removeProcessFromList(process, disabledProcesses);
addProcessToList(process, enabledProcesses);
} else {
P_DEBUG("Enabling ENABLED process " << process->inspect());
}
}
/**
* Marks the given process as disabled. Returns DR_SUCCESS, DR_DEFERRED
* or DR_NOOP. If the result is DR_DEFERRED, then the callback will be
* called later with the result of this action.
*/
DisableResult
Group::disable(const ProcessPtr &process, const DisableCallback &callback) {
assert(process->getGroup() == this);
assert(process->isAlive());
assert(isAlive());
if (process->enabled == Process::ENABLED) {
P_DEBUG("Disabling ENABLED process " << process->inspect() <<
"; enabledCount=" << enabledCount << ", process.sessions=" << process->sessions);
assert(enabledCount >= 0);
if (enabledCount == 1 && !allowSpawn()) {
P_WARN("Cannot disable sole enabled process in group " << info.name <<
" because spawning is not allowed according to the current" <<
" configuration options");
return DR_ERROR;
} else if (enabledCount <= 1 || process->sessions > 0) {
removeProcessFromList(process, enabledProcesses);
addProcessToList(process, disablingProcesses);
disableWaitlist.push_back(DisableWaiter(process, callback));
if (enabledCount == 0) {
/* All processes are going to be disabled, so in order
* to avoid blocking requests we first spawn a new process
* and disable this process after the other one is done
* spawning. We do this irrespective of resource limits
* because this is an exceptional situation.
*/
P_DEBUG("Spawning a new process to avoid the disable action from blocking requests");
spawn();
}
P_DEBUG("Deferring disable command completion");
return DR_DEFERRED;
} else {
removeProcessFromList(process, enabledProcesses);
addProcessToList(process, disabledProcesses);
P_DEBUG("Disable command succeeded immediately");
return DR_SUCCESS;
}
} else if (process->enabled == Process::DISABLING) {
assert(disablingCount > 0);
disableWaitlist.push_back(DisableWaiter(process, callback));
P_DEBUG("Disabling DISABLING process " << process->inspect() <<
info.name << "; command queued, deferring disable command completion");
return DR_DEFERRED;
} else {
assert(disabledCount > 0);
P_DEBUG("Disabling DISABLED process " << process->inspect() <<
info.name << "; disable command succeeded immediately");
return DR_NOOP;
}
}
} // namespace ApplicationPool2
} // namespace Passenger
#endif // _PASSENGER_APPLICATION_POOL_GROUP_PROCESS_LIST_MANAGEMENT_CPP_