Viewing file:  SessionManagement.cpp (10.76 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
/*
* Phusion Passenger - https://www.phusionpassenger.com/
* Copyright (c) 2011-2017 Phusion Holding B.V.
*
* "Passenger", "Phusion Passenger" and "Union Station" are registered
* trademarks of Phusion Holding 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.
*/
#include <Core/ApplicationPool/Group.h>
/*************************************************************************
*
* Session management functions for ApplicationPool2::Group
*
*************************************************************************/
namespace Passenger {
namespace ApplicationPool2 {
using namespace std;
using namespace boost;
/****************************
*
* Private methods
*
****************************/
/* Determines which process to route a get() action to. The returned process
* is guaranteed to be `canBeRoutedTo()`, i.e. not totally busy.
*
* A request is routed to an enabled processes, or if there are none,
* from a disabling process. The rationale is as follows:
* If there are no enabled process, then waiting for one to spawn is too
* expensive. The next best thing is to route to disabling processes
* until more processes have been spawned.
*/
Group::RouteResult
Group::route(const Options &options) const {
if (OXT_LIKELY(enabledCount > 0)) {
if (options.stickySessionId == 0) {
Process *process = findEnabledProcessWithLowestBusyness();
if (process->canBeRoutedTo()) {
return RouteResult(process);
} else {
return RouteResult(NULL, true);
}
} else {
Process *process = findProcessWithStickySessionIdOrLowestBusyness(
options.stickySessionId);
if (process != NULL) {
if (process->canBeRoutedTo()) {
return RouteResult(process);
} else {
return RouteResult(NULL, false);
}
} else {
return RouteResult(NULL, true);
}
}
} else {
Process *process = findProcessWithLowestBusyness(disablingProcesses);
if (process->canBeRoutedTo()) {
return RouteResult(process);
} else {
return RouteResult(NULL, true);
}
}
}
SessionPtr
Group::newSession(Process *process, unsigned long long now) {
bool wasTotallyBusy = process->isTotallyBusy();
SessionPtr session = process->newSession(now);
session->onInitiateFailure = _onSessionInitiateFailure;
session->onClose = _onSessionClose;
if (process->enabled == Process::ENABLED) {
enabledProcessBusynessLevels[process->getIndex()] = process->busyness();
if (!wasTotallyBusy && process->isTotallyBusy()) {
nEnabledProcessesTotallyBusy++;
}
}
return session;
}
void
Group::_onSessionInitiateFailure(Session *session) {
Process *process = session->getProcess();
assert(process != NULL);
process->getGroup()->onSessionInitiateFailure(process, session);
}
void
Group::_onSessionClose(Session *session) {
Process *process = session->getProcess();
assert(process != NULL);
process->getGroup()->onSessionClose(process, session);
}
OXT_FORCE_INLINE void
Group::onSessionInitiateFailure(Process *process, Session *session) {
boost::container::vector<Callback> actions;
TRACE_POINT();
// Standard resource management boilerplate stuff...
Pool *pool = getPool();
boost::unique_lock<boost::mutex> lock(pool->syncher);
assert(process->isAlive());
assert(isAlive() || getLifeStatus() == SHUTTING_DOWN);
UPDATE_TRACE_POINT();
P_DEBUG("Could not initiate a session with process " <<
process->inspect() << ", detaching from pool if possible");
if (!pool->detachProcessUnlocked(process->shared_from_this(), actions)) {
P_DEBUG("Process was already detached");
}
pool->fullVerifyInvariants();
lock.unlock();
runAllActions(actions);
}
OXT_FORCE_INLINE void
Group::onSessionClose(Process *process, Session *session) {
TRACE_POINT();
// Standard resource management boilerplate stuff...
Pool *pool = getPool();
boost::unique_lock<boost::mutex> lock(pool->syncher);
assert(process->isAlive());
assert(isAlive() || getLifeStatus() == SHUTTING_DOWN);
P_TRACE(2, "Session closed for process " << process->inspect());
verifyInvariants();
UPDATE_TRACE_POINT();
/* Update statistics. */
bool wasTotallyBusy = process->isTotallyBusy();
process->sessionClosed(session);
assert(process->getLifeStatus() == Process::ALIVE);
assert(process->enabled == Process::ENABLED
|| process->enabled == Process::DISABLING
|| process->enabled == Process::DETACHED);
if (process->enabled == Process::ENABLED) {
enabledProcessBusynessLevels[process->getIndex()] = process->busyness();
if (wasTotallyBusy) {
assert(nEnabledProcessesTotallyBusy >= 1);
nEnabledProcessesTotallyBusy--;
}
}
/* This group now has a process that's guaranteed to be not
* totally busy.
*/
assert(!process->isTotallyBusy());
bool detachingBecauseOfMaxRequests = false;
bool detachingBecauseCapacityNeeded = false;
bool shouldDetach =
( detachingBecauseOfMaxRequests = (
options.maxRequests > 0
&& process->processed >= options.maxRequests
)) || (
detachingBecauseCapacityNeeded = (
process->sessions == 0
&& getWaitlist.empty()
&& (
!pool->getWaitlist.empty()
|| anotherGroupIsWaitingForCapacity()
)
)
);
bool shouldDisable =
process->enabled == Process::DISABLING
&& process->sessions == 0
&& enabledCount > 0;
if (shouldDetach || shouldDisable) {
UPDATE_TRACE_POINT();
boost::container::vector<Callback> actions;
if (shouldDetach) {
if (detachingBecauseCapacityNeeded) {
/* Someone might be trying to get() a session for a different
* group that couldn't be spawned because of lack of pool capacity.
* If this group isn't under sufficiently load (as apparent by the
* checked conditions) then now's a good time to detach
* this process or group in order to free capacity.
*/
P_DEBUG("Process " << process->inspect() << " is no longer totally "
"busy; detaching it in order to make room in the pool");
} else {
/* This process has processed its maximum number of requests,
* so we detach it.
*/
P_DEBUG("Process " << process->inspect() <<
" has reached its maximum number of requests (" <<
options.maxRequests << "); detaching it");
}
pool->detachProcessUnlocked(process->shared_from_this(), actions);
} else {
ProcessPtr processPtr = process->shared_from_this();
removeProcessFromList(processPtr, disablingProcesses);
addProcessToList(processPtr, disabledProcesses);
removeFromDisableWaitlist(processPtr, DR_SUCCESS, actions);
maybeInitiateOobw(process);
}
pool->fullVerifyInvariants();
lock.unlock();
runAllActions(actions);
} else {
UPDATE_TRACE_POINT();
// This could change process->enabled.
maybeInitiateOobw(process);
if (!getWaitlist.empty() && process->enabled == Process::ENABLED) {
/* If there are clients on this group waiting for a process to
* become available then call them now.
*/
UPDATE_TRACE_POINT();
// Already calls verifyInvariants().
assignSessionsToGetWaitersQuickly(lock);
}
}
}
/****************************
*
* Public methods
*
****************************/
SessionPtr
Group::get(const Options &newOptions, const GetCallback &callback,
boost::container::vector<Callback> &postLockActions)
{
assert(isAlive());
if (OXT_LIKELY(!restarting())) {
if (OXT_UNLIKELY(needsRestart(newOptions))) {
restart(newOptions);
} else {
mergeOptions(newOptions);
}
if (OXT_UNLIKELY(!newOptions.noop && shouldSpawnForGetAction())) {
// If we're trying to spawn the first process for this group, and
// spawning failed because the pool is at full capacity, then we
// try to kill some random idle process in the pool and try again.
if (spawn() == SR_ERR_POOL_AT_FULL_CAPACITY && enabledCount == 0) {
P_INFO("Unable to spawn the the sole process for group " << info.name <<
" because the max pool size has been reached. Trying " <<
"to shutdown another idle process to free capacity...");
if (poolForceFreeCapacity(this, postLockActions) != NULL) {
SpawnResult result = spawn();
assert(result == SR_OK);
(void) result;
} else {
P_INFO("There are no processes right now that are eligible "
"for shutdown. Will try again later.");
}
}
}
}
if (OXT_UNLIKELY(newOptions.noop)) {
return nullProcess->createSessionObject((Socket *) NULL);
}
if (OXT_UNLIKELY(enabledCount == 0)) {
/* We don't have any processes yet, but they're on the way.
*
* We have some choices here. If there are disabling processes
* then we generally want to use them, except:
* - When non-rolling restarting because those disabling processes
* are from the old version.
* - When all disabling processes are totally busy.
*
* Whenever a disabling process cannot be used, call the callback
* after a process has been spawned or has failed to spawn, or
* when a disabling process becomes available.
*/
assert(m_spawning || restarting() || poolAtFullCapacity());
if (disablingCount > 0 && !restarting()) {
Process *process = findProcessWithLowestBusyness(disablingProcesses);
assert(process != NULL);
if (!process->isTotallyBusy()) {
return newSession(process, newOptions.currentTime);
}
}
if (pushGetWaiter(newOptions, callback, postLockActions)) {
P_DEBUG("No session checked out yet: group is spawning or restarting");
}
return SessionPtr();
} else {
RouteResult result = route(newOptions);
if (result.process == NULL) {
/* Looks like all processes are totally busy.
* Wait until a new one has been spawned or until
* resources have become free.
*/
if (pushGetWaiter(newOptions, callback, postLockActions)) {
P_DEBUG("No session checked out yet: all processes are at full capacity");
}
return SessionPtr();
} else {
P_DEBUG("Session checked out from process " << result.process->inspect());
return newSession(result.process, newOptions.currentTime);
}
}
}
} // namespace ApplicationPool2
} // namespace Passenger
|