From f1cde0f1aad3f4c69c66f4deae4d4ad72cc716c4 Mon Sep 17 00:00:00 2001
From: Aaron Schulz <aschulz@wikimedia.org>
Date: Tue, 30 Apr 2013 20:00:20 -0700
Subject: [PATCH] [JobQueue] Allow using gzip on larger jobs in JobQueueRedis.

Change-Id: I74ebdcfa0d3f2bcdc429394f7873fa25b7eb0f58
---
 includes/job/JobQueueRedis.php | 42 +++++++++++++++++++++++++++++++---
 1 file changed, 39 insertions(+), 3 deletions(-)

diff --git a/includes/job/JobQueueRedis.php b/includes/job/JobQueueRedis.php
index a3f847f1f5..1f5b7619bb 100644
--- a/includes/job/JobQueueRedis.php
+++ b/includes/job/JobQueueRedis.php
@@ -61,6 +61,7 @@ class JobQueueRedis extends JobQueue {
 	protected $redisPool;
 
 	protected $server; // string; server address
+	protected $compression; // string; compression method to use
 
 	const MAX_AGE_PRUNE = 604800; // integer; seconds a job can live once claimed (7 days)
 
@@ -73,12 +74,14 @@ class JobQueueRedis extends JobQueue {
 	 *   - redisServer : A hostname/port combination or the absolute path of a UNIX socket.
 	 *                   If a hostname is specified but no port, the standard port number
 	 *                   6379 will be used. Required.
+	 *   - compression : The type of compression to use; one of (none,gzip).
 	 * @param array $params
 	 */
 	public function __construct( array $params ) {
 		parent::__construct( $params );
 		$params['redisConfig']['serializer'] = 'none'; // make it easy to use Lua
 		$this->server = $params['redisServer'];
+		$this->compression = isset( $params['compression'] ) ? $params['compression'] : 'none';
 		$this->redisPool = RedisConnectionPool::singleton( $params['redisConfig'] );
 	}
 
@@ -238,7 +241,7 @@ class JobQueueRedis extends JobQueue {
 			$args[] = (string)$item['uuid'];
 			$args[] = (string)$item['sha1'];
 			$args[] = (string)$item['rtimestamp'];
-			$args[] = (string)serialize( $item );
+			$args[] = (string)$this->serialize( $item );
 		}
 		static $script =
 <<<LUA
@@ -310,7 +313,7 @@ LUA;
 				}
 
 				JobQueue::incrStats( 'job-pop', $this->type );
-				$item = unserialize( $blob );
+				$item = $this->unserialize( $blob );
 				if ( $item === false ) {
 					wfDebugLog( 'JobQueueRedis', "Could not unserialize {$this->type} job." );
 					continue;
@@ -552,7 +555,7 @@ LUA;
 	 */
 	public function getJobFromUidInternal( $uid, RedisConnRef $conn ) {
 		try {
-			$item = unserialize( $conn->hGet( $this->getQueueKey( 'h-data' ), $uid ) );
+			$item = $this->unserialize( $conn->hGet( $this->getQueueKey( 'h-data' ), $uid ) );
 			if ( !is_array( $item ) ) { // this shouldn't happen
 				throw new MWException( "Could not find job with ID '$uid'." );
 			}
@@ -735,6 +738,39 @@ LUA;
 		return false;
 	}
 
+	/**
+	 * @param array $fields
+	 * @return string Serialized and possibly compressed version of $fields
+	 */
+	protected function serialize( array $fields ) {
+		$blob = serialize( $fields );
+		if ( $this->compression === 'gzip'
+			&& strlen( $blob ) >= 1024 && function_exists( 'gzdeflate' ) )
+		{
+			$object = (object)array( 'blob' => gzdeflate( $blob ), 'enc' => 'gzip' );
+			$blobz = serialize( $object );
+			return ( strlen( $blobz ) < strlen( $blob ) ) ? $blobz : $blob;
+		} else {
+			return $blob;
+		}
+	}
+
+	/**
+	 * @param string $blob
+	 * @return array|bool Unserialized version of $blob or false
+	 */
+	protected function unserialize( $blob ) {
+		$fields = unserialize( $blob );
+		if ( is_object( $fields ) ) {
+			if ( $fields->enc === 'gzip' && function_exists( 'gzinflate' ) ) {
+				$fields = unserialize( gzinflate( $fields->blob ) );
+			} else {
+				$fields = false;
+			}
+		}
+		return is_array( $fields ) ? $fields : false;
+	}
+
 	/**
 	 * Get a connection to the server that handles all sub-queues for this queue
 	 *
-- 
2.20.1