Feature: High-level network specification

arbor/CMakeLists.txt

@@ -44,6 +44,8 @@ set(arbor_sources
     morph/segment_tree.cpp
     morph/stitch.cpp
     merge_events.cpp
+    network.cpp
+    network_impl.cpp
     simulation.cpp
     partition_load_balance.cpp
     profile/clock.cpp

arbor/communication/communicator.cpp

@@ -14,6 +14,7 @@
 #include "connection.hpp"
 #include "distributed_context.hpp"
 #include "execution_context.hpp"
+#include "network_impl.hpp"
 #include "profile/profiler_macro.hpp"
 #include "threading/threading.hpp"
 #include "util/partition.hpp"
@@ -24,14 +25,12 @@
 
 namespace arb {
 
-communicator::communicator(const recipe& rec,
-                           const domain_decomposition& dom_dec,
-                           execution_context& ctx):  num_total_cells_{rec.num_cells()},
-                                                     num_local_cells_{dom_dec.num_local_cells()},
-                                                     num_local_groups_{dom_dec.num_groups()},
-                                                     num_domains_{(cell_size_type) ctx.distributed->size()},
-                                                     distributed_{ctx.distributed},
-                                                     thread_pool_{ctx.thread_pool} {}
+communicator::communicator(const recipe& rec, const domain_decomposition& dom_dec, context ctx):
+    num_total_cells_{rec.num_cells()},
+    num_local_cells_{dom_dec.num_local_cells()},
+    num_local_groups_{dom_dec.num_groups()},
+    num_domains_{(cell_size_type)ctx->distributed->size()},
+    ctx_(std::move(ctx)) {}
 
 constexpr inline
 bool is_external(cell_gid_type c) {
@@ -55,7 +54,7 @@ cell_member_type global_cell_of(const cell_member_type& c) {
     return {c.gid | msb, c.index};
 }
 
-void communicator::update_connections(const connectivity& rec,
+void communicator::update_connections(const recipe& rec,
                                       const domain_decomposition& dom_dec,
                                       const label_resolution_map& source_resolution_map,
                                       const label_resolution_map& target_resolution_map) {
@@ -64,8 +63,12 @@ void communicator::update_connections(const connectivity& rec,
     connections_.clear();
     connection_part_.clear();
     index_divisions_.clear();
+    index_on_domain_.clear();
     PL();
 
+    // Construct connections from high-level specification
+    auto generated_connections = generate_connections(rec, ctx_, dom_dec);
+
     // Make a list of local cells' connections
     //   -> gid_connections
     // Count the number of local connections (i.e. connections terminating on this domain)
@@ -80,6 +83,7 @@ void communicator::update_connections(const connectivity& rec,
     PE(init:communicator:update:collect_gids);
     std::vector<cell_gid_type> gids; gids.reserve(num_local_cells_);
     for (const auto& g: dom_dec.groups()) util::append(gids, g.gids);
+    for (const auto index: util::make_span(gids.size())) index_on_domain_.insert({gids[index], index});
     PL();
 
     // Build the connection information for local cells.
@@ -113,9 +117,18 @@ void communicator::update_connections(const connectivity& rec,
         }
         part_ext_connections.push_back(gid_ext_connections.size());
     }
+    for (const auto& c: generated_connections) {
+        auto sgid = c.source.gid;
+        if (sgid >= num_total_cells_) {
+            throw arb::bad_connection_source_gid(c.source.gid, sgid, num_total_cells_);
+        }
+        const auto src = dom_dec.gid_domain(sgid);
+        src_domains.push_back(src);
+        src_counts[src]++;
+    }
 
     util::make_partition(connection_part_, src_counts);
-    auto n_cons = gid_connections.size();
+    auto n_cons = gid_connections.size() + generated_connections.size();
     auto n_ext_cons = gid_ext_connections.size();
     PL();
 
@@ -140,18 +153,23 @@ void communicator::update_connections(const connectivity& rec,
             auto tgt_lid = target_resolver.resolve(tgt_gid, conn.target);
             auto offset  = offsets[*src_domain]++;
             ++src_domain;
-            connections_[offset] = {{src_gid, src_lid}, tgt_lid, conn.weight, conn.delay, index};
+            connections_[offset] = {{src_gid, src_lid}, {tgt_gid, tgt_lid}, conn.weight, conn.delay};
         }
         for (const auto cidx: util::make_span(part_ext_connections[index], part_ext_connections[index+1])) {
             const auto& conn = gid_ext_connections[cidx];
             auto src = global_cell_of(conn.source);
             auto src_gid = conn.source.rid;
             if(is_external(src_gid)) throw arb::source_gid_exceeds_limit(tgt_gid, src_gid);
             auto tgt_lid = target_resolver.resolve(tgt_gid, conn.target);
-            ext_connections_[ext] = {src, tgt_lid, conn.weight, conn.delay, index};
+            ext_connections_[ext] = {src, {tgt_gid, tgt_lid}, conn.weight, conn.delay};
             ++ext;
         }
     }
+    for (const auto& c: generated_connections) {
+        auto offset = offsets[*src_domain]++;
+        ++src_domain;
+        connections_[offset] = c;
+    }
     PL();
 
     PE(init:communicator:update:index);
@@ -166,7 +184,7 @@ void communicator::update_connections(const connectivity& rec,
     // Sort the connections for each domain.
     // This is num_domains_ independent sorts, so it can be parallelized trivially.
     const auto& cp = connection_part_;
-    threading::parallel_for::apply(0, num_domains_, thread_pool_.get(),
+    threading::parallel_for::apply(0, num_domains_, ctx_->thread_pool.get(),
                                    [&](cell_size_type i) {
                                        util::sort(util::subrange_view(connections_, cp[i], cp[i+1]));
                                    });
@@ -187,7 +205,7 @@ time_type communicator::min_delay() {
     res = std::accumulate(ext_connections_.begin(), ext_connections_.end(),
                                 res,
                                 [](auto&& acc, auto&& el) { return std::min(acc, time_type(el.delay)); });
-    res = distributed_->min(res);
+    res = ctx_->distributed->min(res);
     return res;
 }
 
@@ -200,7 +218,7 @@ communicator::exchange(std::vector<spike> local_spikes) {
 
     PE(communication:exchange:gather);
     // global all-to-all to gather a local copy of the global spike list on each node.
-    auto global_spikes = distributed_->gather_spikes(local_spikes);
+    auto global_spikes = ctx_->distributed->gather_spikes(local_spikes);
     num_spikes_ += global_spikes.size();
     PL();
 
@@ -211,7 +229,7 @@ communicator::exchange(std::vector<spike> local_spikes) {
                                           local_spikes.end(),
                                           [this] (const auto& s) { return !remote_spike_filter_(s); }));
     }
-    auto remote_spikes = distributed_->remote_gather_spikes(local_spikes);
+    auto remote_spikes = ctx_->distributed->remote_gather_spikes(local_spikes);
     PL();
 
     PE(communication:exchange:gather:remote:post_process);
@@ -225,13 +243,13 @@ communicator::exchange(std::vector<spike> local_spikes) {
 }
 
 void communicator::set_remote_spike_filter(const spike_predicate& p) { remote_spike_filter_ = p; }
-void communicator::remote_ctrl_send_continue(const epoch& e) { distributed_->remote_ctrl_send_continue(e); }
-void communicator::remote_ctrl_send_done() { distributed_->remote_ctrl_send_done(); }
+void communicator::remote_ctrl_send_continue(const epoch& e) { ctx_->distributed->remote_ctrl_send_continue(e); }
+void communicator::remote_ctrl_send_done() { ctx_->distributed->remote_ctrl_send_done(); }
 
 // Internal helper to append to the event queues
 template<typename S, typename C>
 void append_events_from_domain(C cons,
-                               S spks,
+                               S spks, const std::unordered_map<cell_gid_type, cell_size_type>& index_on_domain,
                                std::vector<pse_vector>& queues) {
     // Predicate for partitioning
     struct spike_pred {
@@ -254,7 +272,7 @@ void append_events_from_domain(C cons,
         while (cn != ce && sp != se) {
             auto sources = std::equal_range(sp, se, cn->source, spike_pred());
             for (auto s: util::make_range(sources)) {
-                queues[cn->index_on_domain].push_back(make_event(*cn, s));
+                queues[index_on_domain.at(cn->destination.gid)].push_back(make_event(*cn, s));
             }
             sp = sources.first;
             ++cn;
@@ -264,7 +282,7 @@ void append_events_from_domain(C cons,
         while (cn != ce && sp != se) {
             auto targets = std::equal_range(cn, ce, sp->source);
             for (auto c: util::make_range(targets)) {
-                queues[c.index_on_domain].push_back(make_event(c, *sp));
+                queues[index_on_domain.at(c.destination.gid)].push_back(make_event(c, *sp));
             }
             cn = targets.first;
             ++sp;
@@ -281,7 +299,7 @@ void communicator::make_event_queues(
     const auto& cp = connection_part_;
     for (auto dom: util::make_span(num_domains_)) {
         append_events_from_domain(util::subrange_view(connections_,           cp[dom], cp[dom+1]),
-                                  util::subrange_view(global_spikes.values(), sp[dom], sp[dom+1]),
+                                  util::subrange_view(global_spikes.values(), sp[dom], sp[dom+1]), index_on_domain_,
                                   queues);
     }
     num_local_events_ = util::sum_by(queues, [](const auto& q) {return q.size();}, num_local_events_);
@@ -291,7 +309,7 @@ void communicator::make_event_queues(
     std::for_each(spikes.begin(),
                   spikes.end(),
                   [](auto& s) { s.source = global_cell_of(s.source); });
-    append_events_from_domain(ext_connections_, spikes, queues);
+    append_events_from_domain(ext_connections_, spikes, index_on_domain_, queues);
 }
 
 std::uint64_t communicator::num_spikes() const {

arbor/communication/communicator.hpp

@@ -1,10 +1,12 @@
 #pragma once
 
 #include <vector>
+#include <unordered_map>
 
-#include <arbor/export.hpp>
 #include <arbor/common_types.hpp>
+#include <arbor/context.hpp>
 #include <arbor/domain_decomposition.hpp>
+#include <arbor/export.hpp>
 #include <arbor/recipe.hpp>
 #include <arbor/spike.hpp>
 
@@ -39,7 +41,7 @@ class ARB_ARBOR_API communicator {
 
     explicit communicator(const recipe& rec,
                           const domain_decomposition& dom_dec,
-                          execution_context& ctx);
+                          context ctx);
 
     /// The range of event queues that belong to cells in group i.
     std::pair<cell_size_type, cell_size_type> group_queue_range(cell_size_type i);
@@ -82,7 +84,7 @@ class ARB_ARBOR_API communicator {
     void remote_ctrl_send_continue(const epoch&);
     void remote_ctrl_send_done();
 
-    void update_connections(const connectivity& rec,
+    void update_connections(const recipe& rec,
                             const domain_decomposition& dom_dec,
                             const label_resolution_map& source_resolution_map,
                             const label_resolution_map& target_resolution_map);
@@ -100,17 +102,17 @@ class ARB_ARBOR_API communicator {
     std::vector<cell_size_type> connection_part_;
     std::vector<cell_size_type> index_divisions_;
     util::partition_view_type<std::vector<cell_size_type>> index_part_;
+    std::unordered_map<cell_gid_type, cell_size_type> index_on_domain_;
 
     spike_predicate remote_spike_filter_;
 
     // Connections from external simulators into Arbor.
     // Currently we have no partitions/indices/acceleration structures
     std::vector<connection> ext_connections_;
 
-    distributed_context_handle distributed_;
-    task_system_handle thread_pool_;
     std::uint64_t num_spikes_ = 0u;
     std::uint64_t num_local_events_ = 0u;
+    context ctx_;
 };
 
 } // namespace arb