remove locks

Author: Bernard Knueven

mpisppy/cylinders/spwindow.py

@@ -282,6 +282,12 @@ def __init__(self, my_fields: dict, strata_comm: MPI.Comm, field_order=None):
         # Gather layouts across ranks
         self.strata_buffer_layouts = strata_comm.allgather(self.buffer_layout)
 
+        # Keep one passive-target access epoch open for the lifetime of the
+        # window. Individual operations complete at their target with Flush;
+        # they do not repeatedly acquire and release target locks.
+        self.window.Lock_all()
+        self._epoch_open = True
+
     def free(self):
         if self.window is not None:
             guard_error = None
@@ -290,6 +296,9 @@ def free(self):
                     self._verify_window_guards(field, "free", "before")
             except RuntimeError as error:
                 guard_error = error
+            if self._epoch_open:
+                self.window.Unlock_all()
+                self._epoch_open = False
             self.window.Free()
             self.buff = None
             self.buffer_layout = None
@@ -465,18 +474,16 @@ def get(self, dest: nptyping.ArrayLike, strata_rank: int, field: Field,
             target_rank=strata_rank,
         )
         window = self.window
-        # The shared epoch may overlap other readers, but it cannot overlap
-        # the publisher's exclusive self-target epoch below. Unlock provides
-        # completion for the Get before its destination and guards are read.
-        window.Lock(strata_rank, MPI.LOCK_SHARED)
         window.Get(
             (transfer, field_layout.transfer_nbytes, MPI.BYTE),
             strata_rank,
             (field_layout.transfer_offset_bytes,
              field_layout.transfer_nbytes,
              MPI.BYTE),
         )
-        window.Unlock(strata_rank)
+        # Flush provides local completion, so the destination and its
+        # transmitted guards are safe to inspect below.
+        window.Flush(strata_rank)
         self._verify_transfer_guards(
             storage, field_layout, field, "get", "after",
             target_rank=strata_rank,
@@ -502,19 +509,18 @@ def put(self, values: nptyping.ArrayLike, field: Field):
         )
         self._verify_window_guards(field, "put", "before")
         window = self.window
-        # Publishing through a self-target Put is intentional: unlike an
-        # ordinary local store, this exclusive epoch participates in the same
-        # MPI lock arbitration as remote readers and makes the field snapshot
-        # indivisible with respect to their shared epochs.
-        window.Lock(self.strata_rank, MPI.LOCK_EXCLUSIVE)
+        # Publishing through a self-target Put is intentional: it keeps local
+        # and remote publication on the same MPI-visible path. Phase 3 will
+        # add the metadata protocol that lets readers reject an overlapping
+        # publication; Phase 2 only changes epoch management.
         window.Put(
             (transfer, field_layout.transfer_nbytes, MPI.BYTE),
             self.strata_rank,
             (field_layout.transfer_offset_bytes,
              field_layout.transfer_nbytes,
              MPI.BYTE),
         )
-        window.Unlock(self.strata_rank)
+        window.Flush(self.strata_rank)
         self._verify_transfer_guards(
             storage, field_layout, field, "put", "after",
             target_rank=self.strata_rank,

mpisppy/tests/test_spwindow_partial_get.py

@@ -28,14 +28,6 @@ def __init__(self, wrapped):
         self.wrapped = wrapped
         self.calls = []
 
-    def Lock(self, rank, lock_type):
-        self.calls.append(("Lock", rank, lock_type))
-        return self.wrapped.Lock(rank, lock_type)
-
-    def Unlock(self, rank):
-        self.calls.append(("Unlock", rank))
-        return self.wrapped.Unlock(rank)
-
     def Put(self, *args):
         self.calls.append(("Put",))
         return self.wrapped.Put(*args)
@@ -44,6 +36,18 @@ def Get(self, *args):
         self.calls.append(("Get",))
         return self.wrapped.Get(*args)
 
+    def Flush(self, rank):
+        self.calls.append(("Flush", rank))
+        return self.wrapped.Flush(rank)
+
+    def Unlock_all(self):
+        self.calls.append(("Unlock_all",))
+        return self.wrapped.Unlock_all()
+
+    def Free(self):
+        self.calls.append(("Free",))
+        return self.wrapped.Free()
+
     def __getattr__(self, name):
         return getattr(self.wrapped, name)
 
@@ -158,7 +162,7 @@ def test_transmitted_canaries_identify_record(self):
                 0, Field.NONANTS_VALS, "left", layout.padded_len + 1),
         )
 
-    def test_put_and_get_preserve_exclusive_shared_locking(self):
+    def test_put_and_get_flush_persistent_epoch(self):
         recording = _RecordingWindow(self.win.window)
         self.win.window = recording
 
@@ -167,12 +171,21 @@ def test_put_and_get_preserve_exclusive_shared_locking(self):
             np.empty(self.padded, dtype="d"), 0, Field.NONANTS_VALS)
 
         self.assertEqual(recording.calls, [
-            ("Lock", 0, MPI.LOCK_EXCLUSIVE),
             ("Put",),
-            ("Unlock", 0),
-            ("Lock", 0, MPI.LOCK_SHARED),
+            ("Flush", 0),
             ("Get",),
-            ("Unlock", 0),
+            ("Flush", 0),
+        ])
+
+    def test_free_closes_epoch_before_freeing_window(self):
+        recording = _RecordingWindow(self.win.window)
+        self.win.window = recording
+
+        self.win.free()
+
+        self.assertEqual(recording.calls, [
+            ("Unlock_all",),
+            ("Free",),
         ])