Add synchronisation of multi-threaded system

multigrid/Cargo.toml

@@ -19,3 +19,4 @@ indexmap = { version = "1.5.2", features = ["serde-1"] }
 argh = "0.1.4"
 evalexpr = "6.3.0"
 crossbeam-channel = "0.5.0"
+crossbeam-utils = "0.8.5"

multigrid/src/main.rs

@@ -108,6 +108,9 @@ fn main() {
     }
 
     let timer = if opt.timings {
+        if let system::System::MultiThreaded(sys) = &sys {
+            sys.synchronise()
+        }
         Some(std::time::Instant::now())
     } else {
         None
@@ -119,25 +122,30 @@ fn main() {
         sys.advance(nexttime - itime);
 
         itime = nexttime;
-        sys.output(itime);
+        if itime != ntime {
+            sys.output(itime);
+        }
     }
 
+    let timer = timer.map(|timer| {
+        if let system::System::MultiThreaded(sys) = &sys {
+            sys.synchronise();
+        }
+
+        timer.elapsed()
+    });
+    sys.output(ntime);
+
+    let mut outinfo = OutputInformation {
+        filename: opt.output,
+        time_elapsed: timer,
+        ..Default::default()
+    };
+
     if !opt.no_progressbar {
         sys.finish_progressbar();
     }
 
-    let mut outinfo = OutputInformation {
-        filename: opt.output,
-        ..Default::default()
-    };
-
-    if let Some(timer) = timer {
-        let duration = timer.elapsed();
-        outinfo.time_elapsed = Some(duration);
-    }
-
-    //output.add_timestep(ntime, &sys.fnow);
-
     if opt.error {
         outinfo.error = Some(sys.error())
     }

multigrid/src/system.rs

@@ -415,6 +415,7 @@ impl System {
                 for tid in &sys.send {
                     tid.send(MsgFromHost::ProgressbarDrop).unwrap();
                 }
+                sys.synchronise();
                 let target = sys.progressbar.take().unwrap();
                 target.clear().unwrap();
             }
@@ -658,6 +659,17 @@ impl DistributedSystem {
         target.set_move_cursor(true);
         self.progressbar = Some(target);
     }
+    fn send_barrier(&self, barrier: &crossbeam_utils::sync::WaitGroup) {
+        for tid in &self.send {
+            tid.send(MsgFromHost::Barrier(barrier.clone())).unwrap()
+        }
+    }
+    pub fn synchronise(&self) {
+        // Syncronise before starting the timer
+        let barrier = crossbeam_utils::sync::WaitGroup::new();
+        self.send_barrier(&barrier);
+        barrier.wait();
+    }
 }
 
 impl Drop for DistributedSystem {
@@ -687,6 +699,8 @@ enum MsgFromHost {
     Stop,
     /// Request the current error
     Error,
+    /// A barrier that must be waited on
+    Barrier(crossbeam_utils::sync::WaitGroup),
     /// Progressbar to report progress
     Progressbar(indicatif::ProgressBar),
     /// Clear and remove the progressbar
@@ -760,6 +774,7 @@ impl DistributedSystemPart {
                 MsgFromHost::Output(ntime) => self.output(ntime),
                 MsgFromHost::Stop => return,
                 MsgFromHost::Error => self.send(MsgToHost::Error(self.error())).unwrap(),
+                MsgFromHost::Barrier(barrier) => barrier.wait(),
                 MsgFromHost::Progressbar(pbar) => self.progressbar = Some(pbar),
                 MsgFromHost::ProgressbarDrop => {
                     let pb = self.progressbar.take().unwrap();