blob: cd04c9f72b7a785f53560ceece081e25d7948700 [file] [log] [blame]
Alex Perrycb7da4b2019-08-28 19:35:56 -07001#include "aos/events/simulated_event_loop.h"
2
3#include <algorithm>
4#include <deque>
Austin Schuh5f1cc5c2019-12-01 18:01:11 -08005#include <string_view>
Alex Perrycb7da4b2019-08-28 19:35:56 -07006
7#include "absl/container/btree_map.h"
8#include "absl/container/btree_set.h"
9#include "aos/json_to_flatbuffer.h"
10#include "aos/util/phased_loop.h"
11
12namespace aos {
13
14// Container for both a message, and the context for it for simulation. This
15// makes tracking the timestamps associated with the data easy.
16struct SimulatedMessage {
17 // Struct to let us force data to be well aligned.
18 struct OveralignedChar {
19 char data alignas(32);
20 };
21
22 // Context for the data.
23 Context context;
24
25 // The data.
26 char *data() { return reinterpret_cast<char *>(&actual_data[0]); }
27
28 // Then the data.
29 OveralignedChar actual_data[];
30};
31
Austin Schuh7d87b672019-12-01 20:23:49 -080032class SimulatedEventLoop;
Alex Perrycb7da4b2019-08-28 19:35:56 -070033class SimulatedFetcher;
Austin Schuh39788ff2019-12-01 18:22:57 -080034class SimulatedChannel;
35
Austin Schuh7d87b672019-12-01 20:23:49 -080036class SimulatedWatcher : public WatcherState {
Austin Schuh39788ff2019-12-01 18:22:57 -080037 public:
Austin Schuh7d87b672019-12-01 20:23:49 -080038 SimulatedWatcher(
39 SimulatedEventLoop *simulated_event_loop, EventScheduler *scheduler,
40 const Channel *channel,
41 std::function<void(const Context &context, const void *message)> fn);
Austin Schuh39788ff2019-12-01 18:22:57 -080042
Austin Schuh7d87b672019-12-01 20:23:49 -080043 ~SimulatedWatcher() override;
Austin Schuh39788ff2019-12-01 18:22:57 -080044
45 void Startup(EventLoop * /*event_loop*/) override {}
46
Austin Schuh7d87b672019-12-01 20:23:49 -080047 void Schedule(std::shared_ptr<SimulatedMessage> message);
48
49 void HandleEvent();
Austin Schuh39788ff2019-12-01 18:22:57 -080050
51 void SetSimulatedChannel(SimulatedChannel *channel) {
52 simulated_channel_ = channel;
53 }
54
55 private:
Austin Schuh7d87b672019-12-01 20:23:49 -080056 void DoSchedule(monotonic_clock::time_point event_time);
57
58 ::std::deque<std::shared_ptr<SimulatedMessage>> msgs_;
59
60 SimulatedEventLoop *simulated_event_loop_;
61 EventHandler<SimulatedWatcher> event_;
62 EventScheduler *scheduler_;
63 EventScheduler::Token token_;
Austin Schuh39788ff2019-12-01 18:22:57 -080064 SimulatedChannel *simulated_channel_ = nullptr;
65};
Alex Perrycb7da4b2019-08-28 19:35:56 -070066
67class SimulatedChannel {
68 public:
69 explicit SimulatedChannel(const Channel *channel, EventScheduler *scheduler)
Austin Schuh39788ff2019-12-01 18:22:57 -080070 : channel_(channel),
Alex Perrycb7da4b2019-08-28 19:35:56 -070071 scheduler_(scheduler),
72 next_queue_index_(ipc_lib::QueueIndex::Zero(channel->max_size())) {}
73
74 ~SimulatedChannel() { CHECK_EQ(0u, fetchers_.size()); }
75
76 // Makes a connected raw sender which calls Send below.
77 ::std::unique_ptr<RawSender> MakeRawSender(EventLoop *event_loop);
78
79 // Makes a connected raw fetcher.
Austin Schuh39788ff2019-12-01 18:22:57 -080080 ::std::unique_ptr<RawFetcher> MakeRawFetcher(EventLoop *event_loop);
Alex Perrycb7da4b2019-08-28 19:35:56 -070081
82 // Registers a watcher for the queue.
Austin Schuh7d87b672019-12-01 20:23:49 -080083 void MakeRawWatcher(SimulatedWatcher *watcher);
Austin Schuh39788ff2019-12-01 18:22:57 -080084
Austin Schuh7d87b672019-12-01 20:23:49 -080085 void RemoveWatcher(SimulatedWatcher *watcher) {
Austin Schuh39788ff2019-12-01 18:22:57 -080086 watchers_.erase(std::find(watchers_.begin(), watchers_.end(), watcher));
87 }
Alex Perrycb7da4b2019-08-28 19:35:56 -070088
89 // Sends the message to all the connected receivers and fetchers.
90 void Send(std::shared_ptr<SimulatedMessage> message);
91
92 // Unregisters a fetcher.
93 void UnregisterFetcher(SimulatedFetcher *fetcher);
94
95 std::shared_ptr<SimulatedMessage> latest_message() { return latest_message_; }
96
Austin Schuh39788ff2019-12-01 18:22:57 -080097 size_t max_size() const { return channel()->max_size(); }
Alex Perrycb7da4b2019-08-28 19:35:56 -070098
Austin Schuh5f1cc5c2019-12-01 18:01:11 -080099 const std::string_view name() const {
Austin Schuh39788ff2019-12-01 18:22:57 -0800100 return channel()->name()->string_view();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700101 }
102
Austin Schuh39788ff2019-12-01 18:22:57 -0800103 const Channel *channel() const { return channel_; }
104
105 ::aos::monotonic_clock::time_point monotonic_now() const {
106 return scheduler_->monotonic_now();
107 }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700108
109 private:
Austin Schuh39788ff2019-12-01 18:22:57 -0800110 const Channel *channel_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700111
112 // List of all watchers.
Austin Schuh7d87b672019-12-01 20:23:49 -0800113 ::std::vector<SimulatedWatcher *> watchers_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700114
115 // List of all fetchers.
116 ::std::vector<SimulatedFetcher *> fetchers_;
117 std::shared_ptr<SimulatedMessage> latest_message_;
118 EventScheduler *scheduler_;
119
120 ipc_lib::QueueIndex next_queue_index_;
121};
122
123namespace {
124
125// Creates a SimulatedMessage with size bytes of storage.
126// This is a shared_ptr so we don't have to implement refcounting or copying.
127std::shared_ptr<SimulatedMessage> MakeSimulatedMessage(size_t size) {
128 SimulatedMessage *message = reinterpret_cast<SimulatedMessage *>(
129 malloc(sizeof(SimulatedMessage) + size));
130 message->context.size = size;
131 message->context.data = message->data();
132
133 return std::shared_ptr<SimulatedMessage>(message, free);
134}
135
136class SimulatedSender : public RawSender {
137 public:
138 SimulatedSender(SimulatedChannel *simulated_channel, EventLoop *event_loop)
Austin Schuh39788ff2019-12-01 18:22:57 -0800139 : RawSender(event_loop, simulated_channel->channel()),
Austin Schuh54cf95f2019-11-29 13:14:18 -0800140 simulated_channel_(simulated_channel),
141 event_loop_(event_loop) {}
Alex Perrycb7da4b2019-08-28 19:35:56 -0700142 ~SimulatedSender() {}
143
144 void *data() override {
145 if (!message_) {
146 message_ = MakeSimulatedMessage(simulated_channel_->max_size());
147 }
148 return message_->data();
149 }
150
151 size_t size() override { return simulated_channel_->max_size(); }
152
Austin Schuh39788ff2019-12-01 18:22:57 -0800153 bool DoSend(size_t length) override {
Alex Perrycb7da4b2019-08-28 19:35:56 -0700154 CHECK_LE(length, size()) << ": Attempting to send too big a message.";
155 message_->context.monotonic_sent_time = event_loop_->monotonic_now();
156 message_->context.realtime_sent_time = event_loop_->realtime_now();
157 CHECK_LE(length, message_->context.size);
158 message_->context.size = length;
159
160 // TODO(austin): Track sending too fast.
161 simulated_channel_->Send(message_);
162
163 // Drop the reference to the message so that we allocate a new message for
164 // next time. Otherwise we will continue to reuse the same memory for all
165 // messages and corrupt it.
166 message_.reset();
167 return true;
168 }
169
Austin Schuh39788ff2019-12-01 18:22:57 -0800170 bool DoSend(const void *msg, size_t size) override {
Alex Perrycb7da4b2019-08-28 19:35:56 -0700171 CHECK_LE(size, this->size()) << ": Attempting to send too big a message.";
172
173 // This is wasteful, but since flatbuffers fill from the back end of the
174 // queue, we need it to be full sized.
175 message_ = MakeSimulatedMessage(simulated_channel_->max_size());
176
177 // Now fill in the message. size is already populated above, and
178 // queue_index will be populated in queue_. Put this at the back of the
179 // data segment.
180 memcpy(message_->data() + simulated_channel_->max_size() - size, msg, size);
181
182 return Send(size);
183 }
184
Alex Perrycb7da4b2019-08-28 19:35:56 -0700185 private:
186 SimulatedChannel *simulated_channel_;
187 EventLoop *event_loop_;
188
189 std::shared_ptr<SimulatedMessage> message_;
190};
191} // namespace
192
193class SimulatedFetcher : public RawFetcher {
194 public:
Austin Schuh39788ff2019-12-01 18:22:57 -0800195 explicit SimulatedFetcher(EventLoop *event_loop, SimulatedChannel *queue)
196 : RawFetcher(event_loop, queue->channel()), queue_(queue) {}
Alex Perrycb7da4b2019-08-28 19:35:56 -0700197 ~SimulatedFetcher() { queue_->UnregisterFetcher(this); }
198
Austin Schuh39788ff2019-12-01 18:22:57 -0800199 std::pair<bool, monotonic_clock::time_point> DoFetchNext() override {
200 if (msgs_.size() == 0) {
201 return std::make_pair(false, monotonic_clock::min_time);
202 }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700203
204 SetMsg(msgs_.front());
205 msgs_.pop_front();
Austin Schuh39788ff2019-12-01 18:22:57 -0800206 return std::make_pair(true, queue_->monotonic_now());
Alex Perrycb7da4b2019-08-28 19:35:56 -0700207 }
208
Austin Schuh39788ff2019-12-01 18:22:57 -0800209 std::pair<bool, monotonic_clock::time_point> DoFetch() override {
Alex Perrycb7da4b2019-08-28 19:35:56 -0700210 if (msgs_.size() == 0) {
Austin Schuh7d87b672019-12-01 20:23:49 -0800211 // TODO(austin): Can we just do this logic unconditionally? It is a lot
212 // simpler. And call clear, obviously.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700213 if (!msg_ && queue_->latest_message()) {
214 SetMsg(queue_->latest_message());
Austin Schuh39788ff2019-12-01 18:22:57 -0800215 return std::make_pair(true, queue_->monotonic_now());
Alex Perrycb7da4b2019-08-28 19:35:56 -0700216 } else {
Austin Schuh39788ff2019-12-01 18:22:57 -0800217 return std::make_pair(false, monotonic_clock::min_time);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700218 }
219 }
220
221 // We've had a message enqueued, so we don't need to go looking for the
222 // latest message from before we started.
223 SetMsg(msgs_.back());
224 msgs_.clear();
Austin Schuh39788ff2019-12-01 18:22:57 -0800225 return std::make_pair(true, queue_->monotonic_now());
Alex Perrycb7da4b2019-08-28 19:35:56 -0700226 }
227
228 private:
229 friend class SimulatedChannel;
230
231 // Updates the state inside RawFetcher to point to the data in msg_.
232 void SetMsg(std::shared_ptr<SimulatedMessage> msg) {
233 msg_ = msg;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700234 context_ = msg_->context;
235 }
236
237 // Internal method for Simulation to add a message to the buffer.
238 void Enqueue(std::shared_ptr<SimulatedMessage> buffer) {
239 msgs_.emplace_back(buffer);
240 }
241
242 SimulatedChannel *queue_;
243 std::shared_ptr<SimulatedMessage> msg_;
244
245 // Messages queued up but not in use.
246 ::std::deque<std::shared_ptr<SimulatedMessage>> msgs_;
247};
248
249class SimulatedTimerHandler : public TimerHandler {
250 public:
251 explicit SimulatedTimerHandler(EventScheduler *scheduler,
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800252 SimulatedEventLoop *simulated_event_loop,
Austin Schuh39788ff2019-12-01 18:22:57 -0800253 ::std::function<void()> fn);
Austin Schuh7d87b672019-12-01 20:23:49 -0800254 ~SimulatedTimerHandler() { Disable(); }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700255
256 void Setup(monotonic_clock::time_point base,
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800257 monotonic_clock::duration repeat_offset) override;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700258
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800259 void HandleEvent();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700260
Austin Schuh7d87b672019-12-01 20:23:49 -0800261 void Disable() override;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700262
263 ::aos::monotonic_clock::time_point monotonic_now() const {
264 return scheduler_->monotonic_now();
265 }
266
267 private:
Austin Schuh7d87b672019-12-01 20:23:49 -0800268 SimulatedEventLoop *simulated_event_loop_;
269 EventHandler<SimulatedTimerHandler> event_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700270 EventScheduler *scheduler_;
271 EventScheduler::Token token_;
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800272
Alex Perrycb7da4b2019-08-28 19:35:56 -0700273 monotonic_clock::time_point base_;
274 monotonic_clock::duration repeat_offset_;
275};
276
277class SimulatedPhasedLoopHandler : public PhasedLoopHandler {
278 public:
279 SimulatedPhasedLoopHandler(EventScheduler *scheduler,
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800280 SimulatedEventLoop *simulated_event_loop,
Alex Perrycb7da4b2019-08-28 19:35:56 -0700281 ::std::function<void(int)> fn,
282 const monotonic_clock::duration interval,
Austin Schuh39788ff2019-12-01 18:22:57 -0800283 const monotonic_clock::duration offset);
Austin Schuh7d87b672019-12-01 20:23:49 -0800284 ~SimulatedPhasedLoopHandler();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700285
Austin Schuh7d87b672019-12-01 20:23:49 -0800286 void HandleEvent();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700287
Austin Schuh7d87b672019-12-01 20:23:49 -0800288 void Schedule(monotonic_clock::time_point sleep_time) override;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700289
290 private:
Austin Schuh39788ff2019-12-01 18:22:57 -0800291 SimulatedEventLoop *simulated_event_loop_;
Austin Schuh7d87b672019-12-01 20:23:49 -0800292 EventHandler<SimulatedPhasedLoopHandler> event_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700293
Austin Schuh39788ff2019-12-01 18:22:57 -0800294 EventScheduler *scheduler_;
295 EventScheduler::Token token_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700296};
297
298class SimulatedEventLoop : public EventLoop {
299 public:
300 explicit SimulatedEventLoop(
301 EventScheduler *scheduler,
302 absl::btree_map<SimpleChannel, std::unique_ptr<SimulatedChannel>>
303 *channels,
304 const Configuration *configuration,
305 std::vector<std::pair<EventLoop *, std::function<void(bool)>>>
Austin Schuh39788ff2019-12-01 18:22:57 -0800306 *raw_event_loops,
Austin Schuh217a9782019-12-21 23:02:50 -0800307 const Node *node, pid_t tid)
Austin Schuh39788ff2019-12-01 18:22:57 -0800308 : EventLoop(CHECK_NOTNULL(configuration)),
Alex Perrycb7da4b2019-08-28 19:35:56 -0700309 scheduler_(scheduler),
310 channels_(channels),
Austin Schuh39788ff2019-12-01 18:22:57 -0800311 raw_event_loops_(raw_event_loops),
Austin Schuh217a9782019-12-21 23:02:50 -0800312 node_(node),
Austin Schuh39788ff2019-12-01 18:22:57 -0800313 tid_(tid) {
314 raw_event_loops_->push_back(std::make_pair(this, [this](bool value) {
315 if (!has_setup_) {
316 Setup();
317 has_setup_ = true;
318 }
319 set_is_running(value);
320 }));
Alex Perrycb7da4b2019-08-28 19:35:56 -0700321 }
322 ~SimulatedEventLoop() override {
Austin Schuh39788ff2019-12-01 18:22:57 -0800323 // Trigger any remaining senders or fetchers to be cleared before destroying
324 // the event loop so the book keeping matches.
325 timing_report_sender_.reset();
326
327 // Force everything with a registered fd with epoll to be destroyed now.
328 timers_.clear();
329 phased_loops_.clear();
330 watchers_.clear();
331
Alex Perrycb7da4b2019-08-28 19:35:56 -0700332 for (auto it = raw_event_loops_->begin(); it != raw_event_loops_->end();
333 ++it) {
334 if (it->first == this) {
335 raw_event_loops_->erase(it);
336 break;
337 }
338 }
339 }
340
Austin Schuh7d87b672019-12-01 20:23:49 -0800341 std::chrono::nanoseconds send_delay() const { return send_delay_; }
342 void set_send_delay(std::chrono::nanoseconds send_delay) {
343 send_delay_ = send_delay;
344 }
345
Alex Perrycb7da4b2019-08-28 19:35:56 -0700346 ::aos::monotonic_clock::time_point monotonic_now() override {
347 return scheduler_->monotonic_now();
348 }
349
350 ::aos::realtime_clock::time_point realtime_now() override {
351 return scheduler_->realtime_now();
352 }
353
354 ::std::unique_ptr<RawSender> MakeRawSender(const Channel *channel) override;
355
356 ::std::unique_ptr<RawFetcher> MakeRawFetcher(const Channel *channel) override;
357
358 void MakeRawWatcher(
359 const Channel *channel,
360 ::std::function<void(const Context &context, const void *message)>
361 watcher) override;
362
363 TimerHandler *AddTimer(::std::function<void()> callback) override {
Austin Schuh39788ff2019-12-01 18:22:57 -0800364 CHECK(!is_running());
365 return NewTimer(::std::unique_ptr<TimerHandler>(
366 new SimulatedTimerHandler(scheduler_, this, callback)));
Alex Perrycb7da4b2019-08-28 19:35:56 -0700367 }
368
369 PhasedLoopHandler *AddPhasedLoop(::std::function<void(int)> callback,
370 const monotonic_clock::duration interval,
371 const monotonic_clock::duration offset =
372 ::std::chrono::seconds(0)) override {
Austin Schuh39788ff2019-12-01 18:22:57 -0800373 return NewPhasedLoop(
374 ::std::unique_ptr<PhasedLoopHandler>(new SimulatedPhasedLoopHandler(
375 scheduler_, this, callback, interval, offset)));
Alex Perrycb7da4b2019-08-28 19:35:56 -0700376 }
377
378 void OnRun(::std::function<void()> on_run) override {
Austin Schuh39788ff2019-12-01 18:22:57 -0800379 scheduler_->ScheduleOnRun(on_run);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700380 }
381
Austin Schuh217a9782019-12-21 23:02:50 -0800382 const Node *node() const override { return node_; }
383
James Kuszmaul3ae42262019-11-08 12:33:41 -0800384 void set_name(const std::string_view name) override {
Alex Perrycb7da4b2019-08-28 19:35:56 -0700385 name_ = std::string(name);
386 }
James Kuszmaul3ae42262019-11-08 12:33:41 -0800387 const std::string_view name() const override { return name_; }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700388
389 SimulatedChannel *GetSimulatedChannel(const Channel *channel);
390
391 void Take(const Channel *channel);
392
Austin Schuh39788ff2019-12-01 18:22:57 -0800393 void SetRuntimeRealtimePriority(int priority) override {
Alex Perrycb7da4b2019-08-28 19:35:56 -0700394 CHECK(!is_running()) << ": Cannot set realtime priority while running.";
Austin Schuh39788ff2019-12-01 18:22:57 -0800395 priority_ = priority;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700396 }
397
Austin Schuh39788ff2019-12-01 18:22:57 -0800398 int priority() const override { return priority_; }
399
400 void Setup() { MaybeScheduleTimingReports(); }
401
Alex Perrycb7da4b2019-08-28 19:35:56 -0700402 private:
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800403 friend class SimulatedTimerHandler;
Austin Schuh7d87b672019-12-01 20:23:49 -0800404 friend class SimulatedPhasedLoopHandler;
405 friend class SimulatedWatcher;
406
407 void HandleEvent() {
408 while (true) {
409 if (EventCount() == 0 || PeekEvent()->event_time() > monotonic_now()) {
410 break;
411 }
412
413 EventLoopEvent *event = PopEvent();
414 event->HandleEvent();
415 }
416 }
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800417
Austin Schuh39788ff2019-12-01 18:22:57 -0800418 pid_t GetTid() override { return tid_; }
419
Alex Perrycb7da4b2019-08-28 19:35:56 -0700420 EventScheduler *scheduler_;
421 absl::btree_map<SimpleChannel, std::unique_ptr<SimulatedChannel>> *channels_;
422 std::vector<std::pair<EventLoop *, std::function<void(bool)>>>
423 *raw_event_loops_;
424 absl::btree_set<SimpleChannel> taken_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700425
426 ::std::string name_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800427
428 int priority_ = 0;
429
430 bool has_setup_ = false;
431
Austin Schuh7d87b672019-12-01 20:23:49 -0800432 std::chrono::nanoseconds send_delay_;
433
Austin Schuh217a9782019-12-21 23:02:50 -0800434 const Node *const node_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800435 const pid_t tid_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700436};
437
Austin Schuh7d87b672019-12-01 20:23:49 -0800438void SimulatedEventLoopFactory::set_send_delay(
439 std::chrono::nanoseconds send_delay) {
440 send_delay_ = send_delay;
441 for (std::pair<EventLoop *, std::function<void(bool)>> &loop :
442 raw_event_loops_) {
443 reinterpret_cast<SimulatedEventLoop *>(loop.first)
444 ->set_send_delay(send_delay_);
445 }
446}
447
Alex Perrycb7da4b2019-08-28 19:35:56 -0700448void SimulatedEventLoop::MakeRawWatcher(
449 const Channel *channel,
450 std::function<void(const Context &channel, const void *message)> watcher) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800451 ChannelIndex(channel);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700452 Take(channel);
Austin Schuh217a9782019-12-21 23:02:50 -0800453
454 if (node() != nullptr) {
455 if (!configuration::ChannelIsReadableOnNode(channel, node())) {
456 LOG(FATAL) << "Channel { \"name\": \"" << channel->name()->string_view()
457 << "\", \"type\": \"" << channel->type()->string_view()
458 << "\" } is not able to be watched on this node. Check your "
459 "configuration.";
460 }
461 }
462
Austin Schuh7d87b672019-12-01 20:23:49 -0800463 std::unique_ptr<SimulatedWatcher> shm_watcher(
464 new SimulatedWatcher(this, scheduler_, channel, std::move(watcher)));
Austin Schuh39788ff2019-12-01 18:22:57 -0800465
466 GetSimulatedChannel(channel)->MakeRawWatcher(shm_watcher.get());
467 NewWatcher(std::move(shm_watcher));
Alex Perrycb7da4b2019-08-28 19:35:56 -0700468}
469
470std::unique_ptr<RawSender> SimulatedEventLoop::MakeRawSender(
471 const Channel *channel) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800472 ChannelIndex(channel);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700473 Take(channel);
474 return GetSimulatedChannel(channel)->MakeRawSender(this);
475}
476
477std::unique_ptr<RawFetcher> SimulatedEventLoop::MakeRawFetcher(
478 const Channel *channel) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800479 ChannelIndex(channel);
Austin Schuh217a9782019-12-21 23:02:50 -0800480
481 if (node() != nullptr) {
482 if (!configuration::ChannelIsReadableOnNode(channel, node())) {
483 LOG(FATAL) << "Channel { \"name\": \"" << channel->name()->string_view()
484 << "\", \"type\": \"" << channel->type()->string_view()
485 << "\" } is not able to be fetched on this node. Check your "
486 "configuration.";
487 }
488 }
489
Austin Schuh39788ff2019-12-01 18:22:57 -0800490 return GetSimulatedChannel(channel)->MakeRawFetcher(this);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700491}
492
493SimulatedChannel *SimulatedEventLoop::GetSimulatedChannel(
494 const Channel *channel) {
495 auto it = channels_->find(SimpleChannel(channel));
496 if (it == channels_->end()) {
497 it = channels_
498 ->emplace(SimpleChannel(channel),
499 std::unique_ptr<SimulatedChannel>(
500 new SimulatedChannel(channel, scheduler_)))
501 .first;
502 }
503 return it->second.get();
504}
505
Austin Schuh7d87b672019-12-01 20:23:49 -0800506SimulatedWatcher::SimulatedWatcher(
507 SimulatedEventLoop *simulated_event_loop, EventScheduler *scheduler,
508 const Channel *channel,
509 std::function<void(const Context &context, const void *message)> fn)
510 : WatcherState(simulated_event_loop, channel, std::move(fn)),
511 simulated_event_loop_(simulated_event_loop),
512 event_(this),
513 scheduler_(scheduler),
514 token_(scheduler_->InvalidToken()) {}
515
516SimulatedWatcher::~SimulatedWatcher() {
517 simulated_event_loop_->RemoveEvent(&event_);
518 if (token_ != scheduler_->InvalidToken()) {
519 scheduler_->Deschedule(token_);
520 }
521 simulated_channel_->RemoveWatcher(this);
522}
523
524void SimulatedWatcher::Schedule(std::shared_ptr<SimulatedMessage> message) {
525 monotonic_clock::time_point event_time = scheduler_->monotonic_now();
526
527 // Messages are queued in order. If we are the first, add ourselves.
528 // Otherwise, don't.
529 if (msgs_.size() == 0) {
530 event_.set_event_time(message->context.monotonic_sent_time);
531 simulated_event_loop_->AddEvent(&event_);
532
533 DoSchedule(event_time);
534 }
535
536 msgs_.emplace_back(message);
537}
538
539void SimulatedWatcher::HandleEvent() {
540 CHECK_NE(msgs_.size(), 0u) << ": No events to handle.";
541
542 const monotonic_clock::time_point monotonic_now =
543 simulated_event_loop_->monotonic_now();
544 DoCallCallback([monotonic_now]() { return monotonic_now; },
545 msgs_.front()->context);
546
547 msgs_.pop_front();
548 if (msgs_.size() != 0) {
549 event_.set_event_time(msgs_.front()->context.monotonic_sent_time);
550 simulated_event_loop_->AddEvent(&event_);
551
552 DoSchedule(event_.event_time());
553 } else {
554 token_ = scheduler_->InvalidToken();
555 }
556}
557
558void SimulatedWatcher::DoSchedule(monotonic_clock::time_point event_time) {
559 token_ =
560 scheduler_->Schedule(event_time + simulated_event_loop_->send_delay(),
561 [this]() { simulated_event_loop_->HandleEvent(); });
562}
563
564void SimulatedChannel::MakeRawWatcher(SimulatedWatcher *watcher) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800565 watcher->SetSimulatedChannel(this);
566 watchers_.emplace_back(watcher);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700567}
568
569::std::unique_ptr<RawSender> SimulatedChannel::MakeRawSender(
570 EventLoop *event_loop) {
571 return ::std::unique_ptr<RawSender>(new SimulatedSender(this, event_loop));
572}
573
Austin Schuh39788ff2019-12-01 18:22:57 -0800574::std::unique_ptr<RawFetcher> SimulatedChannel::MakeRawFetcher(
575 EventLoop *event_loop) {
576 ::std::unique_ptr<SimulatedFetcher> fetcher(
577 new SimulatedFetcher(event_loop, this));
Alex Perrycb7da4b2019-08-28 19:35:56 -0700578 fetchers_.push_back(fetcher.get());
579 return ::std::move(fetcher);
580}
581
582void SimulatedChannel::Send(std::shared_ptr<SimulatedMessage> message) {
583 message->context.queue_index = next_queue_index_.index();
584 message->context.data =
585 message->data() + channel()->max_size() - message->context.size;
586 next_queue_index_ = next_queue_index_.Increment();
587
588 latest_message_ = message;
589 if (scheduler_->is_running()) {
Austin Schuh7d87b672019-12-01 20:23:49 -0800590 for (SimulatedWatcher *watcher : watchers_) {
591 watcher->Schedule(message);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700592 }
593 }
594 for (auto &fetcher : fetchers_) {
595 fetcher->Enqueue(message);
596 }
597}
598
599void SimulatedChannel::UnregisterFetcher(SimulatedFetcher *fetcher) {
600 fetchers_.erase(::std::find(fetchers_.begin(), fetchers_.end(), fetcher));
601}
602
603SimpleChannel::SimpleChannel(const Channel *channel)
604 : name(CHECK_NOTNULL(CHECK_NOTNULL(channel)->name())->str()),
605 type(CHECK_NOTNULL(CHECK_NOTNULL(channel)->type())->str()) {}
606
Austin Schuh39788ff2019-12-01 18:22:57 -0800607SimulatedTimerHandler::SimulatedTimerHandler(
608 EventScheduler *scheduler, SimulatedEventLoop *simulated_event_loop,
609 ::std::function<void()> fn)
610 : TimerHandler(simulated_event_loop, std::move(fn)),
Austin Schuh7d87b672019-12-01 20:23:49 -0800611 simulated_event_loop_(simulated_event_loop),
612 event_(this),
Austin Schuh39788ff2019-12-01 18:22:57 -0800613 scheduler_(scheduler),
614 token_(scheduler_->InvalidToken()) {}
615
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800616void SimulatedTimerHandler::Setup(monotonic_clock::time_point base,
617 monotonic_clock::duration repeat_offset) {
618 Disable();
619 const ::aos::monotonic_clock::time_point monotonic_now =
620 scheduler_->monotonic_now();
621 base_ = base;
622 repeat_offset_ = repeat_offset;
623 if (base < monotonic_now) {
Austin Schuh7d87b672019-12-01 20:23:49 -0800624 token_ = scheduler_->Schedule(
625 monotonic_now, [this]() { simulated_event_loop_->HandleEvent(); });
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800626 } else {
Austin Schuh7d87b672019-12-01 20:23:49 -0800627 token_ = scheduler_->Schedule(
628 base, [this]() { simulated_event_loop_->HandleEvent(); });
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800629 }
Austin Schuh7d87b672019-12-01 20:23:49 -0800630 event_.set_event_time(base_);
631 simulated_event_loop_->AddEvent(&event_);
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800632}
633
634void SimulatedTimerHandler::HandleEvent() {
635 const ::aos::monotonic_clock::time_point monotonic_now =
636 scheduler_->monotonic_now();
637 if (repeat_offset_ != ::aos::monotonic_clock::zero()) {
638 // Reschedule.
639 while (base_ <= monotonic_now) base_ += repeat_offset_;
Austin Schuh7d87b672019-12-01 20:23:49 -0800640 token_ = scheduler_->Schedule(
641 base_, [this]() { simulated_event_loop_->HandleEvent(); });
642 event_.set_event_time(base_);
643 simulated_event_loop_->AddEvent(&event_);
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800644 } else {
645 token_ = scheduler_->InvalidToken();
646 }
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800647
Austin Schuh39788ff2019-12-01 18:22:57 -0800648 Call([monotonic_now]() { return monotonic_now; }, monotonic_now);
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800649}
650
Austin Schuh7d87b672019-12-01 20:23:49 -0800651void SimulatedTimerHandler::Disable() {
652 simulated_event_loop_->RemoveEvent(&event_);
653 if (token_ != scheduler_->InvalidToken()) {
654 scheduler_->Deschedule(token_);
655 token_ = scheduler_->InvalidToken();
656 }
657}
658
Austin Schuh39788ff2019-12-01 18:22:57 -0800659SimulatedPhasedLoopHandler::SimulatedPhasedLoopHandler(
660 EventScheduler *scheduler, SimulatedEventLoop *simulated_event_loop,
661 ::std::function<void(int)> fn, const monotonic_clock::duration interval,
662 const monotonic_clock::duration offset)
663 : PhasedLoopHandler(simulated_event_loop, std::move(fn), interval, offset),
664 simulated_event_loop_(simulated_event_loop),
Austin Schuh7d87b672019-12-01 20:23:49 -0800665 event_(this),
Austin Schuh39788ff2019-12-01 18:22:57 -0800666 scheduler_(scheduler),
667 token_(scheduler_->InvalidToken()) {}
668
Austin Schuh7d87b672019-12-01 20:23:49 -0800669SimulatedPhasedLoopHandler::~SimulatedPhasedLoopHandler() {
670 if (token_ != scheduler_->InvalidToken()) {
671 scheduler_->Deschedule(token_);
672 token_ = scheduler_->InvalidToken();
673 }
674 simulated_event_loop_->RemoveEvent(&event_);
675}
676
677void SimulatedPhasedLoopHandler::HandleEvent() {
Austin Schuh39788ff2019-12-01 18:22:57 -0800678 monotonic_clock::time_point monotonic_now =
679 simulated_event_loop_->monotonic_now();
680 Call(
681 [monotonic_now]() { return monotonic_now; },
682 [this](monotonic_clock::time_point sleep_time) { Schedule(sleep_time); });
683}
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800684
Austin Schuh7d87b672019-12-01 20:23:49 -0800685void SimulatedPhasedLoopHandler::Schedule(
686 monotonic_clock::time_point sleep_time) {
687 token_ = scheduler_->Schedule(
688 sleep_time, [this]() { simulated_event_loop_->HandleEvent(); });
689 event_.set_event_time(sleep_time);
690 simulated_event_loop_->AddEvent(&event_);
691}
692
Alex Perrycb7da4b2019-08-28 19:35:56 -0700693void SimulatedEventLoop::Take(const Channel *channel) {
694 CHECK(!is_running()) << ": Cannot add new objects while running.";
695
696 auto result = taken_.insert(SimpleChannel(channel));
697 CHECK(result.second) << ": " << FlatbufferToJson(channel)
698 << " is already being used.";
699}
700
701SimulatedEventLoopFactory::SimulatedEventLoopFactory(
702 const Configuration *configuration)
Austin Schuh217a9782019-12-21 23:02:50 -0800703 : configuration_(CHECK_NOTNULL(configuration)), node_(nullptr) {
704 CHECK(!configuration_->has_nodes())
705 << ": Got a configuration with multiple nodes and no node was selected.";
706}
707
708SimulatedEventLoopFactory::SimulatedEventLoopFactory(
709 const Configuration *configuration, std::string_view node_name)
710 : configuration_(CHECK_NOTNULL(configuration)),
711 node_(configuration::GetNode(configuration, node_name)) {
712 CHECK(configuration_->has_nodes())
713 << ": Got a configuration with no nodes and node \"" << node_name
714 << "\" was selected.";
715 CHECK(node_ != nullptr) << ": Can't find node \"" << node_name
716 << "\" in the configuration.";
717}
718
Alex Perrycb7da4b2019-08-28 19:35:56 -0700719SimulatedEventLoopFactory::~SimulatedEventLoopFactory() {}
720
Austin Schuh5f1cc5c2019-12-01 18:01:11 -0800721::std::unique_ptr<EventLoop> SimulatedEventLoopFactory::MakeEventLoop(
722 std::string_view name) {
Austin Schuh39788ff2019-12-01 18:22:57 -0800723 pid_t tid = tid_;
724 ++tid_;
Austin Schuh7d87b672019-12-01 20:23:49 -0800725 ::std::unique_ptr<SimulatedEventLoop> result(new SimulatedEventLoop(
Austin Schuh217a9782019-12-21 23:02:50 -0800726 &scheduler_, &channels_, configuration_, &raw_event_loops_, node_, tid));
Austin Schuh5f1cc5c2019-12-01 18:01:11 -0800727 result->set_name(name);
Austin Schuh7d87b672019-12-01 20:23:49 -0800728 result->set_send_delay(send_delay_);
729 return std::move(result);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700730}
731
732void SimulatedEventLoopFactory::RunFor(monotonic_clock::duration duration) {
733 for (const std::pair<EventLoop *, std::function<void(bool)>> &event_loop :
734 raw_event_loops_) {
735 event_loop.second(true);
736 }
737 scheduler_.RunFor(duration);
Austin Schuh39788ff2019-12-01 18:22:57 -0800738 for (const std::pair<EventLoop *, std::function<void(bool)>> &event_loop :
739 raw_event_loops_) {
740 event_loop.second(false);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700741 }
742}
743
744void SimulatedEventLoopFactory::Run() {
745 for (const std::pair<EventLoop *, std::function<void(bool)>> &event_loop :
746 raw_event_loops_) {
747 event_loop.second(true);
748 }
749 scheduler_.Run();
Austin Schuh39788ff2019-12-01 18:22:57 -0800750 for (const std::pair<EventLoop *, std::function<void(bool)>> &event_loop :
751 raw_event_loops_) {
752 event_loop.second(false);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700753 }
754}
755
756} // namespace aos