Handle deconnections and fix requirements

Author: adrien-berchet

.gitignore

@@ -75,6 +75,7 @@ htmlcov/
 .tox/
 .coverage
 .coverage.*
+.coverage-*
 .cache
 nosetests.xml
 coverage.xml

.gitlab-ci.yml

@@ -13,12 +13,14 @@ include:
      file:
         - /ci/jobs/auto-release.yml
         - /ci/lib/pytest-template.yml
-        - /ci/jobs/py36.yml
         - /ci/jobs/py37.yml
         - /ci/jobs/py38.yml
+        - /ci/jobs/py39.yml
         - /ci/jobs/coverage.yml
 
 .tox-template:
   variables:
-    EXTRA_MODULES: archive/2021-05:py-mpi4py
-    bb5_cpus_per_task: 6
+    EXTRA_MODULES:
+      archive/2021-12:py-mpi4py
+    bb5_cpus_per_task: 8
+    bb5_memory: 6G

bluepyparallel/database.py

@@ -58,8 +58,16 @@ def __del__(self):
     @property
     def connection(self):
         """Get a connection to the database."""
+        try:
+            if self._connection.connection.dbapi_connection is None:
+                self._connection.close()
+                self._connection = None
+        except AttributeError:
+            self._connection = None
+
         if self._connection is None:
             self._connection = self.engine.connect()
+
         return self._connection
 
     def get_url(self):

bluepyparallel/parallel.py

@@ -133,8 +133,8 @@ def get_mapper(self, batch_size=None, chunk_size=None, **kwargs):
         """Get a map."""
 
         def _mapper(func, iterable, *func_args, **func_kwargs):
-            func = self.mappable_func(func, *func_args, **func_kwargs)
-            return self._with_batches(map, func, iterable)
+            mapped_func = self.mappable_func(func, *func_args, **func_kwargs)
+            return self._with_batches(map, mapped_func, iterable)
 
         return _mapper
 
@@ -156,10 +156,10 @@ def get_mapper(self, batch_size=None, chunk_size=None, **kwargs):
         self._chunksize_to_kwargs(chunk_size, kwargs, label="chunksize")
 
         def _mapper(func, iterable, *func_args, **func_kwargs):
-            func = self.mappable_func(func, *func_args, **func_kwargs)
+            mapped_func = self.mappable_func(func, *func_args, **func_kwargs)
             return self._with_batches(
                 partial(self.pool.imap_unordered, **kwargs),
-                func,
+                mapped_func,
                 iterable,
             )
 
@@ -195,17 +195,19 @@ def get_mapper(self, batch_size=None, chunk_size=None, **kwargs):
         self._chunksize_to_kwargs(chunk_size, kwargs)
 
         def _mapper(func, iterable, *func_args, **func_kwargs):
-            func = self.mappable_func(func, *func_args, **func_kwargs)
+            mapped_func = self.mappable_func(func, *func_args, **func_kwargs)
             return self._with_batches(
-                partial(self.lview.imap, **kwargs), func, iterable, batch_size=batch_size
+                partial(self.lview.imap, **kwargs), mapped_func, iterable, batch_size=batch_size
             )
 
         return _mapper
 
     def shutdown(self):
         """Remove zmq."""
-        if self.rc is not None:  # pragma: no cover
+        try:
             self.rc.close()
+        except Exception:  # pragma: no cover ; pylint: disable=broad-except
+            pass
 
 
 class DaskFactory(ParallelFactory):
@@ -256,13 +258,13 @@ def get_mapper(self, batch_size=None, chunk_size=None, **kwargs):
         self._chunksize_to_kwargs(chunk_size, kwargs, label="batch_size")
 
         def _mapper(func, iterable, *func_args, **func_kwargs):
-            def _dask_mapper(func, iterable):
-                futures = self.client.map(func, iterable, **kwargs)
+            def _dask_mapper(in_dask_func, iterable):
+                futures = self.client.map(in_dask_func, iterable, **kwargs)
                 for _future, result in dask.distributed.as_completed(futures, with_results=True):
                     yield result
 
-            func = self.mappable_func(func, *func_args, **func_kwargs)
-            return self._with_batches(_dask_mapper, func, iterable, batch_size=batch_size)
+            mapped_func = self.mappable_func(func, *func_args, **func_kwargs)
+            return self._with_batches(_dask_mapper, mapped_func, iterable, batch_size=batch_size)
 
         return _mapper
 

setup.py

@@ -1,23 +1,19 @@
 """Setup for the BluePyParallel package."""
 import imp
-import sys
 
 from setuptools import find_packages
 from setuptools import setup
 
-if sys.version_info < (3, 6):
-    sys.exit("Sorry, Python < 3.6 is not supported")
-
 # Read the contents of the README file
 with open("README.rst", encoding="utf-8") as f:
     README = f.read()
 
 reqs = [
     "pandas",
-    "ipyparallel",
+    "ipyparallel<7",
     "dask[dataframe, distributed]>=2.30",
     "dask-mpi>=2.20",
-    "sqlalchemy<1.4",
+    "sqlalchemy>1.4",
     "sqlalchemy-utils",
     "tqdm",
 ]
@@ -49,12 +45,12 @@
         "docs": doc_reqs,
     },
     packages=find_packages(exclude=["tests"]),
-    python_requires=">=3.6",
+    python_requires=">=3.7",
     classifiers=[
         "Programming Language :: Python",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
     ],
 )

tests/test_evaluator.py

@@ -31,10 +31,12 @@ def _interrupting_function(row, *args, **kwargs):
     return _evaluation_function(row, *args, **kwargs)
 
 
-def _slow_function(row, *args, sleep_time=0.02, **kwargs):
+def _slow_function(row, *args, **kwargs):
     """Mock evaluation function."""
     if "sleep_time" in row:
         sleep_time = row["sleep_time"]
+    else:
+        sleep_time = kwargs.get("sleep_time", 0.02)
     time.sleep(sleep_time)
     return _evaluation_function(row, *args, **kwargs)
 

tox.ini

@@ -7,7 +7,6 @@ testdeps =
     pytest-benchmark
     pytest-cov
     pytest-html
-    pytest-xdist
 
 [tox]
 envlist =
@@ -23,11 +22,10 @@ setenv =
     COVERAGE_FILE = {env:COVERAGE_FILE:.coverage-{envname}}
 deps = {[base]testdeps}
 commands_pre =
-    - ipcluster stop --cluster-id={[base]name}_{envname} --debug
+    - ipcluster stop --cluster-id={[base]name}_{envname} --signal 9 --debug
     ipcluster start -n 2 --daemonize --log-to-file --cluster-id={[base]name}_{envname} --debug
 commands =
     pytest \
-    -n 2 \
     --basetemp={envtmpdir} \
     --cov={[base]name} \
     --cov-branch \
@@ -77,6 +75,7 @@ commands_pre =
 commands_post =
 
 [testenv:lint]
+basepython = python3.7
 deps =
     {[base]testdeps}
     pycodestyle
@@ -94,6 +93,7 @@ commands_pre =
 commands_post =
 
 [testenv:format]
+basepython = python3.7
 skip_install = true
 deps =
     black