Fix resource leaks when nodes and threaded tests end

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

neo/admin/app.py

@@ -79,6 +79,12 @@ class Application(object):
         self.reset()
         registerLiveDebugger(on_log=self.log)
 
+    def close(self):
+        self.listening_conn = None
+        self.nm.close()
+        self.em.close()
+        del self.__dict__
+
     def reset(self):
         self.bootstrapped = False
         self.master_conn = None

neo/admin/handler.py

@@ -113,11 +113,12 @@ class MasterEventHandler(EventHandler):
 
     def _connectionLost(self, conn):
         app = self.app
-        assert app.master_conn in (conn, None)
-        app.dispatcher.clear()
-        app.reset()
-        app.uuid = None
-        raise PrimaryFailure
+        if app.listening_conn: # if running
+            assert app.master_conn in (conn, None)
+            app.dispatcher.clear()
+            app.reset()
+            app.uuid = None
+            raise PrimaryFailure
 
     def connectionFailed(self, conn):
         self._connectionLost(conn)

neo/lib/event.py

@@ -30,6 +30,11 @@ class EpollEventManager(object):
         self.epoll = Epoll()
         self._pending_processing = []
 
+    def close(self):
+        for c in self.connection_dict.values():
+            c.close()
+        del self.__dict__
+
     def getConnectionList(self):
         # XXX: use index
         return [x for x in self.connection_dict.values() if not x.isAborted()]

neo/lib/node.py

@@ -26,6 +26,8 @@ from neo.lib import attributeTracker
 class Node(object):
     """This class represents a node."""
 
+    _connection = None
+
     def __init__(self, manager, address=None, uuid=None,
             state=NodeStates.UNKNOWN):
         self._state = state
@@ -33,7 +35,6 @@ class Node(object):
         self._uuid = uuid
         self._manager = manager
         self._last_state_change = time()
-        self._connection = None
         manager.add(self)
 
     def notify(self, packet):
@@ -88,7 +89,7 @@ class Node(object):
             Callback from node's connection when closed
         """
         assert self._connection is not None
-        self._connection = None
+        del self._connection
         self._manager._updateIdentified(self)
 
     def setConnection(self, connection):
@@ -261,6 +262,8 @@ class NodeManager(object):
         self._state_dict = {}
         self._identified_dict = {}
 
+    close = __init__
+
     def add(self, node):
         if node in self._node_set:
             neo.lib.logging.warning('adding a known node %r, ignoring', node)

neo/master/app.py

@@ -94,6 +94,12 @@ class Application(object):
 
         registerLiveDebugger(on_log=self.log)
 
+    def close(self):
+        self.listening_conn = None
+        self.nm.close()
+        self.em.close()
+        del self.__dict__
+
     def log(self):
         self.em.log()
         self.nm.log()

neo/master/handlers/client.py

@@ -31,12 +31,13 @@ class ClientServiceHandler(MasterHandler):
     def connectionLost(self, conn, new_state):
         # cancel its transactions and forgot the node
         app = self.app
-        node = app.nm.getByUUID(conn.getUUID())
-        assert node is not None
-        app.tm.abortFor(node)
-        node.setState(NodeStates.DOWN)
-        app.broadcastNodesInformation([node])
-        app.nm.remove(node)
+        if app.listening_conn: # if running
+            node = app.nm.getByUUID(conn.getUUID())
+            assert node is not None
+            app.tm.abortFor(node)
+            node.setState(NodeStates.DOWN)
+            app.broadcastNodesInformation([node])
+            app.nm.remove(node)
 
     def askNodeInformation(self, conn):
         # send informations about master and storages only

neo/neoctl/app.py

@@ -39,6 +39,9 @@ class TerminalNeoCTL(object):
     def __init__(self, address):
         self.neoctl = NeoCTL(address)
 
+    def __del__(self):
+        self.neoctl.close()
+
     # Utility methods (could be functions)
     def asNodeState(self, value):
         return NodeStates.getByName(value.upper())

neo/neoctl/neoctl.py

@@ -38,6 +38,10 @@ class NeoCTL(object):
         self.handler = CommandEventHandler(self)
         self.response_queue = []
 
+    def close(self):
+        self.em.close()
+        del self.__dict__
+
     def __getConnection(self):
         if not self.connected:
             self.connection = ClientConnection(self.em, self.handler,

neo/storage/app.py

@@ -88,6 +88,12 @@ class Application(object):
 
         registerLiveDebugger(on_log=self.log)
 
+    def close(self):
+        self.listening_conn = None
+        self.nm.close()
+        self.em.close()
+        del self.__dict__
+
     def _poll(self):
         self.em.poll(1)
 

neo/storage/handlers/__init__.py

@@ -26,7 +26,8 @@ from neo.lib.protocol import NodeStates, NodeTypes, Packets, Errors
 class BaseMasterHandler(EventHandler):
 
     def connectionLost(self, conn, new_state):
-        raise PrimaryFailure('connection lost')
+        if self.app.listening_conn: # if running
+            raise PrimaryFailure('connection lost')
 
     def stopOperation(self, conn):
         raise OperationFailure('operation stopped')

neo/storage/handlers/client.py

@@ -36,8 +36,9 @@ class ClientOperationHandler(BaseClientAndStorageOperationHandler):
     def connectionLost(self, conn, new_state):
         uuid = conn.getUUID()
         node = self.app.nm.getByUUID(uuid)
-        assert node is not None, conn
-        self.app.nm.remove(node)
+        if self.app.listening_conn: # if running
+            assert node is not None, conn
+            self.app.nm.remove(node)
 
     def abortTransaction(self, conn, ttid):
         self.app.tm.abort(ttid)

neo/tests/threaded/__init__.py

@@ -16,7 +16,7 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 
-import os, random, socket, sys, tempfile, threading, time, types
+import os, random, socket, sys, tempfile, threading, time, types, weakref
 from collections import deque
 from functools import wraps
 from Queue import Queue, Empty
@@ -367,8 +367,9 @@ class NEOCluster(object):
         ip = getVirtualIp('master')
         self.master_nodes = ' '.join('%s:%s' % (ip, i)
                                      for i in xrange(master_count))
-        kw = dict(cluster=self, getReplicas=replicas, getPartitions=partitions,
-                  getAdapter=adapter, getReset=clear_databases)
+        weak_self = weakref.proxy(self)
+        kw = dict(cluster=weak_self, getReplicas=replicas, getAdapter=adapter,
+                  getPartitions=partitions, getReset=clear_databases)
         self.master_list = [MasterApplication(address=(ip, i), **kw)
                             for i in xrange(master_count)]
         ip = getVirtualIp('storage')
@@ -383,8 +384,8 @@ class NEOCluster(object):
                              for i, x in enumerate(db_list)]
         ip = getVirtualIp('admin')
         self.admin_list = [AdminApplication(address=(ip, 0), **kw)]
-        self.client = ClientApplication(self)
-        self.neoctl = NeoCTL(self)
+        self.client = ClientApplication(weak_self)
+        self.neoctl = NeoCTL(weak_self)
 
     # A few shortcuts that work when there's only 1 master/storage/admin
     @property
@@ -493,6 +494,13 @@ class NEOCluster(object):
         txn = transaction.TransactionManager()
         return txn, self.db.open(transaction_manager=txn)
 
+    def __del__(self):
+        self.neoctl.close()
+        for node_type in 'admin', 'storage', 'master':
+            for node in getattr(self, node_type + '_list'):
+                node.close()
+        self.client.em.close()
+
 
 class NEOThreadedTest(NeoTestBase):