Recognize events aren't sorted by all keys.

arbor/backends/event_stream_base.hpp

@@ -23,11 +23,9 @@ struct event_stream_base {
 protected: // members
     std::vector<event_data_type> ev_data_;
     std::vector<std::size_t> ev_spans_ = {0};
-    std::vector<std::size_t> lane_spans_;
     std::size_t index_ = 0;
     event_data_type* base_ptr_ = nullptr;
 
-
 public:
     event_stream_base() = default;
 
@@ -106,15 +104,6 @@ struct spike_event_stream_base: event_stream_base<deliverable_event> {
             stream.spikes_.clear();
             // ev_data_ has been cleared during v.clear(), so we use its capacity
             stream.spikes_.reserve(stream.ev_data_.capacity());
-            // record sizes of streams for later merging
-            //
-            // The idea here is that this records the division points `pd` where
-            // `stream` was updated by the lane `lid`. As events within one lane are
-            // sorted, we known that events between two division points are sorted.
-            // Then, we can use `merge_inplace` over `sort` for a small but noticeable
-            // speed-up.
-            stream.lane_spans_.resize(lanes.size() + 1);
-            for (auto& ix: stream.lane_spans_) ix = stream.spikes_.size();
         }
 
         // loop over lanes: group events by mechanism and sort them by time
@@ -135,8 +124,6 @@ struct spike_event_stream_base: event_stream_base<deliverable_event> {
                 stream.spikes_.push_back(spike_data{step, handle.mech_index, time, weight});
                 stream.spike_counter_[step]++;
             }
-            // record current sizes here. putting this into the above loop is slower. significantly
-            for (auto& [id, stream]: streams) stream.lane_spans_[cell + 1] = stream.spikes_.size();
             ++cell;
         }
 
@@ -146,16 +133,8 @@ struct spike_event_stream_base: event_stream_base<deliverable_event> {
             tg.run([&stream]() {
                 // scan over spike_counter_
                 util::make_partition(stream.ev_spans_, stream.spike_counter_);
-                // leverage our earlier partitioning to merge the partitions
-                // theoretically, this could be parallelised, too, practically it didn't pay off
-                auto& part = stream.lane_spans_;
-                for (size_t ix = 0; ix < part.size() - 1; ++ix) {
-                    std::inplace_merge(stream.spikes_.begin(),
-                                       stream.spikes_.begin() + part[ix],
-                                       stream.spikes_.begin() + part[ix + 1]);
-                }
-                // Further optimisation: merge(!) merging, transforming, and appending into one
-                // call.
+                // This is made slightly faster by using pdqsort, if we want to take it on.
+                util::sort(stream.spikes_);
                 // copy temporary deliverable_events into stream's ev_data_
                 stream.ev_data_.reserve(stream.spikes_.size());
                 std::transform(stream.spikes_.begin(), stream.spikes_.end(),

test/unit/test_event_stream.cpp

@@ -17,9 +17,11 @@ void check(Result result) {
 }
 
 TEST(event_stream, single_step) {
-    check(single_step<multicore::spike_event_stream>());
+    auto ctx = arb::make_context();
+    check(single_step<multicore::spike_event_stream>(ctx));
 }
 
 TEST(event_stream, multi_step) {
-    check(multi_step<multicore::spike_event_stream>());
+    auto ctx = arb::make_context();
+    check(multi_step<multicore::spike_event_stream>(ctx));
 }

test/unit/test_event_stream.hpp

@@ -4,6 +4,9 @@
 #include <random>
 #include <gtest/gtest.h>
 
+#include <arbor/spike_event.hpp>
+#include "arbor/context.hpp"
+#include "execution_context.hpp"
 #include "timestep_range.hpp"
 #include "backends/event.hpp"
 #include "util/rangeutil.hpp"
@@ -12,7 +15,8 @@ namespace {
 
 using namespace arb;
 
-void check_result(arb_deliverable_event_data const* results, std::vector<arb_deliverable_event_data> const& expected) {
+inline void
+check_result(arb_deliverable_event_data const* results, std::vector<arb_deliverable_event_data> const& expected) {
     for (std::size_t i=0; i<expected.size(); ++i) {
         EXPECT_EQ(results[i].weight, expected[i].weight);
     }
@@ -26,7 +30,7 @@ struct result {
 };
 
 template<typename Stream>
-result<Stream> single_step() {
+result<Stream> single_step(const arb::context& ctx) {
     // events for 3 cells and 2 mechanisms and according targets
     //
     // target handles                            | events
@@ -87,13 +91,13 @@ result<Stream> single_step() {
 
     // initialize event streams
     auto lanes = util::subrange_view(events, 0u, events.size());
-    initialize(lanes, handles, divs, res.steps, res.streams);
+    initialize(lanes, handles, divs, res.steps, res.streams, ctx->thread_pool);
 
     return res;
 }
 
 template<typename Stream>
-result<Stream> multi_step() {
+result<Stream> multi_step(const arb::context& ctx) {
     // number of events, cells, mechanisms and targets
     std::size_t num_events = 500;
     std::size_t num_cells = 20;
@@ -204,8 +208,7 @@ result<Stream> multi_step() {
 
     // initialize event streams
     auto lanes = util::subrange_view(events, 0u, events.size());
-    initialize(lanes, handles, divs, res.steps, res.streams);
-
+    initialize(lanes, handles, divs, res.steps, res.streams, ctx->thread_pool);
     return res;
 }
 

test/unit/test_lif_cell_group.cpp

@@ -117,8 +117,7 @@ class path_recipe: public arb::recipe {
 };
 
 // LIF cell with probe
-class probe_recipe: public arb::recipe {
-public:
+struct probe_recipe: public arb::recipe {
     probe_recipe(size_t n_conn = 0): n_conn_{n_conn} {}
 
     cell_size_type num_cells() const override {

test/unit/test_synapses.cpp

@@ -153,7 +153,8 @@ TEST(synapses, syn_basic_state) {
     auto lanes = event_lane_subrange(events.begin(), events.end());
     std::vector<target_handle> handles{{0, 1}, {0, 3}, {1, 0}, {1, 2}};
     std::vector<size_t> divs{0, handles.size()};
-    state.begin_epoch(lanes, {}, dts, handles, divs);
+    auto ctx = arb::make_context();
+    state.begin_epoch(lanes, {}, dts, handles, divs, ctx->thread_pool);
     state.mark_events();
 
     state.deliver_events(*expsyn);