blob: dd10f38153b797596a135ef9a888831dd58ad275 [file] [log] [blame]
Austin Schuhe309d2a2019-11-29 13:25:21 -08001#ifndef AOS_EVENTS_LOGGER_H_
2#define AOS_EVENTS_LOGGER_H_
3
Austin Schuh8bd96322020-02-13 21:18:22 -08004#include <chrono>
Austin Schuhe309d2a2019-11-29 13:25:21 -08005#include <deque>
Austin Schuh05b70472020-01-01 17:11:17 -08006#include <string_view>
Austin Schuh2f8fd752020-09-01 22:38:28 -07007#include <tuple>
Austin Schuh6f3babe2020-01-26 20:34:50 -08008#include <vector>
Austin Schuhe309d2a2019-11-29 13:25:21 -08009
Austin Schuh8bd96322020-02-13 21:18:22 -080010#include "Eigen/Dense"
11#include "absl/strings/str_cat.h"
Austin Schuhe309d2a2019-11-29 13:25:21 -080012#include "absl/types/span.h"
13#include "aos/events/event_loop.h"
Austin Schuh2f8fd752020-09-01 22:38:28 -070014#include "aos/events/logging/eigen_mpq.h"
Austin Schuhcb5601b2020-09-10 15:29:59 -070015#include "aos/events/logging/log_namer.h"
Austin Schuha36c8902019-12-30 18:07:15 -080016#include "aos/events/logging/logfile_utils.h"
James Kuszmaul38735e82019-12-07 16:42:06 -080017#include "aos/events/logging/logger_generated.h"
Austin Schuh64fab802020-09-09 22:47:47 -070018#include "aos/events/logging/uuid.h"
Austin Schuh92547522019-12-28 14:33:43 -080019#include "aos/events/simulated_event_loop.h"
Austin Schuh2f8fd752020-09-01 22:38:28 -070020#include "aos/network/message_bridge_server_generated.h"
Austin Schuh8bd96322020-02-13 21:18:22 -080021#include "aos/network/timestamp_filter.h"
Austin Schuhe309d2a2019-11-29 13:25:21 -080022#include "aos/time/time.h"
23#include "flatbuffers/flatbuffers.h"
Austin Schuh2f8fd752020-09-01 22:38:28 -070024#include "third_party/gmp/gmpxx.h"
Austin Schuhe309d2a2019-11-29 13:25:21 -080025
26namespace aos {
27namespace logger {
28
Austin Schuhe309d2a2019-11-29 13:25:21 -080029// Logs all channels available in the event loop to disk every 100 ms.
30// Start by logging one message per channel to capture any state and
31// configuration that is sent rately on a channel and would affect execution.
32class Logger {
33 public:
Austin Schuh0c297012020-09-16 18:41:59 -070034 // Constructs a logger.
Austin Schuh0c297012020-09-16 18:41:59 -070035 // event_loop: The event loop used to read the messages.
Austin Schuh0c297012020-09-16 18:41:59 -070036 // configuration: When provided, this is the configuration to log, and the
37 // configuration to use for the channel list to log. If not provided,
38 // this becomes the configuration from the event loop.
Brian Silverman1f345222020-09-24 21:14:48 -070039 // should_log: When provided, a filter for channels to log. If not provided,
40 // all available channels are logged.
41 Logger(EventLoop *event_loop)
42 : Logger(event_loop, event_loop->configuration()) {}
43 Logger(EventLoop *event_loop, const Configuration *configuration)
44 : Logger(event_loop, configuration,
45 [](const Channel *) { return true; }) {}
46 Logger(EventLoop *event_loop, const Configuration *configuration,
47 std::function<bool(const Channel *)> should_log);
Austin Schuh0c297012020-09-16 18:41:59 -070048 ~Logger();
49
50 // Overrides the name in the log file header.
51 void set_name(std::string_view name) { name_ = name; }
Austin Schuhe309d2a2019-11-29 13:25:21 -080052
Brian Silverman1f345222020-09-24 21:14:48 -070053 // Sets the callback to run after each period of data is logged. Defaults to
54 // doing nothing.
55 //
56 // This callback may safely do things like call Rotate().
57 void set_on_logged_period(std::function<void()> on_logged_period) {
58 on_logged_period_ = std::move(on_logged_period);
59 }
60
61 // Sets the period between polling the data. Defaults to 100ms.
62 //
63 // Changing this while a set of files is being written may result in
64 // unreadable files.
65 void set_polling_period(std::chrono::nanoseconds polling_period) {
66 polling_period_ = polling_period;
67 }
68
Brian Silvermanae7c0332020-09-30 16:58:23 -070069 std::string_view log_start_uuid() const { return log_start_uuid_; }
70
Austin Schuh2f8fd752020-09-01 22:38:28 -070071 // Rotates the log file(s), triggering new part files to be written for each
72 // log file.
73 void Rotate();
Austin Schuhfa895892020-01-07 20:07:41 -080074
Brian Silverman1f345222020-09-24 21:14:48 -070075 // Starts logging to files with the given naming scheme.
Brian Silvermanae7c0332020-09-30 16:58:23 -070076 //
77 // log_start_uuid may be used to tie this log event to other log events across
78 // multiple nodes. The default (empty string) indicates there isn't one
79 // available.
80 void StartLogging(std::unique_ptr<LogNamer> log_namer,
81 std::string_view log_start_uuid = "");
Brian Silverman1f345222020-09-24 21:14:48 -070082
83 // Stops logging. Ensures any messages through end_time make it into the log.
84 //
85 // If you want to stop ASAP, pass min_time to avoid reading any more messages.
86 //
87 // Returns the LogNamer in case the caller wants to do anything else with it
88 // before destroying it.
89 std::unique_ptr<LogNamer> StopLogging(
90 aos::monotonic_clock::time_point end_time);
91
92 // Returns whether a log is currently being written.
93 bool is_started() const { return static_cast<bool>(log_namer_); }
94
95 // Shortcut to call StartLogging with a LocalLogNamer when event processing
96 // starts.
97 void StartLoggingLocalNamerOnRun(std::string base_name) {
98 event_loop_->OnRun([this, base_name]() {
99 StartLogging(
100 std::make_unique<LocalLogNamer>(base_name, event_loop_->node()));
101 });
102 }
103
Austin Schuhe309d2a2019-11-29 13:25:21 -0800104 private:
Austin Schuhe309d2a2019-11-29 13:25:21 -0800105 // Structure to track both a fetcher, and if the data fetched has been
106 // written. We may want to delay writing data to disk so that we don't let
107 // data get too far out of order when written to disk so we can avoid making
108 // it too hard to sort when reading.
109 struct FetcherStruct {
110 std::unique_ptr<RawFetcher> fetcher;
111 bool written = false;
Austin Schuh15649d62019-12-28 16:36:38 -0800112
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700113 // Channel index to log to.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800114 int channel_index = -1;
Brian Silverman1f345222020-09-24 21:14:48 -0700115 const Channel *channel = nullptr;
116 const Node *timestamp_node = nullptr;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800117
118 LogType log_type = LogType::kLogMessage;
119
Brian Silverman1f345222020-09-24 21:14:48 -0700120 // We fill out the metadata at construction, but the actual writers have to
121 // be updated each time we start logging. To avoid duplicating the complex
122 // logic determining whether each writer should be initialized, we just
123 // stash the answer in separate member variables.
124 bool wants_writer = false;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800125 DetachedBufferWriter *writer = nullptr;
Brian Silverman1f345222020-09-24 21:14:48 -0700126 bool wants_timestamp_writer = false;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800127 DetachedBufferWriter *timestamp_writer = nullptr;
Brian Silverman1f345222020-09-24 21:14:48 -0700128 bool wants_contents_writer = false;
Austin Schuh2f8fd752020-09-01 22:38:28 -0700129 DetachedBufferWriter *contents_writer = nullptr;
Brian Silverman1f345222020-09-24 21:14:48 -0700130
Austin Schuh2f8fd752020-09-01 22:38:28 -0700131 int node_index = 0;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800132 };
133
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700134 // Vector mapping from the channel index from the event loop to the logged
135 // channel index.
136 std::vector<int> event_loop_to_logged_channel_index_;
137
Austin Schuh2f8fd752020-09-01 22:38:28 -0700138 struct NodeState {
139 aos::monotonic_clock::time_point monotonic_start_time =
140 aos::monotonic_clock::min_time;
141 aos::realtime_clock::time_point realtime_start_time =
142 aos::realtime_clock::min_time;
143
144 aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> log_file_header =
145 aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader>::Empty();
146 };
Brian Silverman1f345222020-09-24 21:14:48 -0700147
148 void WriteHeader();
149 aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> MakeHeader(
150 const Node *node);
151
152 bool MaybeUpdateTimestamp(
153 const Node *node, int node_index,
154 aos::monotonic_clock::time_point monotonic_start_time,
155 aos::realtime_clock::time_point realtime_start_time);
156
157 void DoLogData(const monotonic_clock::time_point end_time);
158
159 void WriteMissingTimestamps();
160
161 // Fetches from each channel until all the data is logged.
162 void LogUntil(monotonic_clock::time_point t);
163
164 // Sets the start time for a specific node.
165 void SetStartTime(size_t node_index,
166 aos::monotonic_clock::time_point monotonic_start_time,
167 aos::realtime_clock::time_point realtime_start_time);
168
Brian Silvermanae7c0332020-09-30 16:58:23 -0700169 EventLoop *const event_loop_;
Brian Silverman1f345222020-09-24 21:14:48 -0700170 // The configuration to place at the top of the log file.
171 const Configuration *const configuration_;
172
Brian Silvermanae7c0332020-09-30 16:58:23 -0700173 UUID log_event_uuid_ = UUID::Zero();
174 const UUID logger_instance_uuid_ = UUID::Random();
175 std::unique_ptr<LogNamer> log_namer_;
176 // Empty indicates there isn't one.
177 std::string log_start_uuid_;
178 const std::string boot_uuid_;
179
Brian Silverman1f345222020-09-24 21:14:48 -0700180 // Name to save in the log file. Defaults to hostname.
181 std::string name_;
182
183 std::function<void()> on_logged_period_ = []() {};
184
185 std::vector<FetcherStruct> fetchers_;
186 TimerHandler *timer_handler_;
187
188 // Period to poll the channels.
189 std::chrono::nanoseconds polling_period_ = std::chrono::milliseconds(100);
190
191 // Last time that data was written for all channels to disk.
192 monotonic_clock::time_point last_synchronized_time_;
193
194 // Max size that the header has consumed. This much extra data will be
195 // reserved in the builder to avoid reallocating.
196 size_t max_header_size_ = 0;
197
198 // Fetcher for all the statistics from all the nodes.
199 aos::Fetcher<message_bridge::ServerStatistics> server_statistics_fetcher_;
200
Austin Schuh2f8fd752020-09-01 22:38:28 -0700201 std::vector<NodeState> node_state_;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800202};
203
Austin Schuh11d43732020-09-21 17:28:30 -0700204// Datastructure to hold ordered parts.
205struct LogParts {
206 // Monotonic and realtime start times for this set of log files. For log
207 // files which started out unknown and then became known, this is the known
208 // start time.
209 aos::monotonic_clock::time_point monotonic_start_time;
210 aos::realtime_clock::time_point realtime_start_time;
211
212 // UUIDs if available.
Brian Silvermanae7c0332020-09-30 16:58:23 -0700213 std::string log_event_uuid;
Austin Schuh11d43732020-09-21 17:28:30 -0700214 std::string parts_uuid;
215
216 // The node this represents, or empty if we are in a single node world.
217 std::string node;
218
219 // Pre-sorted list of parts.
220 std::vector<std::string> parts;
221};
222
223// Datastructure to hold parts from the same run of the logger which have no
224// ordering constraints relative to each other.
225struct LogFile {
226 // The UUID tying them all together (if available)
Brian Silvermanae7c0332020-09-30 16:58:23 -0700227 std::string log_event_uuid;
Austin Schuh11d43732020-09-21 17:28:30 -0700228
229 // All the parts, unsorted.
230 std::vector<LogParts> parts;
231};
232
233std::ostream &operator<<(std::ostream &stream, const LogFile &file);
234std::ostream &operator<<(std::ostream &stream, const LogParts &parts);
235
Austin Schuh5212cad2020-09-09 23:12:09 -0700236// Takes a bunch of parts and sorts them based on part_uuid and part_index.
Austin Schuh11d43732020-09-21 17:28:30 -0700237std::vector<LogFile> SortParts(const std::vector<std::string> &parts);
238
239std::vector<std::vector<std::string>> ToLogReaderVector(
240 const std::vector<LogFile> &log_files);
Austin Schuh5212cad2020-09-09 23:12:09 -0700241
Austin Schuh6f3babe2020-01-26 20:34:50 -0800242// We end up with one of the following 3 log file types.
243//
244// Single node logged as the source node.
245// -> Replayed just on the source node.
246//
247// Forwarding timestamps only logged from the perspective of the destination
248// node.
249// -> Matched with data on source node and logged.
250//
251// Forwarding timestamps with data logged as the destination node.
252// -> Replayed just as the destination
253// -> Replayed as the source (Much harder, ordering is not defined)
254//
255// Duplicate data logged. -> CHECK that it matches and explode otherwise.
256//
257// This can be boiled down to a set of constraints and tools.
258//
259// 1) Forwarding timestamps and data need to be logged separately.
260// 2) Any forwarded data logged on the destination node needs to be logged
261// separately such that it can be sorted.
262//
263// 1) Log reader needs to be able to sort a list of log files.
264// 2) Log reader needs to be able to merge sorted lists of log files.
265// 3) Log reader needs to be able to match timestamps with messages.
266//
267// We also need to be able to generate multiple views of a log file depending on
268// the target.
269
Austin Schuhe309d2a2019-11-29 13:25:21 -0800270// Replays all the channels in the logfile to the event loop.
271class LogReader {
272 public:
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800273 // If you want to supply a new configuration that will be used for replay
274 // (e.g., to change message rates, or to populate an updated schema), then
275 // pass it in here. It must provide all the channels that the original logged
276 // config did.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800277 //
278 // Log filenames are in the following format:
279 //
280 // {
281 // {log1_part0, log1_part1, ...},
282 // {log2}
283 // }
284 // The inner vector is a list of log file chunks which form up a log file.
285 // The outer vector is a list of log files with subsets of the messages, or
286 // messages from different nodes.
287 //
288 // If the outer vector isn't provided, it is assumed to be of size 1.
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800289 LogReader(std::string_view filename,
290 const Configuration *replay_configuration = nullptr);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800291 LogReader(const std::vector<std::string> &filenames,
292 const Configuration *replay_configuration = nullptr);
293 LogReader(const std::vector<std::vector<std::string>> &filenames,
Austin Schuhfa895892020-01-07 20:07:41 -0800294 const Configuration *replay_configuration = nullptr);
Austin Schuh11d43732020-09-21 17:28:30 -0700295 LogReader(const std::vector<LogFile> &log_files,
296 const Configuration *replay_configuration = nullptr);
James Kuszmaul7daef362019-12-31 18:28:17 -0800297 ~LogReader();
Austin Schuhe309d2a2019-11-29 13:25:21 -0800298
Austin Schuh6331ef92020-01-07 18:28:09 -0800299 // Registers all the callbacks to send the log file data out on an event loop
300 // created in event_loop_factory. This also updates time to be at the start
301 // of the log file by running until the log file starts.
302 // Note: the configuration used in the factory should be configuration()
303 // below, but can be anything as long as the locations needed to send
304 // everything are available.
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800305 void Register(SimulatedEventLoopFactory *event_loop_factory);
Austin Schuh6331ef92020-01-07 18:28:09 -0800306 // Creates an SimulatedEventLoopFactory accessible via event_loop_factory(),
307 // and then calls Register.
308 void Register();
309 // Registers callbacks for all the events after the log file starts. This is
310 // only useful when replaying live.
Austin Schuhe309d2a2019-11-29 13:25:21 -0800311 void Register(EventLoop *event_loop);
Austin Schuh6331ef92020-01-07 18:28:09 -0800312
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800313 // Unregisters the senders. You only need to call this if you separately
314 // supplied an event loop or event loop factory and the lifetimes are such
315 // that they need to be explicitly destroyed before the LogReader destructor
316 // gets called.
Austin Schuhe309d2a2019-11-29 13:25:21 -0800317 void Deregister();
318
Austin Schuh0c297012020-09-16 18:41:59 -0700319 // Returns the configuration being used for replay from the log file.
320 // Note that this may be different from the configuration actually used for
321 // handling events. You should generally only use this to create a
322 // SimulatedEventLoopFactory, and then get the configuration from there for
323 // everything else.
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800324 const Configuration *logged_configuration() const;
Austin Schuh11d43732020-09-21 17:28:30 -0700325 // Returns the configuration being used for replay from the log file.
326 // Note that this may be different from the configuration actually used for
327 // handling events. You should generally only use this to create a
328 // SimulatedEventLoopFactory, and then get the configuration from there for
329 // everything else.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800330 // The pointer is invalidated whenever RemapLoggedChannel is called.
Austin Schuh15649d62019-12-28 16:36:38 -0800331 const Configuration *configuration() const;
332
Austin Schuh6f3babe2020-01-26 20:34:50 -0800333 // Returns the nodes that this log file was created on. This is a list of
334 // pointers to a node in the nodes() list inside configuration(). The
335 // pointers here are invalidated whenever RemapLoggedChannel is called.
336 std::vector<const Node *> Nodes() const;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800337
338 // Returns the starting timestamp for the log file.
Austin Schuh11d43732020-09-21 17:28:30 -0700339 monotonic_clock::time_point monotonic_start_time(
340 const Node *node = nullptr) const;
341 realtime_clock::time_point realtime_start_time(
342 const Node *node = nullptr) const;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800343
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800344 // Causes the logger to publish the provided channel on a different name so
345 // that replayed applications can publish on the proper channel name without
346 // interference. This operates on raw channel names, without any node or
347 // application specific mappings.
348 void RemapLoggedChannel(std::string_view name, std::string_view type,
349 std::string_view add_prefix = "/original");
350 template <typename T>
351 void RemapLoggedChannel(std::string_view name,
352 std::string_view add_prefix = "/original") {
353 RemapLoggedChannel(name, T::GetFullyQualifiedName(), add_prefix);
354 }
355
Austin Schuh01b4c352020-09-21 23:09:39 -0700356 // Remaps the provided channel, though this respects node mappings, and
357 // preserves them too. This makes it so if /aos -> /pi1/aos on one node,
358 // /original/aos -> /original/pi1/aos on the same node after renaming, just
359 // like you would hope.
360 //
361 // TODO(austin): If you have 2 nodes remapping something to the same channel,
362 // this doesn't handle that. No use cases exist yet for that, so it isn't
363 // being done yet.
364 void RemapLoggedChannel(std::string_view name, std::string_view type,
365 const Node *node,
366 std::string_view add_prefix = "/original");
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700367 template <typename T>
Austin Schuh01b4c352020-09-21 23:09:39 -0700368 void RemapLoggedChannel(std::string_view name, const Node *node,
369 std::string_view add_prefix = "/original") {
370 RemapLoggedChannel(name, T::GetFullyQualifiedName(), node, add_prefix);
371 }
372
373 template <typename T>
374 bool HasChannel(std::string_view name, const Node *node = nullptr) {
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700375 return configuration::GetChannel(log_file_header()->configuration(), name,
376 T::GetFullyQualifiedName(), "",
Austin Schuh01b4c352020-09-21 23:09:39 -0700377 node) != nullptr;
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700378 }
379
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800380 SimulatedEventLoopFactory *event_loop_factory() {
381 return event_loop_factory_;
382 }
383
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700384 const LogFileHeader *log_file_header() const {
385 return &log_file_header_.message();
386 }
387
Austin Schuh0c297012020-09-16 18:41:59 -0700388 std::string_view name() const {
389 return log_file_header()->name()->string_view();
390 }
391
Austin Schuhe309d2a2019-11-29 13:25:21 -0800392 private:
Austin Schuh6f3babe2020-01-26 20:34:50 -0800393 const Channel *RemapChannel(const EventLoop *event_loop,
394 const Channel *channel);
395
Austin Schuhe309d2a2019-11-29 13:25:21 -0800396 // Queues at least max_out_of_order_duration_ messages into channels_.
397 void QueueMessages();
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800398 // Handle constructing a configuration with all the additional remapped
399 // channels from calls to RemapLoggedChannel.
400 void MakeRemappedConfig();
Austin Schuhe309d2a2019-11-29 13:25:21 -0800401
Austin Schuh2f8fd752020-09-01 22:38:28 -0700402 // Returns the number of nodes.
403 size_t nodes_count() const {
404 return !configuration::MultiNode(logged_configuration())
405 ? 1u
406 : logged_configuration()->nodes()->size();
407 }
408
Austin Schuh6f3babe2020-01-26 20:34:50 -0800409 const std::vector<std::vector<std::string>> filenames_;
410
411 // This is *a* log file header used to provide the logged config. The rest of
412 // the header is likely distracting.
413 FlatbufferVector<LogFileHeader> log_file_header_;
414
Austin Schuh2f8fd752020-09-01 22:38:28 -0700415 // Returns [ta; tb; ...] = tuple[0] * t + tuple[1]
416 std::tuple<Eigen::Matrix<double, Eigen::Dynamic, 1>,
417 Eigen::Matrix<double, Eigen::Dynamic, 1>>
418 SolveOffsets();
419
420 void LogFit(std::string_view prefix);
Austin Schuh8bd96322020-02-13 21:18:22 -0800421
Austin Schuh6f3babe2020-01-26 20:34:50 -0800422 // State per node.
Austin Schuh858c9f32020-08-31 16:56:12 -0700423 class State {
424 public:
425 State(std::unique_ptr<ChannelMerger> channel_merger);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800426
Austin Schuh858c9f32020-08-31 16:56:12 -0700427 // Returns the timestamps, channel_index, and message from a channel.
428 // update_time (will be) set to true when popping this message causes the
429 // filter to change the time offset estimation function.
430 std::tuple<TimestampMerger::DeliveryTimestamp, int,
431 FlatbufferVector<MessageHeader>>
432 PopOldest(bool *update_time);
433
434 // Returns the monotonic time of the oldest message.
435 monotonic_clock::time_point OldestMessageTime() const;
436
437 // Primes the queues inside State. Should be called before calling
438 // OldestMessageTime.
439 void SeedSortedMessages();
Austin Schuh8bd96322020-02-13 21:18:22 -0800440
Austin Schuh858c9f32020-08-31 16:56:12 -0700441 // Returns the starting time for this node.
442 monotonic_clock::time_point monotonic_start_time() const {
443 return channel_merger_->monotonic_start_time();
444 }
445 realtime_clock::time_point realtime_start_time() const {
446 return channel_merger_->realtime_start_time();
447 }
448
449 // Sets the node event loop factory for replaying into a
450 // SimulatedEventLoopFactory. Returns the EventLoop to use.
451 EventLoop *SetNodeEventLoopFactory(
452 NodeEventLoopFactory *node_event_loop_factory);
453
454 // Sets and gets the event loop to use.
455 void set_event_loop(EventLoop *event_loop) { event_loop_ = event_loop; }
456 EventLoop *event_loop() { return event_loop_; }
457
Austin Schuh858c9f32020-08-31 16:56:12 -0700458 // Sets the current realtime offset from the monotonic clock for this node
459 // (if we are on a simulated event loop).
460 void SetRealtimeOffset(monotonic_clock::time_point monotonic_time,
461 realtime_clock::time_point realtime_time) {
462 if (node_event_loop_factory_ != nullptr) {
463 node_event_loop_factory_->SetRealtimeOffset(monotonic_time,
464 realtime_time);
465 }
466 }
467
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700468 // Returns the MessageHeader sender to log delivery timestamps to for the
469 // provided remote node.
470 aos::Sender<MessageHeader> *RemoteTimestampSender(
471 const Node *delivered_node);
472
Austin Schuh858c9f32020-08-31 16:56:12 -0700473 // Converts a timestamp from the monotonic clock on this node to the
474 // distributed clock.
475 distributed_clock::time_point ToDistributedClock(
476 monotonic_clock::time_point time) {
477 return node_event_loop_factory_->ToDistributedClock(time);
478 }
479
Austin Schuh2f8fd752020-09-01 22:38:28 -0700480 monotonic_clock::time_point FromDistributedClock(
481 distributed_clock::time_point time) {
482 return node_event_loop_factory_->FromDistributedClock(time);
483 }
484
Austin Schuh858c9f32020-08-31 16:56:12 -0700485 // Sets the offset (and slope) from the distributed clock.
486 void SetDistributedOffset(std::chrono::nanoseconds distributed_offset,
487 double distributed_slope) {
488 node_event_loop_factory_->SetDistributedOffset(distributed_offset,
489 distributed_slope);
490 }
491
492 // Returns the current time on the remote node which sends messages on
493 // channel_index.
494 monotonic_clock::time_point monotonic_remote_now(size_t channel_index) {
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700495 return channel_source_state_[channel_index]
496 ->node_event_loop_factory_->monotonic_now();
Austin Schuh858c9f32020-08-31 16:56:12 -0700497 }
498
Austin Schuh2f8fd752020-09-01 22:38:28 -0700499 distributed_clock::time_point RemoteToDistributedClock(
500 size_t channel_index, monotonic_clock::time_point time) {
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700501 return channel_source_state_[channel_index]
502 ->node_event_loop_factory_->ToDistributedClock(time);
Austin Schuh2f8fd752020-09-01 22:38:28 -0700503 }
504
505 const Node *remote_node(size_t channel_index) {
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700506 return channel_source_state_[channel_index]
507 ->node_event_loop_factory_->node();
Austin Schuh2f8fd752020-09-01 22:38:28 -0700508 }
509
510 monotonic_clock::time_point monotonic_now() {
511 return node_event_loop_factory_->monotonic_now();
512 }
513
Austin Schuh858c9f32020-08-31 16:56:12 -0700514 // Sets the node we will be merging as, and returns true if there is any
515 // data on it.
516 bool SetNode() { return channel_merger_->SetNode(event_loop_->node()); }
517
518 // Sets the number of channels.
519 void SetChannelCount(size_t count);
520
521 // Sets the sender, filter, and target factory for a channel.
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700522 void SetChannel(size_t logged_channel_index, size_t factory_channel_index,
523 std::unique_ptr<RawSender> sender,
Austin Schuh2f8fd752020-09-01 22:38:28 -0700524 message_bridge::NoncausalOffsetEstimator *filter,
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700525 aos::Sender<MessageHeader> *remote_timestamp_sender,
526 State *source_state);
Austin Schuh858c9f32020-08-31 16:56:12 -0700527
528 // Returns if we have read all the messages from all the logs.
529 bool at_end() const { return channel_merger_->at_end(); }
530
531 // Unregisters everything so we can destory the event loop.
532 void Deregister();
533
534 // Sets the current TimerHandle for the replay callback.
535 void set_timer_handler(TimerHandler *timer_handler) {
536 timer_handler_ = timer_handler;
537 }
538
539 // Sets the next wakeup time on the replay callback.
540 void Setup(monotonic_clock::time_point next_time) {
541 timer_handler_->Setup(next_time);
542 }
543
544 // Sends a buffer on the provided channel index.
545 bool Send(size_t channel_index, const void *data, size_t size,
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700546 const TimestampMerger::DeliveryTimestamp &delivery_timestamp);
Austin Schuh858c9f32020-08-31 16:56:12 -0700547
548 // Returns a debug string for the channel merger.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700549 std::string DebugString() const {
550 std::stringstream messages;
551 size_t i = 0;
552 for (const auto &message : sorted_messages_) {
553 if (i < 7 || i + 7 > sorted_messages_.size()) {
554 messages << "sorted_messages[" << i
555 << "]: " << std::get<0>(message).monotonic_event_time << " "
556 << configuration::StrippedChannelToString(
557 event_loop_->configuration()->channels()->Get(
558 std::get<2>(message).message().channel_index()))
559 << "\n";
560 } else if (i == 7) {
561 messages << "...\n";
562 }
563 ++i;
564 }
565 return messages.str() + channel_merger_->DebugString();
566 }
Austin Schuh858c9f32020-08-31 16:56:12 -0700567
568 private:
569 // Log file.
570 std::unique_ptr<ChannelMerger> channel_merger_;
571
572 std::deque<std::tuple<TimestampMerger::DeliveryTimestamp, int,
Austin Schuh2f8fd752020-09-01 22:38:28 -0700573 FlatbufferVector<MessageHeader>,
574 message_bridge::NoncausalOffsetEstimator *>>
Austin Schuh858c9f32020-08-31 16:56:12 -0700575 sorted_messages_;
576
577 // Senders.
578 std::vector<std::unique_ptr<RawSender>> channels_;
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700579 std::vector<aos::Sender<MessageHeader> *> remote_timestamp_senders_;
580 // The mapping from logged channel index to sent channel index. Needed for
581 // sending out MessageHeaders.
582 std::vector<int> factory_channel_index_;
583
584 struct SentTimestamp {
585 monotonic_clock::time_point monotonic_event_time =
586 monotonic_clock::min_time;
587 realtime_clock::time_point realtime_event_time = realtime_clock::min_time;
588 uint32_t queue_index = 0xffffffff;
589
590 // The queue index that this message *actually* was sent with.
591 uint32_t actual_queue_index = 0xffffffff;
592 };
593
594 // Stores all the timestamps that have been sent on this channel. This is
595 // only done for channels which are forwarded and on the node which
596 // initially sends the message.
597 //
598 // TODO(austin): This whole concept is a hack. We should be able to
599 // associate state with the message as it gets sorted and recover it.
600 std::vector<std::unique_ptr<std::vector<SentTimestamp>>> queue_index_map_;
Austin Schuh858c9f32020-08-31 16:56:12 -0700601
602 // Factory (if we are in sim) that this loop was created on.
603 NodeEventLoopFactory *node_event_loop_factory_ = nullptr;
604 std::unique_ptr<EventLoop> event_loop_unique_ptr_;
605 // Event loop.
606 EventLoop *event_loop_ = nullptr;
607 // And timer used to send messages.
608 TimerHandler *timer_handler_;
609
Austin Schuh8bd96322020-02-13 21:18:22 -0800610 // Filters (or nullptr if it isn't a forwarded channel) for each channel.
611 // This corresponds to the object which is shared among all the channels
612 // going between 2 nodes. The second element in the tuple indicates if this
613 // is the primary direction or not.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700614 std::vector<message_bridge::NoncausalOffsetEstimator *> filters_;
Austin Schuh8bd96322020-02-13 21:18:22 -0800615
616 // List of NodeEventLoopFactorys (or nullptr if it isn't a forwarded
617 // channel) which correspond to the originating node.
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700618 std::vector<State *> channel_source_state_;
619
620 std::map<const Node *, aos::Sender<MessageHeader>>
621 remote_timestamp_senders_map_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800622 };
623
Austin Schuh8bd96322020-02-13 21:18:22 -0800624 // Node index -> State.
625 std::vector<std::unique_ptr<State>> states_;
626
627 // Creates the requested filter if it doesn't exist, regardless of whether
628 // these nodes can actually communicate directly. The second return value
629 // reports if this is the primary direction or not.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700630 message_bridge::NoncausalOffsetEstimator *GetFilter(const Node *node_a,
631 const Node *node_b);
Austin Schuh8bd96322020-02-13 21:18:22 -0800632
633 // FILE to write offsets to (if populated).
634 FILE *offset_fp_ = nullptr;
635 // Timestamp of the first piece of data used for the horizontal axis on the
636 // plot.
637 aos::realtime_clock::time_point first_time_;
638
639 // List of filters for a connection. The pointer to the first node will be
640 // less than the second node.
641 std::map<std::tuple<const Node *, const Node *>,
Austin Schuh2f8fd752020-09-01 22:38:28 -0700642 std::tuple<message_bridge::NoncausalOffsetEstimator>>
Austin Schuh8bd96322020-02-13 21:18:22 -0800643 filters_;
644
645 // Returns the offset from the monotonic clock for a node to the distributed
Austin Schuh2f8fd752020-09-01 22:38:28 -0700646 // clock. monotonic = distributed * slope() + offset();
647 double slope(int node_index) const {
648 CHECK_LT(node_index, time_slope_matrix_.rows())
James Kuszmaul46d82582020-05-09 19:50:09 -0700649 << ": Got too high of a node index.";
Austin Schuh2f8fd752020-09-01 22:38:28 -0700650 return time_slope_matrix_(node_index);
651 }
652 std::chrono::nanoseconds offset(int node_index) const {
653 CHECK_LT(node_index, time_offset_matrix_.rows())
654 << ": Got too high of a node index.";
655 return std::chrono::duration_cast<std::chrono::nanoseconds>(
656 std::chrono::duration<double>(time_offset_matrix_(node_index)));
Austin Schuh8bd96322020-02-13 21:18:22 -0800657 }
658
659 // Updates the offset matrix solution and sets the per-node distributed
660 // offsets in the factory.
661 void UpdateOffsets();
662
Austin Schuh2f8fd752020-09-01 22:38:28 -0700663 // We have 2 types of equations to do a least squares regression over to fully
664 // constrain our time function.
665 //
666 // One is simple. The distributed clock is the average of all the clocks.
Brian Silverman87ac0402020-09-17 14:47:01 -0700667 // (ta + tb + tc + td) / num_nodes = t_distributed
Austin Schuh2f8fd752020-09-01 22:38:28 -0700668 //
669 // The second is a bit more complicated. Our basic time conversion function
670 // is:
671 // tb = ta + (ta * slope + offset)
672 // We can rewrite this as follows
673 // tb - (1 + slope) * ta = offset
674 //
675 // From here, we have enough equations to solve for t{a,b,c,...} We want to
676 // take as an input the offsets and slope, and solve for the per-node times as
677 // a function of the distributed clock.
678 //
679 // We need to massage our equations to make this work. If we solve for the
680 // per-node times at two set distributed clock times, we will be able to
681 // recreate the linear function (we know it is linear). We can do a similar
682 // thing by breaking our equation up into:
Brian Silverman87ac0402020-09-17 14:47:01 -0700683 //
Austin Schuh2f8fd752020-09-01 22:38:28 -0700684 // [1/3 1/3 1/3 ] [ta] [t_distributed]
685 // [ 1 -1-m1 0 ] [tb] = [oab]
686 // [ 1 0 -1-m2 ] [tc] [oac]
687 //
688 // This solves to:
689 //
690 // [ta] [ a00 a01 a02] [t_distributed]
691 // [tb] = [ a10 a11 a12] * [oab]
692 // [tc] [ a20 a21 a22] [oac]
693 //
694 // and can be split into:
695 //
696 // [ta] [ a00 ] [a01 a02]
697 // [tb] = [ a10 ] * t_distributed + [a11 a12] * [oab]
698 // [tc] [ a20 ] [a21 a22] [oac]
699 //
700 // (map_matrix_ + slope_matrix_) * [ta; tb; tc] = [offset_matrix_];
701 // offset_matrix_ will be in nanoseconds.
702 Eigen::Matrix<mpq_class, Eigen::Dynamic, Eigen::Dynamic> map_matrix_;
703 Eigen::Matrix<mpq_class, Eigen::Dynamic, Eigen::Dynamic> slope_matrix_;
704 Eigen::Matrix<mpq_class, Eigen::Dynamic, 1> offset_matrix_;
705 // Matrix tracking which offsets are valid.
706 Eigen::Matrix<bool, Eigen::Dynamic, 1> valid_matrix_;
707 // Matrix tracking the last valid matrix we used to determine connected nodes.
708 Eigen::Matrix<bool, Eigen::Dynamic, 1> last_valid_matrix_;
709 size_t cached_valid_node_count_ = 0;
Austin Schuh8bd96322020-02-13 21:18:22 -0800710
Austin Schuh2f8fd752020-09-01 22:38:28 -0700711 // [ta; tb; tc] = time_slope_matrix_ * t + time_offset_matrix;
712 // t is in seconds.
713 Eigen::Matrix<double, Eigen::Dynamic, 1> time_slope_matrix_;
714 Eigen::Matrix<double, Eigen::Dynamic, 1> time_offset_matrix_;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800715
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800716 std::unique_ptr<FlatbufferDetachedBuffer<Configuration>>
717 remapped_configuration_buffer_;
718
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800719 std::unique_ptr<SimulatedEventLoopFactory> event_loop_factory_unique_ptr_;
720 SimulatedEventLoopFactory *event_loop_factory_ = nullptr;
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800721
722 // Map of channel indices to new name. The channel index will be an index into
723 // logged_configuration(), and the string key will be the name of the channel
724 // to send on instead of the logged channel name.
725 std::map<size_t, std::string> remapped_channels_;
Austin Schuh01b4c352020-09-21 23:09:39 -0700726 std::vector<MapT> maps_;
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800727
Austin Schuh6f3babe2020-01-26 20:34:50 -0800728 // Number of nodes which still have data to send. This is used to figure out
729 // when to exit.
730 size_t live_nodes_ = 0;
731
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800732 const Configuration *remapped_configuration_ = nullptr;
733 const Configuration *replay_configuration_ = nullptr;
Austin Schuhcde938c2020-02-02 17:30:07 -0800734
735 // If true, the replay timer will ignore any missing data. This is used
736 // during startup when we are bootstrapping everything and trying to get to
737 // the start of all the log files.
738 bool ignore_missing_data_ = false;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800739};
740
741} // namespace logger
742} // namespace aos
743
744#endif // AOS_EVENTS_LOGGER_H_