KOKINIO - MANAGER

Edit File: AgentWatcher.cpp

/*
 *  Phusion Passenger - https://www.phusionpassenger.com/
 *  Copyright (c) 2010-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.
 */

/**
 * Abstract base class for watching agent processes.
 */
class AgentWatcher: public boost::enable_shared_from_this<AgentWatcher> {
private:
	/** The watcher thread. */
	oxt::thread *thr;

void threadMain(boost::shared_ptr<AgentWatcher> self) {
		try {
			pid_t pid, ret;
			int status, e;

while (!boost::this_thread::interruption_requested()) {
				{
					boost::lock_guard<boost::mutex> l(lock);
					pid = this->pid;
				}

// Process can be started before the watcher thread is launched.
				if (pid == 0) {
					pid = start();
				}
				ret = syscalls::waitpid(pid, &status, 0);
				if (ret == -1 && errno == ECHILD) {
					/* If the agent is attached to gdb then waitpid()
					 * here can return -1 with errno == ECHILD.
					 * Fallback to kill() polling for checking
					 * whether the agent is alive.
					 */
					ret = pid;
					status = 0;
					P_WARN("waitpid() on " << name() << " (pid=" << pid <<
						") returned -1 with " <<
						"errno = ECHILD, falling back to kill polling");
					waitpidUsingKillPolling(pid);
					e = 0;
				} else {
					e = errno;
				}

{
					boost::lock_guard<boost::mutex> l(lock);
					this->pid = 0;
				}

boost::this_thread::disable_interruption di;
				boost::this_thread::disable_syscall_interruption dsi;
				if (ret == -1) {
					P_WARN(name() << " (pid=" << pid << ") crashed or killed for "
						"an unknown reason (errno = " <<
						strerror(e) << "), restarting it...");
				} else if (WIFEXITED(status)) {
					if (WEXITSTATUS(status) == 0) {
						/* When the web server is gracefully exiting, it will
						 * tell one or more agents to gracefully exit with exit
						 * status 0. If we see this then it means the watchdog
						 * is gracefully shutting down too and we should stop
						 * watching.
						 */
						return;
					} else {
						P_WARN(name() << " (pid=" << pid <<
							") crashed with exit status " <<
							WEXITSTATUS(status) << ", restarting it...");
					}
				} else {
					P_WARN(name() << " (pid=" << pid <<
						") crashed with signal " <<
						getSignalName(WTERMSIG(status)) <<
						", restarting it...");
				}

const char *sleepTime;
				if ((sleepTime = getenv("PASSENGER_AGENT_RESTART_SLEEP")) != NULL) {
					sleep(atoi(sleepTime));
				}
			}
		} catch (const boost::thread_interrupted &) {
		} catch (const tracable_exception &e) {
			boost::lock_guard<boost::mutex> l(lock);
			threadExceptionMessage = e.what();
			threadExceptionBacktrace = e.backtrace();
			wo->errorEvent.notify();
		} catch (const std::exception &e) {
			boost::lock_guard<boost::mutex> l(lock);
			threadExceptionMessage = e.what();
			wo->errorEvent.notify();
		} catch (...) {
			boost::lock_guard<boost::mutex> l(lock);
			threadExceptionMessage = "Unknown error";
			wo->errorEvent.notify();
		}
	}

protected:
	/** PID of the process we're watching. 0 if no process is started at this time. */
	pid_t pid;

/** If the watcher thread threw an uncaught exception then its information will
	 * be stored here so that the main thread can check whether a watcher encountered
	 * an error. These are empty strings if everything is OK.
	 */
	string threadExceptionMessage;
	string threadExceptionBacktrace;

/** The agent process's feedback fd. */
	FileDescriptor feedbackFd;

/**
	 * Lock for protecting the exchange of data between the main thread and
	 * the watcher thread.
	 */
	mutable boost::mutex lock;

WorkingObjectsPtr wo;

/**
	 * Returns the filename of the agent process's executable. This method may be
	 * called in a forked child process and may therefore not allocate memory.
	 */
	virtual string getExeFilename() const = 0;

/**
	 * This method is to exec() the agent with the right arguments.
	 * It is called from within a forked child process, so don't do any dynamic
	 * memory allocations in here. It must also not throw any exceptions.
	 * It must also preserve the value of errno after exec() is called.
	 */
	virtual void execProgram() const {
		execl(getExeFilename().c_str(),
			getExeFilename().c_str(),
			"3",  // feedback fd
			(char *) 0);
	}

/**
	 * This method is to send startup arguments to the agent process through
	 * the given file descriptor, which is the agent process's feedback fd.
	 * May throw arbitrary exceptions.
	 */
	virtual void sendStartupArguments(pid_t pid, FileDescriptor &fd) = 0;

/**
	 * This method is to process the startup info that the agent process has
	 * sent back. May throw arbitrary exceptions.
	 */
	virtual bool processStartupInfo(pid_t pid, FileDescriptor &fd, const vector<string> &args) = 0;

/**
	 * Kill a process (but not its children) with SIGTERM.
	 * Does not wait until it has quit.
	 */
	static void killAndDontWait(pid_t pid) {
		boost::this_thread::disable_interruption di;
		boost::this_thread::disable_syscall_interruption dsi;
		syscalls::kill(pid, SIGTERM);
	}

/**
	 * Kill a process with SIGKILL, and attempt to kill its children too.
	 * Then wait until it has quit.
	 */
	static void killProcessGroupAndWait(pid_t pid) {
		boost::this_thread::disable_interruption di;
		boost::this_thread::disable_syscall_interruption dsi;
		// If the process is a process group leader then killing the
		// group will likely kill all its child processes too.
		if (syscalls::killpg(pid, SIGKILL) == -1) {
			syscalls::kill(pid, SIGKILL);
		}
		syscalls::waitpid(pid, NULL, 0);
	}

/**
	 * Behaves like <tt>waitpid(pid, status, WNOHANG)</tt>, but waits at most
	 * <em>timeout</em> miliseconds for the process to exit.
	 */
	static int timedWaitPid(pid_t pid, int *status, unsigned long long timeout) {
		Timer<SystemTime::GRAN_10MSEC> timer;
		int ret;

do {
			ret = syscalls::waitpid(pid, status, WNOHANG);
			if (ret > 0 || ret == -1) {
				return ret;
			} else {
				syscalls::usleep(10000);
			}
		} while (timer.elapsed() < timeout);
		return 0; // timed out
	}

static void waitpidUsingKillPolling(pid_t pid) {
		bool done = false;

while (!done) {
			int ret = syscalls::kill(pid, 0);
			done = ret == -1;
			if (!done) {
				syscalls::usleep(20000);
			}
		}
	}

public:
	AgentWatcher(const WorkingObjectsPtr &wo) {
		thr = NULL;
		pid = 0;
		this->wo = wo;
	}

virtual ~AgentWatcher() {
		delete thr;
	}

/**
	 * Store information about the started agent process in the given report object.
	 * May throw arbitrary exceptions.
	 *
	 * @pre start() has been called and succeeded.
	 */
	virtual void reportAgentStartupResult(Json::Value &report) = 0;

/** Returns the name of the agent that this class is watching. */
	virtual const char *name() const = 0;

/**
	 * Starts the agent process. May throw arbitrary exceptions.
	 */
	virtual pid_t start() {
		boost::this_thread::disable_interruption di;
		boost::this_thread::disable_syscall_interruption dsi;
		string exeFilename = getExeFilename();
		SocketPair fds;
		int e, ret;
		pid_t pid;

/* Create feedback fd for this agent process. We'll send some startup
		 * arguments to this agent process through this fd, and we'll receive
		 * startup information through it as well.
		 */
		fds = createUnixSocketPair(__FILE__, __LINE__);

pid = syscalls::fork();
		if (pid == 0) {
			// Child

/* Make sure file descriptor FEEDBACK_FD refers to the newly created
			 * feedback fd (fds[1]) and close all other file descriptors.
			 * In this child process we don't care about the original FEEDBACK_FD
			 * (which is the Watchdog's communication channel to the agents starter.)
			 *
			 * fds[1] is guaranteed to be != FEEDBACK_FD because the watchdog
			 * is started with FEEDBACK_FD already assigned.
			 */
			syscalls::close(fds[0]);

if (syscalls::dup2(fds[1], FEEDBACK_FD) == -1) {
				/* Something went wrong, report error through feedback fd. */
				e = errno;
				try {
					writeArrayMessage(fds[1],
						"system error before exec",
						"dup2() failed",
						toString(e).c_str(),
						NULL);
					_exit(1);
				} catch (...) {
					fprintf(stderr, "PassengerWatchdog: dup2() failed: %s (%d)\n",
						strerror(e), e);
					fflush(stderr);
					_exit(1);
				}
			}

resetSignalHandlersAndMask();
			closeAllFileDescriptors(FEEDBACK_FD);

/* Become the process group leader so that the watchdog can kill the
			 * agent as well as all its descendant processes, and so that a Ctrl-C
			 * only affects the watchdog but not agents. */
			setpgid(getpid(), getpid());

try {
				execProgram();
			} catch (...) {
				fprintf(stderr, "PassengerWatchdog: execProgram() threw an exception\n");
				fflush(stderr);
				_exit(1);
			}
			e = errno;
			try {
				writeArrayMessage(FEEDBACK_FD,
					"exec error",
					toString(e).c_str(),
					NULL);
			} catch (...) {
				fprintf(stderr, "PassengerWatchdog: could not execute %s: %s (%d)\n",
					exeFilename.c_str(), strerror(e), e);
				fflush(stderr);
			}
			_exit(1);
		} else if (pid == -1) {
			// Error
			e = errno;
			throw SystemException("Cannot fork a new process", e);
		} else {
			// Parent
			FileDescriptor feedbackFd(fds[0]);
			vector<string> args;

fds[1].close();
			boost::this_thread::restore_interruption ri(di);
			boost::this_thread::restore_syscall_interruption rsi(dsi);
			ScopeGuard failGuard(boost::bind(killProcessGroupAndWait, pid));

/* Send startup arguments. Ignore EPIPE and ECONNRESET here
			 * because the child process might have sent an feedback message
			 * without reading startup arguments.
			 */
			try {
				sendStartupArguments(pid, feedbackFd);
			} catch (const SystemException &ex) {
				if (ex.code() != EPIPE && ex.code() != ECONNRESET) {
					throw SystemException(string("Unable to start the ") + name() +
						": an error occurred while sending startup arguments",
						ex.code());
				}
			}

// Now read its feedback.
			try {
				ret = readArrayMessage(feedbackFd, args);
			} catch (const SystemException &e) {
				if (e.code() == ECONNRESET) {
					ret = false;
				} else {
					throw SystemException(string("Unable to start the ") + name() +
						": unable to read its startup information",
						e.code());
				}
			}
			if (!ret) {
				boost::this_thread::disable_interruption di2;
				boost::this_thread::disable_syscall_interruption dsi2;
				int status;

/* The feedback fd was prematurely closed for an unknown reason.
				 * Did the agent process crash?
				 *
				 * We use timedWaitPid() here because if the process crashed
				 * because of an uncaught exception, the file descriptor
				 * might be closed before the process has printed an error
				 * message, so we give it some time to print the error
				 * before we kill it.
				 */
				ret = timedWaitPid(pid, &status, 5000);
				if (ret == 0) {
					/* Doesn't look like it; it seems it's still running.
					 * We can't do anything without proper feedback so kill
					 * the agent process and throw an exception.
					 */
					failGuard.runNow();
					throw RuntimeException(string("Unable to start the ") + name() +
						": it froze and reported an unknown error during its startup");
				} else if (ret != -1 && WIFSIGNALED(status)) {
					/* Looks like a crash which caused a signal. */
					throw RuntimeException(string("Unable to start the ") + name() +
						": it seems to have been killed with signal " +
						getSignalName(WTERMSIG(status)) + " during startup");
				} else if (ret == -1) {
					/* Looks like it exited after detecting an error. */
					throw RuntimeException(string("Unable to start the ") + name() +
						": it seems to have crashed during startup for an unknown reason");
				} else {
					/* Looks like it exited after detecting an error, but has an exit code. */
					throw RuntimeException(string("Unable to start the ") + name() +
						": it seems to have crashed during startup for an unknown reason, "
						"with exit code " + toString(WEXITSTATUS(status)));
				}
			}

if (args[0] == "system error before exec") {
				throw SystemException(string("Unable to start the ") + name() +
					": " + args[1], atoi(args[2]));
			} else if (args[0] == "exec error") {
				e = atoi(args[1]);
				if (e == ENOENT) {
					throw RuntimeException(string("Unable to start the ") + name() +
						" because its executable (" + getExeFilename() + ") "
						"doesn't exist. This probably means that your "
						PROGRAM_NAME " installation is broken or "
						"incomplete. Please reinstall " PROGRAM_NAME);
				} else {
					throw SystemException(string("Unable to start the ") + name() +
						" because exec(\"" + getExeFilename() + "\") failed",
						atoi(args[1]));
				}
			} else if (!processStartupInfo(pid, feedbackFd, args)) {
				throw RuntimeException(string("The ") + name() +
					" sent an unknown startup info message '" +
					args[0] + "'");
			}

boost::lock_guard<boost::mutex> l(lock);
			this->feedbackFd = feedbackFd;
			this->pid = pid;
			failGuard.clear();
			return pid;
		}
	}

/**
	 * Begin watching the agent process.
	 *
	 * @pre start() has been called and succeeded.
	 * @pre This watcher isn't already watching.
	 * @throws RuntimeException If a precondition failed.
	 * @throws thread_interrupted
	 * @throws thread_resource_error
	 */
	virtual void beginWatching() {
		boost::lock_guard<boost::mutex> l(lock);
		if (pid == 0) {
			throw RuntimeException("start() hasn't been called yet");
		}
		if (thr != NULL) {
			throw RuntimeException("Already started watching.");
		}

thr = new oxt::thread(boost::bind(&AgentWatcher::threadMain, this, shared_from_this()),
			name(), 256 * 1024);
	}

static void stopWatching(vector< boost::shared_ptr<AgentWatcher> > &watchers) {
		vector< boost::shared_ptr<AgentWatcher> >::const_iterator it;
		vector<oxt::thread *> threads;
		unsigned int i = 0;

for (it = watchers.begin(); it != watchers.end(); it++, i++) {
			threads.push_back((*it)->thr);
			threads[i] = (*it)->thr;
		}

oxt::thread::interrupt_and_join_multiple(&threads[0], threads.size());
		for (it = watchers.begin(); it != watchers.end(); it++, i++) {
			delete (*it)->thr;
			(*it)->thr = NULL;
		}
	}

/**
	 * Tell the agent process to gracefully shut down. Returns true if it
	 * was signaled, or false if it wasn't started.
	 */
	virtual bool signalShutdown() {
		boost::lock_guard<boost::mutex> l(lock);
		if (pid == 0) {
			return false;
		} else {
			killAndDontWait(pid);
			return true;
		}
	}

/**
	 * Force the agent process to shut down. Returns true if it was shut down,
	 * or false if it wasn't started.
	 */
	virtual bool forceShutdown() {
		boost::lock_guard<boost::mutex> l(lock);
		if (pid == 0) {
			return false;
		} else {
			killProcessGroupAndWait(pid);
			this->pid = 0;
			return true;
		}
	}

/**
	 * If the watcher thread has encountered an error, then the error message
	 * will be stored here. If the error message is empty then it means
	 * everything is still OK.
	 */
	string getErrorMessage() const {
		boost::lock_guard<boost::mutex> l(lock);
		return threadExceptionMessage;
	}

/**
	 * The error backtrace, if applicable.
	 */
	string getErrorBacktrace() const {
		boost::lock_guard<boost::mutex> l(lock);
		return threadExceptionBacktrace;
	}

/**
	 * Returns the agent process feedback fd, or -1 if the agent process
	 * hasn't been started yet. Can be used to check whether this agent process
	 * has exited without using waitpid().
	 */
	const FileDescriptor getFeedbackFd() const {
		boost::lock_guard<boost::mutex> l(lock);
		return feedbackFd;
	}
};

typedef boost::shared_ptr<AgentWatcher> AgentWatcherPtr;