Merge pull request #7 from gaillard/bug

Collapse cleaning updates into normal updates.

Author: Jonathan Gaillard

src/Locker.php

@@ -85,10 +85,20 @@ public function readLock($id, $staleDuration)
 
         while (time() < $timeoutTimestamp) {
             $readerId = new \MongoId();
-            $query = ['_id' => $id, 'writing' => false, 'writePending' => false];
+            $query = [
+                '_id' => $id,
+                '$or' => [
+                    //normal case where we have zero or more readers and no writers pending
+                    ['writing' => false, 'writePending' => false],
+                    //case where a pending write is stale
+                    ['writing' => false, 'writePending' => true, 'writeStaleTs' => ['$lte' => new \MongoDate()]],
+                    //case where a writer is stale
+                    ['writing' => true, 'writeStaleTs' => ['$lte' => new \MongoDate()]],
+                ],
+            ];
             $update = [
                 '$push' => ['readers' => ['id' => $readerId, 'staleTs' => $staleTimestamp]],
-                '$set' => ['writeStaleTs' => null],
+                '$set' => ['writing' => false, 'writePending' => false, 'writeStaleTs' => null],
             ];
             try {
                 if ($this->collection->update($query, $update, ['upsert' => true])['n'] === 1) {
@@ -100,10 +110,6 @@ public function readLock($id, $staleDuration)
                 }
             }
 
-            if ($this->clearStuckWrite($id)) {
-                continue;
-            }
-
             usleep($this->pollDuration);
         }
 
@@ -119,7 +125,11 @@ public function readLock($id, $staleDuration)
      */
     public function readUnlock($id, \MongoId $readerId)
     {
-        $this->collection->update(['_id' => $id], ['$pull' => ['readers' => ['id' => $readerId]]]);
+        $this->collection->update(
+            ['_id' => $id],
+            //pull this reader id, or any stale readers
+            ['$pull' => ['readers' => ['$or' => [['id' => $readerId], ['staleTs' => ['$lte' => new \MongoDate()]]]]]]
+        );
         $this->collection->remove(['_id' => $id, 'writing' => false, 'readers' => ['$size' => 0]]);
     }
 
@@ -144,7 +154,17 @@ public function writeLock($id, $staleDuration)
         $staleTimestamp = new \MongoDate((int)min(time() + $staleDuration, PHP_INT_MAX));
 
         while (time() < $timeoutTimestamp) {
-            $query = ['_id' => $id, 'writing' => false, 'readers' => ['$size' => 0]];
+            $query = [
+                '_id' => $id,
+                '$or' => [
+                    //normal case when readers are done
+                    ['writing' => false, 'readers' => ['$size' => 0]],
+                    //to clean where writer is stuck
+                    ['writing' => true, 'writeStaleTs' => ['$lte' => new \MongoDate()]],
+                    //to clean where all readers are stuck
+                    ['writing' => false, 'readers.staleTs' => ['$not' => ['$gt' => new \MongoDate()]]],
+                ],
+            ];
             $update = [
                 '_id' => $id,
                 'writing' => true,
@@ -162,11 +182,10 @@ public function writeLock($id, $staleDuration)
                 }
             }
 
-            if ($this->clearStuckWrite($id) || $this->clearStuckRead($id)) {
-                continue;
-            }
-
-            $this->collection->update(['_id' => $id], ['$set' => ['writePending' => true]]);
+            $this->collection->update(
+                ['_id' => $id],
+                ['$set' => ['writePending' => true, 'writeStaleTs' => $staleTimestamp]]
+            );
 
             usleep($this->pollDuration);
         }
@@ -184,19 +203,4 @@ public function writeUnlock($id)
     {
         $this->collection->remove(['_id' => $id]);
     }
-
-    private function clearStuckWrite($id)
-    {
-        return $this->collection->remove(
-            ['_id' => $id, 'writing' => true, 'writeStaleTs' => ['$lte' => new \MongoDate()]]
-        )['n'] === 1;
-    }
-
-    private function clearStuckRead($id)
-    {
-        $now = new \MongoDate();
-        $query = ['_id' => $id, 'writing' => false, 'readers.staleTs' => ['$lte' => $now]];
-        $update = ['$pull' => ['readers' => ['staleTs' => ['$lte' => $now]]]];
-        return $this->collection->update($query, $update)['n'] === 1;
-    }
 }

tests/LockerTest.php

@@ -134,6 +134,24 @@ public function readLockClearStuck()
         $this->locker->readLock('theId', 1000);
     }
 
+    /**
+     * @test
+     */
+    public function readLockClearStuckWritePending()
+    {
+        $this->locker->readLock('theId', 2);
+
+        //the attempted write lock below only sets writePending since we already have a read lock
+        $locker = new Locker($this->collection, 0, 1);
+        try {
+            $locker->writeLock('theId', 0);
+            $this->fail();
+        } catch (\Exception $e) {
+        }
+
+        $this->locker->readLock('theId', 1000);
+    }
+
     /**
      * @test
      * @expectedException \Exception
@@ -220,6 +238,19 @@ public function readUnlockExistingReader()
         $this->assertGreaterThan(time() + 990, $actualReaders[0]['staleTs']->sec);
     }
 
+    /**
+     * @test
+     */
+    public function readUnlockExistingReaderStale()
+    {
+        $this->locker->readLock('theId', 0);
+
+        $readerId = $this->locker->readLock('theId', 1000);
+        $this->locker->readUnlock('theId', $readerId);
+
+        $this->assertSame(0, $this->collection->count());
+    }
+
     /**
      * @test
      */