/*
 *  Phusion Passenger - https://www.phusionpassenger.com/
 *  Copyright (c) 2010-2013 Phusion
 *
 *  "Phusion Passenger" is a trademark of Hongli Lai & Ninh Bui.
 *
 *  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 enable_shared_from_this<AgentWatcher> {
private:
	/** The watcher thread. */
	oxt::thread *thr;
	
	void threadMain(shared_ptr<AgentWatcher> self) {
		try {
			pid_t pid, ret;
			int status, e;
			
			while (!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;
				}
				
				this_thread::disable_interruption di;
				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();
			errorEvent->notify();
		} catch (const std::exception &e) {
			boost::lock_guard<boost::mutex> l(lock);
			threadExceptionMessage = e.what();
			errorEvent->notify();
		} catch (...) {
			boost::lock_guard<boost::mutex> l(lock);
			threadExceptionMessage = "Unknown error";
			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;
	
	/**
	 * 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 with SIGKILL, and attempt to kill its children too. 
	 * Then wait until it has quit.
	 */
	static void killAndWait(pid_t pid) {
		this_thread::disable_interruption di;
		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 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() {
		thr = NULL;
		pid = 0;
	}
	
	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 reportAgentsInformation(VariantMap &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() {
		this_thread::disable_interruption di;
		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();
		
		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, "Passenger Watchdog: dup2() failed: %s (%d)\n",
						strerror(e), e);
					fflush(stderr);
					_exit(1);
				}
			}
			
			closeAllFileDescriptors(FEEDBACK_FD);
			
			/* Become the process group leader so that the watchdog can kill the
			 * agent as well as all its descendant processes. */
			setpgid(getpid(), getpid());
			
			setOomScore(oldOomScore);
			
			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, "Passenger Watchdog: 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();
			this_thread::restore_interruption ri(di);
			this_thread::restore_syscall_interruption rsi(dsi);
			ScopeGuard failGuard(boost::bind(killAndWait, 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) {
				this_thread::disable_interruption di2;
				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 "
						"Phusion Passenger installation is broken or "
						"incomplete. Please reinstall Phusion Passenger");
				} 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;
		}
	}
	
	/**
	 * Start 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 startWatching() {
		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< shared_ptr<AgentWatcher> > &watchers) {
		vector< shared_ptr<AgentWatcher> >::const_iterator it;
		oxt::thread *threads[watchers.size()];
		unsigned int i = 0;
		
		for (it = watchers.begin(); it != watchers.end(); it++, i++) {
			threads[i] = (*it)->thr;
		}
		
		oxt::thread::interrupt_and_join_multiple(threads, watchers.size());
		for (it = watchers.begin(); it != watchers.end(); it++, i++) {
			delete (*it)->thr;
			(*it)->thr = NULL;
		}
	}
	
	/**
	 * 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 {
			killAndWait(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 shared_ptr<AgentWatcher> AgentWatcherPtr;