From 1402809d7a20e51f34aa3a51605b9d7e387b0ce3 Mon Sep 17 00:00:00 2001
From: JackH <jack@rtr>
Date: Tue, 23 Jun 2020 11:28:33 +0000
Subject: [PATCH] Add stream sync barriers

When blocks blocks write to a ring across the CPU/GPU boundary
this copy is [I think] asynchronous, and needs to be
synchronized before marking the destination buffer as ready
for consumption by downstream consumers.

See https://github.com/ledatelescope/bifrost/issues/138
---
 test-scripts/blocks/copy_block.py        | 4 ++++
 test-scripts/blocks/corr_acc_block.py    | 9 ++++++---
 test-scripts/blocks/corr_subsel_block.py | 3 +++
 3 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/test-scripts/blocks/copy_block.py b/test-scripts/blocks/copy_block.py
index 8b8f00e..e66b4a1 100644
--- a/test-scripts/blocks/copy_block.py
+++ b/test-scripts/blocks/copy_block.py
@@ -6,6 +6,7 @@
 from bifrost.linalg import LinAlg
 from bifrost import map as BFMap
 from bifrost.ndarray import copy_array
+from bifrost.device import stream_synchronize
 
 import time
 import json
@@ -54,3 +55,6 @@ def main(self):
                             #self.log.debug("Copying output")
                             #odata = ospan.data_view('ci4')
                             copy_array(ospan.data, ispan.data)
+                            # The copy is asynchronous, so we must wait for it to finish
+                            # before committing this span
+                            stream_synchronize()
diff --git a/test-scripts/blocks/corr_acc_block.py b/test-scripts/blocks/corr_acc_block.py
index 004e371..1f5114f 100644
--- a/test-scripts/blocks/corr_acc_block.py
+++ b/test-scripts/blocks/corr_acc_block.py
@@ -5,6 +5,8 @@
 from bifrost.ring import WriteSpan
 from bifrost.linalg import LinAlg
 from bifrost import map as BFMap
+from bifrost.ndarray import copy_array
+from bifrost.device import stream_synchronize
 
 import time
 import json
@@ -73,9 +75,10 @@ def main(self):
                             print("Skipping output because oseq isn't open")
                         else:
                             # copy to CPU
-                            odata = ospan.data_view('i32')
-                            odata = self.accdata
-                            print(odata[0:10])
+                            odata = ospan.data_view('i32').reshape(self.accdata.shape)
+                            copy_array(odata, self.accdata)
+                            # Wait for copy to complete before committing span
+                            stream_synchronize()
                             ospan.close()
                             oseq.end()
             # If upstream process stops producing, close things gracefully
diff --git a/test-scripts/blocks/corr_subsel_block.py b/test-scripts/blocks/corr_subsel_block.py
index 121b958..1adb689 100644
--- a/test-scripts/blocks/corr_subsel_block.py
+++ b/test-scripts/blocks/corr_subsel_block.py
@@ -6,6 +6,7 @@
 from bifrost.linalg import LinAlg
 from bifrost import map as BFMap
 from bifrost.ndarray import copy_array
+from bifrost.device import stream_synchronize
 
 import time
 import json
@@ -64,3 +65,5 @@ def main(self):
                                 raise RuntimeError
                             odata = ospan.data_view(dtype='i64').reshape([self.nchans, self.nvis_out])
                             copy_array(odata, self.obuf_gpu)
+                            # Wait for copy to complete before committing span
+                            stream_synchronize()