0
@@ -2,9 +2,9 @@
0
 #define _PASSENGER_APPLICATION_POOL_H_
0
 
0
 #include <boost/shared_ptr.hpp>
0
+#include <string>
0
 #include <map>
0
 
0
-#include "Application.h"
0
 #include "SpawnManager.h"
0
 
0
 namespace Passenger {
0
@@ -12,8 +12,15 @@ namespace Passenger {
0
 using namespace std;
0
 using namespace boost;
0
 
0
-// TODO: document this
0
 class ApplicationPool {
0
+public:
0
+  virtual ~ApplicationPool() {};
0
+  
0
+  virtual ApplicationPtr get(const string &appRoot, const string &user = "", const string &group = "") = 0;
0
+};
0
+
0
+// TODO: document this
0
+class StandardApplicationPool: public ApplicationPool {
0
 private:
0
   typedef map<string, ApplicationPtr> ApplicationMap;
0
 
0
@@ -26,27 +33,23 @@ private:
0
   }
0
   
0
 public:
0
-  ApplicationPool(const string &spawnManagerCommand,
0
-                  const string &logFile = "",
0
-                  const string &environment = "production",
0
-                  const string &rubyCommand = "ruby")
0
+  StandardApplicationPool(const string &spawnManagerCommand,
0
+               const string &logFile = "",
0
+               const string &environment = "production",
0
+               const string &rubyCommand = "ruby")
0
   : spawnManager(spawnManagerCommand, logFile, environment, rubyCommand) {}
0
   
0
-  ApplicationPtr get(const string &appRoot) {
0
-    return get(appRoot, "");
0
-  }
0
-  
0
   // TODO: improve algorithm
0
   // TODO: make thread-safe
0
   // TODO: make it possible to share an ApplicationPool between processes
0
-  ApplicationPtr get(const string &appRoot, const string &username) {
0
+  virtual ApplicationPtr get(const string &appRoot, const string &user = "", const string &group = "") {
0
     string normalizedAppRoot(normalizePath(appRoot));
0
     //scoped_lock l(lock);
0
     
0
     ApplicationPtr app;
0
     //ApplicationMap::iterator it(apps.find(appRoot));
0
     //if (it == apps.end()) {
0
-      app = spawnManager.spawn(appRoot, username);
0
+      app = spawnManager.spawn(appRoot, user, group);
0
     //  apps[appRoot] = app;
0
     //} else {
0
     //  app = it->second;

0
@@ -0,0 +1,348 @@
0
+#ifndef _PASSENGER_APPLICATION_POOL_CLIENT_SERVER_H_
0
+#define _PASSENGER_APPLICATION_POOL_CLIENT_SERVER_H_
0
+
0
+#include <boost/thread/thread.hpp>
0
+#include <set>
0
+
0
+#include <apr_portable.h>
0
+#include <apr_poll.h>
0
+
0
+#include <sys/types.h>
0
+#include <sys/socket.h>
0
+#include <errno.h>
0
+#include <unistd.h>
0
+
0
+#include "ApplicationPool.h"
0
+#include "Exceptions.h"
0
+#include "Utils.h"
0
+
0
+namespace Passenger {
0
+
0
+using namespace boost;
0
+using namespace std;
0
+
0
+class ApplicationPoolServer {
0
+private:
0
+  /*
0
+   * The only thing this struct does is making sure that the
0
+   * ApplicationPoolServer's thread function runs in the context
0
+   * of ApplicationPoolServer. Look at threadMain() instead for
0
+   * the interesting stuff.
0
+   */
0
+  struct ServerThreadDelegator {
0
+    ApplicationPoolServer *self;
0
+
0
+    ServerThreadDelegator(ApplicationPoolServer *self) {
0
+      this->self = self;
0
+    }
0
+    
0
+    void operator()() {
0
+      self->threadMain();
0
+    }
0
+  };
0
+  
0
+  class ApplicationPoolClient: public ApplicationPool {
0
+  private:
0
+    int sock;
0
+    
0
+  public:
0
+    ApplicationPoolClient(int sock) {
0
+      this->sock = sock;
0
+    }
0
+    
0
+    ~ApplicationPoolClient() {
0
+      close(sock);
0
+    }
0
+    
0
+    virtual ApplicationPtr get(const string &appRoot, const string &user = "", const string &group = "") {
0
+      MessageChannel channel(sock);
0
+      vector<string> args;
0
+      int reader, writer;
0
+      
0
+      channel.write(appRoot.c_str(), user.c_str(), group.c_str(), NULL);
0
+      channel.read(args);
0
+      reader = channel.readFileDescriptor();
0
+      writer = channel.readFileDescriptor();
0
+      ApplicationPtr app(new Application(appRoot, atoi(args[0].c_str()), reader, writer));
0
+      return app;
0
+    }
0
+  };
0
+  
0
+  class SocketDemultiplexer {
0
+  private:
0
+    apr_pool_t *pool;
0
+    apr_pollset_t *pollset;
0
+    apr_uint32_t used, capacity;
0
+    set<int> fds;
0
+    int *pollResult;
0
+    
0
+    void createAprPollFd(int fd, apr_pollfd_t *apr_fd) {
0
+      apr_socket_t *sock = NULL;
0
+      apr_status_t ret;
0
+      
0
+      ret = apr_os_sock_put(&sock, &fd, pool);
0
+      if (ret != APR_SUCCESS) {
0
+        throw APRException("Unable to convert a file descriptor to an APR socket", ret);
0
+      }
0
+      
0
+      apr_fd->desc_type = APR_POLL_SOCKET;
0
+      apr_fd->reqevents = APR_POLLIN;
0
+      apr_fd->p = pool;
0
+      apr_fd->desc.s = sock;
0
+      apr_fd->client_data = (void *) fd;
0
+    }
0
+    
0
+    void addToPollset(apr_pollset_t *pollset, int fd) {
0
+      apr_pollfd_t apr_fd;
0
+      apr_status_t ret;
0
+      
0
+      createAprPollFd(fd, &apr_fd);
0
+      ret = apr_pollset_add(pollset, &apr_fd);
0
+      if (ret != APR_SUCCESS) {
0
+        throw APRException("Cannot add file descriptor to poll set", ret);
0
+      }
0
+    }
0
+    
0
+  public:
0
+    SocketDemultiplexer() {
0
+      apr_status_t ret;
0
+      
0
+      ret = apr_pool_create(&pool, NULL);
0
+      if (ret != APR_SUCCESS) {
0
+        throw APRException("Cannot create an APR memory pool", ret);
0
+      }
0
+      used = 0;
0
+      capacity = 8;
0
+      
0
+      ret = apr_pollset_create(&pollset, capacity, pool, 0);
0
+      if (ret != APR_SUCCESS) {
0
+        apr_pool_destroy(pool);
0
+        throw APRException("Cannot create an APR poll set", ret);
0
+      }
0
+      pollResult = (int *) malloc(sizeof(int) * capacity);
0
+    }
0
+    
0
+    ~SocketDemultiplexer() {
0
+      apr_pool_destroy(pool);
0
+      free(pollResult);
0
+      // Apparently APR does not always close the file descriptor,
0
+      // so we do it manually.
0
+      for (set<int>::const_iterator it(fds.begin()); it != fds.end(); it++) {
0
+        close(*it);
0
+      }
0
+    }
0
+  
0
+    void add(int fd) {
0
+      if (used == capacity) {
0
+        apr_pollset_t *newPollset;
0
+        int *newPollResult;
0
+        apr_status_t ret;
0
+        
0
+        ret = apr_pollset_create(&newPollset, capacity * 2, pool, 0);
0
+        if (ret != APR_SUCCESS) {
0
+          throw APRException("Cannot create a new APR poll set", ret);
0
+        }
0
+        newPollResult = (int *) realloc(pollResult, sizeof(int) * capacity * 2);
0
+        if (newPollResult == NULL) {
0
+          throw MemoryException("Cannot allocate memory for result vector.");
0
+        }
0
+        
0
+        apr_pollset_destroy(pollset);
0
+        pollset = newPollset;
0
+        pollResult = newPollResult;
0
+        capacity *= 2;
0
+        
0
+        for (set<int>::const_iterator it(fds.begin()); it != fds.end(); it++) {
0
+          addToPollset(pollset, *it);
0
+        }
0
+      }
0
+      addToPollset(pollset, fd);
0
+      fds.insert(fd);
0
+      used++;
0
+    }
0
+    
0
+    void remove(int fd) {
0
+      apr_pollfd_t apr_fd;
0
+      
0
+      createAprPollFd(fd, &apr_fd);
0
+      if (apr_pollset_remove(pollset, &apr_fd) != APR_SUCCESS) {
0
+        throw MemoryException("Cannot remove a file descriptor from the poll set.");
0
+      }
0
+      fds.erase(fds.find(fd));
0
+      used--;
0
+    }
0
+    
0
+    unsigned int poll(int **fileDescriptors, unsigned int timeout) {
0
+      apr_int32_t num;
0
+      const apr_pollfd_t *result;
0
+      apr_status_t ret;
0
+      
0
+      do {
0
+        ret = apr_pollset_poll(pollset, timeout * 1000, &num, &result);
0
+      } while (ret == APR_EINTR);
0
+      if (ret == APR_TIMEUP) {
0
+        return 0;
0
+      } else if (ret != APR_SUCCESS) {
0
+        throw APRException("Cannot poll file descriptor set", ret);
0
+      }
0
+      for (apr_int32_t i = 0; i < num; i++) {
0
+        pollResult[i] = (int) result[i].client_data;
0
+      }
0
+      *fileDescriptors = pollResult;
0
+      return num;
0
+    }
0
+  };
0
+  
0
+  StandardApplicationPool pool;
0
+  int serverSocket;
0
+  int connectSocket;
0
+  bool done, detached;
0
+  thread *thr;
0
+  SocketDemultiplexer demultiplexer;
0
+  
0
+  void threadMain() {
0
+    demultiplexer.add(serverSocket);
0
+    while (!done) {
0
+      int *fds;
0
+      unsigned int num;
0
+      
0
+      num = demultiplexer.poll(&fds, 500);
0
+      for (unsigned int i = 0; i < num; i++) {
0
+        if (fds[i] == serverSocket) {
0
+          acceptNewClient(demultiplexer);
0
+        } else {
0
+          handleClient(fds[i]);
0
+        }
0
+      }
0
+    }
0
+  }
0
+  
0
+  void acceptNewClient(SocketDemultiplexer &demultiplexer) {
0
+    int fds[2], ret;
0
+    char x;
0
+
0
+    // Discard data, not important. Whatever data was sent only serves
0
+    // to wake up the server socket.
0
+    do {
0
+      ret = read(serverSocket, &x, 1);
0
+    } while (ret == -1 && errno == EINTR);
0
+    socketpair(AF_UNIX, SOCK_STREAM, 0, fds);
0
+    demultiplexer.add(fds[0]);
0
+    MessageChannel(serverSocket).writeFileDescriptor(fds[1]);
0
+    close(fds[1]);
0
+  }
0
+  
0
+  void handleClient(int client) {
0
+    MessageChannel channel(client);
0
+    vector<string> args;
0
+    
0
+    if (channel.read(args) && args.size() == 3) {
0
+      ApplicationPtr app(pool.get(args[0], args[1], args[2]));
0
+      channel.write(toString(app->getPid()).c_str(), NULL);
0
+      channel.writeFileDescriptor(app->getReader());
0
+      channel.writeFileDescriptor(app->getWriter());
0
+    }
0
+  }
0
+  
0
+/*  unsigned int randomNumber() {
0
+    FILE *f = fopen("/dev/urandom", "r");
0
+    if (f == NULL) {
0
+      return rand();
0
+    } else {
0
+      unsigned int result;
0
+      if (fread(&result, 1, sizeof(result), f) == 0) {
0
+        result = rand();
0
+      }
0
+      fclose(f);
0
+      return result;
0
+    }
0
+  }
0
+  
0
+  bool generateUniqueSocketFilename(char *output, size_t size) {
0
+    stringstream base;
0
+    
0
+    if (getenv("TMPDIR") && *getenv("TMPDIR")) {
0
+      base << getenv("TMPDIR");
0
+    } else {
0
+      base << "/tmp";
0
+    }
0
+    base << "/phusion_passenger.tmp." << getpid() << ".";
0
+    
0
+    for (int i = 0; i < 10000; i++) {
0
+      stringstream filename;
0
+      const char *filenameString;
0
+      struct stat buf;
0
+      
0
+      filename << base.str() << randomNumber() << ".sock";
0
+      if (filename.str().size() > size - 1) {
0
+        continue;
0
+      }
0
+      filenameString = filename.str().c_str();
0
+      if (stat(filenameString, &buf) == -1 && errno == ENOENT) {
0
+        strncpy(output, filenameString, size);
0
+        return true;
0
+      }
0
+    }
0
+    return false;
0
+  } */
0
+  
0
+  void finalize() {
0
+    close(serverSocket);
0
+    close(connectSocket);
0
+  }
0
+  
0
+public:
0
+  ApplicationPoolServer(const string &spawnManagerCommand,
0
+               const string &logFile = "",
0
+               const string &environment = "production",
0
+               const string &rubyCommand = "ruby")
0
+  : pool(spawnManagerCommand, logFile, environment, rubyCommand) {
0
+    int fds[2];
0
+    
0
+    if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == -1) {
0
+      throw SystemException("Cannot create a Unix socket pair", errno);
0
+    }
0
+    serverSocket = fds[0];
0
+    connectSocket = fds[1];
0
+    done = false;
0
+    detached = false;
0
+    thr = new thread(ServerThreadDelegator(this));
0
+  }
0
+  
0
+  ~ApplicationPoolServer() {
0
+    if (!detached) {
0
+      done = true;
0
+      thr->join();
0
+      delete thr;
0
+      finalize();
0
+    }
0
+  }
0
+  
0
+  ApplicationPoolPtr connect() {
0
+    int ret;
0
+    do {
0
+      // Write some random data to wake up the server.
0
+      ret = write(connectSocket, "x", 1);
0
+    } while ((ret == -1 && errno == EAGAIN) || ret == 0);
0
+    ret = MessageChannel(connectSocket).readFileDescriptor();
0
+    return ptr(new ApplicationPoolClient(ret));
0
+  }
0
+  
0
+  /**
0
+   * @warning Never call this method in the process in which this
0
+   *          ApplicationPoolServer was created!
0
+   */
0
+  void detach() {
0
+    detached = true;
0
+    #ifdef VALGRIND_FRIENDLY
0
+      delete thr;
0
+    #endif
0
+    finalize();
0
+  }
0
+};
0
+
0
+typedef shared_ptr<ApplicationPoolServer> ApplicationPoolServerPtr;
0
+
0
+} // namespace Passenger
0
+
0
+#endif /* _PASSENGER_APPLICATION_POOL_CLIENT_SERVER_H_ */

0
@@ -1,6 +1,8 @@
0
 #ifndef _PASSENGER_EXCEPTIONS_H_
0
 #define _PASSENGER_EXCEPTIONS_H_
0
 
0
+#include <apr_errno.h>
0
+
0
 #include <exception>
0
 #include <string>
0
 #include <sstream>
0
@@ -15,14 +17,13 @@ private:
0
   int m_code;
0
 public:
0
   SystemException(const string &message, int errorCode) {
0
-    stringstream str(message);
0
-    str << ":" << strerror(errorCode) << " (" << errorCode << ")";
0
+    stringstream str;
0
+    str << message << ": " << strerror(errorCode) << " (" << errorCode << ")";
0
     msg = str.str();
0
     m_code = errorCode;
0
   }
0
   
0
-  virtual ~SystemException() throw() {
0
-  }
0
+  virtual ~SystemException() throw() {}
0
   
0
   virtual const char *what() const throw() {
0
     return msg.c_str();
0
@@ -34,11 +35,17 @@ public:
0
 };
0
 
0
 class MemoryException: public exception {
0
+private:
0
+  string message;
0
 public:
0
-  MemoryException() {}
0
+  MemoryException(): message("Unable to allocate memory.") {}
0
+  
0
+  MemoryException(const string &msg): message(msg) {}
0
+  
0
   virtual ~MemoryException() throw() {}
0
+  
0
   virtual const char *what() const throw() {
0
-    return "Unable to allocate memory.";
0
+    return message.c_str();
0
   }
0
 };
0
 
0
@@ -51,6 +58,24 @@ public:
0
   virtual const char *what() const throw() { return msg.c_str(); }
0
 };
0
 
0
+class APRException: public exception {
0
+private:
0
+  string msg;
0
+public:
0
+  APRException(const string &message, apr_status_t status) {
0
+    stringstream str;
0
+    char buf[1024];
0
+    str << message << ": " << apr_strerror(status, buf, sizeof(buf)) << " (" << status << ")";
0
+    msg = str.str();
0
+  }
0
+  
0
+  virtual ~APRException() throw() {}
0
+  
0
+  virtual const char *what() const throw() {
0
+    return msg.c_str();
0
+  }
0
+};
0
+
0
 } // namespace Passenger
0
 
0
 #endif /* _PASSENGER_EXCEPTIONS_H_ */

0
@@ -22,7 +22,7 @@
0
 #include "Types.h"
0
 #include "Utils.h"
0
 #include "DispatcherBucket.h"
0
-#include "ApplicationPool.h"
0
+#include "ApplicationPoolClientServer.h"
0
 #include "MessageChannel.h"
0
 
0
 using namespace std;
0
@@ -33,6 +33,7 @@ extern "C" module AP_MODULE_DECLARE_DATA rails_module;
0
 
0
 class Hooks {
0
 private:
0
+  ApplicationPoolServerPtr applicationPoolServer;
0
   ApplicationPoolPtr applicationPool;
0
   
0
   RailsConfig *getConfig(request_rec *r) {
0
@@ -217,7 +218,7 @@ public:
0
     ap_add_version_component(pconf, "Phusion_Passenger/" PASSENGER_VERSION);
0
     
0
     const char *spawnManagerCommand = "/home/hongli/Projects/mod_rails/lib/mod_rails/spawn_manager.rb";
0
-    applicationPool = ApplicationPoolPtr(new ApplicationPool(spawnManagerCommand, "", "production"));
0
+    applicationPoolServer = ptr(new ApplicationPoolServer(spawnManagerCommand, "", "production"));
0
   }
0
   
0
   ~Hooks() {
0
@@ -225,6 +226,8 @@ public:
0
   }
0
   
0
   void initChild(apr_pool_t *pchild, server_rec *s) {
0
+    applicationPool = applicationPoolServer->connect();
0
+    applicationPoolServer->detach();
0
   }
0
   
0
   int handleRequest(request_rec *r) {

0
@@ -16,7 +16,7 @@ install: $(OBJECTS)
0
 Configuration.o: Configuration.cpp
0
   $(CXX) $(CXXFLAGS) -c Configuration.cpp
0
 
0
-Hooks.o: Hooks.cpp Hooks.h Types.h DispatcherBucket.h ApplicationPool.h SpawnManager.h Exceptions.h Application.h MessageChannel.h Utils.h
0
+Hooks.o: Hooks.cpp Hooks.h Types.h DispatcherBucket.h ApplicationPool.h ApplicationPoolClientServer.h SpawnManager.h Exceptions.h Application.h MessageChannel.h Utils.h
0
   $(CXX) $(CXXFLAGS) -c Hooks.cpp
0
 
0
 DispatcherBucket.o: DispatcherBucket.h DispatcherBucket.cpp Utils.h

0
@@ -11,6 +11,7 @@
0
 #include <arpa/inet.h>
0
 #include <errno.h>
0
 #include <unistd.h>
0
+#include <cstdarg>
0
 
0
 #include "Exceptions.h"
0
 

0
@@ -10,9 +10,9 @@
0
 #include <sys/wait.h>
0
 #include <arpa/inet.h>
0
 #include <cstdio>
0
+#include <cstdarg>
0
 #include <unistd.h>
0
 #include <errno.h>
0
-#include <stdarg.h>
0
 
0
 #include "Application.h"
0
 #include "MessageChannel.h"

0
@@ -16,6 +16,9 @@ initDebugging(const char *logFile) {
0
       if (stream->fail()) {
0
         delete stream;
0
       } else {
0
+        if (_debugStream != NULL && _debugStream != &cerr) {
0
+          delete _debugStream;
0
+        }
0
         _debugStream = stream;
0
       }
0
     } else {

0
@@ -1,7 +1,9 @@
0
 #ifndef _PASSENGER_UTILS_H_
0
 #define _PASSENGER_UTILS_H_
0
 
0
+#include <boost/shared_ptr.hpp>
0
 #include <ostream>
0
+#include <sstream>
0
 
0
 namespace Passenger {
0
 
0
@@ -20,10 +22,22 @@ namespace Passenger {
0
   #define P_TRACE P_DEBUG
0
 #endif
0
 
0
-  // Internal; do not use directly.
0
-  extern std::ostream *_debugStream;
0
+void initDebugging(const char *logFile = NULL);
0
 
0
-  void initDebugging(const char *logFile = NULL);
0
+template<typename T> boost::shared_ptr<T>
0
+ptr(T *pointer) {
0
+  return boost::shared_ptr<T>(pointer);
0
+}
0
+
0
+template<typename T> std::string
0
+toString(T something) {
0
+  std::stringstream s;
0
+  s << something;
0
+  return s.str();
0
+}
0
+
0
+// Internal; do not use directly.
0
+extern std::ostream *_debugStream;
0
 
0
 } // namespace Passenger
0
 

0
@@ -0,0 +1,56 @@
0
+#include "tut.h"
0
+#include "ApplicationPoolClientServer.h"
0
+#include "Utils.cpp"
0
+#include <cstring>
0
+#include <unistd.h>
0
+
0
+using namespace Passenger;
0
+
0
+namespace tut {
0
+  struct ApplicationPoolClientServerTest {
0
+    ApplicationPoolServerPtr server;
0
+    
0
+    ApplicationPoolClientServerTest() {
0
+      server = ptr(new ApplicationPoolServer("support/spawn_server_mock.rb"));