blob: 161300c06c773247239d9ed4582904ecc1af3ba0 [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 Schuh6f3babe2020-01-26 20:34:50 -0800113 int channel_index = -1;
Brian Silverman1f345222020-09-24 21:14:48 -0700114 const Channel *channel = nullptr;
115 const Node *timestamp_node = nullptr;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800116
117 LogType log_type = LogType::kLogMessage;
118
Brian Silverman1f345222020-09-24 21:14:48 -0700119 // We fill out the metadata at construction, but the actual writers have to
120 // be updated each time we start logging. To avoid duplicating the complex
121 // logic determining whether each writer should be initialized, we just
122 // stash the answer in separate member variables.
123 bool wants_writer = false;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800124 DetachedBufferWriter *writer = nullptr;
Brian Silverman1f345222020-09-24 21:14:48 -0700125 bool wants_timestamp_writer = false;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800126 DetachedBufferWriter *timestamp_writer = nullptr;
Brian Silverman1f345222020-09-24 21:14:48 -0700127 bool wants_contents_writer = false;
Austin Schuh2f8fd752020-09-01 22:38:28 -0700128 DetachedBufferWriter *contents_writer = nullptr;
Brian Silverman1f345222020-09-24 21:14:48 -0700129
Austin Schuh2f8fd752020-09-01 22:38:28 -0700130 int node_index = 0;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800131 };
132
Austin Schuh2f8fd752020-09-01 22:38:28 -0700133 struct NodeState {
134 aos::monotonic_clock::time_point monotonic_start_time =
135 aos::monotonic_clock::min_time;
136 aos::realtime_clock::time_point realtime_start_time =
137 aos::realtime_clock::min_time;
138
139 aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> log_file_header =
140 aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader>::Empty();
141 };
Brian Silverman1f345222020-09-24 21:14:48 -0700142
143 void WriteHeader();
144 aos::SizePrefixedFlatbufferDetachedBuffer<LogFileHeader> MakeHeader(
145 const Node *node);
146
147 bool MaybeUpdateTimestamp(
148 const Node *node, int node_index,
149 aos::monotonic_clock::time_point monotonic_start_time,
150 aos::realtime_clock::time_point realtime_start_time);
151
152 void DoLogData(const monotonic_clock::time_point end_time);
153
154 void WriteMissingTimestamps();
155
156 // Fetches from each channel until all the data is logged.
157 void LogUntil(monotonic_clock::time_point t);
158
159 // Sets the start time for a specific node.
160 void SetStartTime(size_t node_index,
161 aos::monotonic_clock::time_point monotonic_start_time,
162 aos::realtime_clock::time_point realtime_start_time);
163
Brian Silvermanae7c0332020-09-30 16:58:23 -0700164 EventLoop *const event_loop_;
Brian Silverman1f345222020-09-24 21:14:48 -0700165 // The configuration to place at the top of the log file.
166 const Configuration *const configuration_;
167
Brian Silvermanae7c0332020-09-30 16:58:23 -0700168 UUID log_event_uuid_ = UUID::Zero();
169 const UUID logger_instance_uuid_ = UUID::Random();
170 std::unique_ptr<LogNamer> log_namer_;
171 // Empty indicates there isn't one.
172 std::string log_start_uuid_;
173 const std::string boot_uuid_;
174
Brian Silverman1f345222020-09-24 21:14:48 -0700175 // Name to save in the log file. Defaults to hostname.
176 std::string name_;
177
178 std::function<void()> on_logged_period_ = []() {};
179
180 std::vector<FetcherStruct> fetchers_;
181 TimerHandler *timer_handler_;
182
183 // Period to poll the channels.
184 std::chrono::nanoseconds polling_period_ = std::chrono::milliseconds(100);
185
186 // Last time that data was written for all channels to disk.
187 monotonic_clock::time_point last_synchronized_time_;
188
189 // Max size that the header has consumed. This much extra data will be
190 // reserved in the builder to avoid reallocating.
191 size_t max_header_size_ = 0;
192
193 // Fetcher for all the statistics from all the nodes.
194 aos::Fetcher<message_bridge::ServerStatistics> server_statistics_fetcher_;
195
Austin Schuh2f8fd752020-09-01 22:38:28 -0700196 std::vector<NodeState> node_state_;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800197};
198
Austin Schuh11d43732020-09-21 17:28:30 -0700199// Datastructure to hold ordered parts.
200struct LogParts {
201 // Monotonic and realtime start times for this set of log files. For log
202 // files which started out unknown and then became known, this is the known
203 // start time.
204 aos::monotonic_clock::time_point monotonic_start_time;
205 aos::realtime_clock::time_point realtime_start_time;
206
207 // UUIDs if available.
Brian Silvermanae7c0332020-09-30 16:58:23 -0700208 std::string log_event_uuid;
Austin Schuh11d43732020-09-21 17:28:30 -0700209 std::string parts_uuid;
210
211 // The node this represents, or empty if we are in a single node world.
212 std::string node;
213
214 // Pre-sorted list of parts.
215 std::vector<std::string> parts;
216};
217
218// Datastructure to hold parts from the same run of the logger which have no
219// ordering constraints relative to each other.
220struct LogFile {
221 // The UUID tying them all together (if available)
Brian Silvermanae7c0332020-09-30 16:58:23 -0700222 std::string log_event_uuid;
Austin Schuh11d43732020-09-21 17:28:30 -0700223
224 // All the parts, unsorted.
225 std::vector<LogParts> parts;
226};
227
228std::ostream &operator<<(std::ostream &stream, const LogFile &file);
229std::ostream &operator<<(std::ostream &stream, const LogParts &parts);
230
Austin Schuh5212cad2020-09-09 23:12:09 -0700231// Takes a bunch of parts and sorts them based on part_uuid and part_index.
Austin Schuh11d43732020-09-21 17:28:30 -0700232std::vector<LogFile> SortParts(const std::vector<std::string> &parts);
233
234std::vector<std::vector<std::string>> ToLogReaderVector(
235 const std::vector<LogFile> &log_files);
Austin Schuh5212cad2020-09-09 23:12:09 -0700236
Austin Schuh6f3babe2020-01-26 20:34:50 -0800237// We end up with one of the following 3 log file types.
238//
239// Single node logged as the source node.
240// -> Replayed just on the source node.
241//
242// Forwarding timestamps only logged from the perspective of the destination
243// node.
244// -> Matched with data on source node and logged.
245//
246// Forwarding timestamps with data logged as the destination node.
247// -> Replayed just as the destination
248// -> Replayed as the source (Much harder, ordering is not defined)
249//
250// Duplicate data logged. -> CHECK that it matches and explode otherwise.
251//
252// This can be boiled down to a set of constraints and tools.
253//
254// 1) Forwarding timestamps and data need to be logged separately.
255// 2) Any forwarded data logged on the destination node needs to be logged
256// separately such that it can be sorted.
257//
258// 1) Log reader needs to be able to sort a list of log files.
259// 2) Log reader needs to be able to merge sorted lists of log files.
260// 3) Log reader needs to be able to match timestamps with messages.
261//
262// We also need to be able to generate multiple views of a log file depending on
263// the target.
264
Austin Schuhe309d2a2019-11-29 13:25:21 -0800265// Replays all the channels in the logfile to the event loop.
266class LogReader {
267 public:
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800268 // If you want to supply a new configuration that will be used for replay
269 // (e.g., to change message rates, or to populate an updated schema), then
270 // pass it in here. It must provide all the channels that the original logged
271 // config did.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800272 //
273 // Log filenames are in the following format:
274 //
275 // {
276 // {log1_part0, log1_part1, ...},
277 // {log2}
278 // }
279 // The inner vector is a list of log file chunks which form up a log file.
280 // The outer vector is a list of log files with subsets of the messages, or
281 // messages from different nodes.
282 //
283 // If the outer vector isn't provided, it is assumed to be of size 1.
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800284 LogReader(std::string_view filename,
285 const Configuration *replay_configuration = nullptr);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800286 LogReader(const std::vector<std::string> &filenames,
287 const Configuration *replay_configuration = nullptr);
288 LogReader(const std::vector<std::vector<std::string>> &filenames,
Austin Schuhfa895892020-01-07 20:07:41 -0800289 const Configuration *replay_configuration = nullptr);
Austin Schuh11d43732020-09-21 17:28:30 -0700290 LogReader(const std::vector<LogFile> &log_files,
291 const Configuration *replay_configuration = nullptr);
James Kuszmaul7daef362019-12-31 18:28:17 -0800292 ~LogReader();
Austin Schuhe309d2a2019-11-29 13:25:21 -0800293
Austin Schuh6331ef92020-01-07 18:28:09 -0800294 // Registers all the callbacks to send the log file data out on an event loop
295 // created in event_loop_factory. This also updates time to be at the start
296 // of the log file by running until the log file starts.
297 // Note: the configuration used in the factory should be configuration()
298 // below, but can be anything as long as the locations needed to send
299 // everything are available.
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800300 void Register(SimulatedEventLoopFactory *event_loop_factory);
Austin Schuh6331ef92020-01-07 18:28:09 -0800301 // Creates an SimulatedEventLoopFactory accessible via event_loop_factory(),
302 // and then calls Register.
303 void Register();
304 // Registers callbacks for all the events after the log file starts. This is
305 // only useful when replaying live.
Austin Schuhe309d2a2019-11-29 13:25:21 -0800306 void Register(EventLoop *event_loop);
Austin Schuh6331ef92020-01-07 18:28:09 -0800307
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800308 // Unregisters the senders. You only need to call this if you separately
309 // supplied an event loop or event loop factory and the lifetimes are such
310 // that they need to be explicitly destroyed before the LogReader destructor
311 // gets called.
Austin Schuhe309d2a2019-11-29 13:25:21 -0800312 void Deregister();
313
Austin Schuh0c297012020-09-16 18:41:59 -0700314 // Returns the configuration being used for replay from the log file.
315 // Note that this may be different from the configuration actually used for
316 // handling events. You should generally only use this to create a
317 // SimulatedEventLoopFactory, and then get the configuration from there for
318 // everything else.
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800319 const Configuration *logged_configuration() const;
Austin Schuh11d43732020-09-21 17:28:30 -0700320 // Returns the configuration being used for replay from the log file.
321 // Note that this may be different from the configuration actually used for
322 // handling events. You should generally only use this to create a
323 // SimulatedEventLoopFactory, and then get the configuration from there for
324 // everything else.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800325 // The pointer is invalidated whenever RemapLoggedChannel is called.
Austin Schuh15649d62019-12-28 16:36:38 -0800326 const Configuration *configuration() const;
327
Austin Schuh6f3babe2020-01-26 20:34:50 -0800328 // Returns the nodes that this log file was created on. This is a list of
329 // pointers to a node in the nodes() list inside configuration(). The
330 // pointers here are invalidated whenever RemapLoggedChannel is called.
331 std::vector<const Node *> Nodes() const;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800332
333 // Returns the starting timestamp for the log file.
Austin Schuh11d43732020-09-21 17:28:30 -0700334 monotonic_clock::time_point monotonic_start_time(
335 const Node *node = nullptr) const;
336 realtime_clock::time_point realtime_start_time(
337 const Node *node = nullptr) const;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800338
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800339 // Causes the logger to publish the provided channel on a different name so
340 // that replayed applications can publish on the proper channel name without
341 // interference. This operates on raw channel names, without any node or
342 // application specific mappings.
343 void RemapLoggedChannel(std::string_view name, std::string_view type,
344 std::string_view add_prefix = "/original");
345 template <typename T>
346 void RemapLoggedChannel(std::string_view name,
347 std::string_view add_prefix = "/original") {
348 RemapLoggedChannel(name, T::GetFullyQualifiedName(), add_prefix);
349 }
350
Austin Schuh01b4c352020-09-21 23:09:39 -0700351 // Remaps the provided channel, though this respects node mappings, and
352 // preserves them too. This makes it so if /aos -> /pi1/aos on one node,
353 // /original/aos -> /original/pi1/aos on the same node after renaming, just
354 // like you would hope.
355 //
356 // TODO(austin): If you have 2 nodes remapping something to the same channel,
357 // this doesn't handle that. No use cases exist yet for that, so it isn't
358 // being done yet.
359 void RemapLoggedChannel(std::string_view name, std::string_view type,
360 const Node *node,
361 std::string_view add_prefix = "/original");
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700362 template <typename T>
Austin Schuh01b4c352020-09-21 23:09:39 -0700363 void RemapLoggedChannel(std::string_view name, const Node *node,
364 std::string_view add_prefix = "/original") {
365 RemapLoggedChannel(name, T::GetFullyQualifiedName(), node, add_prefix);
366 }
367
368 template <typename T>
369 bool HasChannel(std::string_view name, const Node *node = nullptr) {
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700370 return configuration::GetChannel(log_file_header()->configuration(), name,
371 T::GetFullyQualifiedName(), "",
Austin Schuh01b4c352020-09-21 23:09:39 -0700372 node) != nullptr;
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700373 }
374
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800375 SimulatedEventLoopFactory *event_loop_factory() {
376 return event_loop_factory_;
377 }
378
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700379 const LogFileHeader *log_file_header() const {
380 return &log_file_header_.message();
381 }
382
Austin Schuh0c297012020-09-16 18:41:59 -0700383 std::string_view name() const {
384 return log_file_header()->name()->string_view();
385 }
386
Austin Schuhe309d2a2019-11-29 13:25:21 -0800387 private:
Austin Schuh6f3babe2020-01-26 20:34:50 -0800388 const Channel *RemapChannel(const EventLoop *event_loop,
389 const Channel *channel);
390
Austin Schuhe309d2a2019-11-29 13:25:21 -0800391 // Queues at least max_out_of_order_duration_ messages into channels_.
392 void QueueMessages();
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800393 // Handle constructing a configuration with all the additional remapped
394 // channels from calls to RemapLoggedChannel.
395 void MakeRemappedConfig();
Austin Schuhe309d2a2019-11-29 13:25:21 -0800396
Austin Schuh2f8fd752020-09-01 22:38:28 -0700397 // Returns the number of nodes.
398 size_t nodes_count() const {
399 return !configuration::MultiNode(logged_configuration())
400 ? 1u
401 : logged_configuration()->nodes()->size();
402 }
403
Austin Schuh6f3babe2020-01-26 20:34:50 -0800404 const std::vector<std::vector<std::string>> filenames_;
405
406 // This is *a* log file header used to provide the logged config. The rest of
407 // the header is likely distracting.
408 FlatbufferVector<LogFileHeader> log_file_header_;
409
Austin Schuh2f8fd752020-09-01 22:38:28 -0700410 // Returns [ta; tb; ...] = tuple[0] * t + tuple[1]
411 std::tuple<Eigen::Matrix<double, Eigen::Dynamic, 1>,
412 Eigen::Matrix<double, Eigen::Dynamic, 1>>
413 SolveOffsets();
414
415 void LogFit(std::string_view prefix);
Austin Schuh8bd96322020-02-13 21:18:22 -0800416
Austin Schuh6f3babe2020-01-26 20:34:50 -0800417 // State per node.
Austin Schuh858c9f32020-08-31 16:56:12 -0700418 class State {
419 public:
420 State(std::unique_ptr<ChannelMerger> channel_merger);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800421
Austin Schuh858c9f32020-08-31 16:56:12 -0700422 // Returns the timestamps, channel_index, and message from a channel.
423 // update_time (will be) set to true when popping this message causes the
424 // filter to change the time offset estimation function.
425 std::tuple<TimestampMerger::DeliveryTimestamp, int,
426 FlatbufferVector<MessageHeader>>
427 PopOldest(bool *update_time);
428
429 // Returns the monotonic time of the oldest message.
430 monotonic_clock::time_point OldestMessageTime() const;
431
432 // Primes the queues inside State. Should be called before calling
433 // OldestMessageTime.
434 void SeedSortedMessages();
Austin Schuh8bd96322020-02-13 21:18:22 -0800435
Austin Schuh858c9f32020-08-31 16:56:12 -0700436 // Returns the starting time for this node.
437 monotonic_clock::time_point monotonic_start_time() const {
438 return channel_merger_->monotonic_start_time();
439 }
440 realtime_clock::time_point realtime_start_time() const {
441 return channel_merger_->realtime_start_time();
442 }
443
444 // Sets the node event loop factory for replaying into a
445 // SimulatedEventLoopFactory. Returns the EventLoop to use.
446 EventLoop *SetNodeEventLoopFactory(
447 NodeEventLoopFactory *node_event_loop_factory);
448
449 // Sets and gets the event loop to use.
450 void set_event_loop(EventLoop *event_loop) { event_loop_ = event_loop; }
451 EventLoop *event_loop() { return event_loop_; }
452
Austin Schuh858c9f32020-08-31 16:56:12 -0700453 // Sets the current realtime offset from the monotonic clock for this node
454 // (if we are on a simulated event loop).
455 void SetRealtimeOffset(monotonic_clock::time_point monotonic_time,
456 realtime_clock::time_point realtime_time) {
457 if (node_event_loop_factory_ != nullptr) {
458 node_event_loop_factory_->SetRealtimeOffset(monotonic_time,
459 realtime_time);
460 }
461 }
462
463 // Converts a timestamp from the monotonic clock on this node to the
464 // distributed clock.
465 distributed_clock::time_point ToDistributedClock(
466 monotonic_clock::time_point time) {
467 return node_event_loop_factory_->ToDistributedClock(time);
468 }
469
Austin Schuh2f8fd752020-09-01 22:38:28 -0700470 monotonic_clock::time_point FromDistributedClock(
471 distributed_clock::time_point time) {
472 return node_event_loop_factory_->FromDistributedClock(time);
473 }
474
Austin Schuh858c9f32020-08-31 16:56:12 -0700475 // Sets the offset (and slope) from the distributed clock.
476 void SetDistributedOffset(std::chrono::nanoseconds distributed_offset,
477 double distributed_slope) {
478 node_event_loop_factory_->SetDistributedOffset(distributed_offset,
479 distributed_slope);
480 }
481
482 // Returns the current time on the remote node which sends messages on
483 // channel_index.
484 monotonic_clock::time_point monotonic_remote_now(size_t channel_index) {
485 return channel_target_event_loop_factory_[channel_index]->monotonic_now();
486 }
487
Austin Schuh2f8fd752020-09-01 22:38:28 -0700488 distributed_clock::time_point RemoteToDistributedClock(
489 size_t channel_index, monotonic_clock::time_point time) {
490 return channel_target_event_loop_factory_[channel_index]
491 ->ToDistributedClock(time);
492 }
493
494 const Node *remote_node(size_t channel_index) {
495 return channel_target_event_loop_factory_[channel_index]->node();
496 }
497
498 monotonic_clock::time_point monotonic_now() {
499 return node_event_loop_factory_->monotonic_now();
500 }
501
Austin Schuh858c9f32020-08-31 16:56:12 -0700502 // Sets the node we will be merging as, and returns true if there is any
503 // data on it.
504 bool SetNode() { return channel_merger_->SetNode(event_loop_->node()); }
505
506 // Sets the number of channels.
507 void SetChannelCount(size_t count);
508
509 // Sets the sender, filter, and target factory for a channel.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700510 void SetChannel(size_t channel, std::unique_ptr<RawSender> sender,
511 message_bridge::NoncausalOffsetEstimator *filter,
512 NodeEventLoopFactory *channel_target_event_loop_factory);
Austin Schuh858c9f32020-08-31 16:56:12 -0700513
514 // Returns if we have read all the messages from all the logs.
515 bool at_end() const { return channel_merger_->at_end(); }
516
517 // Unregisters everything so we can destory the event loop.
518 void Deregister();
519
520 // Sets the current TimerHandle for the replay callback.
521 void set_timer_handler(TimerHandler *timer_handler) {
522 timer_handler_ = timer_handler;
523 }
524
525 // Sets the next wakeup time on the replay callback.
526 void Setup(monotonic_clock::time_point next_time) {
527 timer_handler_->Setup(next_time);
528 }
529
530 // Sends a buffer on the provided channel index.
531 bool Send(size_t channel_index, const void *data, size_t size,
532 aos::monotonic_clock::time_point monotonic_remote_time,
533 aos::realtime_clock::time_point realtime_remote_time,
534 uint32_t remote_queue_index) {
535 return channels_[channel_index]->Send(data, size, monotonic_remote_time,
536 realtime_remote_time,
537 remote_queue_index);
538 }
539
540 // Returns a debug string for the channel merger.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700541 std::string DebugString() const {
542 std::stringstream messages;
543 size_t i = 0;
544 for (const auto &message : sorted_messages_) {
545 if (i < 7 || i + 7 > sorted_messages_.size()) {
546 messages << "sorted_messages[" << i
547 << "]: " << std::get<0>(message).monotonic_event_time << " "
548 << configuration::StrippedChannelToString(
549 event_loop_->configuration()->channels()->Get(
550 std::get<2>(message).message().channel_index()))
551 << "\n";
552 } else if (i == 7) {
553 messages << "...\n";
554 }
555 ++i;
556 }
557 return messages.str() + channel_merger_->DebugString();
558 }
Austin Schuh858c9f32020-08-31 16:56:12 -0700559
560 private:
561 // Log file.
562 std::unique_ptr<ChannelMerger> channel_merger_;
563
564 std::deque<std::tuple<TimestampMerger::DeliveryTimestamp, int,
Austin Schuh2f8fd752020-09-01 22:38:28 -0700565 FlatbufferVector<MessageHeader>,
566 message_bridge::NoncausalOffsetEstimator *>>
Austin Schuh858c9f32020-08-31 16:56:12 -0700567 sorted_messages_;
568
569 // Senders.
570 std::vector<std::unique_ptr<RawSender>> channels_;
571
572 // Factory (if we are in sim) that this loop was created on.
573 NodeEventLoopFactory *node_event_loop_factory_ = nullptr;
574 std::unique_ptr<EventLoop> event_loop_unique_ptr_;
575 // Event loop.
576 EventLoop *event_loop_ = nullptr;
577 // And timer used to send messages.
578 TimerHandler *timer_handler_;
579
Austin Schuh8bd96322020-02-13 21:18:22 -0800580 // Filters (or nullptr if it isn't a forwarded channel) for each channel.
581 // This corresponds to the object which is shared among all the channels
582 // going between 2 nodes. The second element in the tuple indicates if this
583 // is the primary direction or not.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700584 std::vector<message_bridge::NoncausalOffsetEstimator *> filters_;
Austin Schuh8bd96322020-02-13 21:18:22 -0800585
586 // List of NodeEventLoopFactorys (or nullptr if it isn't a forwarded
587 // channel) which correspond to the originating node.
Austin Schuh858c9f32020-08-31 16:56:12 -0700588 std::vector<NodeEventLoopFactory *> channel_target_event_loop_factory_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800589 };
590
Austin Schuh8bd96322020-02-13 21:18:22 -0800591 // Node index -> State.
592 std::vector<std::unique_ptr<State>> states_;
593
594 // Creates the requested filter if it doesn't exist, regardless of whether
595 // these nodes can actually communicate directly. The second return value
596 // reports if this is the primary direction or not.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700597 message_bridge::NoncausalOffsetEstimator *GetFilter(const Node *node_a,
598 const Node *node_b);
Austin Schuh8bd96322020-02-13 21:18:22 -0800599
600 // FILE to write offsets to (if populated).
601 FILE *offset_fp_ = nullptr;
602 // Timestamp of the first piece of data used for the horizontal axis on the
603 // plot.
604 aos::realtime_clock::time_point first_time_;
605
606 // List of filters for a connection. The pointer to the first node will be
607 // less than the second node.
608 std::map<std::tuple<const Node *, const Node *>,
Austin Schuh2f8fd752020-09-01 22:38:28 -0700609 std::tuple<message_bridge::NoncausalOffsetEstimator>>
Austin Schuh8bd96322020-02-13 21:18:22 -0800610 filters_;
611
612 // Returns the offset from the monotonic clock for a node to the distributed
Austin Schuh2f8fd752020-09-01 22:38:28 -0700613 // clock. monotonic = distributed * slope() + offset();
614 double slope(int node_index) const {
615 CHECK_LT(node_index, time_slope_matrix_.rows())
James Kuszmaul46d82582020-05-09 19:50:09 -0700616 << ": Got too high of a node index.";
Austin Schuh2f8fd752020-09-01 22:38:28 -0700617 return time_slope_matrix_(node_index);
618 }
619 std::chrono::nanoseconds offset(int node_index) const {
620 CHECK_LT(node_index, time_offset_matrix_.rows())
621 << ": Got too high of a node index.";
622 return std::chrono::duration_cast<std::chrono::nanoseconds>(
623 std::chrono::duration<double>(time_offset_matrix_(node_index)));
Austin Schuh8bd96322020-02-13 21:18:22 -0800624 }
625
626 // Updates the offset matrix solution and sets the per-node distributed
627 // offsets in the factory.
628 void UpdateOffsets();
629
Austin Schuh2f8fd752020-09-01 22:38:28 -0700630 // We have 2 types of equations to do a least squares regression over to fully
631 // constrain our time function.
632 //
633 // One is simple. The distributed clock is the average of all the clocks.
Brian Silverman87ac0402020-09-17 14:47:01 -0700634 // (ta + tb + tc + td) / num_nodes = t_distributed
Austin Schuh2f8fd752020-09-01 22:38:28 -0700635 //
636 // The second is a bit more complicated. Our basic time conversion function
637 // is:
638 // tb = ta + (ta * slope + offset)
639 // We can rewrite this as follows
640 // tb - (1 + slope) * ta = offset
641 //
642 // From here, we have enough equations to solve for t{a,b,c,...} We want to
643 // take as an input the offsets and slope, and solve for the per-node times as
644 // a function of the distributed clock.
645 //
646 // We need to massage our equations to make this work. If we solve for the
647 // per-node times at two set distributed clock times, we will be able to
648 // recreate the linear function (we know it is linear). We can do a similar
649 // thing by breaking our equation up into:
Brian Silverman87ac0402020-09-17 14:47:01 -0700650 //
Austin Schuh2f8fd752020-09-01 22:38:28 -0700651 // [1/3 1/3 1/3 ] [ta] [t_distributed]
652 // [ 1 -1-m1 0 ] [tb] = [oab]
653 // [ 1 0 -1-m2 ] [tc] [oac]
654 //
655 // This solves to:
656 //
657 // [ta] [ a00 a01 a02] [t_distributed]
658 // [tb] = [ a10 a11 a12] * [oab]
659 // [tc] [ a20 a21 a22] [oac]
660 //
661 // and can be split into:
662 //
663 // [ta] [ a00 ] [a01 a02]
664 // [tb] = [ a10 ] * t_distributed + [a11 a12] * [oab]
665 // [tc] [ a20 ] [a21 a22] [oac]
666 //
667 // (map_matrix_ + slope_matrix_) * [ta; tb; tc] = [offset_matrix_];
668 // offset_matrix_ will be in nanoseconds.
669 Eigen::Matrix<mpq_class, Eigen::Dynamic, Eigen::Dynamic> map_matrix_;
670 Eigen::Matrix<mpq_class, Eigen::Dynamic, Eigen::Dynamic> slope_matrix_;
671 Eigen::Matrix<mpq_class, Eigen::Dynamic, 1> offset_matrix_;
672 // Matrix tracking which offsets are valid.
673 Eigen::Matrix<bool, Eigen::Dynamic, 1> valid_matrix_;
674 // Matrix tracking the last valid matrix we used to determine connected nodes.
675 Eigen::Matrix<bool, Eigen::Dynamic, 1> last_valid_matrix_;
676 size_t cached_valid_node_count_ = 0;
Austin Schuh8bd96322020-02-13 21:18:22 -0800677
Austin Schuh2f8fd752020-09-01 22:38:28 -0700678 // [ta; tb; tc] = time_slope_matrix_ * t + time_offset_matrix;
679 // t is in seconds.
680 Eigen::Matrix<double, Eigen::Dynamic, 1> time_slope_matrix_;
681 Eigen::Matrix<double, Eigen::Dynamic, 1> time_offset_matrix_;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800682
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800683 std::unique_ptr<FlatbufferDetachedBuffer<Configuration>>
684 remapped_configuration_buffer_;
685
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800686 std::unique_ptr<SimulatedEventLoopFactory> event_loop_factory_unique_ptr_;
687 SimulatedEventLoopFactory *event_loop_factory_ = nullptr;
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800688
689 // Map of channel indices to new name. The channel index will be an index into
690 // logged_configuration(), and the string key will be the name of the channel
691 // to send on instead of the logged channel name.
692 std::map<size_t, std::string> remapped_channels_;
Austin Schuh01b4c352020-09-21 23:09:39 -0700693 std::vector<MapT> maps_;
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800694
Austin Schuh6f3babe2020-01-26 20:34:50 -0800695 // Number of nodes which still have data to send. This is used to figure out
696 // when to exit.
697 size_t live_nodes_ = 0;
698
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800699 const Configuration *remapped_configuration_ = nullptr;
700 const Configuration *replay_configuration_ = nullptr;
Austin Schuhcde938c2020-02-02 17:30:07 -0800701
702 // If true, the replay timer will ignore any missing data. This is used
703 // during startup when we are bootstrapping everything and trying to get to
704 // the start of all the log files.
705 bool ignore_missing_data_ = false;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800706};
707
708} // namespace logger
709} // namespace aos
710
711#endif // AOS_EVENTS_LOGGER_H_