Make a bunch of incompatible changes to the PoolCounter.

It wasn't finished, so it's not a big deal.

* Use the term workers instead of threads, which fits better for a multiserver setup.
* The API is now more accurate for our goals (I hope).
* Add support for using the parse from another worker.
* Use child classes instead of array callbacks.
* The daemon is written in C using libevent instead of python using twistd.
* The hash function used is that of Bob Jenkins, with files hash.c and hash.h directly copied from memcached 1.4.5
* Although similar in a few aspects to memcached assoc.c hash table, this is a different hash table implementation. Most important:
** The usage of a double linked list in the hash table.
** Growing is not performed using a maintenance thread. Since the entries are shortlived, it just waits for the old hash table to disappear.
* Note: valgrind 3.5.0 (2009-8-19) does not support accept4 (added in r10955, 2009-11-25). In the meantime you need to use HAVE_ACCEPT4=0 for running with valgrind (as you would need for a non-linux system).
* Sending SIGUSR1 to the daemon gracefully restarts it. The maximum limits will be doubled until the old instance finishes (ie. all its client connections expire).
* Do not try to test it with instances calling an ?action=purge They will serialize on the "UPDATE  `page` SET page_touched" query instead of being serialized by the PoolCounter.
* The workers parameter is not stored by the poolcounter. It is expected that all requests with the same key will also have the same value. A reduction in new entries will not take effect if that number is working (not even when they end, if there are waiting entries). But an increase will increase throughput even for old queued requests.

1 files changed, 156 insertions, 39 deletions

diff --git a/includes/PoolCounter.php b/includes/PoolCounter.php
index 2564fbc67f02..825ae34ca154 100644
--- a/includes/PoolCounter.php
+++ b/includes/PoolCounter.php
@@ -1,6 +1,66 @@
 <?php
 
+/**
+ *  When you have many workers (threads/servers) giving service, and a 
+ * cached item expensive to produce expires, you may get several workers
+ * doing the job at the same time.
+ *  Given enough requests and the item expiring fast (non-cacheable, 
+ * lots of edits...) that single work can end up unfairly using most (all)
+ * of the cpu of the pool. This is also known as 'Michael Jackson effect'.
+ *  The PoolCounter provides semaphore semantics for restricting the number
+ * of workers that may be concurrently performing such single task.
+ */
+
 abstract class PoolCounter {
+	
+	/* Return codes */
+	const LOCKED = 1; 	/* Lock acquired */
+	const RELEASED = 2; /* Lock released */
+	const DONE = 3;		/* Another one did the work for you */
+	
+	const ERROR = -1;		/* Indeterminate error */
+	const NOT_LOCKED = -2;	/* Called release() with no lock held */
+	const QUEUE_FULL = -3;	/* There are already maxqueue workers on this lock */
+	const TIMEOUT = -4;		/* Timeout exceeded */
+	const LOCK_HELD = -5;	/* Cannot acquire another lock while you have one lock held */
+
+	/**
+	 * I want to do this task and I need to do it myself.
+	 * 
+	 * @return Locked/Error
+	 */
+	abstract function acquireForMe();
+
+	/**
+	 * I want to do this task, but if anyone else does it 
+	 * instead, it's also fine for me. I will read its cached data.
+	 * 
+	 * @return Locked/Done/Error
+	 */
+	abstract function acquireForAnyone();
+
+	/**
+	 * I have successfully finished my task.
+	 * Lets another one grab the lock, and returns the workers 
+	 * waiting on acquireForAnyone()
+	 * 
+	 * @return Released/NotLocked/Error
+	 */
+	abstract function release();
+	
+	/**
+	 *  $key: All workers with the same key share the lock.
+	 *  $workers: It wouldn't be a good idea to have more than this number of 
+	 * workers doing the task simultaneously.
+	 *  $maxqueue: If this number of workers are already working/waiting, 
+	 * fail instead of wait.
+	 *  $timeout: Maximum time in seconds to wait for the lock.
+	 */
+	protected $key, $workers, $maxqueue, $timeout;
+	
+	/**
+	 * Create a Pool counter. This should only be called from the PoolWorks.
+	 */
 	public static function factory( $type, $key ) {
 		global $wgPoolCounterConf;
 		if ( !isset( $wgPoolCounterConf[$type] ) ) {
@@ -8,57 +68,114 @@ abstract class PoolCounter {
 		}
 		$conf = $wgPoolCounterConf[$type];
 		$class = $conf['class'];
+		
 		return new $class( $conf, $type, $key );
 	}
-
-	abstract public function acquire();
-	abstract public function release();
-	abstract public function wait();
-
-	public function executeProtected( $mainCallback, $dirtyCallback = false ) {
-		$status = $this->acquire();
-		if ( !$status->isOK() ) {
-			return $status;
-		}
-		if ( !empty( $status->value['overload'] ) ) {
-			# Overloaded. Try a dirty cache entry.
-			if ( $dirtyCallback ) {
-				if ( call_user_func( $dirtyCallback ) ) {
-					$this->release();
-					return Status::newGood();
-				}
-			}
-
-			# Wait for a thread
-			$status = $this->wait();
-			if ( !$status->isOK() ) {
-				$this->release();
-				return $status;
-			}
-		}
-		# Call the main callback
-		call_user_func( $mainCallback );
-		return $this->release();
+	
+	protected function __construct( $conf, $type, $key ) {
+		$this->key = $key;
+		$this->workers  = $conf['workers'];
+		$this->maxqueue = $conf['maxqueue'];
+		$this->timeout  = $conf['timeout'];
 	}
 }
 
 class PoolCounter_Stub extends PoolCounter {
-	public function acquire() {
-		return Status::newGood();
+	function acquireForMe() {
+		return PoolCounter::LOCKED;
 	}
 
-	public function release() {
-		return Status::newGood();
+	function acquireForAnyone() {
+		return PoolCounter::LOCKED;
 	}
 
-	public function wait() {
-		return Status::newGood();
+	function release() {
+		return PoolCounter::RELEASED;
 	}
-
-	public function executeProtected( $mainCallback, $dirtyCallback = false ) {
-		call_user_func( $mainCallback );
-		return Status::newGood();
+	
+	public function __construct() {
+		/* No parameters needed */
 	}
 }
 
+/**
+ * Handy class for dealing with PoolCounters using class members instead of callbacks.
+ */
+abstract class PoolCounterWork {
+	protected $cacheable = false; //Does this override getCachedWork() ?
+	
+	/**
+	 * Actually perform the work, caching it if needed.
+	 */
+	abstract function doWork();
 
+	/**
+	 * Retrieve the work from cache
+	 * @return mixed work result or false
+	 */
+	function getCachedWork() {
+		return false;
+	}
+
+	/**
+	 * A work not so good (eg. expired one) but better than an error 
+	 * message.
+	 * @return mixed work result or false
+	 */
+	function fallback() {
+		return false;
+	}
+	
+	/**
+	 * Do something with the error, like showing it to the user.
+	 */
+	function error( $status ) {	
+		return false;
+	}
+	
+	/**
+	 * Get the result of the work (whatever it is), or false.
+	 */
+	function execute( $skipcache = false ) {
+		if ( $this->cacheable && !$skipcache ) {
+			$status = $this->poolCounter->acquireForAnyone();
+		} else {
+			$status = $this->poolCounter->acquireForMe();
+		}
+		
+		$result = false;
+		switch ( is_int( $status ) ? $status : PoolCounter::ERROR ) {
+			case PoolCounter::LOCKED:
+				$result = $this->doWork();
+				$this->poolCounter->release();
+				return $result;
+			
+			case PoolCounter::DONE:
+				$result = $this->getCachedWork();
+				if ( $result === false ) {
+					/* That someone else work didn't serve us.
+					 * Acquire the lock for me
+					 */
+					return $this->execute( true );
+				}
+				return $result;
+				
+			case PoolCounter::QUEUE_FULL:
+			case PoolCounter::TIMEOUT:
+				$result = $this->fallback();
+				
+				if ( $result !== false ) {
+					return $result;
+				}
+				/* no break */
+			
+			case PoolCounter::ERROR:
+			default:
+				return $this->error( $status );
+		}
+	}
+	
+	function __construct( $type, $key ) {
+		$this->poolCounter = PoolCounter::factory( $type, $key );
+	}
+}