| | | 1 | | // Copyright 2025 Digital Holography Foundation |
| | | 2 | | // |
| | | 3 | | // Licensed under the Apache License, Version 2.0 (the "License"); |
| | | 4 | | // you may not use this file except in compliance with the License. |
| | | 5 | | // You may obtain a copy of the License at |
| | | 6 | | // |
| | | 7 | | // http://www.apache.org/licenses/LICENSE-2.0 |
| | | 8 | | // |
| | | 9 | | // Unless required by applicable law or agreed to in writing, software |
| | | 10 | | // distributed under the License is distributed on an "AS IS" BASIS, |
| | | 11 | | // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| | | 12 | | // See the License for the specific language governing permissions and |
| | | 13 | | // limitations under the License. |
| | | 14 | | |
| | | 15 | | #include "holoflow_event/router.hh" |
| | | 16 | | |
| | | 17 | | #include <iostream> |
| | | 18 | | |
| | | 19 | | namespace holoflow_event { |
| | | 20 | | |
| | 1 | 21 | | MailboxCounters::Snapshot MailboxCounters::snapshot() const noexcept { |
| | 1 | 22 | | return Snapshot{received_events.load(std::memory_order_relaxed), |
| | | 23 | | dropped_events.load(std::memory_order_relaxed), |
| | | 24 | | sent_events.load(std::memory_order_relaxed)}; |
| | 1 | 25 | | } |
| | | 26 | | |
| | | 27 | | EventReader::EventReader(BoundedMPMC<Event> *queue, MailboxCounters *counters) |
| | 1 | 28 | | : queue_(queue), counters_(counters) {} |
| | | 29 | | |
| | 1 | 30 | | std::optional<Event> EventReader::try_pop() noexcept { |
| | 1 | 31 | | auto event = queue_->try_pop(); |
| | 1 | 32 | | if (event) { |
| | 1 | 33 | | counters_->received_events.fetch_add(1, std::memory_order_relaxed); |
| | | 34 | | } |
| | 1 | 35 | | return event; |
| | 1 | 36 | | } |
| | | 37 | | |
| | 1 | 38 | | size_t EventReader::drop_all() noexcept { |
| | 1 | 39 | | size_t dropped = 0; |
| | 1 | 40 | | while (true) { |
| | 1 | 41 | | auto event = queue_->try_pop(); |
| | 1 | 42 | | if (!event) { |
| | 1 | 43 | | break; |
| | | 44 | | } |
| | 1 | 45 | | dropped++; |
| | 1 | 46 | | } |
| | 1 | 47 | | counters_->dropped_events.fetch_add(dropped, std::memory_order_relaxed); |
| | 1 | 48 | | return dropped; |
| | 1 | 49 | | } |
| | | 50 | | |
| | 1 | 51 | | MailboxCounters::Snapshot EventReader::counters() const noexcept { return counters_->snapshot(); } |
| | | 52 | | |
| | | 53 | | EventWriter::EventWriter(BoundedMPMC<Event> *queue, MailboxCounters *counters) |
| | 1 | 54 | | : queue_(queue), counters_(counters) {} |
| | | 55 | | |
| | 1 | 56 | | bool EventWriter::try_push(Event &&event) noexcept { |
| | 1 | 57 | | bool ok = queue_->try_push(std::move(event)); |
| | 1 | 58 | | counters_->sent_events.fetch_add(ok, std::memory_order_relaxed); |
| | 1 | 59 | | counters_->dropped_events.fetch_add(!ok, std::memory_order_relaxed); |
| | 1 | 60 | | return ok; |
| | 1 | 61 | | } |
| | | 62 | | |
| | 0 | 63 | | MailboxCounters::Snapshot EventWriter::counters() const noexcept { return counters_->snapshot(); } |
| | | 64 | | |
| | | 65 | | Router::Router(const Config &config) |
| | 1 | 66 | | : config_(config), ui_to_router_(config.ui_to_router_capacity), |
| | 1 | 67 | | router_to_ui_(config.router_to_ui_capacity), |
| | 1 | 68 | | nodes_to_router_(config.nodes_to_router_capacity) {} |
| | | 69 | | |
| | 1 | 70 | | Router::NodeHandles Router::bind_node(const NodeId &node_id) { |
| | 1 | 71 | | auto queue = std::make_unique<BoundedMPMC<Event>>(config_.router_to_node_capacity); |
| | 1 | 72 | | auto [it, inserted] = router_to_nodes_.try_emplace(node_id, std::move(queue)); |
| | 1 | 73 | | auto &counters = router_to_node_counters_[node_id]; |
| | | 74 | | |
| | 1 | 75 | | if (!inserted) { |
| | 0 | 76 | | std::cerr << "[holoflow_event::Router] Error: Node '" << node_id |
| | | 77 | | << "' is already bound to the router." << std::endl; |
| | 0 | 78 | | std::abort(); |
| | | 79 | | } |
| | | 80 | | |
| | 1 | 81 | | EventWriter w{&nodes_to_router_, &nodes_to_router_counters_}; |
| | 1 | 82 | | EventReader r{it->second.get(), &counters}; |
| | 1 | 83 | | return NodeHandles{r, w}; |
| | 1 | 84 | | } |
| | | 85 | | |
| | 1 | 86 | | bool Router::ui_try_send(const NodeId &node_id, nlohmann::json &&data, time_point ts) noexcept { |
| | 1 | 87 | | Event event{EventDirection::ToNode, node_id, std::move(data), ts}; |
| | 1 | 88 | | bool ok = ui_to_router_.try_push(std::move(event)); |
| | 1 | 89 | | ui_to_router_counters_.received_events.fetch_add(ok, std::memory_order_relaxed); |
| | 1 | 90 | | ui_to_router_counters_.dropped_events.fetch_add(!ok, std::memory_order_relaxed); |
| | 1 | 91 | | return ok; |
| | 1 | 92 | | } |
| | | 93 | | |
| | 1 | 94 | | std::optional<Event> Router::ui_try_receive() noexcept { |
| | 1 | 95 | | auto event = router_to_ui_.try_pop(); |
| | 1 | 96 | | if (event) { |
| | 1 | 97 | | router_to_ui_counters_.sent_events.fetch_add(1, std::memory_order_relaxed); |
| | | 98 | | } |
| | 1 | 99 | | return event; |
| | 1 | 100 | | } |
| | | 101 | | |
| | 1 | 102 | | void Router::tick(size_t budget) { |
| | | 103 | | // Nodes->Router to Router->UI |
| | 1 | 104 | | for (size_t i = 0; i < budget; i++) { |
| | 1 | 105 | | auto event = nodes_to_router_.try_pop(); |
| | 1 | 106 | | if (!event) { |
| | 1 | 107 | | break; |
| | | 108 | | } |
| | 1 | 109 | | nodes_to_router_counters_.received_events.fetch_add(1, std::memory_order_relaxed); |
| | | 110 | | |
| | 1 | 111 | | bool ok = router_to_ui_.try_push(std::move(*event)); |
| | 1 | 112 | | router_to_ui_counters_.sent_events.fetch_add(ok, std::memory_order_relaxed); |
| | 1 | 113 | | router_to_ui_counters_.dropped_events.fetch_add(!ok, std::memory_order_relaxed); |
| | 1 | 114 | | } |
| | | 115 | | |
| | | 116 | | // UI->Router to Router->Nodes |
| | 1 | 117 | | for (size_t i = 0; i < budget; i++) { |
| | 1 | 118 | | auto event = ui_to_router_.try_pop(); |
| | 1 | 119 | | if (!event) { |
| | 1 | 120 | | break; |
| | | 121 | | } |
| | 1 | 122 | | ui_to_router_counters_.received_events.fetch_add(1, std::memory_order_relaxed); |
| | | 123 | | |
| | 1 | 124 | | auto it = router_to_nodes_.find(event->node_id); |
| | 1 | 125 | | if (it == router_to_nodes_.end()) { |
| | 1 | 126 | | std::cerr << "[holoflow_event::Router] Warning: Dropping event for unknown node '" |
| | | 127 | | << event->node_id << "'." << std::endl; |
| | 1 | 128 | | ui_to_router_counters_.dropped_events.fetch_add(1, std::memory_order_relaxed); |
| | 1 | 129 | | continue; |
| | | 130 | | } |
| | | 131 | | |
| | 1 | 132 | | auto &queue = it->second; |
| | 1 | 133 | | auto &counters = router_to_node_counters_[event->node_id]; |
| | 1 | 134 | | bool ok = queue->try_push(std::move(*event)); |
| | 1 | 135 | | counters.sent_events.fetch_add(ok, std::memory_order_relaxed); |
| | 1 | 136 | | counters.dropped_events.fetch_add(!ok, std::memory_order_relaxed); |
| | 1 | 137 | | } |
| | 1 | 138 | | } |
| | | 139 | | |
| | 1 | 140 | | MailboxCounters::Snapshot Router::ui_to_router_counters() const noexcept { |
| | 1 | 141 | | return ui_to_router_counters_.snapshot(); |
| | 1 | 142 | | } |
| | | 143 | | |
| | 0 | 144 | | MailboxCounters::Snapshot Router::router_to_ui_counters() const noexcept { |
| | 0 | 145 | | return router_to_ui_counters_.snapshot(); |
| | 0 | 146 | | } |
| | | 147 | | |
| | 0 | 148 | | MailboxCounters::Snapshot Router::nodes_to_router_counters() const noexcept { |
| | 0 | 149 | | return nodes_to_router_counters_.snapshot(); |
| | 0 | 150 | | } |
| | | 151 | | |
| | 0 | 152 | | MailboxCounters::Snapshot Router::router_to_node_counters(const NodeId &node_id) const noexcept { |
| | 0 | 153 | | auto it = router_to_node_counters_.find(node_id); |
| | 0 | 154 | | if (it == router_to_node_counters_.end()) { |
| | 0 | 155 | | std::cerr << "[holoflow_event::Router] Warning: No counters for unknown node '" << node_id |
| | | 156 | | << "'." << std::endl; |
| | 0 | 157 | | return MailboxCounters::Snapshot{0, 0, 0}; |
| | | 158 | | } |
| | 0 | 159 | | return it->second.snapshot(); |
| | 0 | 160 | | } |
| | | 161 | | |
| | | 162 | | } // namespace holoflow_event |