Merge pull request #65 from manover/fix_connection_leak

Fix connection leaks caused by a stale job queue

asyncmongo/connection.py

@@ -128,6 +128,9 @@ def _socket_close(self):
  if callback:
  callback(None, InterfaceError('connection closed'))
  finally:
+ # Flush the job queue, don't call the callbacks associated with the remaining jobs
+ # since they have already been called as error callback on connection closing
+ self.__job_queue = []
  self.__alive = False
  self.__pool.cache(self)
 
@@ -139,6 +142,9 @@ def _close(self):
  if callback:
  callback(None, InterfaceError('connection closed'))
  finally:
+ # Flush the job queue, don't call the callbacks associated with the remaining jobs
+ # since they have already been called as error callback on connection closing
+ self.__job_queue = []
  self.__alive = False
  self.__stream.close()
 
@@ -232,9 +238,9 @@ def _parse_response(self, response):
  self.__request_id = None
  self.__callback = None
  if not self.__job_queue:
- # skip adding to the cache because there is something else 
+ # skip adding to the cache because there is something else
  # that needs to be called on this connection for this request
- # (ie: we authenticted, but still have to send the real req)
+ # (ie: we authenticated, but still have to send the real req)
  self.__pool.cache(self)
 
  try:

asyncmongo/pool.py

@@ -34,6 +34,7 @@ def get_connection_pool(self, pool_id, *args, **kwargs):
  self._pools = {}
  if pool_id not in self._pools:
  self._pools[pool_id] = ConnectionPool(*args, **kwargs)
+ # logging.debug("%s: _connections = %d", pool_id, self._pools[pool_id]._connections)
  return self._pools[pool_id]
 
  @classmethod