refactor!: mpi hdf5 (#902)

* fix: cleanup jobpool with mpi

* fix: make Engine.recognizes static and pass MPIService comm as attribute.

* fix: use MPILock instead of single rank selection.

* fix: change variable name to match parent class

* fix: clear of stored data

* fix: change recognizes signature to use MPI communicator instead of MPILock

* fix: test_connectivity.py to take into account new clear system

* fix: bump pull request to allow test for breaking changes

Author: drodarie

.github/workflows/pull_request.yml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: PR Conventional Commit Validation
-        uses:  ytanikin/PRConventionalCommits@1.2.0
+        uses:  ytanikin/PRConventionalCommits@1.3.0
         with:
           task_types: '["feat","fix","docs","test","ci","refactor","perf","revert"]'
 

bsb/core.py

@@ -1,6 +1,7 @@
 import itertools
 import os
 import sys
+import time
 import typing
 
 import numpy as np
@@ -201,8 +202,6 @@ def _bootstrap(self, config, storage, clear=False):
         # Then, `storage` is initialized for the scaffold, and `config` is stored (happens
         # inside the `storage` property).
         self.storage = storage
-        # Synchronize the JobPool static variable so that each core use the same ids.
-        JobPool._next_pool_id = self._comm.bcast(JobPool._next_pool_id, root=0)
 
     storage_cfg = _config_property("storage")
     for attr in _cfg_props:
@@ -388,8 +387,7 @@ def compile(
                 report("Clearing data", level=2)
                 # Clear the placement and connectivity data, but leave any cached files
                 # and morphologies intact.
-                self.clear_placement()
-                self.clear_connectivity()
+                self.clear()
             elif redo:
                 # In order to properly redo things, we clear some placement and connection
                 # data, but since multiple placement/connection strategies can contribute
@@ -791,8 +789,9 @@ def _load_cs_types(
         return cs
 
     def create_job_pool(self, fail_fast=None, quiet=False):
+        id_pool = self._comm.bcast(int(time.time()), root=0)
         pool = JobPool(
-            self, fail_fast=fail_fast, workflow=getattr(self, "_workflow", None)
+            id_pool, self, fail_fast=fail_fast, workflow=getattr(self, "_workflow", None)
         )
         try:
             # Check whether stdout is a TTY, and that it is larger than 0x0

bsb/services/pool.py

@@ -540,9 +540,9 @@ class JobPool:
     _pool_owners = {}
     _tmp_folders = {}
 
-    def __init__(self, scaffold, fail_fast=False, workflow: "Workflow" = None):
+    def __init__(self, id, scaffold, fail_fast=False, workflow: "Workflow" = None):
         self._schedulers: list[concurrent.futures.Future] = []
-        self.id: int = None
+        self.id: int = id
         self._scaffold = scaffold
         self._comm = scaffold._comm
         self._unhandled_errors = []
@@ -563,8 +563,6 @@ def __enter__(self):
         self._context = ExitStack()
         tmp_dirname = self._context.enter_context(tempfile.TemporaryDirectory())
 
-        self.id = JobPool._next_pool_id
-        JobPool._next_pool_id += 1
         JobPool._pool_owners[self.id] = self._scaffold
         JobPool._pools[self.id] = self
         JobPool._tmp_folders[self.id] = tmp_dirname

bsb/storage/__init__.py

@@ -19,6 +19,7 @@
 
 from .. import plugins
 from ..exceptions import UnknownStorageEngineError
+from ..services import MPILock
 from ..services.mpi import MPIService
 
 if typing.TYPE_CHECKING:
@@ -389,7 +390,7 @@ def open_storage(root, comm=None):
     """
     engines = get_engines()
     for name, engine in engines.items():
-        if engine.peek_exists(root) and engine.recognizes(root):
+        if engine.peek_exists(root) and engine.recognizes(root, comm):
             return Storage(name, root, comm, missing_ok=False)
     else:
         for name, engine in engines.items():
@@ -417,7 +418,7 @@ def view_support(engine=None):
     """
     if engine is None:
         return {
-            # Loop over all enginges
+            # Loop over all engines
             engine_name: {
                 # Loop over all features, check whether they're supported
                 feature_name: not isinstance(feature, NotSupported)

bsb/storage/fs/__init__.py

@@ -29,8 +29,8 @@ def versions(self):
         path = Path(self._root) / "versions.txt"
         return json.loads(path.read_text())
 
-    @classmethod
-    def recognizes(cls, root):
+    @staticmethod
+    def recognizes(root, comm):
         try:
             return os.path.exists(root) and os.path.isdir(root)
         except Exception:

bsb/storage/interfaces.py

@@ -114,10 +114,15 @@ def root_slug(self):
         """
         pass
 
+    @staticmethod
     @abc.abstractmethod
-    def recognizes(self, root):
+    def recognizes(root, comm):
         """
-        Must return whether the given argument is recognized as a valid storage object.
+        Must return whether the given root argument is recognized as a valid storage object.
+
+        :param root: The unique identifier for the storage
+        :param mpi4py.MPI.Comm comm: MPI communicator that shares control
+          over the Storage.
         """
         pass
 

tests/test_connectivity.py

@@ -639,7 +639,7 @@ def test_contacts(self):
         conns = len(self.network.get_connectivity_set("intersect"))
         self.assertGreater(conns, 0, "no connections formed")
         self.network.connectivity.intersect.contacts = 2
-        self.network.compile(clear=True)
+        self.network.compile(redo=True)
         new_conns = len(self.network.get_connectivity_set("intersect"))
         self.assertEqual(conns * 2, new_conns, "Expected double contacts")
 
@@ -651,7 +651,7 @@ def test_zero_contacts(self):
         conns = len(self.network.get_connectivity_set("intersect"))
         self.assertEqual(0, conns, "expected no contacts")
         self.network.connectivity.intersect.contacts = -3
-        self.network.compile(clear=True)
+        self.network.compile(redo=True)
         conns = len(self.network.get_connectivity_set("intersect"))
         self.assertEqual(0, conns, "expected no contacts")