fixed a majour bug in distributed lock: when using a pool of Redis clients, user can use

Aaron Boxer · Aaron Boxer · commit 5c2a3e5a06a8 · 2011-09-09T13:51:06.000-04:00
one client to lock, and another to unlock. This was a problem because the client stored the
distributed lock, and the lock timeout, which is used to release the lock. Solution is to move
lock out of the client.
diff --git a/src/ServiceStack.Redis/Support/Locking/DisposableDistributedLock.cs b/src/ServiceStack.Redis/Support/Locking/DisposableDistributedLock.cs
@@ -9,6 +9,8 @@ public class DisposableDistributedLock : IDisposable
     {
         private readonly IDistributedLock myLock;
         private readonly long lockState;
+        private readonly long lockExpire;
+        private readonly IRedisClient myClient;
 
         /// <summary>
         /// Lock
@@ -19,8 +21,9 @@ public class DisposableDistributedLock : IDisposable
         /// <param name="lockTimeout">in seconds</param>
         public DisposableDistributedLock(IRedisClient client, string globalLockKey, int acquisitionTimeout, int lockTimeout)
         {
-            myLock = new DistributedLock(client);
-            lockState = myLock.Lock(globalLockKey, acquisitionTimeout, lockTimeout);
+            myLock = new DistributedLock();
+            myClient = client;
+            lockState = myLock.Lock(globalLockKey, acquisitionTimeout, lockTimeout, out lockExpire, myClient);
         }
 
 
@@ -29,12 +32,18 @@ public long LockState
             get { return lockState; } 
         }
 
+        public long LockExpire
+        {
+            get { return lockExpire; }
+        }
+
+
         /// <summary>
         /// unlock
         /// </summary>
         public void Dispose()
         {
-            myLock.Unlock();
+            myLock.Unlock(lockExpire, myClient);
         }
     }
 }
diff --git a/src/ServiceStack.Redis/Support/Locking/DistributedLock.cs b/src/ServiceStack.Redis/Support/Locking/DistributedLock.cs
@@ -10,23 +10,20 @@ public class DistributedLock : IDistributedLock
         public const int LOCK_RECOVERED = 2;
 
         protected string lockKey;
-        protected long lockExpire;
 
-        protected IRedisClient myClient;
-
-        public DistributedLock(IRedisClient client)
-        {
-            myClient = client;
-        }
 
 		/// <summary>
 		/// acquire distributed, non-reentrant lock on key
 		/// </summary>
 	    /// <param name="key">global key for this lock</param>
 		/// <param name="acquisitionTimeout">timeout for acquiring lock</param>
 		/// <param name="lockTimeout">timeout for lock, in seconds (stored as value against lock key) </param>
-        public virtual long Lock(string key, int acquisitionTimeout, int lockTimeout)
+        /// <param name="client"></param>
+        /// <param name="lockExpire"></param>
+        public virtual long Lock(string key, int acquisitionTimeout, int lockTimeout, out long lockExpire, IRedisClient client)
 		{
+		    lockExpire = 0;
+
             // cannot lock on a null key
             if (key == null)
                 return LOCK_NOT_ACQUIRED;
@@ -42,7 +39,7 @@ public virtual long Lock(string key, int acquisitionTimeout, int lockTimeout)
 			var ts = (DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0));
 			var newLockExpire = CalculateLockExpire(ts, lockTimeout);
 
-		    var localClient = (RedisClient)myClient;
+            var localClient = (RedisClient)client;
             int wasSet = localClient.SetNX(key, BitConverter.GetBytes(newLockExpire));
 			int totalTime = 0;
             while (wasSet == LOCK_NOT_ACQUIRED && totalTime < acquisitionTimeout)
@@ -105,12 +102,12 @@ public virtual long Lock(string key, int acquisitionTimeout, int lockTimeout)
 		/// <summary>
 		/// unlock key
 		/// </summary>
-		public virtual bool Unlock()
+		public virtual bool Unlock(long lockExpire, IRedisClient client)
 		{
 			if (lockExpire <= 0)
 				return false;
 		    long lockVal = 0;
-            var localClient = AcquireClient();
+            var localClient = (RedisClient)client;
             using (var pipe = localClient.CreatePipeline())
             {
                
@@ -153,10 +150,5 @@ private static long CalculateLockExpire(TimeSpan ts, int timeout)
             return (long)(ts.TotalSeconds + timeout + 1.5);
         }
 
-        protected virtual RedisClient AcquireClient()
-        {
-            return (RedisClient)myClient;
-        }
-
 	}
 }
diff --git a/src/ServiceStack.Redis/Support/Locking/IDistributedLock.cs b/src/ServiceStack.Redis/Support/Locking/IDistributedLock.cs
@@ -5,9 +5,7 @@
     /// </summary>
 	public interface IDistributedLock
 	{
-
-	    long Lock(string key, int acquisitionTimeout, int lockTimeout);
-
-	    bool Unlock();
+	    long Lock(string key, int acquisitionTimeout, int lockTimeout, out long lockExpire, IRedisClient client);
+	    bool Unlock(long lockExpire, IRedisClient client);
 	}
 }
diff --git a/src/ServiceStack.Redis/Support/Queue/Implementation/RedisSequentialWorkQueue.Locks.cs b/src/ServiceStack.Redis/Support/Queue/Implementation/RedisSequentialWorkQueue.Locks.cs
@@ -11,37 +11,39 @@ namespace ServiceStack.Redis.Support.Queue.Implementation
     /// </summary>
     public partial class RedisSequentialWorkQueue<T> 
     {
-        public class DequeueManager : DistributedLock
+        public class DequeueManager
         {
- 
-            private bool ownsClient;
             protected readonly RedisSequentialWorkQueue<T> workQueue;
             protected readonly string workItemId;
             protected readonly PooledRedisClientManager clientManager;
             protected readonly int numberOfDequeuedItems;
             protected int numberOfProcessedItems;
-            public DequeueManager(IRedisClient client,    PooledRedisClientManager clientManager, RedisSequentialWorkQueue<T> workQueue, string workItemId, int numberOfDequeuedItems) : base(client)
+            private readonly DistributedLock myLock;
+            private readonly string dequeueLockKey;
+            private int dequeueLockTimeout = 300;
+            private long lockExpire;
+
+            public DequeueManager(PooledRedisClientManager clientManager, RedisSequentialWorkQueue<T> workQueue, string workItemId, string dequeueLockKey, int numberOfDequeuedItems, int dequeueLockTimeout) 
             {
                 this.workQueue = workQueue;
                 this.workItemId = workItemId;
                 this.clientManager = clientManager;
-                ownsClient = false;
                 this.numberOfDequeuedItems = numberOfDequeuedItems;
-            }
-
-            public override long Lock(string key, int acquisitionTimeout, int lockTimeout)
-            {
-                long rc = base.Lock(key, acquisitionTimeout, lockTimeout);
-                // do not hang on to the client reference. This lock may be held for a long time.
-                ReleaseClient();
-                return rc;
+                myLock = new DistributedLock();
+                this.dequeueLockKey = dequeueLockKey;
+                this.dequeueLockTimeout = dequeueLockTimeout;
             }
 
             public void DoneProcessedWorkItem()
             {
                 numberOfProcessedItems++;
                 if (numberOfProcessedItems == numberOfDequeuedItems)
-                    Unlock();
+                {
+                    using (var disposable = new PooledRedisClientManager.DisposablePooledClient<SerializingRedisClient>(clientManager))
+                    {
+                        Unlock(disposable.Client);
+                    }
+                }
             }
             /// <summary>
             /// 
@@ -52,17 +54,23 @@ public void UpdateNextUnprocessed(T newWorkItem)
                 workQueue.Update(workItemId, numberOfProcessedItems, newWorkItem);
             }
 
-            public override bool Unlock()
+            public long Lock(int acquisitionTimeout, IRedisClient client)
             {
-                return PopAndUnlock(numberOfDequeuedItems);
+                return myLock.Lock(dequeueLockKey, acquisitionTimeout, dequeueLockTimeout, out lockExpire, client);
+            }
+
+            public bool Unlock(IRedisClient client)
+            {
+                return PopAndUnlock(numberOfDequeuedItems, client);
             }
 
             /// <summary>
             /// 
             /// </summary>
             /// <param name="numProcessed"></param>
+            /// <param name="client"></param>
             /// <returns></returns>
-            public bool PopAndUnlock(int numProcessed)
+            public bool PopAndUnlock(int numProcessed, IRedisClient client)
             {
                 if (numProcessed < 0)
                     numProcessed = 0;
@@ -74,27 +82,20 @@ public bool PopAndUnlock(int numProcessed)
 
                 // unlock work queue id
                 workQueue.Unlock(workItemId);
-                bool rc = base.Unlock();
-
-                ReleaseClient();
-                return rc;
-            }
-            protected void ReleaseClient()
-            {
-                if (ownsClient && myClient != null)
-                    clientManager.DisposeClient((RedisNativeClient)myClient);
-                myClient = null;
-                ownsClient = false;
+                return  myLock.Unlock(lockExpire, client);
             }
 
-            protected override RedisClient AcquireClient()
+            /// <summary>
+            /// 
+            /// </summary>
+            /// <param name="numProcessed"></param>
+            /// <returns></returns>
+            public bool PopAndUnlock(int numProcessed)
             {
-                if (myClient == null)
+                using (var disposable = new PooledRedisClientManager.DisposablePooledClient<SerializingRedisClient>(clientManager))
                 {
-                    myClient = clientManager.GetClient();
-                    ownsClient = true;
+                    return PopAndUnlock(numProcessed, disposable.Client);
                 }
-                return (RedisClient) myClient;
             }
         }
     }
diff --git a/src/ServiceStack.Redis/Support/Queue/Implementation/RedisSequentialWorkQueue.cs b/src/ServiceStack.Redis/Support/Queue/Implementation/RedisSequentialWorkQueue.cs
@@ -1,5 +1,6 @@
 ﻿using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using ServiceStack.Redis.Support.Locking;
 
 
@@ -17,13 +18,12 @@ public partial class RedisSequentialWorkQueue<T> : RedisWorkQueue<T>, ISequentia
         private DateTime harvestTime = DateTime.UtcNow;
         private int lockAcquisitionTimeout = 2;
         private int lockTimeout = 2;
-        private int dequeueLockTimeout = 300;
         protected const double  CONVENIENTLY_SIZED_FLOAT = 18014398509481984.0;
 
         // store list of work item ids that have been dequeued
         // this list is checked regularly in harvest zombies method
         private string dequeueIdSet;
-
+        private int dequeueLockTimeout = 300;
 
         private string workItemIdPriorityQueue;
 
@@ -37,9 +37,9 @@ public RedisSequentialWorkQueue(int maxReadPoolSize, int maxWritePoolSize, strin
         public RedisSequentialWorkQueue(int maxReadPoolSize, int maxWritePoolSize, string host, int port, string queueName, int dequeueLockTimeout) 
             : base(maxReadPoolSize, maxWritePoolSize, host, port, queueName)
         {
-            this.dequeueLockTimeout = dequeueLockTimeout;
             dequeueIdSet = queueNamespace.GlobalCacheKey("DequeueIdSet");
             workItemIdPriorityQueue = queueNamespace.GlobalCacheKey("WorkItemIdPriorityQueue");
+            this.dequeueLockTimeout = dequeueLockTimeout;
         }
 
         /// <summary>
@@ -158,12 +158,10 @@ public ISequentialData<T> Dequeue(int maxBatchSize)
                             pipe.Flush();
                         }
 
-                        workItemDequeueManager = new DequeueManager(client, clientManager, this, workItemId, workItems.Count);
-                        string dequeueLockKey = null;
+                        workItemDequeueManager = new DequeueManager(clientManager, this, workItemId, GlobalDequeueLockKey(workItemId), workItems.Count, dequeueLockTimeout);
                         // don't lock if there are no work items to be processed (can't lock on null lock key)
                         if (workItems.Count > 0)
-                            dequeueLockKey = GlobalDequeueLockKey(workItemId);
-                        workItemDequeueManager.Lock(dequeueLockKey, lockAcquisitionTimeout, dequeueLockTimeout);
+                           workItemDequeueManager.Lock(lockAcquisitionTimeout, client);
 
                     }
                     return new SequentialData<T>(workItemId, workItems, workItemDequeueManager);
@@ -173,7 +171,7 @@ public ISequentialData<T> Dequeue(int maxBatchSize)
                 {
                     //release resources
                     if (workItemDequeueManager != null)
-                        workItemDequeueManager.Unlock();
+                        workItemDequeueManager.Unlock(client);
 
                     throw;
                 }
@@ -228,7 +226,7 @@ public bool HarvestZombies()
             bool rc = false;
             using (var disposableClient = clientManager.GetDisposableClient<SerializingRedisClient>())
             {
-            var client = disposableClient.Client;
+                var client = disposableClient.Client;
                 var dequeueWorkItemIds = client.SMembers(dequeueIdSet);
                 if (dequeueWorkItemIds.Length == 0) return false;
 
diff --git a/src/ServiceStack.Redis/Support/Queue/Implementation/SerializingRedisClient.cs b/src/ServiceStack.Redis/Support/Queue/Implementation/SerializingRedisClient.cs
@@ -6,34 +6,13 @@ namespace ServiceStack.Redis.Support.Queue.Implementation
 {
     public class SerializingRedisClient : RedisClient
     {
-        private readonly DistributedLock myLock;
         private ISerializer serializer = new ObjectSerializer();
    
         public SerializingRedisClient(string host, int port)
             : base(host, port)
         {
-             myLock = new DistributedLock(this);
         }
         
-        /// <summary>
-        /// acquire distributed, non-reentrant lock on key
-        /// </summary>
-        /// <param name="key">global key for this lock</param>
-        /// <param name="acquisitionTimeout">timeout for acquiring lock</param>
-        /// <param name="lockTimeout">timeout for lock, in seconds (stored as value against lock key) </param>
-        public long Lock(string key, int acquisitionTimeout, int lockTimeout)
-        {
-            return myLock.Lock(key, acquisitionTimeout, lockTimeout);
-
-        }
-        /// <summary>
-        /// unlock key
-        /// </summary>
-        public bool Unlock()
-        {
-            return myLock.Unlock();
-        }
-
         /// <summary>
         /// customize the client serializer
         /// </summary>
diff --git a/tests/ServiceStack.Redis.Tests/QueueTests.cs b/tests/ServiceStack.Redis.Tests/QueueTests.cs
@@ -29,7 +29,7 @@ public override void OnBeforeEachTest()
         [Test]
         public void TestSequentialWorkQueueUpdate()
         {
-            using (var queue = new RedisSequentialWorkQueue<string>(10, 10, "127.0.0.1", 6379, 1))
+            using (var queue = new RedisSequentialWorkQueue<string>(10, 10, "127.0.0.1", 6379,1))
             {
 
                 for (int i = 0; i < numMessages; ++i)
@@ -70,6 +70,7 @@ public void TestSequentialWorkQueue()
                 // check that half of patient[0] messages are returned
                 for (int i = 0; i < numMessages/2; ++i)
                     Assert.AreEqual(batch.DequeueItems[i], messages0[i]);
+                Assert.AreEqual(numMessages/2, batch.DequeueItems.Count);
                 Thread.Sleep(5000);
                 Assert.IsTrue(queue.HarvestZombies());
                 for (int i = 0; i < batch.DequeueItems.Count; ++i)
@@ -98,18 +99,14 @@ public void TestSequentialWorkQueue()
                 int remaining = batch.DequeueItems.Count-1;
                 batch.PopAndUnlock();
 
-                queue.PrepareNextWorkItem();
-                batch = queue.Dequeue(numMessages);
-                Assert.AreEqual(batch.DequeueItems.Count, remaining);
-
                 //process remaining items
-                queue.PrepareNextWorkItem();
+                Assert.IsTrue(queue.PrepareNextWorkItem());
                 batch = queue.Dequeue(remaining);
                 Assert.AreEqual(batch.DequeueItems.Count, remaining);
-                for (int i = 0; i < numMessages; ++i)
+                for (int i = 0; i < batch.DequeueItems.Count; ++i)
                     batch.DoneProcessedWorkItem();
 
-                queue.PrepareNextWorkItem();
+                Assert.IsFalse(queue.PrepareNextWorkItem());
                 batch = queue.Dequeue(remaining);
                 Assert.AreEqual(batch.DequeueItems.Count, 0);
            
diff --git a/tests/ServiceStack.Redis.Tests/RedisClientTests.cs b/tests/ServiceStack.Redis.Tests/RedisClientTests.cs

Original file line number	Diff line number	Diff line change
`@@ -9,6 +9,8 @@ public class DisposableDistributedLock : IDisposable`
`9`	`9`	`{`
`10`	`10`	`private readonly IDistributedLock myLock;`
`11`	`11`	`private readonly long lockState;`
	`12`	`+ private readonly long lockExpire;`
	`13`	`+ private readonly IRedisClient myClient;`
`12`	`14`
`13`	`15`	`/// <summary>`
`14`	`16`	`/// Lock`
`@@ -19,8 +21,9 @@ public class DisposableDistributedLock : IDisposable`
`19`	`21`	`/// <param name="lockTimeout">in seconds</param>`
`20`	`22`	`public DisposableDistributedLock(IRedisClient client, string globalLockKey, int acquisitionTimeout, int lockTimeout)`
`21`	`23`	`{`
`22`		`- myLock = new DistributedLock(client);`
`23`		`- lockState = myLock.Lock(globalLockKey, acquisitionTimeout, lockTimeout);`
	`24`	`+ myLock = new DistributedLock();`
	`25`	`+ myClient = client;`
	`26`	`+ lockState = myLock.Lock(globalLockKey, acquisitionTimeout, lockTimeout, out lockExpire, myClient);`
`24`	`27`	`}`
`25`	`28`
`26`	`29`
`@@ -29,12 +32,18 @@ public long LockState`
`29`	`32`	`get { return lockState; }`
`30`	`33`	`}`
`31`	`34`
	`35`	`+ public long LockExpire`
	`36`	`+ {`
	`37`	`+ get { return lockExpire; }`
	`38`	`+ }`
	`39`	`+`
	`40`	`+`
`32`	`41`	`/// <summary>`
`33`	`42`	`/// unlock`
`34`	`43`	`/// </summary>`
`35`	`44`	`public void Dispose()`
`36`	`45`	`{`
`37`		`- myLock.Unlock();`
	`46`	`+ myLock.Unlock(lockExpire, myClient);`
`38`	`47`	`}`
`39`	`48`	`}`
`40`	`49`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,9 +5,7 @@`
`5`	`5`	`/// </summary>`
`6`	`6`	`public interface IDistributedLock`
`7`	`7`	`{`
`8`		`-`
`9`		`- long Lock(string key, int acquisitionTimeout, int lockTimeout);`
`10`		`-`
`11`		`- bool Unlock();`
	`8`	`+ long Lock(string key, int acquisitionTimeout, int lockTimeout, out long lockExpire, IRedisClient client);`
	`9`	`+ bool Unlock(long lockExpire, IRedisClient client);`
`12`	`10`	`}`
`13`	`11`	`}`