+ self._waitThreads()
+ self.assertEqual(self.done.get_nowait(), ['six'])
+ self._addThread(target=self._doRemoveSet, args=(['two']))
+ self._waitThreads()
+ self.assertEqual(self.done.get_nowait(), ['two'])
+ self.ls.release()
+ # reset lockset
+ self._setUpLS()
+
+ @_Repeat
+ def testConcurrentSharedSetLock(self):
+ # share the set-lock...
+ self.ls.acquire(None, shared=1)
+ # ...another thread can share it too
+ self._addThread(target=self._doLockSet, args=(None, 1))
+ self._waitThreads()
+ self.assertEqual(self.done.get_nowait(), 'DONE')
+ # ...or just share some elements
+ self._addThread(target=self._doLockSet, args=(['one', 'three'], 1))
+ self._waitThreads()
+ self.assertEqual(self.done.get_nowait(), 'DONE')
+ # ...but not add new ones or remove any
+ t = self._addThread(target=self._doAddSet, args=(['nine']))
+ self._addThread(target=self._doRemoveSet, args=(['two'], ))
+ self.assertRaises(Queue.Empty, self.done.get_nowait)
+ # this just releases the set-lock
+ self.ls.release([])
+ t.join(60)
+ self.assertEqual(self.done.get_nowait(), 'DONE')
+ # release the lock on the actual elements so remove() can proceed too
+ self.ls.release()
+ self._waitThreads()
+ self.failUnlessEqual(self.done.get_nowait(), ['two'])
+ # reset lockset
+ self._setUpLS()
+
+ @_Repeat
+ def testConcurrentExclusiveSetLock(self):
+ # acquire the set-lock...
+ self.ls.acquire(None, shared=0)
+ # ...no one can do anything else
+ self._addThread(target=self._doLockSet, args=(None, 1))
+ self._addThread(target=self._doLockSet, args=(None, 0))
+ self._addThread(target=self._doLockSet, args=(['three'], 0))
+ self._addThread(target=self._doLockSet, args=(['two'], 1))
+ self._addThread(target=self._doAddSet, args=(['nine']))
+ self.assertRaises(Queue.Empty, self.done.get_nowait)
+ self.ls.release()
+ self._waitThreads()
+ for _ in range(5):
+ self.assertEqual(self.done.get(True, 1), 'DONE')
+ # cleanup
+ self._setUpLS()
+
+ @_Repeat
+ def testConcurrentSetLockAdd(self):
+ self.ls.acquire('one')
+ # Another thread wants the whole SetLock
+ self._addThread(target=self._doLockSet, args=(None, 0))
+ self._addThread(target=self._doLockSet, args=(None, 1))
+ self.assertRaises(Queue.Empty, self.done.get_nowait)
+ self.assertRaises(AssertionError, self.ls.add, 'four')
+ self.ls.release()
+ self._waitThreads()
+ self.assertEqual(self.done.get_nowait(), 'DONE')
+ self.assertEqual(self.done.get_nowait(), 'DONE')
+ self.ls.acquire(None)
+ self._addThread(target=self._doLockSet, args=(None, 0))
+ self._addThread(target=self._doLockSet, args=(None, 1))
+ self.assertRaises(Queue.Empty, self.done.get_nowait)
+ self.ls.add('four')
+ self.ls.add('five', acquired=1)
+ self.ls.add('six', acquired=1, shared=1)
+ self.assertEquals(self.ls._list_owned(),
+ set(['one', 'two', 'three', 'five', 'six']))
+ self.assertEquals(self.ls._is_owned(), True)
+ self.assertEquals(self.ls._names(),
+ set(['one', 'two', 'three', 'four', 'five', 'six']))
+ self.ls.release()
+ self._waitThreads()
+ self.assertEqual(self.done.get_nowait(), 'DONE')
+ self.assertEqual(self.done.get_nowait(), 'DONE')
+ self._setUpLS()
+
+ @_Repeat
+ def testEmptyLockSet(self):
+ # get the set-lock
+ self.assertEqual(self.ls.acquire(None), set(['one', 'two', 'three']))
+ # now empty it...
+ self.ls.remove(['one', 'two', 'three'])
+ # and adds/locks by another thread still wait
+ self._addThread(target=self._doAddSet, args=(['nine']))
+ self._addThread(target=self._doLockSet, args=(None, 1))
+ self._addThread(target=self._doLockSet, args=(None, 0))
+ self.assertRaises(Queue.Empty, self.done.get_nowait)