Implemented RedisLuaJournal

This journal takes advantage of Redis's scripting feature. You can create scripts in lua and Redis will execute them for you. This lowers latency and dramatically decreases number of requests that has to be send.

src/Kdyby/Redis/DI/RedisExtension.php

@@ -78,7 +78,7 @@ public function loadConfiguration()
 
 		if ($config['journal']) {
 			$builder->addDefinition($this->prefix('cacheJournal'))
-				->setClass('Kdyby\Redis\RedisJournal');
+				->setClass('Kdyby\Redis\RedisLuaJournal');
 
 			// overwrite
 			$builder->removeDefinition('nette.cacheJournal');

src/Kdyby/Redis/RedisClient.php

@@ -46,8 +46,6 @@
  * @method discard() Discard all commands issued after MULTI
  * @method dump(string $key) Return a serialized version of the value stored at the specified key.
  * @method echo(string $message) Echo the given string
- * @method eval(string $script, int $numkeys, string $key1, string $key2 = NULL, string $arg1 = NULL, string $arg2 = NULL) Execute a Lua script server side
- * @method evalSha(string $sha1, int $numkeys, string $key1, string $key2 = NULL, string $arg1 = NULL, string $arg2 = NULL) Execute a Lua script server side
  * @method exists(string $key) Determine if a key exists
  * @method expire(string $key, int $seconds) Set a key's time to live in seconds
  * @method expireAt(string $key, int $timestamp) Set the expiration for a key as a UNIX timestamp
@@ -208,6 +206,8 @@ class RedisClient extends Nette\Object implements \ArrayAccess
 	 */
 	private $lock;
 
+	private static $exceptionCmd = array('evalsha' => 0);
+
 
 
 	/**
@@ -313,6 +313,14 @@ protected function send($cmd, array $args = array())
 
 			if ($result instanceof \Redis) {
 				$result = strtolower($cmd) === 'multi' ? 'OK' : 'QUEUED';
+
+			} elseif ($result === FALSE && ($msg = $this->driver->getLastError())) {
+				if (!isset(self::$exceptionCmd[strtolower($cmd)])) {
+					throw new \RedisException($msg);
+				}
+
+			} else {
+				$this->driver->clearLastError();
 			}
 
 			if ($this->panel) {
@@ -326,7 +334,6 @@ protected function send($cmd, array $args = array())
 			throw new RedisClientException($e->getMessage(), $e->getCode(), $e);
 		}
 
-
 		return $result;
 	}
 
@@ -397,6 +404,25 @@ public function exec()
 
 
 
+	/**
+	 * Execute a Lua script server side
+	 */
+	public function evalScript($script, array $keys = array(), array $args = array())
+	{
+		$script = trim($script);
+
+		$result = $this->send('evalsha', array(sha1($script), array_merge($keys, $args), count($keys)));
+		if ($result === FALSE && stripos($this->driver->getLastError(), 'NOSCRIPT') !== FALSE) {
+			$this->driver->clearLastError();
+			$sha = $this->driver->script('load', $script);
+			$result = $this->send('evalsha', array($sha, array_merge($keys, $args), count($keys)));
+		}
+
+		return $result;
+	}
+
+
+
 	/**
 	 * @return ExclusiveLock
 	 */

src/Kdyby/Redis/RedisLuaJournal.php

@@ -0,0 +1,141 @@
+<?php
+
+/**
+ * This file is part of the Kdyby (http://www.kdyby.org)
+ *
+ * Copyright (c) 2008 Filip Procházka ([email protected])
+ *
+ * For the full copyright and license information, please view the file license.md that was distributed with this source code.
+ */
+
+namespace Kdyby\Redis;
+
+use Kdyby;
+use Nette;
+use Nette\Caching\Cache;
+
+
+
+/**
+ * Redis journal for tags and priorities of cached values.
+ *
+ * @author Filip Procházka <[email protected]>
+ */
+class RedisLuaJournal extends Nette\Object implements Nette\Caching\Storages\IJournal
+{
+
+	/** @internal cache structure */
+	const NS_NETTE = 'Nette.Journal';
+
+	/** dependency */
+	const PRIORITY = 'priority',
+		TAGS = 'tags',
+		KEYS = 'keys';
+
+	/**
+	 * @var RedisClient
+	 */
+	private $client;
+
+	/**
+	 * @var array
+	 */
+	private $scriptSha = array();
+
+	/**
+	 * @var array
+	 */
+	private $script = array();
+
+
+
+	/**
+	 * @param RedisClient $client
+	 */
+	public function __construct(RedisClient $client)
+	{
+		$this->client = $client;
+	}
+
+
+
+	/**
+	 * Writes entry information into the journal.
+	 *
+	 * @param  string $key
+	 * @param  array  $dp
+	 *
+	 * @return void
+	 */
+	public function write($key, array $dp)
+	{
+		$args = self::flattenDp($dp);
+
+		$result = $this->client->evalScript($this->getScript('write'), array($key), $args);
+		if ($result !== TRUE) {
+			throw new RedisClientException("Failed to successfully execute lua script journal.write($key)");
+		}
+	}
+
+
+
+	/**
+	 * Cleans entries from journal.
+	 *
+	 * @param  array  $conds
+	 *
+	 * @return array of removed items or NULL when performing a full cleanup
+	 */
+	public function clean(array $conds)
+	{
+		$args = self::flattenDp($conds);
+
+		$result = $this->client->evalScript($this->getScript('clean'), array(), $args);
+		if (!is_array($result) && $result !== TRUE) {
+			throw new RedisClientException("Failed to successfully execute lua script journal.clean()");
+		}
+
+		return is_array($result) ? array_unique($result) : NULL;
+	}
+
+
+
+	private static function flattenDp($array)
+	{
+		if (isset($array[Cache::TAGS])) {
+			$array[Cache::TAGS] = (array) $array[Cache::TAGS];
+		}
+
+		$res = array();
+		foreach (array_intersect_key($array, array_flip(array(Cache::TAGS, Cache::PRIORITY, Cache::ALL))) as $key => $value) {
+			$res[] = $key;
+			if (is_array($value)) {
+				$res[] = count($value);
+				foreach ($value as $item) {
+					$res[] = $item;
+				}
+
+			} else {
+				$res[] = -1;
+				$res[] = $value;
+			}
+		}
+
+		return $res;
+	}
+
+
+
+	private function getScript($name)
+	{
+		if (isset($this->script[$name])) {
+			return $this->script[$name];
+		}
+
+		$script = file_get_contents(__DIR__ . '/scripts/common.lua');
+		$script .= file_get_contents(__DIR__ . '/scripts/journal.' . $name . '.lua');
+
+		return $this->script[$name] = $script;
+	}
+
+}

src/Kdyby/Redis/scripts/common.lua

@@ -0,0 +1,67 @@
+
+local formatKey = function (key, suffix)
+    local res = "Nette.Journal:" .. key:gsub("\x00", ":")
+    if suffix ~= nil then
+        res = res .. ":" .. suffix
+    end
+
+    return res
+end
+
+local priorityEntries = function (priority)
+    return redis.call('zRangeByScore', formatKey("priority"), 0, priority)
+end
+
+local entryTags = function (key)
+    return redis.call('lRange', formatKey(key, "tags"), 0, -1)
+end
+
+local tagEntries = function (tag)
+    return redis.call('lRange', formatKey(tag, "keys"), 0, -1)
+end
+
+local cleanEntry = function (keys)
+    for i, key in pairs(keys) do
+        local tags = entryTags(key)
+
+        -- redis.call('multi')
+        for i, tag in pairs(tags) do
+            redis.call('lRem', formatKey(tag, "keys"), 0, key)
+        end
+
+        -- drop tags of entry and priority, in case there are some
+        redis.call('del', formatKey(key, "tags"), formatKey(key, "priority"))
+        redis.call('zRem', formatKey("priority"), key)
+
+        -- redis.call('exec')
+    end
+end
+
+-- builds table from serialized arguments
+local readArgs = function (args)
+    local res = {}
+    local counter = 0
+    local key
+    local tmp
+
+    for i, item in pairs(args) do
+        if counter > 0 then
+            if res[key] == nil then res[key] = {} end
+
+            tmp = res[key]
+            res[key][#tmp + 1] = item
+            counter = counter - 1
+
+            if counter == 0 then key = nil end
+
+        elseif counter < 0 then
+            res[key] = item
+            key = nil
+
+        else
+            if key == nil then key = item else counter = tonumber(item); end
+        end
+    end
+
+    return res
+end

src/Kdyby/Redis/scripts/journal.clean.lua

@@ -0,0 +1,37 @@
+
+local conds = readArgs(ARGV)
+
+if conds["all"] ~= nil then
+    -- redis.call('multi')
+    for i, value in pairs(redis.call('keys', "Nette.Journal:*")) do
+        redis.call('del', value)
+    end
+    -- redis.call('exec')
+
+    return redis.status_reply("Ok")
+end
+
+local entries = {}
+if conds["tags"] ~= nil then
+    for i, tag in pairs(conds["tags"]) do
+        local found = tagEntries(tag)
+        if #found > 0 then
+            cleanEntry(found)
+            for i, key in pairs(found) do
+                entries[#entries + 1] = key
+            end
+        end
+    end
+end
+
+if conds["priority"] ~= nil then
+    local found = priorityEntries(conds["priority"])
+    if #found > 0 then
+        cleanEntry(found)
+        for i, key in pairs(found) do
+            entries[#entries + 1] = key
+        end
+    end
+end
+
+return entries

src/Kdyby/Redis/scripts/journal.write.lua

@@ -0,0 +1,24 @@
+
+local dp = readArgs(ARGV)
+
+-- clean the entry key
+cleanEntry({KEYS[1]})
+
+-- write the entry key
+-- redis.call('multi')
+
+-- add entry to each tag & tag to entry
+if dp["tags"] ~= nil then
+    for i, tag in pairs(dp["tags"]) do
+        redis.call('rPush', formatKey(tag, "keys") , KEYS[1])
+        redis.call('rPush', formatKey(KEYS[1], "tags") , tag)
+    end
+end
+
+if dp["priority"] ~= nil then
+    redis.call('zAdd', formatKey("priority"), dp["priority"], KEYS[1])
+end
+
+-- redis.call('exec')
+
+return redis.status_reply("Ok")