0
@@ -101,6 +101,13 @@
0
   Invariant:
0
      active <= count
0
 
0
+- waiting: integer
0
+  The number of get() calls that are currently waiting until a new application
0
+  can be spawned or an existing can be connected to.
0
+
0
+- waiting_lock: mutex
0
+  A mutex for the _waiting_ instance variable.
0
+
0
 - inactive_apps: list<AppContainer>
0
   A linked list of AppContainer objects. All application instances in this list
0
   are inactive.
0
0
@@ -124,8 +131,9 @@
0
 
0
   Invariant:
0
      app_instance_count.keys == apps.keys
0
-     for all keys app_root in app_instance_count:
0
+     for all (app_root, app_count) in app_instance_count:
0
         app_instance_count[app_root] < count
0
+        app_count == apps[app_root].size()
0
      (sum of all values in app_instance_count) == count.
0
 
0
 
0
0
0
0
@@ -135,35 +143,52 @@
0
 # - All wait commands are to unlock the lock during waiting.
0
 
0
 function get(app_root):
0
-	MAX_ATTEMPTS = 5
0
-	attempt = 0
0
+	MAX_ATTEMPTS = 10
0
+	attempt = 1
0
   while (true):
0
-		attempt++
0
     lock.synchronize:
0
       container, list = spawn_or_use_existing(app_root)
0
-			container.last_used = current_time()
0
-			container.sessions++
0
-			try:
0
-				return container.app.connect()
0
-			on exception:
0
-				container.sessions--
0
-				if (attempt == MAX_ATTEMPTS):
0
-					propagate exception
0
-				else:
0
-					# The app instance seems to have crashed. So we remove this
0
-					# instance from our data structures.
0
-					list.remove(container.iterator)
0
-					if list.empty():
0
-						apps.remove(app_root)
0
-					app_instance_count.remove(app_root)
0
-					count--
0
-					active--
0
+			if (container != null):
0
+				container.last_used = current_time()
0
+				container.sessions++
0
+				try:
0
+					return container.app.connect()
0
+				on exception:
0
+					container.sessions--
0
+					if (attempt == MAX_ATTEMPTS):
0
+						propagate exception
0
+					else:
0
+						# The app instance seems to have crashed.
0
+						# So we remove this instance from our data
0
+						# structures.
0
+						list.remove(container.iterator)
0
+						if list.empty():
0
+							apps.remove(app_root)
0
+						app_instance_count.remove(app_root)
0
+						count--
0
+						active--
0
+						attempt++
0
+		if (container == null):
0
+			# Unable to spawn or connect to an application right
0
+			# now. Try again later, unless we've already tried too
0
+			# many times.
0
+			if (attempt == MAX_ATTEMPTS):
0
+				raise exception
0
+			waiting_lock.synchronize:
0
+				waiting++
0
+			usleep(attempt * 2000)
0
+			waiting_lock.synchronize:
0
+				waiting--
0
+			attempt++
0
 
0
 
0
 # Returns a pair of [AppContainer, list<AppContainer>] that matches the
0
 # given application root. If no such AppContainer exists, then it is created
0
 # and a new application instance is spawned. All exceptions that occur are
0
 # propagated.
0
+#
0
+# Returns [null, null] if no new application can be spawned and no existing
0
+# application can be connected to, at the moment.
0
 function spawn_or_use_existing(app_root):
0
   list = apps[app_root]
0
   
0
0
0
@@ -182,25 +207,24 @@
0
   
0
   if list != nil:
0
     # There are apps for this app root.
0
-		if (list.front.sessions == 0) or (count >= max) or (
0
-		   (max_per_app != 0) and (app_instance_count[app_root] >= max_per_app)
0
-		   ):
0
+		if (list.front.sessions == 0):
0
       # There is an inactive app, so we use it.
0
-			#  -OR-
0
-			# All apps are active, and the pool is full. 
0
-			#  -OR-
0
-			# All apps are active and the number of max instances
0
-			# spawned for this rails application has been reached.
0
-			#
0
-			# We're not allowed to spawn a new app, so if there are
0
-			# no inactive apps, we try to connect to an existing one.
0
-			# Our connection request will be put into that app's
0
-			# connection queue.
0
       container = list.front
0
       list.move_to_back(container.iterator)
0
       if container.sessions == 0:
0
         inactive_apps.remove(container.ia_iterator)
0
       active++
0
+		else if	(count >= max) or (
0
+			(max_per_app != 0) and (app_instance_count[app_root] >= max_per_app)
0
+			):
0
+			# All apps are active, and the pool is full. 
0
+			#  -OR-
0
+			# All apps are active and the number of max instances
0
+			# spawned for this application has been reached.
0
+			#
0
+			# We can't do anything, so we return null. One should
0
+			# try later.
0
+			return [null, null]
0
     else:
0
       # All apps are active, but the pool hasn't reached its
0
       # maximum yet. So we spawn a new app.

0
@@ -69,7 +69,7 @@
0
  * @endcode
0
  *
0
  * Internally, ApplicationPool::get() will keep spawned applications instances in
0
- * memory, and reuse them if possible. It will try to keep spawning to a minimum.
0
+ * memory, and reuse them if possible. It wil* @throw l try to keep spawning to a minimum.
0
  * Furthermore, if an application instance hasn't been used for a while, it
0
  * will be automatically shutdown in order to save memory. Restart requests are
0
  * honored: if an application has the file 'restart.txt' in its 'tmp' folder,
0
@@ -110,6 +110,8 @@
0
     *                    See the Ruby class SpawnManager for details.
0
    * @return A session object.
0
    * @throw SpawnException An attempt was made to spawn a new application instance, but that attempt failed.
0
+	 * @throw BusyException The application pool is too busy right now, and cannot
0
+	 *       satisfy the request. One should either abort, or try again later.
0
    * @throw IOException Something else went wrong.
0
    * @note Applications are uniquely identified with the application root
0
    *       string. So although <tt>appRoot</tt> does not have to be absolute, it

0
@@ -330,6 +330,8 @@
0
         } else {
0
           throw SpawnException(args[1]);
0
         }
0
+			} else if (args[0] == "BusyExeption") {
0
+				throw BusyException(args[1]);
0
       } else if (args[0] == "IOException") {
0
         throw IOException(args[1]);
0
       } else {
0
0
@@ -573,10 +575,10 @@
0
     try {
0
       MessageChannel channel(serverSocket);
0
       int clientConnection;
0
-				
0
+				
0
       // Write some random data to wake up the server.
0
       channel.writeRaw("x", 1);
0
-				
0
+				
0
       clientConnection = channel.readFileDescriptor();
0
       return ptr(new Client(clientConnection));
0
     } catch (const SystemException &e) {

0
@@ -219,6 +219,9 @@
0
         channel.write("SpawnException", e.what(), "false", NULL);
0
       }
0
       failed = true;
0
+		} catch (const BusyException &e) {
0
+			channel.write("BusyException", e.what(), NULL);
0
+			failed = true;
0
     } catch (const IOException &e) {
0
       channel.write("IOException", e.what(), NULL);
0
       failed = true;
0
@@ -326,9 +329,11 @@
0
           break;
0
         }
0
       } catch (const exception &e) {
0
-				P_WARN("Uncaught exception in ApplicationPoolServer client thread:\n"
0
-					<< "   message: " << toString(args) << "\n"
0
-					<< "   exception: " << e.what());
0
+				if (!serverDone) {
0
+					P_WARN("Uncaught exception in ApplicationPoolServer client thread:\n"
0
+						<< "   message: " << toString(args) << "\n"
0
+						<< "   exception: " << e.what());
0
+				}
0
         break;
0
       }
0
     }

0
@@ -175,6 +175,20 @@
0
   }
0
 };
0
 
0
+/**
0
+ * The application pool is too busy and cannot fulfill a get() request.
0
+ *
0
+ * @ingroup Exceptions
0
+ */
0
+class BusyException: public exception {
0
+private:
0
+	string msg;
0
+public:
0
+	BusyException(const string &message): msg(message) {}
0
+	virtual ~BusyException() throw() {}
0
+	virtual const char *what() const throw() { return msg.c_str(); }
0
+};
0
+
0
 } // namespace Passenger
0
 
0
 #endif /* _PASSENGER_EXCEPTIONS_H_ */

0
@@ -72,6 +72,12 @@
0
     return (ServerConfig *) ap_get_module_config(s->module_config, &passenger_module);
0
   }
0
   
0
+	int reportBusyException(request_rec *r) {
0
+		ap_custom_response(r, HTTP_SERVICE_UNAVAILABLE,
0
+			"This website is too busy right now.  Please try again later.");
0
+		return HTTP_SERVICE_UNAVAILABLE;
0
+	}
0
+	
0
   /**
0
    * Determine whether the given HTTP request falls under one of the specified
0
    * RailsBaseURIs. If yes, then the first matching base URI will be returned.
0
@@ -430,7 +436,7 @@
0
         }
0
         session = applicationPool->get(canonicalizePath(railsDir + "/.."),
0
           true, defaultUser, environment, spawnMethod);
0
-				P_DEBUG("Forwarding " << r->uri << " to PID " << session->getPid());
0
+				//P_DEBUG("Forwarding " << r->uri << " to PID " << session->getPid());
0
       } catch (const SpawnException &e) {
0
         if (e.hasErrorPage()) {
0
           ap_set_content_type(r, "text/html; charset=utf-8");
0
@@ -441,6 +447,8 @@
0
         } else {
0
           throw;
0
         }
0
+			} catch (const BusyException &e) {
0
+				return reportBusyException(r);
0
       }
0
       sendHeaders(r, session, railsBaseURI);
0
       sendRequestBody(r, session);

0
@@ -73,8 +73,18 @@
0
         } \
0
       } \
0
     } while (false)
0
+	
0
+	#define P_ASSERT(expr, result_if_failed, message) \
0
+		do { \
0
+			if (!(expr)) { \
0
+				P_ERROR("Assertion failed: " << message); \
0
+				return result_if_failed; \
0
+			} \
0
+		} while (false)
0
 #else
0
   #define P_TRACE(level, expr) do { /* nothing */ } while (false)
0
+	
0
+	#define P_ASSERT(expr, result_if_failed, message) do { /* nothing */ } while (false)
0
 #endif
0
 
0
 } // namespace Passenger

0
@@ -268,6 +268,7 @@
0
       char control_data[CMSG_SPACE(sizeof(int))];
0
     #endif
0
     struct cmsghdr *control_header;
0
+		int ret;
0
   
0
     msg.msg_name = NULL;
0
     msg.msg_namelen = 0;
0
@@ -294,7 +295,10 @@
0
       memcpy(CMSG_DATA(control_header), &fileDescriptor, sizeof(int));
0
     #endif
0
     
0
-		if (sendmsg(fd, &msg, 0) == -1) {
0
+		do {
0
+			ret = sendmsg(fd, &msg, 0);
0
+		} while (ret == -1 && errno == EINTR);
0
+		if (ret == -1) {
0
       throw SystemException("Cannot send file descriptor with sendmsg()", errno);
0
     }
0
   }
0
@@ -448,6 +452,7 @@
0
       #define EXPECTED_CMSG_LEN CMSG_LEN(sizeof(int))
0
     #endif
0
     struct cmsghdr *control_header;
0
+		int ret;
0
 
0
     msg.msg_name    = NULL;
0
     msg.msg_namelen = 0;
0
@@ -462,7 +467,10 @@
0
     msg.msg_controllen = sizeof(control_data);
0
     msg.msg_flags      = 0;
0
     
0
-		if (recvmsg(fd, &msg, 0) == -1) {
0
+		do {
0
+			ret = recvmsg(fd, &msg, 0);
0
+		} while (ret == -1 && errno == EINTR);
0
+		if (ret == -1) {
0
       throw SystemException("Cannot read file descriptor with recvmsg()", errno);
0
     }
0
     

0
@@ -92,6 +92,7 @@
0
   static const int DEFAULT_MAX_POOL_SIZE = 20;
0
   static const int DEFAULT_MAX_INSTANCES_PER_APP = 0;
0
   static const int CLEANER_THREAD_STACK_SIZE = 1024 * 128;
0
+	static const unsigned int MAX_GET_ATTEMPTS = 10;
0
 
0
   friend class ApplicationPoolServer;
0
   struct AppContainer;
0
0
@@ -111,12 +112,14 @@
0
   
0
   struct SharedData {
0
     mutex lock;
0
+		mutex waitingLock;
0
     condition activeOrMaxChanged;
0
     
0
     ApplicationMap apps;
0
     unsigned int max;
0
     unsigned int count;
0
     unsigned int active;
0
+		unsigned int waiting;
0
     unsigned int maxPerApp;
0
     AppContainerList inactiveApps;
0
     map<string, time_t> restartFileTimes;
0
0
0
@@ -177,16 +180,49 @@
0
   
0
   // Shortcuts for instance variables in SharedData. Saves typing in get().
0
   mutex &lock;
0
+	mutex &waitingLock;
0
   condition &activeOrMaxChanged;
0
   ApplicationMap &apps;
0
   unsigned int &max;
0
   unsigned int &count;
0
   unsigned int &active;
0
+	unsigned int &waiting;
0
   unsigned int &maxPerApp;
0
   AppContainerList &inactiveApps;
0
   map<string, time_t> &restartFileTimes;
0
   map<string, unsigned int> &appInstanceCount;
0
   
0
+	/**
0
+	 * Verify that all the invariants are correct.
0
+	 */
0
+	bool inline verifyState() {
0
+	#if PASSENGER_DEBUG
0
+		// Invariant for _apps_.
0
+		ApplicationMap::const_iterator it;
0
+		for (it = apps.begin(); it != apps.end(); it++) {
0
+			AppContainerList *list = it->second.get();
0
+			P_ASSERT(!list->empty(), false, "List for '" << it->first << "' is nonempty.");
0
+			
0
+			AppContainerList::const_iterator prev_lit;
0
+			AppContainerList::const_iterator lit;
0
+			prev_lit = list->begin();
0
+			lit = prev_lit;
0
+			lit++;
0
+			for (; lit != list->end(); lit++) {
0
+				if ((*prev_lit)->sessions > 0) {
0
+					P_ASSERT((*lit)->sessions > 0, false,
0
+						"List for '" << it->first <<
0
+						"' is sorted from nonactive to active");
0
+				}
0
+			}
0
+		}
0
+		
0
+		P_ASSERT(active <= count, false,
0
+			"active (" << active << ") < count (" << count << ")");
0
+	#endif
0
+		return true;
0
+	}
0
+	
0
   bool needsRestart(const string &appRoot) {
0
     string restartFile(appRoot);
0
     restartFile.append("/tmp/restart.txt");
0
@@ -316,8 +352,7 @@
0
       if (it != apps.end()) {
0
         list = it->second.get();
0
     
0
-				if (list->front()->sessions == 0 || count >= max
0
-				 || ( maxPerApp != 0 && appInstanceCount[appRoot] >= maxPerApp )) {
0
+				if (list->front()->sessions == 0) {
0
           container = list->front();
0
           list->pop_front();
0
           list->push_back(container);
0
@@ -327,6 +362,10 @@
0
             inactiveApps.erase(container->ia_iterator);
0
           }
0
           active++;
0
+				} else if (count >= max || (
0
+					maxPerApp != 0 && appInstanceCount[appRoot] >= maxPerApp )
0
+					) {
0
+					return make_pair(AppContainerPtr(), (AppContainerList *) 0);
0
         } else {
0
           container = ptr(new AppContainer());
0
           container->app = spawnManager.spawn(appRoot,
0
0
@@ -434,11 +473,13 @@
0
     #endif
0
     data(new SharedData()),
0
     lock(data->lock),
0
+		waitingLock(data->waitingLock),
0
     activeOrMaxChanged(data->activeOrMaxChanged),
0
     apps(data->apps),
0
     max(data->max),
0
     count(data->count),
0
     active(data->active),
0
+		waiting(data->waiting),
0
     maxPerApp(data->maxPerApp),
0
     inactiveApps(data->inactiveApps),
0
     restartFileTimes(data->restartFileTimes),
0
@@ -449,6 +490,7 @@
0
     max = DEFAULT_MAX_POOL_SIZE;
0
     count = 0;
0
     active = 0;
0
+		waiting = 0;
0
     maxPerApp = DEFAULT_MAX_INSTANCES_PER_APP;
0
     maxIdleTime = DEFAULT_MAX_IDLE_TIME;
0
     cleanerThread = new thread(
0
0
0
@@ -477,12 +519,9 @@
0
     const string &spawnMethod = "smart"
0
   ) {
0
     unsigned int attempt;
0
-		const unsigned int MAX_ATTEMPTS = 5;
0
     
0
-		attempt = 0;
0
+		attempt = 1;
0
     while (true) {
0
-			attempt++;
0
-			
0
       mutex::scoped_lock l(lock);
0
       pair<AppContainerPtr, AppContainerList *> p(
0
         spawnOrUseExisting(l, appRoot, lowerPrivilege, lowestUser,
0
0
0
@@ -491,33 +530,56 @@
0
       AppContainerPtr &container(p.first);
0
       AppContainerList &list(*p.second);
0
       
0
-			container->lastUsed = time(NULL);
0
-			container->sessions++;
0
-			try {
0
-				return container->app->connect(SessionCloseCallback(data, container));
0
-			} catch (const exception &e) {
0
-				container->sessions--;
0
-				if (attempt == MAX_ATTEMPTS) {
0
-					string message("Cannot connect to an existing application instance for '");
0
-					message.append(appRoot);
0
-					message.append("': ");
0
-					try {
0
-						const SystemException &syse = dynamic_cast<const SystemException &>(e);
0
-						message.append(syse.sys());
0
-					} catch (const bad_cast &) {
0
-						message.append(e.what());
0
+			if (container != NULL) {
0
+				container->lastUsed = time(NULL);
0
+				container->sessions++;
0
+			
0
+				P_ASSERT(verifyState(), Application::SessionPtr(),
0
+					"State is valid:\n" << toString(false));
0
+				try {
0
+					return container->app->connect(SessionCloseCallback(data, container));
0
+				} catch (const exception &e) {
0
+					container->sessions--;
0
+					if (attempt == MAX_GET_ATTEMPTS) {
0
+						string message("Cannot connect to an existing "
0
+							"application instance for '");
0
+						message.append(appRoot);
0
+						message.append("': ");
0
+						try {
0
+							const SystemException &syse = dynamic_cast<const SystemException &>(e);
0
+							message.append(syse.sys());
0
+						} catch (const bad_cast &) {
0
+							message.append(e.what());
0
+						}
0
+						throw IOException(message);
0
+					} else {
0
+						list.erase(container->iterator);
0
+						if (list.empty()) {
0
+							apps.erase(appRoot);
0
+						}
0
+						appInstanceCount.erase(appRoot);
0
+						count--;
0
+						active--;
0
+						attempt++;
0
+						P_ASSERT(verifyState(), Application::SessionPtr(), "State is valid.");
0
           }
0
-					throw IOException(message);
0
-				} else {
0
-					list.erase(container->iterator);
0
-					if (list.empty()) {
0
-						apps.erase(appRoot);
0
-					}
0
-					appInstanceCount.erase(appRoot);
0
-					count--;
0
-					active--;
0
         }
0
       }
0
+			if (container == NULL) {
0
+				l.unlock();
0
+				if (attempt == MAX_GET_ATTEMPTS) {
0
+					throw BusyException("Cannot satisfy get() request.");
0
+				}
0
+				{
0
+					mutex::scoped_lock wl(waitingLock);
0
+					waiting++;
0
+				}
0
+				usleep(attempt * 20000);
0
+				{
0
+					mutex::scoped_lock wl(waitingLock);
0
+					waiting--;
0
+				}
0
+			}
0
     }
0
     // Never reached; shut up compiler warning
0
     return Application::SessionPtr();
0
0
@@ -567,14 +629,15 @@
0
    * Returns a textual description of the internal state of
0
    * the application pool.
0
    */
0
-	virtual string toString() const {
0
-		mutex::scoped_lock l(lock);
0
+	virtual string toString(bool lockMutex = true) const {
0
+		mutex::scoped_lock l(lock, lockMutex);
0
     stringstream result;
0
     
0
     result << "----------- General information -----------" << endl;
0
     result << "max    = " << max << endl;
0
     result << "count  = " << count << endl;
0
     result << "active = " << active << endl;
0
+		result << "waiting = " << waiting << endl;
0
     result << endl;
0
     
0
     result << "----------- Applications -----------" << endl;

0
@@ -121,13 +121,10 @@
0
   TEST_METHOD(7) {
0
     // If we call get() even though the pool is already full
0
     // (active == max), and the application root is already
0
-		// in the pool, then the pool should have tried to open
0
-		// a session in an already active app.
0
+		// in the pool, then the pool must wait until there's an
0
+		// inactive application.
0
     pool->setMax(1);
0
-		Application::SessionPtr session1(pool->get("stub/railsapp"));
0
-		Application::SessionPtr session2(pool->get("stub/railsapp"));
0
-		ensure_equals("An attempt to open a session on an already busy app was made", pool->getActive(), 2u);
0
-		ensure_equals("No new app has been spawned", pool->getCount(), 1u);
0
+		// TODO: How do we test this?
0
   }
0
   
0
   TEST_METHOD(8) {
0
@@ -349,10 +346,7 @@
0
     // MaxPerApp must be respected.
0
     pool->setMax(3);
0
     pool->setMaxPerApp(1);
0
-		
0
-		Application::SessionPtr session1 = pool->get("stub/minimal-railsapp");
0
-		Application::SessionPtr session2 = pool2->get("stub/minimal-railsapp");
0
-		ensure_equals("Only 1 application should have been spawned", pool->getCount(), 1u);
0
+		// TODO: how do we test this?
0
   }
0
 
0
 #endif /* USE_TEMPLATE */