Sequence cache

.gitignore

@@ -7,3 +7,5 @@ test/meta/Proto/
 test/main/data/urls/parser.dump.new
 .idea
 test/onphp
+/nbproject/
+

core/Base/Assert.class.php

@@ -277,6 +277,14 @@ public static function isUnreachable($message = 'unreachable code reached')
 			throw new WrongArgumentException($message);
 		}
 
+		public static function isObject($object, $message = null)
+		{
+			if (!is_object($object))
+				throw new WrongArgumentException(
+					$message.' not object given'
+				);
+		}
+
 		/// exceptionless methods
 		//@{
 		public static function checkInteger($value)

core/Cache/PeclMemcached.class.php

@@ -1,6 +1,6 @@
 <?php
 /***************************************************************************
- *   Copyright (C) 2006-2008 by Konstantin V. Arkhipov                     *
+ *   Copyright (C) 2006-2012 by Konstantin V. Arkhipov                     *
  *                                                                         *
  *   This program is free software; you can redistribute it and/or modify  *
  *   it under the terms of the GNU Lesser General Public License as        *
@@ -21,38 +21,36 @@ class PeclMemcached extends CachePeer
 	{
 		const DEFAULT_PORT		= 11211;
 		const DEFAULT_HOST		= '127.0.0.1';
+		const DEFAULT_TIMEOUT	= 1;
 
-		private $instance = null;
+		private $instance		= null;
+		private $requestTimeout = null;
+		private $connectTimeout = null;
+		private $host			= null;
+		private $port			= null;
+		private $triedConnect	= false;
 
 		/**
 		 * @return PeclMemcached
 		**/
 		public static function create(
 			$host = Memcached::DEFAULT_HOST,
-			$port = Memcached::DEFAULT_PORT
+			$port = Memcached::DEFAULT_PORT,
+			$connectTimeout = PeclMemcached::DEFAULT_TIMEOUT
 		)
 		{
-			return new self($host, $port);
+			return new self($host, $port, $connectTimeout);
 		}
 
 		public function __construct(
 			$host = Memcached::DEFAULT_HOST,
-			$port = Memcached::DEFAULT_PORT
+			$port = Memcached::DEFAULT_PORT,
+			$connectTimeout = PeclMemcached::DEFAULT_TIMEOUT
 		)
 		{
-			$this->instance = new Memcache();
-
-			try {
-				try {
-					$this->instance->pconnect($host, $port);
-				} catch (BaseException $e) {
-					$this->instance->connect($host, $port);
-				}
-
-				$this->alive = true;
-			} catch (BaseException $e) {
-				// bad luck.
-			}
+			$this->host = $host;
+			$this->port = $port;
+			$this->connectTimeout = $connectTimeout;
 		}
 
 		public function __destruct()
@@ -66,11 +64,20 @@ public function __destruct()
 			}
 		}
 
+		public function isAlive()
+		{
+			$this->ensureTriedToConnect();
+
+			return parent::isAlive();
+		}
+
 		/**
 		 * @return PeclMemcached
 		**/
 		public function clean()
 		{
+			$this->ensureTriedToConnect();
+
 			try {
 				$this->instance->flush();
 			} catch (BaseException $e) {
@@ -82,6 +89,8 @@ public function clean()
 
 		public function increment($key, $value)
 		{
+			$this->ensureTriedToConnect();
+
 			try {
 				return $this->instance->increment($key, $value);
 			} catch (BaseException $e) {
@@ -91,6 +100,8 @@ public function increment($key, $value)
 
 		public function decrement($key, $value)
 		{
+			$this->ensureTriedToConnect();
+
 			try {
 				return $this->instance->decrement($key, $value);
 			} catch (BaseException $e) {
@@ -100,6 +111,8 @@ public function decrement($key, $value)
 
 		public function getList($indexes)
 		{
+			$this->ensureTriedToConnect();
+
 			return
 				($return = $this->get($indexes))
 					? $return
@@ -108,6 +121,8 @@ public function getList($indexes)
 
 		public function get($index)
 		{
+			$this->ensureTriedToConnect();
+
 			try {
 				return $this->instance->get($index);
 			} catch (BaseException $e) {
@@ -124,6 +139,8 @@ public function get($index)
 
 		public function delete($index)
 		{
+			$this->ensureTriedToConnect();
+
 			try {
 				// second parameter required, wrt new memcached protocol:
 				// delete key 0 (see process_delete_command in the memcached.c)
@@ -138,6 +155,8 @@ public function delete($index)
 
 		public function append($key, $data)
 		{
+			$this->ensureTriedToConnect();
+
 			try {
 				return $this->instance->append($key, $data);
 			} catch (BaseException $e) {
@@ -147,10 +166,58 @@ public function append($key, $data)
 			Assert::isUnreachable();
 		}
 
+		/**
+		 * @param float $requestTimeout time in seconds
+		 * @return PeclMemcached
+		 */
+		public function setTimeout($requestTimeout)
+		{
+			$this->ensureTriedToConnect();
+			$this->requestTimeout = $requestTimeout;
+			$this->instance->setServerParams($this->host, $this->port, $requestTimeout);
+
+			return $this;
+		}
+
+		/**
+		 * @return float 
+		 */
+		public function getTimeout()
+		{
+			return $this->requestTimeout;
+		}
+
+		protected function ensureTriedToConnect()
+		{
+			if ($this->triedConnect) 
+				return $this;
+
+			$this->triedConnect = true;
+			$this->instance = new Memcache();
+
+			try {
+
+				try {
+					$this->instance->pconnect($this->host, $this->port, $this->connectTimeout);
+				} catch (BaseException $e) {
+					$this->instance->connect($this->host, $this->port, $this->connectTimeout);
+				}
+
+				$this->alive = true;
+
+			} catch (BaseException $e) {
+				// bad luck.
+			}
+
+			return $this;
+		}
+
 		protected function store(
 			$action, $key, $value, $expires = Cache::EXPIRES_MEDIUM
 		)
 		{
+			$this->ensureTriedToConnect();
+
 			try {
 				return
 					$this->instance->$action(
@@ -167,5 +234,5 @@ protected function store(
 
 			Assert::isUnreachable();
 		}
+
 	}
-?>

core/Cache/SequentialCache.class.php

@@ -0,0 +1,132 @@
+<?php
+/****************************************************************************
+ *   Copyright (C) 2012 by Artem Naumenko                                   *
+ *                                                                          *
+ *   This program is free software; you can redistribute it and/or modify   *
+ *   it under the terms of the GNU Lesser General Public License as         *
+ *   published by the Free Software Foundation; either version 3 of the     *
+ *   License, or (at your option) any later version.                        *
+ *                                                                          *
+ ****************************************************************************/
+
+	final class SequentialCache extends CachePeer
+	{
+		/**
+		 * List of all peers, including master
+		 * @var array of CachePeer
+		 */
+		private $list		= array();
+
+		/**
+		 * List of slaves only
+		 * @var array of CachePeer
+		 */
+		private $slaves	= array();
+
+		/**
+		 * @var CachePeer
+		 */
+		private $master	= null;
+
+		/**
+		 * @param CachePeer $master
+		 * @param array $slaves or CachePeer
+		 * @return SequentialCache 
+		 */
+		public static function create(CachePeer $master, array $slaves = array())
+		{
+			return new self($master, $slaves);
+		}
+
+		/**
+		 * @param CachePeer $master
+		 * @param array $slaves or CachePeer
+		 */
+		public function __construct(CachePeer $master, array $slaves = array())
+		{
+			$this->setMaster($master);
+
+			foreach ($slaves as $cache) {
+				$this->addPeer($cache);
+			}
+		}
+
+		/**
+		 * @param CachePeer $master
+		 * @return \SequentialCache 
+		 */
+		public function setMaster(CachePeer $master)
+		{
+			$this->master = $master;
+			$this->list = $this->slaves;
+			array_unshift($this->list, $this->master);
+
+			return $this;
+		}
+
+		/**
+		 * @param CachePeer $master
+		 * @return \SequentialCache 
+		 */
+		public function addPeer(CachePeer $peer)
+		{
+			$this->list[]	= $peer;
+			$this->slaves[]	= $peer;
+
+			return $this;
+		}
+
+		public function get($key)
+		{
+			foreach ($this->list as $val) {
+				/* @var $val CachePeer */
+				$result = $val->get($key);
+
+				if (
+					!empty($result)
+					|| $val->isAlive()
+				) {
+					return $result;
+				}
+			}
+
+			throw new RuntimeException('All peers are dead');
+		}
+
+		public function append($key, $data)
+		{
+			return $this->foreachItem(__METHOD__, func_get_args());
+		}
+
+		public function decrement($key, $value)
+		{
+			throw new UnsupportedMethodException('decrement is not supported');
+		}
+
+		public function delete($key)
+		{
+			return $this->foreachItem(__METHOD__, func_get_args());
+		}
+
+		public function increment($key, $value)
+		{
+			throw new UnsupportedMethodException('increment is not supported');
+		}
+
+		protected function store($action, $key, $value, $expires = Cache::EXPIRES_MEDIUM)
+		{
+			return $this->foreachItem(__METHOD__, func_get_args());
+		}
+
+		private function foreachItem($method, array $args)
+		{
+			$result = true;
+
+			foreach ($this->list as $peer) {
+				/* @var $peer CachePeer */
+				$result = call_user_func_array(array($peer, $method), $args) && $result;
+			}
+
+			return $result;
+		}
+	}

main/Monitoring/PinbedPeclMemcached.class.php

@@ -22,15 +22,17 @@ final class PinbedPeclMemcached extends PeclMemcached
 		**/
 		public static function create(
 			$host = Memcached::DEFAULT_HOST,
-			$port = Memcached::DEFAULT_PORT
+			$port = Memcached::DEFAULT_PORT,
+			$connectTimeout = PeclMemcached::DEFAULT_TIMEOUT
 		)
 		{
 			return new self($host, $port);
 		}
 
 		public function __construct(
 			$host = Memcached::DEFAULT_HOST,
-			$port = Memcached::DEFAULT_PORT
+			$port = Memcached::DEFAULT_PORT,
+			$connectTimeout = PeclMemcached::DEFAULT_TIMEOUT
 		)
 		{
 			$this->host = $host;
@@ -42,7 +44,7 @@ public function __construct(
 					array('pecl_memcached_connect' => $host.'_'.$port)
 				);
 
-			parent::__construct($host, $port);
+			parent::__construct($host, $port, $connectTimeout);
 
 			if (PinbaClient::isEnabled())
 				PinbaClient::me()->timerStop(