blob: c922f2ec924516ebe4e18c8cb0832ab85ff011c4 [file] [log] [blame]
Alex Perrycb7da4b2019-08-28 19:35:56 -07001#ifndef AOS_EVENTS_EVENT_LOOP_H_
2#define AOS_EVENTS_EVENT_LOOP_H_
3
Brian Silverman6a54ff32020-04-28 16:41:39 -07004#include <sched.h>
5
Alex Perrycb7da4b2019-08-28 19:35:56 -07006#include <atomic>
7#include <string>
James Kuszmaul3ae42262019-11-08 12:33:41 -08008#include <string_view>
Alex Perrycb7da4b2019-08-28 19:35:56 -07009
Alex Perrycb7da4b2019-08-28 19:35:56 -070010#include "aos/configuration.h"
11#include "aos/configuration_generated.h"
Austin Schuh1af273d2020-03-07 20:11:34 -080012#include "aos/events/channel_preallocated_allocator.h"
Austin Schuh7d87b672019-12-01 20:23:49 -080013#include "aos/events/event_loop_event.h"
Austin Schuh39788ff2019-12-01 18:22:57 -080014#include "aos/events/event_loop_generated.h"
15#include "aos/events/timing_statistics.h"
Alex Perrycb7da4b2019-08-28 19:35:56 -070016#include "aos/flatbuffers.h"
Brian Silverman79ec7fc2020-06-08 20:11:22 -050017#include "aos/ftrace.h"
Brian Silvermana1652f32020-01-29 20:41:44 -080018#include "aos/ipc_lib/data_alignment.h"
Alex Perrycb7da4b2019-08-28 19:35:56 -070019#include "aos/json_to_flatbuffer.h"
20#include "aos/time/time.h"
Austin Schuh39788ff2019-12-01 18:22:57 -080021#include "aos/util/phased_loop.h"
Brian Silverman0fc69932020-01-24 21:54:02 -080022
23#include "absl/container/btree_set.h"
Alex Perrycb7da4b2019-08-28 19:35:56 -070024#include "flatbuffers/flatbuffers.h"
25#include "glog/logging.h"
26
Austin Schuh39788ff2019-12-01 18:22:57 -080027DECLARE_bool(timing_reports);
28DECLARE_int32(timing_report_ms);
29
Alex Perrycb7da4b2019-08-28 19:35:56 -070030namespace aos {
31
Austin Schuh39788ff2019-12-01 18:22:57 -080032class EventLoop;
33class WatcherState;
34
Austin Schuh6231cc32019-12-07 13:06:15 -080035// Struct available on Watchers, Fetchers, Timers, and PhasedLoops with context
36// about the current message.
Alex Perrycb7da4b2019-08-28 19:35:56 -070037struct Context {
Austin Schuhad154822019-12-27 15:45:13 -080038 // Time that the message was sent on this node, or the timer was triggered.
39 monotonic_clock::time_point monotonic_event_time;
40 // Realtime the message was sent on this node. This is set to min_time for
41 // Timers and PhasedLoops.
42 realtime_clock::time_point realtime_event_time;
43
44 // For a single-node configuration, these two are identical to *_event_time.
45 // In a multinode configuration, these are the times that the message was
46 // sent on the original node.
47 monotonic_clock::time_point monotonic_remote_time;
48 realtime_clock::time_point realtime_remote_time;
49
Austin Schuh6231cc32019-12-07 13:06:15 -080050 // The rest are only valid for Watchers and Fetchers.
Brian Silverman4f4e0612020-08-12 19:54:41 -070051
Alex Perrycb7da4b2019-08-28 19:35:56 -070052 // Index in the queue.
53 uint32_t queue_index;
Austin Schuhad154822019-12-27 15:45:13 -080054 // Index into the remote queue. Useful to determine if data was lost. In a
55 // single-node configuration, this will match queue_index.
56 uint32_t remote_queue_index;
57
Alex Perrycb7da4b2019-08-28 19:35:56 -070058 // Size of the data sent.
59 size_t size;
60 // Pointer to the data.
Brian Silvermaneaa41d62020-07-08 19:47:35 -070061 const void *data;
Austin Schuh678078e2020-08-01 14:30:36 -070062
Brian Silverman4f4e0612020-08-12 19:54:41 -070063 // Index of the message buffer. This will be in [0, NumberBuffers) on
64 // read_method=PIN channels, and -1 for other channels.
65 //
66 // This only tells you about the underlying storage for this message, not
67 // anything about its position in the queue. This is only useful for advanced
68 // zero-copy use cases, on read_method=PIN channels.
69 //
70 // This will uniquely identify a message on this channel at a point in time.
71 // For senders, this point in time is while the sender has the message. With
72 // read_method==PIN, this point in time includes while the caller has access
73 // to this context. For other read_methods, this point in time may be before
74 // the caller has access to this context, which makes this pretty useless.
75 int buffer_index;
76
Austin Schuh678078e2020-08-01 14:30:36 -070077 // Efficiently coppies the flatbuffer into a FlatbufferVector, allocating
78 // memory in the process. It is vital that T matches the type of the
79 // underlying flatbuffer.
80 template <typename T>
81 FlatbufferVector<T> CopyFlatBuffer() const {
82 return FlatbufferVector<T>(
83 std::vector<uint8_t>(reinterpret_cast<const uint8_t *>(data),
84 reinterpret_cast<const uint8_t *>(data) + size));
85 }
Alex Perrycb7da4b2019-08-28 19:35:56 -070086};
87
88// Raw version of fetcher. Contains a local variable that the fetcher will
89// update. This is used for reflection and as an interface to implement typed
90// fetchers.
91class RawFetcher {
92 public:
Austin Schuh39788ff2019-12-01 18:22:57 -080093 RawFetcher(EventLoop *event_loop, const Channel *channel);
Alex Perrycb7da4b2019-08-28 19:35:56 -070094 RawFetcher(const RawFetcher &) = delete;
95 RawFetcher &operator=(const RawFetcher &) = delete;
Austin Schuh39788ff2019-12-01 18:22:57 -080096 virtual ~RawFetcher();
Alex Perrycb7da4b2019-08-28 19:35:56 -070097
Austin Schuh39788ff2019-12-01 18:22:57 -080098 // Fetches the next message in the queue without blocking. Returns true if
99 // there was a new message and we got it.
100 bool FetchNext();
101
102 // Fetches the latest message without blocking.
103 bool Fetch();
104
105 // Returns the channel this fetcher uses.
106 const Channel *channel() const { return channel_; }
107 // Returns the context for the current message.
108 const Context &context() const { return context_; }
109
110 protected:
111 EventLoop *event_loop() { return event_loop_; }
112
Alex Perrycb7da4b2019-08-28 19:35:56 -0700113 Context context_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800114
115 private:
116 friend class EventLoop;
117 // Implementation
118 virtual std::pair<bool, monotonic_clock::time_point> DoFetchNext() = 0;
119 virtual std::pair<bool, monotonic_clock::time_point> DoFetch() = 0;
120
Brian Silverman79ec7fc2020-06-08 20:11:22 -0500121 EventLoop *const event_loop_;
122 const Channel *const channel_;
123 const std::string ftrace_prefix_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800124
125 internal::RawFetcherTiming timing_;
Brian Silverman79ec7fc2020-06-08 20:11:22 -0500126 Ftrace ftrace_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700127};
128
129// Raw version of sender. Sends a block of data. This is used for reflection
130// and as a building block to implement typed senders.
131class RawSender {
132 public:
Austin Schuh39788ff2019-12-01 18:22:57 -0800133 RawSender(EventLoop *event_loop, const Channel *channel);
134 RawSender(const RawSender &) = delete;
135 RawSender &operator=(const RawSender &) = delete;
136
137 virtual ~RawSender();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700138
139 // Sends a message without copying it. The users starts by copying up to
140 // size() bytes into the data backed by data(). They then call Send to send.
141 // Returns true on a successful send.
Austin Schuhad154822019-12-27 15:45:13 -0800142 // If provided, monotonic_remote_time, realtime_remote_time, and
143 // remote_queue_index are attached to the message and are available in the
144 // context on the read side. If they are not populated, the read side will
145 // get the sent times instead.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700146 virtual void *data() = 0;
147 virtual size_t size() = 0;
Austin Schuhad154822019-12-27 15:45:13 -0800148 bool Send(size_t size,
149 aos::monotonic_clock::time_point monotonic_remote_time =
150 aos::monotonic_clock::min_time,
151 aos::realtime_clock::time_point realtime_remote_time =
152 aos::realtime_clock::min_time,
153 uint32_t remote_queue_index = 0xffffffffu);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700154
155 // Sends a single block of data by copying it.
Austin Schuhad154822019-12-27 15:45:13 -0800156 // The remote arguments have the same meaning as in Send above.
157 bool Send(const void *data, size_t size,
158 aos::monotonic_clock::time_point monotonic_remote_time =
159 aos::monotonic_clock::min_time,
160 aos::realtime_clock::time_point realtime_remote_time =
161 aos::realtime_clock::min_time,
162 uint32_t remote_queue_index = 0xffffffffu);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700163
Austin Schuh54cf95f2019-11-29 13:14:18 -0800164 const Channel *channel() const { return channel_; }
165
Austin Schuhad154822019-12-27 15:45:13 -0800166 // Returns the time_points that the last message was sent at.
167 aos::monotonic_clock::time_point monotonic_sent_time() const {
168 return monotonic_sent_time_;
169 }
170 aos::realtime_clock::time_point realtime_sent_time() const {
171 return realtime_sent_time_;
172 }
173 // Returns the queue index that this was sent with.
174 uint32_t sent_queue_index() const { return sent_queue_index_; }
175
Brian Silvermana1652f32020-01-29 20:41:44 -0800176 // Returns the associated flatbuffers-style allocator. This must be
177 // deallocated before the message is sent.
Austin Schuh1af273d2020-03-07 20:11:34 -0800178 ChannelPreallocatedAllocator *fbb_allocator() {
179 fbb_allocator_ = ChannelPreallocatedAllocator(
180 reinterpret_cast<uint8_t *>(data()), size(), channel());
Brian Silvermana1652f32020-01-29 20:41:44 -0800181 return &fbb_allocator_;
182 }
183
Brian Silverman4f4e0612020-08-12 19:54:41 -0700184 // Index of the buffer which is currently exposed by data() and the various
185 // other accessors. This is the message the caller should be filling out.
186 virtual int buffer_index() = 0;
187
Alex Perrycb7da4b2019-08-28 19:35:56 -0700188 protected:
Austin Schuh39788ff2019-12-01 18:22:57 -0800189 EventLoop *event_loop() { return event_loop_; }
Austin Schuh54cf95f2019-11-29 13:14:18 -0800190
Austin Schuhad154822019-12-27 15:45:13 -0800191 aos::monotonic_clock::time_point monotonic_sent_time_ =
192 aos::monotonic_clock::min_time;
193 aos::realtime_clock::time_point realtime_sent_time_ =
194 aos::realtime_clock::min_time;
195 uint32_t sent_queue_index_ = 0xffffffff;
196
Austin Schuh39788ff2019-12-01 18:22:57 -0800197 private:
198 friend class EventLoop;
199
Austin Schuhad154822019-12-27 15:45:13 -0800200 virtual bool DoSend(const void *data, size_t size,
201 aos::monotonic_clock::time_point monotonic_remote_time,
202 aos::realtime_clock::time_point realtime_remote_time,
203 uint32_t remote_queue_index) = 0;
204 virtual bool DoSend(size_t size,
205 aos::monotonic_clock::time_point monotonic_remote_time,
206 aos::realtime_clock::time_point realtime_remote_time,
207 uint32_t remote_queue_index) = 0;
Austin Schuh39788ff2019-12-01 18:22:57 -0800208
Brian Silverman79ec7fc2020-06-08 20:11:22 -0500209 EventLoop *const event_loop_;
210 const Channel *const channel_;
211 const std::string ftrace_prefix_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700212
Austin Schuh39788ff2019-12-01 18:22:57 -0800213 internal::RawSenderTiming timing_;
Brian Silverman79ec7fc2020-06-08 20:11:22 -0500214 Ftrace ftrace_;
Brian Silvermana1652f32020-01-29 20:41:44 -0800215
Austin Schuh1af273d2020-03-07 20:11:34 -0800216 ChannelPreallocatedAllocator fbb_allocator_{nullptr, 0, nullptr};
Austin Schuh39788ff2019-12-01 18:22:57 -0800217};
Alex Perrycb7da4b2019-08-28 19:35:56 -0700218
219// Fetches the newest message from a channel.
220// This provides a polling based interface for channels.
221template <typename T>
222class Fetcher {
223 public:
224 Fetcher() {}
225
226 // Fetches the next message. Returns true if it fetched a new message. This
227 // method will only return messages sent after the Fetcher was created.
Brian Silvermana1652f32020-01-29 20:41:44 -0800228 bool FetchNext() {
229 const bool result = fetcher_->FetchNext();
230 if (result) {
231 CheckChannelDataAlignment(fetcher_->context().data,
232 fetcher_->context().size);
233 }
234 return result;
235 }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700236
237 // Fetches the most recent message. Returns true if it fetched a new message.
238 // This will return the latest message regardless of if it was sent before or
239 // after the fetcher was created.
Brian Silvermana1652f32020-01-29 20:41:44 -0800240 bool Fetch() {
241 const bool result = fetcher_->Fetch();
242 if (result) {
243 CheckChannelDataAlignment(fetcher_->context().data,
244 fetcher_->context().size);
245 }
246 return result;
247 }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700248
249 // Returns a pointer to the contained flatbuffer, or nullptr if there is no
250 // available message.
251 const T *get() const {
Austin Schuh39788ff2019-12-01 18:22:57 -0800252 return fetcher_->context().data != nullptr
253 ? flatbuffers::GetRoot<T>(
254 reinterpret_cast<const char *>(fetcher_->context().data))
Alex Perrycb7da4b2019-08-28 19:35:56 -0700255 : nullptr;
256 }
257
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700258 // Returns the channel this fetcher uses
259 const Channel *channel() const { return fetcher_->channel(); }
260
Alex Perrycb7da4b2019-08-28 19:35:56 -0700261 // Returns the context holding timestamps and other metadata about the
262 // message.
263 const Context &context() const { return fetcher_->context(); }
264
265 const T &operator*() const { return *get(); }
266 const T *operator->() const { return get(); }
267
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700268 // Returns true if this fetcher is valid and connected to a channel.
269 operator bool() const { return static_cast<bool>(fetcher_); }
270
Alex Perrycb7da4b2019-08-28 19:35:56 -0700271 private:
272 friend class EventLoop;
273 Fetcher(::std::unique_ptr<RawFetcher> fetcher)
274 : fetcher_(::std::move(fetcher)) {}
275 ::std::unique_ptr<RawFetcher> fetcher_;
276};
277
278// Sends messages to a channel.
279template <typename T>
280class Sender {
281 public:
282 Sender() {}
283
284 // Represents a single message about to be sent to the queue.
285 // The lifecycle goes:
286 //
287 // Builder builder = sender.MakeBuilder();
288 // T::Builder t_builder = builder.MakeBuilder<T>();
289 // Populate(&t_builder);
290 // builder.Send(t_builder.Finish());
291 class Builder {
292 public:
Austin Schuh1af273d2020-03-07 20:11:34 -0800293 Builder(RawSender *sender, ChannelPreallocatedAllocator *allocator)
Brian Silverman9dd793b2020-01-31 23:52:21 -0800294 : fbb_(allocator->size(), allocator),
295 allocator_(allocator),
296 sender_(sender) {
Brian Silvermana1652f32020-01-29 20:41:44 -0800297 CheckChannelDataAlignment(allocator->data(), allocator->size());
Austin Schuhd7b15da2020-02-17 15:06:11 -0800298 fbb_.ForceDefaults(true);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700299 }
Brian Silvermana1652f32020-01-29 20:41:44 -0800300 Builder() {}
301 Builder(const Builder &) = delete;
302 Builder(Builder &&) = default;
303
304 Builder &operator=(const Builder &) = delete;
305 Builder &operator=(Builder &&) = default;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700306
307 flatbuffers::FlatBufferBuilder *fbb() { return &fbb_; }
308
309 template <typename T2>
310 typename T2::Builder MakeBuilder() {
311 return typename T2::Builder(fbb_);
312 }
313
314 bool Send(flatbuffers::Offset<T> offset) {
315 fbb_.Finish(offset);
Brian Silverman9dd793b2020-01-31 23:52:21 -0800316 const bool result = sender_->Send(fbb_.GetSize());
317 // Ensure fbb_ knows it shouldn't access the memory any more.
318 fbb_ = flatbuffers::FlatBufferBuilder();
319 return result;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700320 }
321
322 // CHECKs that this message was sent.
Brian Silverman9dd793b2020-01-31 23:52:21 -0800323 void CheckSent() {
324 CHECK(!allocator_->is_allocated()) << ": Message was not sent yet";
325 }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700326
Brian Silverman341b57e2020-06-23 16:23:18 -0700327 // Detaches a buffer, for later use calling Sender::Send directly.
328 //
329 // Note that the underlying memory remains with the Sender, so creating
330 // another Builder before destroying the FlatbufferDetachedBuffer will fail.
331 FlatbufferDetachedBuffer<T> Detach(flatbuffers::Offset<T> offset) {
332 fbb_.Finish(offset);
333 return fbb_.Release();
334 }
335
Alex Perrycb7da4b2019-08-28 19:35:56 -0700336 private:
Alex Perrycb7da4b2019-08-28 19:35:56 -0700337 flatbuffers::FlatBufferBuilder fbb_;
Austin Schuh1af273d2020-03-07 20:11:34 -0800338 ChannelPreallocatedAllocator *allocator_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700339 RawSender *sender_;
340 };
341
342 // Constructs an above builder.
Brian Silverman9dd793b2020-01-31 23:52:21 -0800343 //
344 // Only a single one of these may be "alive" for this object at any point in
345 // time. After calling Send on the result, it is no longer "alive". This means
346 // that you must manually reset a variable holding the return value (by
347 // assigning a default-constructed Builder to it) before calling this method
348 // again to overwrite the value in the variable.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700349 Builder MakeBuilder();
350
Austin Schuha28cbc32019-12-27 16:28:04 -0800351 // Sends a prebuilt flatbuffer.
352 bool Send(const Flatbuffer<T> &flatbuffer);
353
Brian Silverman341b57e2020-06-23 16:23:18 -0700354 // Sends a prebuilt flatbuffer which was detached from a Builder created via
355 // MakeBuilder() on this object.
356 bool SendDetached(FlatbufferDetachedBuffer<T> detached);
357
Austin Schuh39788ff2019-12-01 18:22:57 -0800358 // Returns the name of the underlying queue.
Austin Schuh1e869472019-12-01 13:36:10 -0800359 const Channel *channel() const { return sender_->channel(); }
Alex Perrycb7da4b2019-08-28 19:35:56 -0700360
Brian Silverman454bc112020-03-05 14:21:25 -0800361 operator bool() { return sender_ ? true : false; }
Tyler Chatow67ddb032020-01-12 14:30:04 -0800362
Austin Schuh7bc59052020-02-16 23:48:33 -0800363 // Returns the time_points that the last message was sent at.
364 aos::monotonic_clock::time_point monotonic_sent_time() const {
365 return sender_->monotonic_sent_time();
366 }
367 aos::realtime_clock::time_point realtime_sent_time() const {
368 return sender_->realtime_sent_time();
369 }
370 // Returns the queue index that this was sent with.
371 uint32_t sent_queue_index() const { return sender_->sent_queue_index(); }
372
Brian Silverman4f4e0612020-08-12 19:54:41 -0700373 // Returns the buffer index which MakeBuilder() will expose access to. This is
374 // the buffer the caller can fill out.
375 int buffer_index() const { return sender_->buffer_index(); }
376
Alex Perrycb7da4b2019-08-28 19:35:56 -0700377 private:
378 friend class EventLoop;
379 Sender(std::unique_ptr<RawSender> sender) : sender_(std::move(sender)) {}
380 std::unique_ptr<RawSender> sender_;
381};
382
Brian Silverman4f4e0612020-08-12 19:54:41 -0700383// Interface for timers.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700384class TimerHandler {
385 public:
Austin Schuh39788ff2019-12-01 18:22:57 -0800386 virtual ~TimerHandler();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700387
388 // Timer should sleep until base, base + offset, base + offset * 2, ...
389 // If repeat_offset isn't set, the timer only expires once.
390 virtual void Setup(monotonic_clock::time_point base,
391 monotonic_clock::duration repeat_offset =
392 ::aos::monotonic_clock::zero()) = 0;
393
394 // Stop future calls to callback().
395 virtual void Disable() = 0;
Austin Schuh1540c2f2019-11-29 21:59:29 -0800396
397 // Sets and gets the name of the timer. Set this if you want a descriptive
398 // name in the timing report.
399 void set_name(std::string_view name) { name_ = std::string(name); }
400 const std::string_view name() const { return name_; }
401
Austin Schuh39788ff2019-12-01 18:22:57 -0800402 protected:
403 TimerHandler(EventLoop *event_loop, std::function<void()> fn);
404
Austin Schuhcde39fd2020-02-22 20:58:24 -0800405 monotonic_clock::time_point Call(
406 std::function<monotonic_clock::time_point()> get_time,
407 monotonic_clock::time_point event_time);
Austin Schuh39788ff2019-12-01 18:22:57 -0800408
Austin Schuh1540c2f2019-11-29 21:59:29 -0800409 private:
Austin Schuh39788ff2019-12-01 18:22:57 -0800410 friend class EventLoop;
411
412 EventLoop *event_loop_;
413 // The function to call when Call is called.
414 std::function<void()> fn_;
Austin Schuh1540c2f2019-11-29 21:59:29 -0800415 std::string name_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800416
417 internal::TimerTiming timing_;
Brian Silverman79ec7fc2020-06-08 20:11:22 -0500418 Ftrace ftrace_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700419};
420
421// Interface for phased loops. They are built on timers.
422class PhasedLoopHandler {
423 public:
Austin Schuh39788ff2019-12-01 18:22:57 -0800424 virtual ~PhasedLoopHandler();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700425
426 // Sets the interval and offset. Any changes to interval and offset only take
427 // effect when the handler finishes running.
Austin Schuh39788ff2019-12-01 18:22:57 -0800428 void set_interval_and_offset(const monotonic_clock::duration interval,
429 const monotonic_clock::duration offset) {
430 phased_loop_.set_interval_and_offset(interval, offset);
431 }
Austin Schuh1540c2f2019-11-29 21:59:29 -0800432
433 // Sets and gets the name of the timer. Set this if you want a descriptive
434 // name in the timing report.
435 void set_name(std::string_view name) { name_ = std::string(name); }
436 const std::string_view name() const { return name_; }
437
Austin Schuh39788ff2019-12-01 18:22:57 -0800438 protected:
439 void Call(std::function<monotonic_clock::time_point()> get_time,
440 std::function<void(monotonic_clock::time_point)> schedule);
441
442 PhasedLoopHandler(EventLoop *event_loop, std::function<void(int)> fn,
443 const monotonic_clock::duration interval,
444 const monotonic_clock::duration offset);
445
Austin Schuh1540c2f2019-11-29 21:59:29 -0800446 private:
Austin Schuh39788ff2019-12-01 18:22:57 -0800447 friend class EventLoop;
448
449 void Reschedule(std::function<void(monotonic_clock::time_point)> schedule,
450 monotonic_clock::time_point monotonic_now) {
451 cycles_elapsed_ += phased_loop_.Iterate(monotonic_now);
452 schedule(phased_loop_.sleep_time());
453 }
454
455 virtual void Schedule(monotonic_clock::time_point sleep_time) = 0;
456
457 EventLoop *event_loop_;
458 std::function<void(int)> fn_;
Austin Schuh1540c2f2019-11-29 21:59:29 -0800459 std::string name_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800460 time::PhasedLoop phased_loop_;
461
462 int cycles_elapsed_ = 0;
463
464 internal::TimerTiming timing_;
Brian Silverman79ec7fc2020-06-08 20:11:22 -0500465 Ftrace ftrace_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700466};
467
Brian Silverman6a54ff32020-04-28 16:41:39 -0700468inline cpu_set_t MakeCpusetFromCpus(std::initializer_list<int> cpus) {
469 cpu_set_t result;
470 CPU_ZERO(&result);
471 for (int cpu : cpus) {
472 CPU_SET(cpu, &result);
473 }
474 return result;
475}
476
Alex Perrycb7da4b2019-08-28 19:35:56 -0700477class EventLoop {
478 public:
Tyler Chatow67ddb032020-01-12 14:30:04 -0800479 EventLoop(const Configuration *configuration);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700480
Austin Schuh39788ff2019-12-01 18:22:57 -0800481 virtual ~EventLoop();
Alex Perrycb7da4b2019-08-28 19:35:56 -0700482
483 // Current time.
484 virtual monotonic_clock::time_point monotonic_now() = 0;
485 virtual realtime_clock::time_point realtime_now() = 0;
486
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700487 // Returns true if the channel exists in the configuration.
488 template <typename T>
489 bool HasChannel(const std::string_view channel_name) {
490 return configuration::GetChannel(configuration_, channel_name,
491 T::GetFullyQualifiedName(), name(),
492 node()) != nullptr;
493 }
494
Alex Perrycb7da4b2019-08-28 19:35:56 -0700495 // Note, it is supported to create:
496 // multiple fetchers, and (one sender or one watcher) per <name, type>
497 // tuple.
498
499 // Makes a class that will always fetch the most recent value
500 // sent to the provided channel.
501 template <typename T>
James Kuszmaul3ae42262019-11-08 12:33:41 -0800502 Fetcher<T> MakeFetcher(const std::string_view channel_name) {
Austin Schuhbca6cf02019-12-22 17:28:34 -0800503 const Channel *channel =
504 configuration::GetChannel(configuration_, channel_name,
505 T::GetFullyQualifiedName(), name(), node());
Alex Perrycb7da4b2019-08-28 19:35:56 -0700506 CHECK(channel != nullptr)
507 << ": Channel { \"name\": \"" << channel_name << "\", \"type\": \""
508 << T::GetFullyQualifiedName() << "\" } not found in config.";
509
Austin Schuhca4828c2019-12-28 14:21:35 -0800510 if (!configuration::ChannelIsReadableOnNode(channel, node())) {
511 LOG(FATAL) << "Channel { \"name\": \"" << channel_name
512 << "\", \"type\": \"" << T::GetFullyQualifiedName()
513 << "\" } is not able to be fetched on this node. Check your "
514 "configuration.";
Austin Schuh217a9782019-12-21 23:02:50 -0800515 }
516
Alex Perrycb7da4b2019-08-28 19:35:56 -0700517 return Fetcher<T>(MakeRawFetcher(channel));
518 }
519
520 // Makes class that allows constructing and sending messages to
521 // the provided channel.
522 template <typename T>
James Kuszmaul3ae42262019-11-08 12:33:41 -0800523 Sender<T> MakeSender(const std::string_view channel_name) {
Austin Schuhbca6cf02019-12-22 17:28:34 -0800524 const Channel *channel =
525 configuration::GetChannel(configuration_, channel_name,
526 T::GetFullyQualifiedName(), name(), node());
Alex Perrycb7da4b2019-08-28 19:35:56 -0700527 CHECK(channel != nullptr)
528 << ": Channel { \"name\": \"" << channel_name << "\", \"type\": \""
Austin Schuh28fedcb2020-02-08 15:59:58 -0800529 << T::GetFullyQualifiedName() << "\" } not found in config for "
530 << name() << ".";
Alex Perrycb7da4b2019-08-28 19:35:56 -0700531
Austin Schuhca4828c2019-12-28 14:21:35 -0800532 if (!configuration::ChannelIsSendableOnNode(channel, node())) {
533 LOG(FATAL) << "Channel { \"name\": \"" << channel_name
534 << "\", \"type\": \"" << T::GetFullyQualifiedName()
535 << "\" } is not able to be sent on this node. Check your "
536 "configuration.";
Austin Schuh217a9782019-12-21 23:02:50 -0800537 }
538
Alex Perrycb7da4b2019-08-28 19:35:56 -0700539 return Sender<T>(MakeRawSender(channel));
540 }
541
542 // This will watch messages sent to the provided channel.
543 //
Brian Silverman454bc112020-03-05 14:21:25 -0800544 // w must have a non-polymorphic operator() (aka it can only be called with a
545 // single set of arguments; no overloading or templates). It must be callable
546 // with this signature:
547 // void(const MessageType &);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700548 //
Brian Silverman454bc112020-03-05 14:21:25 -0800549 // Lambdas are a common form for w. A std::function will work too.
550 //
551 // Note that bind expressions have polymorphic call operators, so they are not
552 // allowed.
553 //
554 // We template Watch as a whole instead of using std::function<void(const T
555 // &)> to allow deducing MessageType from lambdas and other things which are
556 // implicitly convertible to std::function, but not actually std::function
557 // instantiations. Template deduction guides might allow solving this
558 // differently in newer versions of C++, but those have their own corner
559 // cases.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700560 template <typename Watch>
Brian Silverman454bc112020-03-05 14:21:25 -0800561 void MakeWatcher(const std::string_view channel_name, Watch &&w);
562
563 // Like MakeWatcher, but doesn't have access to the message data. This may be
564 // implemented to use less resources than an equivalent MakeWatcher.
565 //
Brian Silverman6b8a3c32020-03-06 11:26:14 -0800566 // The function will still have access to context(), although that will have
567 // its data field set to nullptr.
Brian Silverman454bc112020-03-05 14:21:25 -0800568 template <typename MessageType>
569 void MakeNoArgWatcher(const std::string_view channel_name,
570 std::function<void()> w);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700571
572 // The passed in function will be called when the event loop starts.
573 // Use this to run code once the thread goes into "real-time-mode",
574 virtual void OnRun(::std::function<void()> on_run) = 0;
575
Austin Schuh217a9782019-12-21 23:02:50 -0800576 // Gets the name of the event loop. This is the application name.
James Kuszmaul3ae42262019-11-08 12:33:41 -0800577 virtual const std::string_view name() const = 0;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700578
Austin Schuh217a9782019-12-21 23:02:50 -0800579 // Returns the node that this event loop is running on. Returns nullptr if we
580 // are running in single-node mode.
581 virtual const Node *node() const = 0;
582
Alex Perrycb7da4b2019-08-28 19:35:56 -0700583 // Creates a timer that executes callback when the timer expires
584 // Returns a TimerHandle for configuration of the timer
585 virtual TimerHandler *AddTimer(::std::function<void()> callback) = 0;
586
587 // Creates a timer that executes callback periodically at the specified
588 // interval and offset. Returns a PhasedLoopHandler for interacting with the
589 // timer.
590 virtual PhasedLoopHandler *AddPhasedLoop(
591 ::std::function<void(int)> callback,
592 const monotonic_clock::duration interval,
593 const monotonic_clock::duration offset = ::std::chrono::seconds(0)) = 0;
594
Austin Schuh217a9782019-12-21 23:02:50 -0800595 // TODO(austin): OnExit for cleanup.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700596
597 // Threadsafe.
598 bool is_running() const { return is_running_.load(); }
599
600 // Sets the scheduler priority to run the event loop at. This may not be
601 // called after we go into "real-time-mode".
602 virtual void SetRuntimeRealtimePriority(int priority) = 0;
Austin Schuh39788ff2019-12-01 18:22:57 -0800603 virtual int priority() const = 0;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700604
Brian Silverman6a54ff32020-04-28 16:41:39 -0700605 // Sets the scheduler affinity to run the event loop with. This may only be
606 // called before Run().
607 virtual void SetRuntimeAffinity(const cpu_set_t &cpuset) = 0;
608
Austin Schuh217a9782019-12-21 23:02:50 -0800609 // Fetches new messages from the provided channel (path, type).
610 //
611 // Note: this channel must be a member of the exact configuration object this
612 // was built with.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700613 virtual std::unique_ptr<RawFetcher> MakeRawFetcher(
614 const Channel *channel) = 0;
615
Austin Schuh217a9782019-12-21 23:02:50 -0800616 // Watches channel (name, type) for new messages.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700617 virtual void MakeRawWatcher(
618 const Channel *channel,
619 std::function<void(const Context &context, const void *message)>
620 watcher) = 0;
621
Brian Silverman454bc112020-03-05 14:21:25 -0800622 // Watches channel (name, type) for new messages, without needing to extract
623 // the message contents. Default implementation simply re-uses MakeRawWatcher.
624 virtual void MakeRawNoArgWatcher(
625 const Channel *channel,
626 std::function<void(const Context &context)> watcher) {
627 MakeRawWatcher(channel, [watcher](const Context &context, const void *) {
Brian Silverman6b8a3c32020-03-06 11:26:14 -0800628 Context new_context = context;
629 new_context.data = nullptr;
Brian Silverman4f4e0612020-08-12 19:54:41 -0700630 new_context.buffer_index = -1;
Brian Silverman6b8a3c32020-03-06 11:26:14 -0800631 watcher(new_context);
Brian Silverman454bc112020-03-05 14:21:25 -0800632 });
633 }
634
Austin Schuh217a9782019-12-21 23:02:50 -0800635 // Creates a raw sender for the provided channel. This is used for reflection
636 // based sending.
637 // Note: this ignores any node constraints. Ignore at your own peril.
638 virtual std::unique_ptr<RawSender> MakeRawSender(const Channel *channel) = 0;
639
Austin Schuh6231cc32019-12-07 13:06:15 -0800640 // Returns the context for the current callback.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700641 const Context &context() const { return context_; }
642
643 // Returns the configuration that this event loop was built with.
644 const Configuration *configuration() const { return configuration_; }
645
Austin Schuh39788ff2019-12-01 18:22:57 -0800646 // Prevents the event loop from sending a timing report.
647 void SkipTimingReport() { skip_timing_report_ = true; }
648
Brian Silverman4f4e0612020-08-12 19:54:41 -0700649 // Prevents AOS_LOG being sent to message on /aos.
Tyler Chatow67ddb032020-01-12 14:30:04 -0800650 void SkipAosLog() { skip_logger_ = true; }
651
Brian Silverman4f4e0612020-08-12 19:54:41 -0700652 // Returns the number of buffers for this channel. This corresponds with the
653 // range of Context::buffer_index values for this channel.
654 virtual int NumberBuffers(const Channel *channel) = 0;
655
Alex Perrycb7da4b2019-08-28 19:35:56 -0700656 protected:
Austin Schuh217a9782019-12-21 23:02:50 -0800657 // Sets the name of the event loop. This is the application name.
658 virtual void set_name(const std::string_view name) = 0;
659
Alex Perrycb7da4b2019-08-28 19:35:56 -0700660 void set_is_running(bool value) { is_running_.store(value); }
661
Austin Schuh39788ff2019-12-01 18:22:57 -0800662 // Validates that channel exists inside configuration_ and finds its index.
663 int ChannelIndex(const Channel *channel);
Alex Perrycb7da4b2019-08-28 19:35:56 -0700664
Brian Silverman5120afb2020-01-31 17:44:35 -0800665 // Returns the state for the watcher on the corresponding channel. This
666 // watcher must exist before calling this.
667 WatcherState *GetWatcherState(const Channel *channel);
668
Brian Silverman6d2b3592020-06-18 14:40:15 -0700669 // Returns a Sender's protected RawSender.
Brian Silverman5120afb2020-01-31 17:44:35 -0800670 template <typename T>
671 static RawSender *GetRawSender(aos::Sender<T> *sender) {
672 return sender->sender_.get();
673 }
674
Brian Silverman6d2b3592020-06-18 14:40:15 -0700675 // Returns a Fetcher's protected RawFetcher.
676 template <typename T>
677 static RawFetcher *GetRawFetcher(aos::Fetcher<T> *fetcher) {
678 return fetcher->fetcher_.get();
679 }
680
Austin Schuh6231cc32019-12-07 13:06:15 -0800681 // Context available for watchers, timers, and phased loops.
Alex Perrycb7da4b2019-08-28 19:35:56 -0700682 Context context_;
683
Austin Schuh39788ff2019-12-01 18:22:57 -0800684 friend class RawSender;
685 friend class TimerHandler;
686 friend class RawFetcher;
687 friend class PhasedLoopHandler;
688 friend class WatcherState;
689
690 // Methods used to implement timing reports.
691 void NewSender(RawSender *sender);
692 void DeleteSender(RawSender *sender);
693 TimerHandler *NewTimer(std::unique_ptr<TimerHandler> timer);
694 PhasedLoopHandler *NewPhasedLoop(
695 std::unique_ptr<PhasedLoopHandler> phased_loop);
696 void NewFetcher(RawFetcher *fetcher);
697 void DeleteFetcher(RawFetcher *fetcher);
698 WatcherState *NewWatcher(std::unique_ptr<WatcherState> watcher);
699
Brian Silverman0fc69932020-01-24 21:54:02 -0800700 // Tracks that we have a (single) watcher on the given channel.
701 void TakeWatcher(const Channel *channel);
702 // Tracks that we have at least one sender on the given channel.
703 void TakeSender(const Channel *channel);
704
Austin Schuh39788ff2019-12-01 18:22:57 -0800705 std::vector<RawSender *> senders_;
Austin Schuh7d87b672019-12-01 20:23:49 -0800706 std::vector<RawFetcher *> fetchers_;
707
Austin Schuh39788ff2019-12-01 18:22:57 -0800708 std::vector<std::unique_ptr<TimerHandler>> timers_;
709 std::vector<std::unique_ptr<PhasedLoopHandler>> phased_loops_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800710 std::vector<std::unique_ptr<WatcherState>> watchers_;
711
712 void SendTimingReport();
713 void UpdateTimingReport();
714 void MaybeScheduleTimingReports();
715
716 std::unique_ptr<RawSender> timing_report_sender_;
717
Austin Schuh7d87b672019-12-01 20:23:49 -0800718 // Tracks which event sources (timers and watchers) have data, and which
719 // don't. Added events may not change their event_time().
720 // TODO(austin): Test case 1: timer triggers at t1, handler takes until after
721 // t2 to run, t2 should then be picked up without a context switch.
722 void AddEvent(EventLoopEvent *event);
723 void RemoveEvent(EventLoopEvent *event);
724 size_t EventCount() const { return events_.size(); }
725 EventLoopEvent *PopEvent();
726 EventLoopEvent *PeekEvent() { return events_.front(); }
727 void ReserveEvents();
728
729 std::vector<EventLoopEvent *> events_;
Brian Silvermanbd405c02020-06-23 16:25:23 -0700730 size_t event_generation_ = 1;
Austin Schuh7d87b672019-12-01 20:23:49 -0800731
Tyler Chatow67ddb032020-01-12 14:30:04 -0800732 // If true, don't send AOS_LOG to /aos
733 bool skip_logger_ = false;
734
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800735 private:
Austin Schuh39788ff2019-12-01 18:22:57 -0800736 virtual pid_t GetTid() = 0;
737
738 FlatbufferDetachedBuffer<timing::Report> timing_report_;
739
Austin Schuhde8a8ff2019-11-30 15:25:36 -0800740 ::std::atomic<bool> is_running_{false};
741
Alex Perrycb7da4b2019-08-28 19:35:56 -0700742 const Configuration *configuration_;
Austin Schuh39788ff2019-12-01 18:22:57 -0800743
744 // If true, don't send out timing reports.
745 bool skip_timing_report_ = false;
Brian Silverman0fc69932020-01-24 21:54:02 -0800746
747 absl::btree_set<const Channel *> taken_watchers_, taken_senders_;
Alex Perrycb7da4b2019-08-28 19:35:56 -0700748};
749
750} // namespace aos
751
752#include "aos/events/event_loop_tmpl.h"
753
754#endif // AOS_EVENTS_EVENT_LOOP_H