Minor incremental sorting perf optimisations

Author: juntyr

necsim/impls/no-std/src/parallelisation/independent/monolithic/reporter/live.rs

@@ -12,7 +12,7 @@ use necsim_partitioning_core::LocalPartition;
 
 use super::WaterLevelReporterProxy;
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug)]
 struct Run {
     start: usize,
     len: usize,
@@ -49,6 +49,7 @@ impl<'l, 'p, R: Reporter, P: LocalPartition<'p, R>> fmt::Debug
             .field("water_level", &self.water_level)
             .field("runs", &self.runs.len())
             .field("overflow", &self.overflow.len())
+            .field("run", &self.run)
             .field("slow_events", &EventBufferLen(self.slow_events.len()))
             .field("fast_events", &EventBufferLen(self.fast_events.len()))
             .finish()
@@ -319,17 +320,35 @@ impl<'l, 'p, R: Reporter, P: LocalPartition<'p, R>> WaterLevelReporterProxy<'l,
     }
 
     fn advance_water_level(&mut self, water_level: NonNegativeF64) {
-        let mut i = 0;
+        info!("{:?}", self);
+
+        if self.run.len > 0 {
+            let last_run = core::mem::replace(
+                &mut self.run,
+                Run {
+                    start: self.slow_events.len(),
+                    len: 0,
+                },
+            );
+            self.overflow.push_back(last_run);
+        }
+
+        if let (Some(Run { start, .. }), Some(Run { start: middle, len })) = (
+            self.overflow.front().copied(),
+            self.overflow.back().copied(),
+        ) {
+            self.slow_events[start..middle + len].sort_unstable();
+            self.overflow.clear();
+            self.runs.push(Run {
+                start,
+                len: middle + len - start,
+            });
+        }
 
         // Report all events below the water level in sorted order
         // TODO: Should we detect if no partial sort steps were taken
         //       and revert to a full unstable sort in that case?
-        while let ControlFlow::Continue(()) = self.sort_slow_events_step(true) {
-            if (i % 100) == 0 {
-                info!("{:?}", self);
-            }
-            i += 1;
-        }
+        while let ControlFlow::Continue(()) = self.sort_slow_events_step(true) {}
 
         debug_assert!(self.slow_events.is_sorted());
 
@@ -387,6 +406,31 @@ impl<'l, 'p, R: Reporter, P: LocalPartition<'p, R>> Drop
     for LiveWaterLevelReporterProxy<'l, 'p, R, P>
 {
     fn drop(&mut self) {
+        info!("{:?}", self);
+
+        if self.run.len > 0 {
+            let last_run = core::mem::replace(
+                &mut self.run,
+                Run {
+                    start: self.slow_events.len(),
+                    len: 0,
+                },
+            );
+            self.overflow.push_back(last_run);
+        }
+
+        if let (Some(Run { start, .. }), Some(Run { start: middle, len })) = (
+            self.overflow.front().copied(),
+            self.overflow.back().copied(),
+        ) {
+            self.slow_events[start..middle + len].sort_unstable();
+            self.overflow.clear();
+            self.runs.push(Run {
+                start,
+                len: middle + len - start,
+            });
+        }
+
         // Report all events below the water level in sorted order
         while let ControlFlow::Continue(()) = self.sort_slow_events_step(true) {}
 
@@ -411,7 +455,7 @@ impl<'l, 'p, R: Reporter, P: LocalPartition<'p, R>> Drop
         self.run = Run { start: 0, len: 0 };
 
         // Report all events above the water level in sorted order
-        self.fast_events.sort();
+        self.fast_events.sort_unstable();
 
         for event in self.fast_events.drain(..) {
             match event.into() {
@@ -427,5 +471,7 @@ impl<'l, 'p, R: Reporter, P: LocalPartition<'p, R>> Drop
                 },
             }
         }
+
+        info!("{:?}", self);
     }
 }

necsim/impls/no-std/src/parallelisation/monolithic/optimistic.rs

@@ -135,7 +135,7 @@ pub fn simulate<
                 ));
             }
 
-            immigrants.sort();
+            immigrants.sort_unstable();
 
             // A global rollback is required if at least one partition received unexpected
             // immigration

necsim/impls/std/src/event_log/recorder.rs

@@ -177,7 +177,7 @@ impl EventLogRecorder {
     }
 
     fn sort_and_write_segment(&mut self) -> Result<()> {
-        self.buffer.sort();
+        self.buffer.sort_unstable();
 
         let segment_path = self.directory.join(format!("{}", self.segment_index));
         self.segment_index += 1;

necsim/impls/std/src/event_log/replay/mod.rs

@@ -43,7 +43,7 @@ impl Serialize for EventLogReplay {
             }
         }
 
-        segments.sort();
+        segments.sort_unstable();
 
         EventLog { segments, capacity }.serialize(serializer)
     }

rustcoalescence/algorithms/gillespie/src/event_skipping/initialiser/fixup.rs

@@ -199,7 +199,7 @@ where
                 );
             },
             CoalescenceStrategy::Coalescence => {
-                coalescence.sort();
+                coalescence.sort_unstable();
 
                 for (coalescing_lineage, parent) in coalescence {
                     local_partition.get_reporter().report_dispersal(
@@ -244,7 +244,7 @@ where
             return Err(ResumeError::Sample(exceptional_lineages));
         }
 
-        fixable_lineages.sort();
+        fixable_lineages.sort_unstable();
 
         let active_lineage_sampler = RestartFixUpActiveLineageSampler::new(
             active_lineage_sampler,

rustcoalescence/algorithms/gillespie/src/gillespie/classical/initialiser/fixup.rs

@@ -151,7 +151,7 @@ impl<
                 );
             },
             CoalescenceStrategy::Coalescence => {
-                coalescence.sort();
+                coalescence.sort_unstable();
 
                 for (coalescing_lineage, parent) in coalescence {
                     local_partition.get_reporter().report_dispersal(
@@ -196,7 +196,7 @@ impl<
             return Err(ResumeError::Sample(exceptional_lineages));
         }
 
-        fixable_lineages.sort();
+        fixable_lineages.sort_unstable();
 
         let dispersal_sampler = TrespassingDispersalSampler::new(
             dispersal_sampler,

rustcoalescence/algorithms/gillespie/src/gillespie/turnover/initialiser/fixup.rs

@@ -173,7 +173,7 @@ impl<
                 );
             },
             CoalescenceStrategy::Coalescence => {
-                coalescence.sort();
+                coalescence.sort_unstable();
 
                 for (coalescing_lineage, parent) in coalescence {
                     local_partition.get_reporter().report_dispersal(
@@ -218,7 +218,7 @@ impl<
             return Err(ResumeError::Sample(exceptional_lineages));
         }
 
-        fixable_lineages.sort();
+        fixable_lineages.sort_unstable();
 
         let dispersal_sampler = TrespassingDispersalSampler::new(
             dispersal_sampler,

simulate.ron

@@ -1,19 +1,19 @@
 (
-    speciation: 0.1,
+    speciation: 0.0001,
     sample: Sample(percentage: 1.0),
-    rng: Entropy,
+    rng: Seed(42),
 
-    algorithm: EventSkipping(),
+    algorithm: CUDA(step_slice: 150, block_size: 64, grid_size: 128, sort_block_size: 512, sort_mode: None, sort_batch_size: 3500),
 
     scenario: NonSpatial(
         area: (100, 100),
-        deme: 100,
+        deme: 1000,
     ),
 
     reporters: [
         Plugin(
             library: "target/release/deps/libnecsim_plugins_common.so",
-            reporters: [Biodiversity(), Execution(), Progress()],
+            reporters: [Biodiversity(), Execution(), Progress(), Counter()],
         ),
     ],
 )