Many fixes for r2564.

- Pop queued transaction from the beginning instead of end
- Restore test for begin with a tid
- Add test to check transactions unlocking mechanism
- Add/fix comments

git-svn-id: https://svn.erp5.org/repos/neo/trunk@2569 71dcc9de-d417-0410-9af5-da40c76e7ee4

neo/master/transactions.py

@@ -389,6 +389,7 @@ class TransactionManager(object):
             A storage node has been lost, don't expect a reply from it for
             current transactions
         """
+        # iterate over a copy because _unlockPending may alter the dict
         for tid, txn in self._tid_dict.items():
             if txn.forget(uuid):
                 self._unlockPending()
@@ -396,34 +397,32 @@ class TransactionManager(object):
     def _unlockPending(self):
         # unlock pending transactions
         while self._queue:
-            tid = self._queue[0][1]
-            # _queue can contain un-prepared transactions
+            uuid, tid = self._queue.pop(0)
             txn = self._tid_dict.get(tid, None)
+            # _queue can contain un-prepared transactions
             if txn is not None and txn.locked():
-                self._queue.pop()
                 self._on_commit(tid, txn)
             else:
+                self._queue.insert(0, (uuid, tid))
                 break
 
     def abortFor(self, node):
         """
             Abort pending transactions initiated by a node
         """
-        neo.logging.debug('Abort for %s', node)
-        # nothing to do
-        if node not in self._node_dict:
-            return
-        # remove transactions
+        neo.logging.debug('Abort TXN for %s', node)
         uuid = node.getUUID()
-        remove = self.remove
-        for tid in self._node_dict[node].keys():
-            remove(uuid, tid)
-        # the code below is usefull only during an import
+        # XXX: this loop is usefull only during an import
         for nuuid, ntid in list(self._queue):
             if nuuid == uuid:
-                self._queue.remove((uuid, tid))
-        # discard node entry
-        del self._node_dict[node]
+                self._queue.remove((uuid, ntid))
+        if node in self._node_dict:
+            # remove transactions
+            remove = self.remove
+            for tid in self._node_dict[node].keys():
+                remove(uuid, tid)
+            # discard node entry
+            del self._node_dict[node]
 
     def log(self):
         neo.logging.info('Transactions:')

neo/tests/__init__.py

@@ -375,6 +375,9 @@ class NeoUnitTestBase(NeoTestBase):
     def checkAnswerTransactionInformation(self, conn, **kw):
         return self.checkAnswerPacket(conn, Packets.AnswerTransactionInformation, **kw)
 
+    def checkAnswerBeginTransaction(self, conn, **kw):
+        return self.checkAnswerPacket(conn, Packets.AnswerBeginTransaction, **kw)
+
     def checkAnswerTids(self, conn, **kw):
         return self.checkAnswerPacket(conn, Packets.AnswerTIDs, **kw)
 

neo/tests/master/testClientHandler.py

@@ -79,9 +79,16 @@ class MasterClientHandlerTests(NeoUnitTestBase):
         calls = tm.mockGetNamedCalls('begin')
         self.assertEqual(len(calls), 1)
         calls[0].checkArgs(client_uuid, None)
+        self.checkAnswerBeginTransaction(conn)
         # Client asks for a TID
+        conn = self.getFakeConnection(client_uuid, self.client_address)
         self.app.tm = tm_org
         service.askBeginTransaction(conn, tid1)
+        calls = tm.mockGetNamedCalls('begin')
+        self.assertEqual(len(calls), 1)
+        calls[0].checkArgs(client_uuid, None)
+        args = self.checkAnswerBeginTransaction(conn, decode=True)
+        self.assertEqual(args, (tid1, ))
 
     def test_08_askNewOIDs(self):
         service = self.service

neo/tests/master/testTransactions.py

@@ -35,6 +35,11 @@ class testTransactionManager(NeoUnitTestBase):
     def makeUUID(self, i):
         return '\0' * 12 + pack('!Q', i)
 
+    def makeNode(self, i):
+        uuid = self.makeUUID(i)
+        node = Mock({'getUUID': uuid, '__hash__': 0})
+        return uuid, node
+
     def testTransaction(self):
         # test data
         node = Mock({'__repr__': 'Node'})
@@ -108,7 +113,7 @@ class testTransactionManager(NeoUnitTestBase):
         self.assertEqual(txnman.getPendingList(), [])
         self.assertFalse(txnman.hasPending())
         # ...and the lock is available
-        txnman.begin(self.getNextTID())
+        txnman.begin(client_uuid, self.getNextTID())
 
     def test_getNextOIDList(self):
         txnman = TransactionManager(lambda tid, txn: None)
@@ -226,14 +231,30 @@ class testTransactionManager(NeoUnitTestBase):
 
     def testClientDisconectsAfterBegin(self):
         client1_uuid = self.makeUUID(1)
-        client2_uuid = self.makeUUID(2)
         tm = TransactionManager(lambda tid, txn: None)
         tid1 = self.getNextTID()
         tid2 = self.getNextTID()
-        tm.begin(tid1)
+        tm.begin(client1_uuid, tid1)
         node1 = Mock({'getUUID': client1_uuid, '__hash__': 0})
         tm.abortFor(node1)
-        tm.begin(tid2)
+        self.assertTrue(tid1 not in tm)
+
+    def testUnlockPending(self):
+        callback = Mock()
+        uuid1, node1 = self.makeNode(1)
+        uuid2, node2 = self.makeNode(2)
+        storage_uuid = self.makeUUID(3)
+        tm = TransactionManager(callback)
+        ttid1 = tm.begin(uuid1)
+        ttid2 = tm.begin(uuid2)
+        tid1 = tm.prepare(node1, ttid1, 1, [], [storage_uuid], 0)
+        tid2 = tm.prepare(node2, ttid2, 1, [], [storage_uuid], 0)
+        tm.lock(tid2, storage_uuid)
+        # txn 2 is still blocked by txn 1
+        self.assertEqual(len(callback.getNamedCalls('__call__')), 0)
+        tm.lock(tid1, storage_uuid)
+        # both transactions are unlocked when txn 1 is fully locked
+        self.assertEqual(len(callback.getNamedCalls('__call__')), 2)
 
 if __name__ == '__main__':
     unittest.main()