ApplicationPoolStatusReporter.h in passenger-2.2.3

- old
+ new

@@ -31,17 +31,20 @@
 #include <oxt/backtrace.hpp>
 #include <oxt/system_calls.hpp>
 
 #include <string>
 #include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
 #include <sys/stat.h>
 #include <cstdio>
 #include <unistd.h>
 #include <errno.h>
 
 #include "StandardApplicationPool.h"
 #include "MessageChannel.h"
+#include "Exceptions.h"
 #include "Logging.h"
 #include "Utils.h"
 
 namespace Passenger {
 
@@ -49,76 +52,180 @@
 using namespace oxt;
 using namespace std;
 
 /**
  * An ApplicationPoolStatusReporter allows commandline admin tools to inspect
- * the status of a StandardApplicationPool. It does so by creating a FIFO
- * in the Passenger temp folder.
+ * the status of a StandardApplicationPool. It does so by creating a Unix socket
+ * in the Passenger temp folder, which tools can connect to to query for
+ * information.
  *
- * An ApplicationPoolStatusReporter creates a background thread, which
- * continuously sends new information through the FIFO. This thread will
- * be automatically cleaned up upon destroying the ApplicationPoolStatusReporter
- * object.
+ * An ApplicationPoolStatusReporter creates a background thread for handling
+ * connections on the socket. This thread will be automatically cleaned up upon
+ * destroying the ApplicationPoolStatusReporter object.
  */
 class ApplicationPoolStatusReporter {
 private:
+	/**
+	 * Wrapper class around a file descriptor integer, for RAII behavior.
+	 *
+	 * A FileDescriptor object behaves just like an int, so that you can pass it to
+	 * system calls such as read(). It performs reference counting. When the last
+	 * copy of a FileDescriptor has been destroyed, the underlying file descriptor
+	 * will be automatically closed.
+	 */
+	class FileDescriptor {
+	private:
+		struct SharedData {
+			int fd;
+
+			/**
+			 * Constructor to assign this file descriptor's handle.
+			 */
+			SharedData(int fd) {
+				this->fd = fd;
+			}
+
+			/**
+			 * Attempts to close this file descriptor. When created on the stack,
+			 * this destructor will automatically be invoked as a result of C++
+			 * semantics when exiting the scope this object was created in. This
+			 * ensures that stack created objects with destructors like these will
+			 * de-allocate their resources upon leaving their corresponding scope.
+			 * This pattern is also known Resource Acquisition Is Initialization (RAII).
+			 *
+			 * @throws SystemException File descriptor could not be closed.
+			 */
+			~SharedData() {
+				this_thread::disable_syscall_interruption dsi;
+				if (syscalls::close(fd) == -1) {
+					throw SystemException("Cannot close file descriptor", errno);
+				}
+			}
+		};
+
+		/* Shared pointer for reference counting on this file descriptor */
+		shared_ptr<SharedData> data;
+
+	public:
+		FileDescriptor() {
+			// Do nothing.
+		}
+
+		/**
+		 * Creates a new FileDescriptor instance with the given fd as a handle.
+		 */
+		FileDescriptor(int fd) {
+			data = ptr(new SharedData(fd));
+		}
+
+		/**
+		 * Overloads the integer cast operator so that it will return the file
+		 * descriptor handle as an integer.
+		 *
+		 * @return This file descriptor's handle as an integer.
+		 */
+		operator int () const {
+			return data->fd;
+		}
+	};
+	
 	/** The application pool to monitor. */
 	StandardApplicationPoolPtr pool;
 	
-	/** The FIFO's filename. */
+	/** The socket's filename. */
 	char filename[PATH_MAX];
 	
-	/** The background thread. */
-	oxt::thread *thr;
+	/** The socket's file descriptor. */
+	int serverFd;
 	
-	void threadMain() {
+	/** The main thread. */
+	oxt::thread *mainThread;
+	
+	/** The mutex which protects the 'threads' member. */
+	boost::mutex threadsLock;
+	
+	/** A map which maps a client file descriptor to its handling thread. */
+	map< int, shared_ptr<oxt::thread> > threads;
+	
+	void writeScalarAndIgnoreErrors(MessageChannel &channel, const string &data) {
+		try {
+			channel.writeScalar(data);
+		} catch (const SystemException &e) {
+			// Don't care about write errors.
+		}
+	}
+	
+	void mainThreadFunction() {
 		TRACE_POINT();
 		try {
 			while (!this_thread::interruption_requested()) {
-				struct stat buf;
-				int ret;
-				
 				UPDATE_TRACE_POINT();
-				do {
-					ret = stat(filename, &buf);
-				} while (ret == -1 && errno == EINTR);
-				if (ret == -1 || !S_ISFIFO(buf.st_mode)) {
-					// Something bad happened with the status
-					// report FIFO, so we bail out.
-					break;
-				}
+				sockaddr_un addr;
+				socklen_t addr_len = sizeof(addr);
 				
-				UPDATE_TRACE_POINT();
-				FILE *f = syscalls::fopen(filename, "w");
-				if (f == NULL) {
+				FileDescriptor fd(syscalls::accept(serverFd, (struct sockaddr *) &addr, &addr_len));
+				if (fd == -1) {
 					int e = errno;
-					P_ERROR("Cannot open status report FIFO " <<
-						filename << ": " <<
+					P_ERROR("Cannot accept new client on status reporter socket: " <<
 						strerror(e) << " (" << e << ")");
 					break;
 				}
 				
-				UPDATE_TRACE_POINT();
-				MessageChannel channel(fileno(f));
-				string report;
-				report.append("----------- Backtraces -----------\n");
-				report.append(oxt::thread::all_backtraces());
-				report.append("\n\n");
-				report.append(pool->toString());
+				boost::lock_guard<boost::mutex> l(threadsLock);
+				this_thread::disable_syscall_interruption dsi;
+				this_thread::disable_interruption di;
+				shared_ptr<oxt::thread> thread(new oxt::thread(
+					bind(&ApplicationPoolStatusReporter::clientThreadFunction, this, fd),
+					"Status reporter client thread " + toString(fd),
+					1024 * 128
+				));
+				threads[fd] = thread;
+			}
+		} catch (const boost::thread_interrupted &) {
+			P_TRACE(2, "Status reporter main thread interrupted.");
+		} catch (const exception &e) {
+			P_ERROR("Error in status reporter main thread: " << e.what());
+		}
+	}
+	
+	void clientThreadFunction(FileDescriptor fd) {
+		TRACE_POINT();
+		MessageChannel channel(fd);
+		
+		try {
+			while (!this_thread::interruption_requested()) {
+				vector<string> args;
 				
 				UPDATE_TRACE_POINT();
-				try {
-					channel.writeScalar(report);
-					channel.writeScalar(pool->toXml());
-				} catch (...) {
-					// Ignore write errors.
+				if (!channel.read(args) || args.size() < 1) {
+					break;
 				}
-				syscalls::fclose(f);
+				
+				if (args[0] == "backtraces") {
+					UPDATE_TRACE_POINT();
+					writeScalarAndIgnoreErrors(channel, oxt::thread::all_backtraces());
+				} else if (args[0] == "status") {
+					UPDATE_TRACE_POINT();
+					writeScalarAndIgnoreErrors(channel, pool->toString());
+				} else if (args[0] == "status_xml") {
+					UPDATE_TRACE_POINT();
+					writeScalarAndIgnoreErrors(channel, pool->toXml());
+				} else {
+					P_ERROR("Error in status reporter client thread: unknown query '" <<
+						args[0] << "'.");
+				}
 			}
 		} catch (const boost::thread_interrupted &) {
-			P_TRACE(2, "Status report thread interrupted.");
+			P_TRACE(2, "Status reporter client thread " << fd << " interrupted.");
+		} catch (const exception &e) {
+			P_ERROR("Error in status reporter client thread: " << e.what());
 		}
+		
+		boost::lock_guard<boost::mutex> l(threadsLock);
+		this_thread::disable_syscall_interruption dsi;
+		this_thread::disable_interruption di;
+		threads.erase(fd);
 	}
 
 public:
 	/**
 	 * Creates a new ApplicationPoolStatusReporter.
@@ -132,13 +239,15 @@
 	 *                    be created.
 	 * @param uid The UID of the user who should own the FIFO file, or
 	 *            -1 if the current user should be set as owner.
 	 * @param gid The GID of the user who should own the FIFO file, or
 	 *            -1 if the current group should be set as group.
-	 * @throws SystemException An error occurred while creating the FIFO.
+	 * @throws RuntimeException An error occurred.
+	 * @throws SystemException An error occurred while creating the server socket.
 	 * @throws boost::thread_resource_error Something went wrong during
 	 *     creation of the thread.
+	 * @throws boost::thread_interrupted A system call has been interrupted.
 	 */
 	ApplicationPoolStatusReporter(StandardApplicationPoolPtr &pool,
 	                              bool userSwitching,
 	                              mode_t permissions = S_IRUSR | S_IWUSR,
 	                              uid_t uid = -1, gid_t gid = -1) {
@@ -147,64 +256,79 @@
 		this->pool = pool;
 		
 		createPassengerTempDir(getSystemTempDir(), userSwitching,
 			"nobody", geteuid(), getegid());
 		
-		snprintf(filename, sizeof(filename) - 1, "%s/info/status.fifo",
+		snprintf(filename, sizeof(filename) - 1, "%s/info/status.socket",
 			getPassengerTempDir().c_str());
 		filename[PATH_MAX - 1] = '\0';
+		serverFd = createUnixServer(filename, 10);
 		
+		/* Set the socket file's permissions... */
 		do {
-			ret = mkfifo(filename, permissions);
-		} while (ret == -1 && errno == EINTR);
-		if (ret == -1 && errno != EEXIST) {
-			int e = errno;
-			string message("Cannot create FIFO '");
-			message.append(filename);
-			message.append("'");
-			throw SystemException(message, e);
-		}
-		
-		// It seems that the permissions passed to mkfifo()
-		// aren't respected, so here we chmod the file.
-		do {
 			ret = chmod(filename, permissions);
 		} while (ret == -1 && errno == EINTR);
 		
+		/* ...and ownership. */
 		if (uid != (uid_t) -1 && gid != (gid_t) -1) {
 			do {
 				ret = chown(filename, uid, gid);
 			} while (ret == -1 && errno == EINTR);
 			if (errno == -1) {
 				int e = errno;
 				char message[1024];
 				
 				snprintf(message, sizeof(message) - 1,
-					"Cannot set the FIFO file '%s' its owner to %lld and group to %lld",
+					"Cannot set the owner for socket file '%s' to %lld and its group to %lld",
 					filename, (long long) uid, (long long) gid);
 				message[sizeof(message) - 1] = '\0';
+				do {
+					ret = close(serverFd);
+				} while (ret == -1 && errno == EINTR);
 				throw SystemException(message, e);
 			}
 		}
 		
-		thr = new oxt::thread(
-			bind(&ApplicationPoolStatusReporter::threadMain, this),
-			"Status report thread",
-			1024 * 128
-		);
+		try {
+			mainThread = new oxt::thread(
+				bind(&ApplicationPoolStatusReporter::mainThreadFunction, this),
+				"Status reporter main thread",
+				1024 * 128
+			);
+		} catch (...) {
+			do {
+				ret = close(serverFd);
+			} while (ret == -1 && errno == EINTR);
+			throw;
+		}
 	}
 	
 	~ApplicationPoolStatusReporter() {
 		this_thread::disable_syscall_interruption dsi;
 		this_thread::disable_interruption di;
-		
-		thr->interrupt_and_join();
-		delete thr;
-		
 		int ret;
+		
 		do {
 			ret = unlink(filename);
 		} while (ret == -1 && errno == EINTR);
+		
+		mainThread->interrupt_and_join();
+		delete mainThread;
+		
+		do {
+			ret = close(serverFd);
+		} while (ret == -1 && errno == EINTR);
+		
+		/* We make a copy of the data structure here to avoid deadlocks. */
+		map< int, shared_ptr<oxt::thread> > threadsCopy;
+		{
+			boost::lock_guard<boost::mutex> l(threadsLock);
+			threadsCopy = threads;
+		}
+		map< int, shared_ptr<oxt::thread> >::iterator it;
+		for (it = threadsCopy.begin(); it != threadsCopy.end(); it++) {
+			it->second->interrupt_and_join();
+		}
 	}
 };
 
 } // namespace Passenger
 