blob: 2c31e0f5866038d0a353bd6225d29b4984bc2bc2 [file] [log] [blame]
Austin Schuhb06f03b2021-02-17 22:00:37 -08001#ifndef AOS_EVENTS_LOGGING_LOG_READER_H_
2#define AOS_EVENTS_LOGGING_LOG_READER_H_
Austin Schuhe309d2a2019-11-29 13:25:21 -08003
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 Schuhe309d2a2019-11-29 13:25:21 -080010#include "aos/events/event_loop.h"
Austin Schuhf6f9bf32020-10-11 14:37:43 -070011#include "aos/events/logging/logfile_sorting.h"
Austin Schuha36c8902019-12-30 18:07:15 -080012#include "aos/events/logging/logfile_utils.h"
James Kuszmaul38735e82019-12-07 16:42:06 -080013#include "aos/events/logging/logger_generated.h"
James Kuszmaul09632422022-05-25 15:56:19 -070014#include "aos/events/shm_event_loop.h"
Austin Schuh92547522019-12-28 14:33:43 -080015#include "aos/events/simulated_event_loop.h"
Austin Schuh2f8fd752020-09-01 22:38:28 -070016#include "aos/network/message_bridge_server_generated.h"
Austin Schuh0ca1fd32020-12-18 22:53:05 -080017#include "aos/network/multinode_timestamp_filter.h"
Austin Schuh0de30f32020-12-06 12:44:28 -080018#include "aos/network/remote_message_generated.h"
Austin Schuh8bd96322020-02-13 21:18:22 -080019#include "aos/network/timestamp_filter.h"
Austin Schuhe309d2a2019-11-29 13:25:21 -080020#include "aos/time/time.h"
Austin Schuh4385b142021-03-14 21:31:13 -070021#include "aos/uuid.h"
Austin Schuhe309d2a2019-11-29 13:25:21 -080022#include "flatbuffers/flatbuffers.h"
23
24namespace aos {
25namespace logger {
26
Austin Schuhe33c08d2022-02-03 18:15:21 -080027class EventNotifier;
28
Austin Schuh6f3babe2020-01-26 20:34:50 -080029// We end up with one of the following 3 log file types.
30//
31// Single node logged as the source node.
32// -> Replayed just on the source node.
33//
34// Forwarding timestamps only logged from the perspective of the destination
35// node.
36// -> Matched with data on source node and logged.
37//
38// Forwarding timestamps with data logged as the destination node.
39// -> Replayed just as the destination
40// -> Replayed as the source (Much harder, ordering is not defined)
41//
42// Duplicate data logged. -> CHECK that it matches and explode otherwise.
43//
44// This can be boiled down to a set of constraints and tools.
45//
46// 1) Forwarding timestamps and data need to be logged separately.
47// 2) Any forwarded data logged on the destination node needs to be logged
48// separately such that it can be sorted.
49//
50// 1) Log reader needs to be able to sort a list of log files.
51// 2) Log reader needs to be able to merge sorted lists of log files.
52// 3) Log reader needs to be able to match timestamps with messages.
53//
54// We also need to be able to generate multiple views of a log file depending on
55// the target.
56
Austin Schuhe309d2a2019-11-29 13:25:21 -080057// Replays all the channels in the logfile to the event loop.
58class LogReader {
59 public:
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -080060 // If you want to supply a new configuration that will be used for replay
61 // (e.g., to change message rates, or to populate an updated schema), then
62 // pass it in here. It must provide all the channels that the original logged
63 // config did.
Austin Schuh6f3babe2020-01-26 20:34:50 -080064 //
Austin Schuh287d43d2020-12-04 20:19:33 -080065 // The single file constructor calls SortParts internally.
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -080066 LogReader(std::string_view filename,
67 const Configuration *replay_configuration = nullptr);
Austin Schuh287d43d2020-12-04 20:19:33 -080068 LogReader(std::vector<LogFile> log_files,
Austin Schuh11d43732020-09-21 17:28:30 -070069 const Configuration *replay_configuration = nullptr);
James Kuszmaul7daef362019-12-31 18:28:17 -080070 ~LogReader();
Austin Schuhe309d2a2019-11-29 13:25:21 -080071
Austin Schuh6331ef92020-01-07 18:28:09 -080072 // Registers all the callbacks to send the log file data out on an event loop
73 // created in event_loop_factory. This also updates time to be at the start
74 // of the log file by running until the log file starts.
75 // Note: the configuration used in the factory should be configuration()
76 // below, but can be anything as long as the locations needed to send
77 // everything are available.
James Kuszmaul84ff3e52020-01-03 19:48:53 -080078 void Register(SimulatedEventLoopFactory *event_loop_factory);
Austin Schuhe33c08d2022-02-03 18:15:21 -080079
Austin Schuh58646e22021-08-23 23:51:46 -070080 // Registers all the callbacks to send the log file data out to an event loop
81 // factory. This does not start replaying or change the current distributed
82 // time of the factory. It does change the monotonic clocks to be right.
83 void RegisterWithoutStarting(SimulatedEventLoopFactory *event_loop_factory);
Austin Schuhe33c08d2022-02-03 18:15:21 -080084 // Runs the log until the last start time. Register above is defined as:
85 // Register(...) {
86 // RegisterWithoutStarting
87 // StartAfterRegister
88 // }
89 // This should generally be considered as a stepping stone to convert from
90 // Register() to RegisterWithoutStarting() incrementally.
91 void StartAfterRegister(SimulatedEventLoopFactory *event_loop_factory);
92
Austin Schuh6331ef92020-01-07 18:28:09 -080093 // Creates an SimulatedEventLoopFactory accessible via event_loop_factory(),
94 // and then calls Register.
95 void Register();
James Kuszmaul09632422022-05-25 15:56:19 -070096
Austin Schuh6331ef92020-01-07 18:28:09 -080097 // Registers callbacks for all the events after the log file starts. This is
98 // only useful when replaying live.
Austin Schuhe309d2a2019-11-29 13:25:21 -080099 void Register(EventLoop *event_loop);
Austin Schuh6331ef92020-01-07 18:28:09 -0800100
Austin Schuh58646e22021-08-23 23:51:46 -0700101 // Called whenever a log file starts for a node.
102 void OnStart(std::function<void()> fn);
103 void OnStart(const Node *node, std::function<void()> fn);
104 // Called whenever a log file ends for a node.
105 void OnEnd(std::function<void()> fn);
106 void OnEnd(const Node *node, std::function<void()> fn);
107
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800108 // Unregisters the senders. You only need to call this if you separately
109 // supplied an event loop or event loop factory and the lifetimes are such
110 // that they need to be explicitly destroyed before the LogReader destructor
111 // gets called.
Austin Schuhe309d2a2019-11-29 13:25:21 -0800112 void Deregister();
113
Austin Schuh0c297012020-09-16 18:41:59 -0700114 // Returns the configuration being used for replay from the log file.
115 // Note that this may be different from the configuration actually used for
116 // handling events. You should generally only use this to create a
117 // SimulatedEventLoopFactory, and then get the configuration from there for
118 // everything else.
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800119 const Configuration *logged_configuration() const;
Austin Schuh11d43732020-09-21 17:28:30 -0700120 // Returns the configuration being used for replay from the log file.
121 // Note that this may be different from the configuration actually used for
122 // handling events. You should generally only use this to create a
123 // SimulatedEventLoopFactory, and then get the configuration from there for
124 // everything else.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800125 // The pointer is invalidated whenever RemapLoggedChannel is called.
Austin Schuh15649d62019-12-28 16:36:38 -0800126 const Configuration *configuration() const;
127
Austin Schuh6f3babe2020-01-26 20:34:50 -0800128 // Returns the nodes that this log file was created on. This is a list of
Austin Schuh07676622021-01-21 18:59:17 -0800129 // pointers to a node in the nodes() list inside logged_configuration().
130 std::vector<const Node *> LoggedNodes() const;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800131
132 // Returns the starting timestamp for the log file.
Austin Schuh11d43732020-09-21 17:28:30 -0700133 monotonic_clock::time_point monotonic_start_time(
134 const Node *node = nullptr) const;
135 realtime_clock::time_point realtime_start_time(
136 const Node *node = nullptr) const;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800137
Austin Schuhe33c08d2022-02-03 18:15:21 -0800138 // Sets the start and end times to replay data until for all nodes. This
139 // overrides the --start_time and --end_time flags. The default is to replay
140 // all data.
141 void SetStartTime(std::string start_time);
142 void SetStartTime(realtime_clock::time_point start_time);
143 void SetEndTime(std::string end_time);
144 void SetEndTime(realtime_clock::time_point end_time);
145
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800146 // Causes the logger to publish the provided channel on a different name so
147 // that replayed applications can publish on the proper channel name without
148 // interference. This operates on raw channel names, without any node or
149 // application specific mappings.
150 void RemapLoggedChannel(std::string_view name, std::string_view type,
Austin Schuh0de30f32020-12-06 12:44:28 -0800151 std::string_view add_prefix = "/original",
152 std::string_view new_type = "");
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800153 template <typename T>
154 void RemapLoggedChannel(std::string_view name,
Austin Schuh0de30f32020-12-06 12:44:28 -0800155 std::string_view add_prefix = "/original",
156 std::string_view new_type = "") {
157 RemapLoggedChannel(name, T::GetFullyQualifiedName(), add_prefix, new_type);
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800158 }
159
Austin Schuh01b4c352020-09-21 23:09:39 -0700160 // Remaps the provided channel, though this respects node mappings, and
161 // preserves them too. This makes it so if /aos -> /pi1/aos on one node,
162 // /original/aos -> /original/pi1/aos on the same node after renaming, just
Austin Schuh0de30f32020-12-06 12:44:28 -0800163 // like you would hope. If new_type is not empty, the new channel will use
164 // the provided type instead. This allows for renaming messages.
Austin Schuh01b4c352020-09-21 23:09:39 -0700165 //
166 // TODO(austin): If you have 2 nodes remapping something to the same channel,
167 // this doesn't handle that. No use cases exist yet for that, so it isn't
168 // being done yet.
169 void RemapLoggedChannel(std::string_view name, std::string_view type,
170 const Node *node,
Austin Schuh0de30f32020-12-06 12:44:28 -0800171 std::string_view add_prefix = "/original",
172 std::string_view new_type = "");
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700173 template <typename T>
Austin Schuh01b4c352020-09-21 23:09:39 -0700174 void RemapLoggedChannel(std::string_view name, const Node *node,
Austin Schuh0de30f32020-12-06 12:44:28 -0800175 std::string_view add_prefix = "/original",
176 std::string_view new_type = "") {
177 RemapLoggedChannel(name, T::GetFullyQualifiedName(), node, add_prefix,
178 new_type);
Austin Schuh01b4c352020-09-21 23:09:39 -0700179 }
180
181 template <typename T>
182 bool HasChannel(std::string_view name, const Node *node = nullptr) {
Austin Schuh0ca51f32020-12-25 21:51:45 -0800183 return configuration::GetChannel(logged_configuration(), name,
Austin Schuh0de30f32020-12-06 12:44:28 -0800184 T::GetFullyQualifiedName(), "", node,
185 true) != nullptr;
Brian Silvermande9f3ff2020-04-28 16:56:58 -0700186 }
187
Austin Schuh82529062021-12-08 12:09:52 -0800188 template <typename T>
189 void MaybeRemapLoggedChannel(std::string_view name,
190 const Node *node = nullptr) {
191 if (HasChannel<T>(name, node)) {
192 RemapLoggedChannel<T>(name, node);
193 }
194 }
195
James Kuszmaul4f106fb2021-01-05 20:53:02 -0800196 // Returns true if the channel exists on the node and was logged.
197 template <typename T>
198 bool HasLoggedChannel(std::string_view name, const Node *node = nullptr) {
Austin Schuh5ee56872021-01-30 16:53:34 -0800199 const Channel *channel =
200 configuration::GetChannel(logged_configuration(), name,
201 T::GetFullyQualifiedName(), "", node, true);
James Kuszmaul4f106fb2021-01-05 20:53:02 -0800202 if (channel == nullptr) return false;
203 return channel->logger() != LoggerConfig::NOT_LOGGED;
204 }
205
Austin Schuh1c227352021-09-17 12:53:54 -0700206 // Returns a list of all the original channels from remapping.
207 std::vector<const Channel *> RemappedChannels() const;
208
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800209 SimulatedEventLoopFactory *event_loop_factory() {
210 return event_loop_factory_;
211 }
212
Austin Schuh0ca51f32020-12-25 21:51:45 -0800213 std::string_view name() const { return log_files_[0].name; }
Austin Schuh0c297012020-09-16 18:41:59 -0700214
James Kuszmaul71a81932020-12-15 21:08:01 -0800215 // Set whether to exit the SimulatedEventLoopFactory when we finish reading
216 // the logfile.
217 void set_exit_on_finish(bool exit_on_finish) {
218 exit_on_finish_ = exit_on_finish;
219 }
220
Austin Schuhe309d2a2019-11-29 13:25:21 -0800221 private:
Austin Schuh58646e22021-08-23 23:51:46 -0700222 void Register(EventLoop *event_loop, const Node *node);
223
224 void RegisterDuringStartup(EventLoop *event_loop, const Node *node);
225
226 const Channel *RemapChannel(const EventLoop *event_loop, const Node *node,
Austin Schuh6f3babe2020-01-26 20:34:50 -0800227 const Channel *channel);
228
Austin Schuhe309d2a2019-11-29 13:25:21 -0800229 // Queues at least max_out_of_order_duration_ messages into channels_.
230 void QueueMessages();
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800231 // Handle constructing a configuration with all the additional remapped
232 // channels from calls to RemapLoggedChannel.
233 void MakeRemappedConfig();
Austin Schuhe309d2a2019-11-29 13:25:21 -0800234
Austin Schuh2f8fd752020-09-01 22:38:28 -0700235 // Returns the number of nodes.
236 size_t nodes_count() const {
237 return !configuration::MultiNode(logged_configuration())
238 ? 1u
239 : logged_configuration()->nodes()->size();
240 }
241
Austin Schuh287d43d2020-12-04 20:19:33 -0800242 const std::vector<LogFile> log_files_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800243
Austin Schuh969cd602021-01-03 00:09:45 -0800244 // Class to manage sending RemoteMessages on the provided node after the
245 // correct delay.
Austin Schuh5ee56872021-01-30 16:53:34 -0800246 class RemoteMessageSender {
Austin Schuh969cd602021-01-03 00:09:45 -0800247 public:
248 RemoteMessageSender(aos::Sender<message_bridge::RemoteMessage> sender,
249 EventLoop *event_loop);
250 RemoteMessageSender(RemoteMessageSender const &) = delete;
251 RemoteMessageSender &operator=(RemoteMessageSender const &) = delete;
252
253 // Sends the provided message. If monotonic_timestamp_time is min_time,
254 // send it immediately.
255 void Send(
256 FlatbufferDetachedBuffer<message_bridge::RemoteMessage> remote_message,
Austin Schuh58646e22021-08-23 23:51:46 -0700257 BootTimestamp monotonic_timestamp_time, size_t source_boot_count);
Austin Schuh969cd602021-01-03 00:09:45 -0800258
259 private:
260 // Handles actually sending the timestamp if we were delayed.
261 void SendTimestamp();
262 // Handles scheduling the timer to send at the correct time.
263 void ScheduleTimestamp();
264
265 EventLoop *event_loop_;
266 aos::Sender<message_bridge::RemoteMessage> sender_;
267 aos::TimerHandler *timer_;
268
269 // Time we are scheduled for, or min_time if we aren't scheduled.
270 monotonic_clock::time_point scheduled_time_ = monotonic_clock::min_time;
271
272 struct Timestamp {
273 Timestamp(FlatbufferDetachedBuffer<message_bridge::RemoteMessage>
274 new_remote_message,
275 monotonic_clock::time_point new_monotonic_timestamp_time)
276 : remote_message(std::move(new_remote_message)),
277 monotonic_timestamp_time(new_monotonic_timestamp_time) {}
278 FlatbufferDetachedBuffer<message_bridge::RemoteMessage> remote_message;
279 monotonic_clock::time_point monotonic_timestamp_time;
280 };
281
282 // List of messages to send. The timer works through them and then disables
283 // itself automatically.
284 std::deque<Timestamp> remote_timestamps_;
285 };
286
Austin Schuh6f3babe2020-01-26 20:34:50 -0800287 // State per node.
Austin Schuh858c9f32020-08-31 16:56:12 -0700288 class State {
289 public:
James Kuszmaul09632422022-05-25 15:56:19 -0700290 State(std::unique_ptr<TimestampMapper> timestamp_mapper,
291 message_bridge::MultiNodeNoncausalOffsetEstimator *multinode_filters,
292 const Node *node);
Austin Schuh287d43d2020-12-04 20:19:33 -0800293
294 // Connects up the timestamp mappers.
295 void AddPeer(State *peer);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800296
Austin Schuhe639ea12021-01-25 13:00:22 -0800297 TimestampMapper *timestamp_mapper() { return timestamp_mapper_.get(); }
298
Austin Schuhdda74ec2021-01-03 19:30:37 -0800299 // Returns the next sorted message with all the timestamps extracted and
300 // matched.
301 TimestampedMessage PopOldest();
Austin Schuh188eabe2020-12-29 23:41:13 -0800302
Austin Schuh858c9f32020-08-31 16:56:12 -0700303 // Returns the monotonic time of the oldest message.
Austin Schuhe33c08d2022-02-03 18:15:21 -0800304 BootTimestamp OldestMessageTime();
Austin Schuh58646e22021-08-23 23:51:46 -0700305
306 size_t boot_count() const {
307 // If we are replaying directly into an event loop, we can't reboot. So
308 // we will stay stuck on the 0th boot.
James Kuszmaul09632422022-05-25 15:56:19 -0700309 if (!node_event_loop_factory_) {
310 if (event_loop_ == nullptr) {
311 // If boot_count is being checked after startup for any of the
312 // non-primary nodes, then returning 0 may not be accurate (since
313 // remote nodes *can* reboot even if the EventLoop being played to
314 // can't).
315 CHECK(!started_);
316 CHECK(!stopped_);
317 }
318 return 0u;
319 }
Austin Schuh58646e22021-08-23 23:51:46 -0700320 return node_event_loop_factory_->boot_count();
321 }
Austin Schuh858c9f32020-08-31 16:56:12 -0700322
323 // Primes the queues inside State. Should be called before calling
324 // OldestMessageTime.
325 void SeedSortedMessages();
Austin Schuh8bd96322020-02-13 21:18:22 -0800326
Austin Schuh58646e22021-08-23 23:51:46 -0700327 void SetupStartupTimer() {
328 const monotonic_clock::time_point start_time =
329 monotonic_start_time(boot_count());
330 if (start_time == monotonic_clock::min_time) {
331 LOG(ERROR)
332 << "No start time, skipping, please figure out when this happens";
Austin Schuhe33c08d2022-02-03 18:15:21 -0800333 NotifyLogfileStart();
Austin Schuh58646e22021-08-23 23:51:46 -0700334 return;
335 }
James Kuszmaul09632422022-05-25 15:56:19 -0700336 if (node_event_loop_factory_) {
337 CHECK_GE(start_time + clock_offset(), event_loop_->monotonic_now());
338 }
339 startup_timer_->Setup(start_time + clock_offset());
Austin Schuh58646e22021-08-23 23:51:46 -0700340 }
341
342 void set_startup_timer(TimerHandler *timer_handler) {
343 startup_timer_ = timer_handler;
344 if (startup_timer_) {
345 if (event_loop_->node() != nullptr) {
346 startup_timer_->set_name(absl::StrCat(
347 event_loop_->node()->name()->string_view(), "_startup"));
348 } else {
349 startup_timer_->set_name("startup");
350 }
351 }
352 }
353
Austin Schuh858c9f32020-08-31 16:56:12 -0700354 // Returns the starting time for this node.
Austin Schuh2dc8c7d2021-07-01 17:41:28 -0700355 monotonic_clock::time_point monotonic_start_time(size_t boot_count) const {
356 return timestamp_mapper_
357 ? timestamp_mapper_->monotonic_start_time(boot_count)
358 : monotonic_clock::min_time;
Austin Schuh858c9f32020-08-31 16:56:12 -0700359 }
Austin Schuh2dc8c7d2021-07-01 17:41:28 -0700360 realtime_clock::time_point realtime_start_time(size_t boot_count) const {
361 return timestamp_mapper_
362 ? timestamp_mapper_->realtime_start_time(boot_count)
363 : realtime_clock::min_time;
Austin Schuh858c9f32020-08-31 16:56:12 -0700364 }
365
366 // Sets the node event loop factory for replaying into a
367 // SimulatedEventLoopFactory. Returns the EventLoop to use.
Austin Schuh60e77942022-05-16 17:48:24 -0700368 void SetNodeEventLoopFactory(NodeEventLoopFactory *node_event_loop_factory,
369 SimulatedEventLoopFactory *event_loop_factory);
Austin Schuh858c9f32020-08-31 16:56:12 -0700370
371 // Sets and gets the event loop to use.
372 void set_event_loop(EventLoop *event_loop) { event_loop_ = event_loop; }
373 EventLoop *event_loop() { return event_loop_; }
374
Austin Schuh58646e22021-08-23 23:51:46 -0700375 const Node *node() const { return node_; }
376
377 void Register(EventLoop *event_loop);
378
379 void OnStart(std::function<void()> fn);
380 void OnEnd(std::function<void()> fn);
381
Austin Schuh858c9f32020-08-31 16:56:12 -0700382 // Sets the current realtime offset from the monotonic clock for this node
383 // (if we are on a simulated event loop).
384 void SetRealtimeOffset(monotonic_clock::time_point monotonic_time,
385 realtime_clock::time_point realtime_time) {
386 if (node_event_loop_factory_ != nullptr) {
387 node_event_loop_factory_->SetRealtimeOffset(monotonic_time,
388 realtime_time);
389 }
390 }
391
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700392 // Returns the MessageHeader sender to log delivery timestamps to for the
393 // provided remote node.
Austin Schuh61e973f2021-02-21 21:43:56 -0800394 RemoteMessageSender *RemoteTimestampSender(const Channel *channel,
395 const Connection *connection);
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700396
Austin Schuh858c9f32020-08-31 16:56:12 -0700397 // Converts a timestamp from the monotonic clock on this node to the
398 // distributed clock.
399 distributed_clock::time_point ToDistributedClock(
400 monotonic_clock::time_point time) {
James Kuszmaul09632422022-05-25 15:56:19 -0700401 CHECK(node_event_loop_factory_);
Austin Schuh858c9f32020-08-31 16:56:12 -0700402 return node_event_loop_factory_->ToDistributedClock(time);
403 }
404
Austin Schuh858c9f32020-08-31 16:56:12 -0700405 // Returns the current time on the remote node which sends messages on
406 // channel_index.
Austin Schuh58646e22021-08-23 23:51:46 -0700407 BootTimestamp monotonic_remote_now(size_t channel_index) {
408 State *s = channel_source_state_[channel_index];
409 return BootTimestamp{
410 .boot = s->boot_count(),
411 .time = s->node_event_loop_factory_->monotonic_now()};
Austin Schuh858c9f32020-08-31 16:56:12 -0700412 }
413
Austin Schuh5ee56872021-01-30 16:53:34 -0800414 // Returns the start time of the remote for the provided channel.
415 monotonic_clock::time_point monotonic_remote_start_time(
Austin Schuh58646e22021-08-23 23:51:46 -0700416 size_t boot_count, size_t channel_index) {
Austin Schuh2dc8c7d2021-07-01 17:41:28 -0700417 return channel_source_state_[channel_index]->monotonic_start_time(
418 boot_count);
Austin Schuh5ee56872021-01-30 16:53:34 -0800419 }
420
Austin Schuh58646e22021-08-23 23:51:46 -0700421 void DestroyEventLoop() { event_loop_unique_ptr_.reset(); }
422
423 EventLoop *MakeEventLoop() {
424 CHECK(!event_loop_unique_ptr_);
James Kuszmaul890c2492022-04-06 14:59:31 -0700425 // TODO(james): Enable exclusive senders on LogReader to allow us to
426 // ensure we are remapping channels correctly.
427 event_loop_unique_ptr_ = node_event_loop_factory_->MakeEventLoop(
428 "log_reader", {NodeEventLoopFactory::CheckSentTooFast::kNo,
429 NodeEventLoopFactory::ExclusiveSenders::kNo});
Austin Schuh58646e22021-08-23 23:51:46 -0700430 return event_loop_unique_ptr_.get();
431 }
432
Austin Schuh2f8fd752020-09-01 22:38:28 -0700433 distributed_clock::time_point RemoteToDistributedClock(
434 size_t channel_index, monotonic_clock::time_point time) {
James Kuszmaul09632422022-05-25 15:56:19 -0700435 CHECK(node_event_loop_factory_);
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700436 return channel_source_state_[channel_index]
437 ->node_event_loop_factory_->ToDistributedClock(time);
Austin Schuh2f8fd752020-09-01 22:38:28 -0700438 }
439
440 const Node *remote_node(size_t channel_index) {
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700441 return channel_source_state_[channel_index]
442 ->node_event_loop_factory_->node();
Austin Schuh2f8fd752020-09-01 22:38:28 -0700443 }
444
Stephan Pleines559fa6c2022-01-06 17:23:51 -0800445 monotonic_clock::time_point monotonic_now() const {
James Kuszmaul09632422022-05-25 15:56:19 -0700446 return event_loop_->monotonic_now();
Austin Schuh2f8fd752020-09-01 22:38:28 -0700447 }
448
Austin Schuh858c9f32020-08-31 16:56:12 -0700449 // Sets the number of channels.
450 void SetChannelCount(size_t count);
451
452 // Sets the sender, filter, and target factory for a channel.
Austin Schuh969cd602021-01-03 00:09:45 -0800453 void SetChannel(size_t logged_channel_index, size_t factory_channel_index,
454 std::unique_ptr<RawSender> sender,
455 message_bridge::NoncausalOffsetEstimator *filter,
Austin Schuh58646e22021-08-23 23:51:46 -0700456 bool is_forwarded, State *source_state);
457
458 void SetRemoteTimestampSender(size_t logged_channel_index,
459 RemoteMessageSender *remote_timestamp_sender);
460
461 void RunOnStart();
462 void RunOnEnd();
Austin Schuh858c9f32020-08-31 16:56:12 -0700463
Austin Schuhe33c08d2022-02-03 18:15:21 -0800464 // Handles a logfile start event to potentially call the OnStart callbacks.
465 void NotifyLogfileStart();
466 // Handles a start time flag start event to potentially call the OnStart
467 // callbacks.
468 void NotifyFlagStart();
469
470 // Handles a logfile end event to potentially call the OnEnd callbacks.
471 void NotifyLogfileEnd();
472 // Handles a end time flag start event to potentially call the OnEnd
473 // callbacks.
474 void NotifyFlagEnd();
475
Austin Schuh858c9f32020-08-31 16:56:12 -0700476 // Unregisters everything so we can destory the event loop.
Austin Schuh58646e22021-08-23 23:51:46 -0700477 // TODO(austin): Is this needed? OnShutdown should be able to serve this
478 // need.
Austin Schuh858c9f32020-08-31 16:56:12 -0700479 void Deregister();
480
481 // Sets the current TimerHandle for the replay callback.
482 void set_timer_handler(TimerHandler *timer_handler) {
483 timer_handler_ = timer_handler;
Austin Schuh58646e22021-08-23 23:51:46 -0700484 if (timer_handler_) {
485 if (event_loop_->node() != nullptr) {
486 timer_handler_->set_name(absl::StrCat(
487 event_loop_->node()->name()->string_view(), "_main"));
488 } else {
489 timer_handler_->set_name("main");
490 }
491 }
Austin Schuh858c9f32020-08-31 16:56:12 -0700492 }
493
Austin Schuhe33c08d2022-02-03 18:15:21 -0800494 // Creates and registers the --start_time and --end_time event callbacks.
495 void SetStartTimeFlag(realtime_clock::time_point start_time);
496 void SetEndTimeFlag(realtime_clock::time_point end_time);
497
498 // Notices the next message to update the start/end time callbacks.
499 void ObserveNextMessage(monotonic_clock::time_point monotonic_event,
500 realtime_clock::time_point realtime_event);
501
502 // Clears the start and end time flag handlers so we can delete the event
503 // loop.
504 void ClearTimeFlags();
505
Austin Schuh858c9f32020-08-31 16:56:12 -0700506 // Sets the next wakeup time on the replay callback.
507 void Setup(monotonic_clock::time_point next_time) {
James Kuszmaul09632422022-05-25 15:56:19 -0700508 timer_handler_->Setup(next_time + clock_offset());
Austin Schuh858c9f32020-08-31 16:56:12 -0700509 }
510
511 // Sends a buffer on the provided channel index.
Austin Schuh287d43d2020-12-04 20:19:33 -0800512 bool Send(const TimestampedMessage &timestamped_message);
Austin Schuh858c9f32020-08-31 16:56:12 -0700513
James Kuszmaul09632422022-05-25 15:56:19 -0700514 void SetClockOffset();
515 std::chrono::nanoseconds clock_offset() const { return clock_offset_; }
516
Austin Schuh858c9f32020-08-31 16:56:12 -0700517 // Returns a debug string for the channel merger.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700518 std::string DebugString() const {
Austin Schuh287d43d2020-12-04 20:19:33 -0800519 if (!timestamp_mapper_) {
Austin Schuhe639ea12021-01-25 13:00:22 -0800520 return "";
Austin Schuh287d43d2020-12-04 20:19:33 -0800521 }
Austin Schuhe639ea12021-01-25 13:00:22 -0800522 return timestamp_mapper_->DebugString();
Austin Schuh2f8fd752020-09-01 22:38:28 -0700523 }
Austin Schuh858c9f32020-08-31 16:56:12 -0700524
Austin Schuh58646e22021-08-23 23:51:46 -0700525 void ClearRemoteTimestampSenders() {
526 channel_timestamp_loggers_.clear();
527 timestamp_loggers_.clear();
528 }
529
Austin Schuhbd5f74a2021-11-11 20:55:38 -0800530 void SetFoundLastMessage(bool val) {
531 found_last_message_ = val;
532 last_message_.resize(factory_channel_index_.size(), false);
533 }
534 bool found_last_message() const { return found_last_message_; }
535
536 void set_last_message(size_t channel_index) {
537 CHECK_LT(channel_index, last_message_.size());
538 last_message_[channel_index] = true;
539 }
540
541 bool last_message(size_t channel_index) {
542 CHECK_LT(channel_index, last_message_.size());
543 return last_message_[channel_index];
544 }
545
Austin Schuh858c9f32020-08-31 16:56:12 -0700546 private:
547 // Log file.
Austin Schuh287d43d2020-12-04 20:19:33 -0800548 std::unique_ptr<TimestampMapper> timestamp_mapper_;
Austin Schuh858c9f32020-08-31 16:56:12 -0700549
Austin Schuh858c9f32020-08-31 16:56:12 -0700550 // Senders.
551 std::vector<std::unique_ptr<RawSender>> channels_;
Austin Schuh969cd602021-01-03 00:09:45 -0800552 std::vector<RemoteMessageSender *> remote_timestamp_senders_;
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700553 // The mapping from logged channel index to sent channel index. Needed for
554 // sending out MessageHeaders.
555 std::vector<int> factory_channel_index_;
556
Austin Schuh9942bae2021-01-07 22:06:44 -0800557 struct ContiguousSentTimestamp {
558 // Most timestamps make it through the network, so it saves a ton of
559 // memory and CPU to store the start and end, and search for valid ranges.
560 // For one of the logs I looked at, we had 2 ranges for 4 days.
561 //
562 // Save monotonic times as well to help if a queue index ever wraps. Odds
563 // are very low, but doesn't hurt.
564 //
565 // The starting time and matching queue index.
566 monotonic_clock::time_point starting_monotonic_event_time =
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700567 monotonic_clock::min_time;
Austin Schuh9942bae2021-01-07 22:06:44 -0800568 uint32_t starting_queue_index = 0xffffffff;
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700569
Austin Schuh9942bae2021-01-07 22:06:44 -0800570 // Ending time and queue index.
571 monotonic_clock::time_point ending_monotonic_event_time =
572 monotonic_clock::max_time;
573 uint32_t ending_queue_index = 0xffffffff;
574
575 // The queue index that the first message was *actually* sent with. The
576 // queue indices are assumed to be contiguous through this range.
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700577 uint32_t actual_queue_index = 0xffffffff;
578 };
579
580 // Stores all the timestamps that have been sent on this channel. This is
581 // only done for channels which are forwarded and on the node which
Austin Schuh9942bae2021-01-07 22:06:44 -0800582 // initially sends the message. Compress using ranges and offsets.
583 std::vector<std::unique_ptr<std::vector<ContiguousSentTimestamp>>>
584 queue_index_map_;
Austin Schuh858c9f32020-08-31 16:56:12 -0700585
586 // Factory (if we are in sim) that this loop was created on.
587 NodeEventLoopFactory *node_event_loop_factory_ = nullptr;
Austin Schuhe33c08d2022-02-03 18:15:21 -0800588 SimulatedEventLoopFactory *event_loop_factory_ = nullptr;
589
Austin Schuh858c9f32020-08-31 16:56:12 -0700590 std::unique_ptr<EventLoop> event_loop_unique_ptr_;
591 // Event loop.
Austin Schuh58646e22021-08-23 23:51:46 -0700592 const Node *node_ = nullptr;
Austin Schuh858c9f32020-08-31 16:56:12 -0700593 EventLoop *event_loop_ = nullptr;
594 // And timer used to send messages.
Austin Schuh58646e22021-08-23 23:51:46 -0700595 TimerHandler *timer_handler_ = nullptr;
596 TimerHandler *startup_timer_ = nullptr;
Austin Schuh858c9f32020-08-31 16:56:12 -0700597
Austin Schuhe33c08d2022-02-03 18:15:21 -0800598 std::unique_ptr<EventNotifier> start_event_notifier_;
599 std::unique_ptr<EventNotifier> end_event_notifier_;
600
Austin Schuh8bd96322020-02-13 21:18:22 -0800601 // Filters (or nullptr if it isn't a forwarded channel) for each channel.
602 // This corresponds to the object which is shared among all the channels
603 // going between 2 nodes. The second element in the tuple indicates if this
604 // is the primary direction or not.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700605 std::vector<message_bridge::NoncausalOffsetEstimator *> filters_;
James Kuszmaul09632422022-05-25 15:56:19 -0700606 message_bridge::MultiNodeNoncausalOffsetEstimator *multinode_filters_;
Austin Schuh8bd96322020-02-13 21:18:22 -0800607
608 // List of NodeEventLoopFactorys (or nullptr if it isn't a forwarded
609 // channel) which correspond to the originating node.
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700610 std::vector<State *> channel_source_state_;
611
Austin Schuh61e973f2021-02-21 21:43:56 -0800612 // This is a cache for channel, connection mapping to the corresponding
613 // sender.
614 absl::btree_map<std::pair<const Channel *, const Connection *>,
615 std::shared_ptr<RemoteMessageSender>>
616 channel_timestamp_loggers_;
617
618 // Mapping from resolved RemoteMessage channel to RemoteMessage sender. This
619 // is the channel that timestamps are published to.
620 absl::btree_map<const Channel *, std::shared_ptr<RemoteMessageSender>>
621 timestamp_loggers_;
Austin Schuh58646e22021-08-23 23:51:46 -0700622
James Kuszmaul09632422022-05-25 15:56:19 -0700623 // Time offset between the log's monotonic clock and the current event
624 // loop's monotonic clock. Useful when replaying logs with non-simulated
625 // event loops.
626 std::chrono::nanoseconds clock_offset_{0};
627
Austin Schuh58646e22021-08-23 23:51:46 -0700628 std::vector<std::function<void()>> on_starts_;
629 std::vector<std::function<void()>> on_ends_;
630
631 bool stopped_ = false;
632 bool started_ = false;
Austin Schuhbd5f74a2021-11-11 20:55:38 -0800633
634 bool found_last_message_ = false;
635 std::vector<bool> last_message_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800636 };
637
Austin Schuh8bd96322020-02-13 21:18:22 -0800638 // Node index -> State.
639 std::vector<std::unique_ptr<State>> states_;
640
641 // Creates the requested filter if it doesn't exist, regardless of whether
642 // these nodes can actually communicate directly. The second return value
643 // reports if this is the primary direction or not.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700644 message_bridge::NoncausalOffsetEstimator *GetFilter(const Node *node_a,
645 const Node *node_b);
Austin Schuh8bd96322020-02-13 21:18:22 -0800646
Austin Schuh8bd96322020-02-13 21:18:22 -0800647 // List of filters for a connection. The pointer to the first node will be
648 // less than the second node.
Austin Schuh0ca1fd32020-12-18 22:53:05 -0800649 std::unique_ptr<message_bridge::MultiNodeNoncausalOffsetEstimator> filters_;
Austin Schuh8bd96322020-02-13 21:18:22 -0800650
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800651 std::unique_ptr<FlatbufferDetachedBuffer<Configuration>>
652 remapped_configuration_buffer_;
653
James Kuszmaul84ff3e52020-01-03 19:48:53 -0800654 std::unique_ptr<SimulatedEventLoopFactory> event_loop_factory_unique_ptr_;
655 SimulatedEventLoopFactory *event_loop_factory_ = nullptr;
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800656
657 // Map of channel indices to new name. The channel index will be an index into
658 // logged_configuration(), and the string key will be the name of the channel
659 // to send on instead of the logged channel name.
Austin Schuh0de30f32020-12-06 12:44:28 -0800660 struct RemappedChannel {
661 std::string remapped_name;
662 std::string new_type;
663 };
664 std::map<size_t, RemappedChannel> remapped_channels_;
Austin Schuh01b4c352020-09-21 23:09:39 -0700665 std::vector<MapT> maps_;
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800666
Austin Schuh6f3babe2020-01-26 20:34:50 -0800667 // Number of nodes which still have data to send. This is used to figure out
668 // when to exit.
669 size_t live_nodes_ = 0;
670
James Kuszmaulc7bbb3e2020-01-03 20:01:00 -0800671 const Configuration *remapped_configuration_ = nullptr;
672 const Configuration *replay_configuration_ = nullptr;
Austin Schuhcde938c2020-02-02 17:30:07 -0800673
674 // If true, the replay timer will ignore any missing data. This is used
675 // during startup when we are bootstrapping everything and trying to get to
676 // the start of all the log files.
677 bool ignore_missing_data_ = false;
James Kuszmaul71a81932020-12-15 21:08:01 -0800678
679 // Whether to exit the SimulatedEventLoop when we finish reading the logs.
680 bool exit_on_finish_ = true;
Austin Schuhe33c08d2022-02-03 18:15:21 -0800681
682 realtime_clock::time_point start_time_ = realtime_clock::min_time;
683 realtime_clock::time_point end_time_ = realtime_clock::max_time;
Austin Schuhe309d2a2019-11-29 13:25:21 -0800684};
685
686} // namespace logger
687} // namespace aos
688
Austin Schuhb06f03b2021-02-17 22:00:37 -0800689#endif // AOS_EVENTS_LOGGING_LOG_READER_H_