blob: fddda729f84752f52fcafedce3cf0f370db962ec [file] [log] [blame]
Austin Schuha36c8902019-12-30 18:07:15 -08001#ifndef AOS_EVENTS_LOGGING_LOGFILE_UTILS_H_
2#define AOS_EVENTS_LOGGING_LOGFILE_UTILS_H_
3
4#include <sys/uio.h>
5
Austin Schuh97789fc2020-08-01 14:42:45 -07006#include <chrono>
Austin Schuh05b70472020-01-01 17:11:17 -08007#include <deque>
Austin Schuh97789fc2020-08-01 14:42:45 -07008#include <limits>
9#include <memory>
Austin Schuh05b70472020-01-01 17:11:17 -080010#include <optional>
Austin Schuhfa895892020-01-07 20:07:41 -080011#include <string>
Austin Schuha36c8902019-12-30 18:07:15 -080012#include <string_view>
Brian Silverman98360e22020-04-28 16:51:20 -070013#include <tuple>
Austin Schuh97789fc2020-08-01 14:42:45 -070014#include <utility>
Austin Schuha36c8902019-12-30 18:07:15 -080015#include <vector>
16
Austin Schuh05b70472020-01-01 17:11:17 -080017#include "absl/types/span.h"
Brian Silvermanf51499a2020-09-21 12:49:08 -070018#include "aos/containers/resizeable_buffer.h"
Austin Schuha36c8902019-12-30 18:07:15 -080019#include "aos/events/event_loop.h"
Brian Silvermanf51499a2020-09-21 12:49:08 -070020#include "aos/events/logging/buffer_encoder.h"
Austin Schuha36c8902019-12-30 18:07:15 -080021#include "aos/events/logging/logger_generated.h"
Brian Silvermanf51499a2020-09-21 12:49:08 -070022#include "aos/flatbuffers.h"
Austin Schuha36c8902019-12-30 18:07:15 -080023#include "flatbuffers/flatbuffers.h"
24
Brian Silvermanf51499a2020-09-21 12:49:08 -070025namespace aos::logger {
Austin Schuha36c8902019-12-30 18:07:15 -080026
27enum class LogType : uint8_t {
28 // The message originated on this node and should be logged here.
29 kLogMessage,
30 // The message originated on another node, but only the delivery times are
31 // logged here.
32 kLogDeliveryTimeOnly,
33 // The message originated on another node. Log it and the delivery times
34 // together. The message_gateway is responsible for logging any messages
35 // which didn't get delivered.
Austin Schuh6f3babe2020-01-26 20:34:50 -080036 kLogMessageAndDeliveryTime,
37 // The message originated on the other node and should be logged on this node.
38 kLogRemoteMessage
Austin Schuha36c8902019-12-30 18:07:15 -080039};
40
Austin Schuha36c8902019-12-30 18:07:15 -080041// This class manages efficiently writing a sequence of detached buffers to a
Brian Silvermanf51499a2020-09-21 12:49:08 -070042// file. It encodes them, queues them up, and batches the write operation.
Austin Schuha36c8902019-12-30 18:07:15 -080043class DetachedBufferWriter {
44 public:
Brian Silvermana9f2ec92020-10-06 18:00:53 -070045 // Marker struct for one of our constructor overloads.
46 struct already_out_of_space_t {};
47
Brian Silvermanf51499a2020-09-21 12:49:08 -070048 DetachedBufferWriter(std::string_view filename,
49 std::unique_ptr<DetachedBufferEncoder> encoder);
Brian Silvermana9f2ec92020-10-06 18:00:53 -070050 // Creates a dummy instance which won't even open a file. It will act as if
51 // opening the file ran out of space immediately.
52 DetachedBufferWriter(already_out_of_space_t) : ran_out_of_space_(true) {}
Austin Schuh2f8fd752020-09-01 22:38:28 -070053 DetachedBufferWriter(DetachedBufferWriter &&other);
54 DetachedBufferWriter(const DetachedBufferWriter &) = delete;
55
Austin Schuha36c8902019-12-30 18:07:15 -080056 ~DetachedBufferWriter();
57
Austin Schuh2f8fd752020-09-01 22:38:28 -070058 DetachedBufferWriter &operator=(DetachedBufferWriter &&other);
Brian Silverman98360e22020-04-28 16:51:20 -070059 DetachedBufferWriter &operator=(const DetachedBufferWriter &) = delete;
60
Austin Schuh6f3babe2020-01-26 20:34:50 -080061 std::string_view filename() const { return filename_; }
62
Brian Silvermana9f2ec92020-10-06 18:00:53 -070063 // This will be true until Close() is called, unless the file couldn't be
64 // created due to running out of space.
65 bool is_open() const { return fd_ != -1; }
66
Brian Silvermanf51499a2020-09-21 12:49:08 -070067 // Queues up a finished FlatBufferBuilder to be encoded and written.
68 //
69 // Triggers a flush if there's enough data queued up.
70 //
71 // Steals the detached buffer from it.
72 void QueueSizedFlatbuffer(flatbuffers::FlatBufferBuilder *fbb) {
73 QueueSizedFlatbuffer(fbb->Release());
74 }
75 // May steal the backing storage of buffer, or may leave it alone.
76 void QueueSizedFlatbuffer(flatbuffers::DetachedBuffer &&buffer) {
Brian Silvermana9f2ec92020-10-06 18:00:53 -070077 if (ran_out_of_space_) {
78 return;
79 }
Brian Silvermanf51499a2020-09-21 12:49:08 -070080 encoder_->Encode(std::move(buffer));
81 FlushAtThreshold();
82 }
Austin Schuha36c8902019-12-30 18:07:15 -080083
Brian Silvermanf51499a2020-09-21 12:49:08 -070084 // Queues up data in span. May copy or may write it to disk immediately.
85 void QueueSpan(absl::Span<const uint8_t> span);
Austin Schuha36c8902019-12-30 18:07:15 -080086
Brian Silverman0465fcf2020-09-24 00:29:18 -070087 // Indicates we got ENOSPC when trying to write. After this returns true, no
88 // further data is written.
89 bool ran_out_of_space() const { return ran_out_of_space_; }
90
91 // To avoid silently failing to write logfiles, you must call this before
92 // destruction if ran_out_of_space() is true and the situation has been
93 // handled.
94 void acknowledge_out_of_space() {
95 CHECK(ran_out_of_space_);
96 acknowledge_ran_out_of_space_ = true;
97 }
98
99 // Fully flushes and closes the underlying file now. No additional data may be
100 // enqueued after calling this.
101 //
102 // This will be performed in the destructor automatically.
103 //
104 // Note that this may set ran_out_of_space().
105 void Close();
106
Brian Silvermanf51499a2020-09-21 12:49:08 -0700107 // Returns the total number of bytes written and currently queued.
108 size_t total_bytes() const { return encoder_->total_bytes(); }
Austin Schuha36c8902019-12-30 18:07:15 -0800109
Brian Silvermanf51499a2020-09-21 12:49:08 -0700110 // The maximum time for a single write call, or 0 if none have been performed.
111 std::chrono::nanoseconds max_write_time() const { return max_write_time_; }
112 // The number of bytes in the longest write call, or -1 if none have been
113 // performed.
114 int max_write_time_bytes() const { return max_write_time_bytes_; }
115 // The number of buffers in the longest write call, or -1 if none have been
116 // performed.
117 int max_write_time_messages() const { return max_write_time_messages_; }
118 // The total time spent in write calls.
119 std::chrono::nanoseconds total_write_time() const {
120 return total_write_time_;
121 }
122 // The total number of writes which have been performed.
123 int total_write_count() const { return total_write_count_; }
124 // The total number of messages which have been written.
125 int total_write_messages() const { return total_write_messages_; }
126 // The total number of bytes which have been written.
127 int total_write_bytes() const { return total_write_bytes_; }
128 void ResetStatistics() {
129 max_write_time_ = std::chrono::nanoseconds::zero();
130 max_write_time_bytes_ = -1;
131 max_write_time_messages_ = -1;
132 total_write_time_ = std::chrono::nanoseconds::zero();
133 total_write_count_ = 0;
134 total_write_messages_ = 0;
135 total_write_bytes_ = 0;
136 }
Brian Silverman98360e22020-04-28 16:51:20 -0700137
Austin Schuha36c8902019-12-30 18:07:15 -0800138 private:
Brian Silvermanf51499a2020-09-21 12:49:08 -0700139 // Performs a single writev call with as much of the data we have queued up as
140 // possible.
141 //
142 // This will normally take all of the data we have queued up, unless an
143 // encoder has spit out a big enough chunk all at once that we can't manage
144 // all of it.
145 void Flush();
146
Brian Silverman0465fcf2020-09-24 00:29:18 -0700147 // write_return is what write(2) or writev(2) returned. write_size is the
148 // number of bytes we expected it to write.
149 void HandleWriteReturn(ssize_t write_return, size_t write_size);
150
Brian Silvermanf51499a2020-09-21 12:49:08 -0700151 void UpdateStatsForWrite(aos::monotonic_clock::duration duration,
152 ssize_t written, int iovec_size);
153
154 // Flushes data if we've reached the threshold to do that as part of normal
155 // operation.
156 void FlushAtThreshold();
157
Austin Schuh2f8fd752020-09-01 22:38:28 -0700158 std::string filename_;
Brian Silvermanf51499a2020-09-21 12:49:08 -0700159 std::unique_ptr<DetachedBufferEncoder> encoder_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800160
Austin Schuha36c8902019-12-30 18:07:15 -0800161 int fd_ = -1;
Brian Silverman0465fcf2020-09-24 00:29:18 -0700162 bool ran_out_of_space_ = false;
163 bool acknowledge_ran_out_of_space_ = false;
Austin Schuha36c8902019-12-30 18:07:15 -0800164
Austin Schuha36c8902019-12-30 18:07:15 -0800165 // List of iovecs to use with writev. This is a member variable to avoid
166 // churn.
167 std::vector<struct iovec> iovec_;
Brian Silvermanf51499a2020-09-21 12:49:08 -0700168
169 std::chrono::nanoseconds max_write_time_ = std::chrono::nanoseconds::zero();
170 int max_write_time_bytes_ = -1;
171 int max_write_time_messages_ = -1;
172 std::chrono::nanoseconds total_write_time_ = std::chrono::nanoseconds::zero();
173 int total_write_count_ = 0;
174 int total_write_messages_ = 0;
175 int total_write_bytes_ = 0;
Austin Schuha36c8902019-12-30 18:07:15 -0800176};
177
178// Packes a message pointed to by the context into a MessageHeader.
179flatbuffers::Offset<MessageHeader> PackMessage(
180 flatbuffers::FlatBufferBuilder *fbb, const Context &context,
181 int channel_index, LogType log_type);
182
Austin Schuh6f3babe2020-01-26 20:34:50 -0800183FlatbufferVector<LogFileHeader> ReadHeader(std::string_view filename);
Austin Schuh5212cad2020-09-09 23:12:09 -0700184FlatbufferVector<MessageHeader> ReadNthMessage(std::string_view filename,
185 size_t n);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800186
Austin Schuh05b70472020-01-01 17:11:17 -0800187// Class to read chunks out of a log file.
188class SpanReader {
189 public:
190 SpanReader(std::string_view filename);
Austin Schuha36c8902019-12-30 18:07:15 -0800191
Austin Schuh6f3babe2020-01-26 20:34:50 -0800192 std::string_view filename() const { return filename_; }
193
Austin Schuh05b70472020-01-01 17:11:17 -0800194 // Returns a span with the data for a message from the log file, excluding
195 // the size.
196 absl::Span<const uint8_t> ReadMessage();
197
198 // Returns true if there is a full message available in the buffer, or if we
199 // will have to read more data from disk.
200 bool MessageAvailable();
201
202 private:
203 // TODO(austin): Optimization:
204 // Allocate the 256k blocks like we do today. But, refcount them with
205 // shared_ptr pointed to by the messageheader that is returned. This avoids
206 // the copy. Need to do more benchmarking.
Brian Silvermanf51499a2020-09-21 12:49:08 -0700207 // And (Brian): Consider just mmapping the file and handing out refcounted
208 // pointers into that too.
Austin Schuh05b70472020-01-01 17:11:17 -0800209
210 // Reads a chunk of data into data_. Returns false if no data was read.
211 bool ReadBlock();
212
Austin Schuh6f3babe2020-01-26 20:34:50 -0800213 const std::string filename_;
214
Brian Silvermanf51499a2020-09-21 12:49:08 -0700215 // File reader and data decoder.
216 std::unique_ptr<DataDecoder> decoder_;
Austin Schuh05b70472020-01-01 17:11:17 -0800217
Brian Silvermanf51499a2020-09-21 12:49:08 -0700218 // Vector to read into.
219 ResizeableBuffer data_;
Austin Schuh05b70472020-01-01 17:11:17 -0800220
221 // Amount of data consumed already in data_.
222 size_t consumed_data_ = 0;
Austin Schuh05b70472020-01-01 17:11:17 -0800223};
224
225// Class which handles reading the header and messages from the log file. This
226// handles any per-file state left before merging below.
227class MessageReader {
228 public:
229 MessageReader(std::string_view filename);
230
Austin Schuh6f3babe2020-01-26 20:34:50 -0800231 std::string_view filename() const { return span_reader_.filename(); }
232
Austin Schuh05b70472020-01-01 17:11:17 -0800233 // Returns the header from the log file.
234 const LogFileHeader *log_file_header() const {
Austin Schuh97789fc2020-08-01 14:42:45 -0700235 return &raw_log_file_header_.message();
236 }
237
238 // Returns the raw data of the header from the log file.
239 const FlatbufferVector<LogFileHeader> &raw_log_file_header() const {
240 return raw_log_file_header_;
Austin Schuh05b70472020-01-01 17:11:17 -0800241 }
242
243 // Returns the minimum maount of data needed to queue up for sorting before
244 // ware guarenteed to not see data out of order.
245 std::chrono::nanoseconds max_out_of_order_duration() const {
246 return max_out_of_order_duration_;
247 }
248
Austin Schuhcde938c2020-02-02 17:30:07 -0800249 // Returns the newest timestamp read out of the log file.
Austin Schuh05b70472020-01-01 17:11:17 -0800250 monotonic_clock::time_point newest_timestamp() const {
251 return newest_timestamp_;
252 }
253
254 // Returns the next message if there is one.
255 std::optional<FlatbufferVector<MessageHeader>> ReadMessage();
256
257 // The time at which we need to read another chunk from the logfile.
258 monotonic_clock::time_point queue_data_time() const {
259 return newest_timestamp() - max_out_of_order_duration();
260 }
261
262 private:
263 // Log chunk reader.
264 SpanReader span_reader_;
265
Austin Schuh97789fc2020-08-01 14:42:45 -0700266 // Vector holding the raw data for the log file header.
267 FlatbufferVector<LogFileHeader> raw_log_file_header_;
Austin Schuh05b70472020-01-01 17:11:17 -0800268
269 // Minimum amount of data to queue up for sorting before we are guarenteed
270 // to not see data out of order.
271 std::chrono::nanoseconds max_out_of_order_duration_;
272
273 // Timestamp of the newest message in a channel queue.
274 monotonic_clock::time_point newest_timestamp_ = monotonic_clock::min_time;
275};
276
Austin Schuh6f3babe2020-01-26 20:34:50 -0800277class TimestampMerger;
Austin Schuh05b70472020-01-01 17:11:17 -0800278
Austin Schuh6f3babe2020-01-26 20:34:50 -0800279// A design requirement is that the relevant data for a channel is not more than
280// max_out_of_order_duration out of order. We approach sorting in layers.
281//
282// 1) Split each (maybe chunked) log file into one queue per channel. Read this
283// log file looking for data pertaining to a specific node.
284// (SplitMessageReader)
285// 2) Merge all the data per channel from the different log files into a sorted
286// list of timestamps and messages. (TimestampMerger)
287// 3) Combine the timestamps and messages. (TimestampMerger)
288// 4) Merge all the channels to produce the next message on a node.
289// (ChannelMerger)
290// 5) Duplicate this entire stack per node.
291
292// This class splits messages and timestamps up into a queue per channel, and
293// handles reading data from multiple chunks.
294class SplitMessageReader {
295 public:
296 SplitMessageReader(const std::vector<std::string> &filenames);
297
298 // Sets the TimestampMerger that gets notified for each channel. The node
299 // that the TimestampMerger is merging as needs to be passed in.
300 void SetTimestampMerger(TimestampMerger *timestamp_merger, int channel,
301 const Node *target_node);
302
Austin Schuh2f8fd752020-09-01 22:38:28 -0700303 // Returns the (timestamp, queue_index, message_header) for the oldest message
304 // in a channel, or max_time if there is nothing in the channel.
Austin Schuhcde938c2020-02-02 17:30:07 -0800305 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
306 oldest_message(int channel) {
Austin Schuh6f3babe2020-01-26 20:34:50 -0800307 return channels_[channel].data.front_timestamp();
308 }
309
Austin Schuh2f8fd752020-09-01 22:38:28 -0700310 // Returns the (timestamp, queue_index, message_header) for the oldest
311 // delivery time in a channel, or max_time if there is nothing in the channel.
Austin Schuhcde938c2020-02-02 17:30:07 -0800312 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
313 oldest_message(int channel, int destination_node) {
Austin Schuh6f3babe2020-01-26 20:34:50 -0800314 return channels_[channel].timestamps[destination_node].front_timestamp();
315 }
316
317 // Returns the timestamp, queue_index, and message for the oldest data on a
318 // channel. Requeues data as needed.
319 std::tuple<monotonic_clock::time_point, uint32_t,
320 FlatbufferVector<MessageHeader>>
321 PopOldest(int channel_index);
322
323 // Returns the timestamp, queue_index, and message for the oldest timestamp on
324 // a channel delivered to a node. Requeues data as needed.
325 std::tuple<monotonic_clock::time_point, uint32_t,
326 FlatbufferVector<MessageHeader>>
Austin Schuh2f8fd752020-09-01 22:38:28 -0700327 PopOldestTimestamp(int channel, int node_index);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800328
329 // Returns the header for the log files.
Austin Schuh05b70472020-01-01 17:11:17 -0800330 const LogFileHeader *log_file_header() const {
Austin Schuhfa895892020-01-07 20:07:41 -0800331 return &log_file_header_.message();
Austin Schuh05b70472020-01-01 17:11:17 -0800332 }
333
Austin Schuh97789fc2020-08-01 14:42:45 -0700334 const FlatbufferVector<LogFileHeader> &raw_log_file_header() const {
335 return log_file_header_;
336 }
337
Austin Schuh6f3babe2020-01-26 20:34:50 -0800338 // Returns the starting time for this set of log files.
Austin Schuh05b70472020-01-01 17:11:17 -0800339 monotonic_clock::time_point monotonic_start_time() {
340 return monotonic_clock::time_point(
341 std::chrono::nanoseconds(log_file_header()->monotonic_start_time()));
342 }
343 realtime_clock::time_point realtime_start_time() {
344 return realtime_clock::time_point(
345 std::chrono::nanoseconds(log_file_header()->realtime_start_time()));
346 }
347
Austin Schuh6f3babe2020-01-26 20:34:50 -0800348 // Returns the configuration from the log file header.
349 const Configuration *configuration() const {
350 return log_file_header()->configuration();
351 }
352
Austin Schuh05b70472020-01-01 17:11:17 -0800353 // Returns the node who's point of view this log file is from. Make sure this
354 // is a pointer in the configuration() nodes list so it can be consumed
355 // elsewhere.
356 const Node *node() const {
357 if (configuration()->has_nodes()) {
Austin Schuh6f3babe2020-01-26 20:34:50 -0800358 return configuration::GetNodeOrDie(configuration(),
359 log_file_header()->node());
Austin Schuh05b70472020-01-01 17:11:17 -0800360 } else {
361 CHECK(!log_file_header()->has_node());
362 return nullptr;
363 }
364 }
365
Austin Schuh6f3babe2020-01-26 20:34:50 -0800366 // Returns the timestamp of the newest message read from the log file, and the
367 // timestamp that we need to re-queue data.
368 monotonic_clock::time_point newest_timestamp() const {
Austin Schuhcde938c2020-02-02 17:30:07 -0800369 return newest_timestamp_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800370 }
371
Austin Schuhcde938c2020-02-02 17:30:07 -0800372 // Returns the next time to trigger a requeue.
373 monotonic_clock::time_point time_to_queue() const { return time_to_queue_; }
374
375 // Returns the minimum amount of data needed to queue up for sorting before
376 // ware guarenteed to not see data out of order.
377 std::chrono::nanoseconds max_out_of_order_duration() const {
378 return message_reader_->max_out_of_order_duration();
379 }
380
381 std::string_view filename() const { return message_reader_->filename(); }
Austin Schuh6f3babe2020-01-26 20:34:50 -0800382
383 // Adds more messages to the sorted list. This reads enough data such that
384 // oldest_message_time can be replayed safely. Returns false if the log file
385 // has all been read.
386 bool QueueMessages(monotonic_clock::time_point oldest_message_time);
Austin Schuh05b70472020-01-01 17:11:17 -0800387
Austin Schuhcde938c2020-02-02 17:30:07 -0800388 // Returns debug strings for a channel, and timestamps for a node.
389 std::string DebugString(int channel) const;
390 std::string DebugString(int channel, int node_index) const;
391
Austin Schuh8bd96322020-02-13 21:18:22 -0800392 // Returns true if all the messages have been queued from the last log file in
393 // the list of log files chunks.
394 bool at_end() const { return at_end_; }
395
Austin Schuh05b70472020-01-01 17:11:17 -0800396 private:
Austin Schuh6f3babe2020-01-26 20:34:50 -0800397 // TODO(austin): Need to copy or refcount the message instead of running
398 // multiple copies of the reader. Or maybe have a "as_node" index and hide it
399 // inside.
400
Austin Schuhfa895892020-01-07 20:07:41 -0800401 // Moves to the next log file in the list.
402 bool NextLogFile();
403
Austin Schuh6f3babe2020-01-26 20:34:50 -0800404 // Filenames of the log files.
405 std::vector<std::string> filenames_;
406 // And the index of the next file to open.
407 size_t next_filename_index_ = 0;
Austin Schuh05b70472020-01-01 17:11:17 -0800408
Austin Schuhee711052020-08-24 16:06:09 -0700409 // Node we are reading as.
410 const Node *target_node_ = nullptr;
411
Austin Schuh6f3babe2020-01-26 20:34:50 -0800412 // Log file header to report. This is a copy.
Austin Schuh97789fc2020-08-01 14:42:45 -0700413 FlatbufferVector<LogFileHeader> log_file_header_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800414 // Current log file being read.
415 std::unique_ptr<MessageReader> message_reader_;
Austin Schuh05b70472020-01-01 17:11:17 -0800416
417 // Datastructure to hold the list of messages, cached timestamp for the
418 // oldest message, and sender to send with.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800419 struct MessageHeaderQueue {
420 // If true, this is a timestamp queue.
421 bool timestamps = false;
Austin Schuh05b70472020-01-01 17:11:17 -0800422
Austin Schuh6f3babe2020-01-26 20:34:50 -0800423 // Returns a reference to the the oldest message.
424 FlatbufferVector<MessageHeader> &front() {
425 CHECK_GT(data_.size(), 0u);
426 return data_.front();
Austin Schuh05b70472020-01-01 17:11:17 -0800427 }
Austin Schuh6f3babe2020-01-26 20:34:50 -0800428
Austin Schuhcde938c2020-02-02 17:30:07 -0800429 // Adds a message to the back of the queue. Returns true if it was actually
430 // emplaced.
431 bool emplace_back(FlatbufferVector<MessageHeader> &&msg);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800432
433 // Drops the front message. Invalidates the front() reference.
Austin Schuh2f8fd752020-09-01 22:38:28 -0700434 void PopFront();
Austin Schuh6f3babe2020-01-26 20:34:50 -0800435
436 // The size of the queue.
437 size_t size() { return data_.size(); }
438
Austin Schuhcde938c2020-02-02 17:30:07 -0800439 // Returns a debug string with info about each message in the queue.
440 std::string DebugString() const;
441
Austin Schuh2f8fd752020-09-01 22:38:28 -0700442 // Returns the (timestamp, queue_index, message_header) for the oldest
443 // message.
Austin Schuhcde938c2020-02-02 17:30:07 -0800444 const std::tuple<monotonic_clock::time_point, uint32_t,
445 const MessageHeader *>
446 front_timestamp() {
Austin Schuh2f8fd752020-09-01 22:38:28 -0700447 const MessageHeader &message = front().message();
Austin Schuh6f3babe2020-01-26 20:34:50 -0800448 return std::make_tuple(
Austin Schuh2f8fd752020-09-01 22:38:28 -0700449 monotonic_clock::time_point(
450 std::chrono::nanoseconds(message.monotonic_sent_time())),
451 message.queue_index(), &message);
Austin Schuh6f3babe2020-01-26 20:34:50 -0800452 }
453
454 // Pointer to the timestamp merger for this queue if available.
455 TimestampMerger *timestamp_merger = nullptr;
456 // Pointer to the reader which feeds this queue.
457 SplitMessageReader *split_reader = nullptr;
458
459 private:
460 // The data.
461 std::deque<FlatbufferVector<MessageHeader>> data_;
Austin Schuh05b70472020-01-01 17:11:17 -0800462 };
463
Austin Schuh6f3babe2020-01-26 20:34:50 -0800464 // All the queues needed for a channel. There isn't going to be data in all
465 // of these.
466 struct ChannelData {
467 // The data queue for the channel.
468 MessageHeaderQueue data;
469 // Queues for timestamps for each node.
470 std::vector<MessageHeaderQueue> timestamps;
471 };
Austin Schuhfa895892020-01-07 20:07:41 -0800472
Austin Schuh6f3babe2020-01-26 20:34:50 -0800473 // Data for all the channels.
Austin Schuh05b70472020-01-01 17:11:17 -0800474 std::vector<ChannelData> channels_;
475
Austin Schuh6f3babe2020-01-26 20:34:50 -0800476 // Once we know the node that this SplitMessageReader will be writing as,
477 // there will be only one MessageHeaderQueue that a specific channel matches.
478 // Precompute this here for efficiency.
479 std::vector<MessageHeaderQueue *> channels_to_write_;
480
Austin Schuhcde938c2020-02-02 17:30:07 -0800481 monotonic_clock::time_point time_to_queue_ = monotonic_clock::min_time;
482
483 // Latches true when we hit the end of the last log file and there is no sense
484 // poking it further.
485 bool at_end_ = false;
486
487 // Timestamp of the newest message that was read and actually queued. We want
488 // to track this independently from the log file because we need the
489 // timestamps here to be timestamps of messages that are queued.
490 monotonic_clock::time_point newest_timestamp_ = monotonic_clock::min_time;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800491};
492
493class ChannelMerger;
494
495// Sorts channels (and timestamps) from multiple log files for a single channel.
496class TimestampMerger {
497 public:
498 TimestampMerger(const Configuration *configuration,
499 std::vector<SplitMessageReader *> split_message_readers,
500 int channel_index, const Node *target_node,
501 ChannelMerger *channel_merger);
502
503 // Metadata used to schedule the message.
504 struct DeliveryTimestamp {
505 monotonic_clock::time_point monotonic_event_time =
506 monotonic_clock::min_time;
507 realtime_clock::time_point realtime_event_time = realtime_clock::min_time;
Austin Schuh8d7e0bb2020-10-02 17:57:00 -0700508 uint32_t queue_index = 0xffffffff;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800509
510 monotonic_clock::time_point monotonic_remote_time =
511 monotonic_clock::min_time;
512 realtime_clock::time_point realtime_remote_time = realtime_clock::min_time;
513 uint32_t remote_queue_index = 0xffffffff;
514 };
515
516 // Pushes SplitMessageReader onto the timestamp heap. This should only be
517 // called when timestamps are placed in the channel this class is merging for
518 // the reader.
519 void UpdateTimestamp(
520 SplitMessageReader *split_message_reader,
Austin Schuhcde938c2020-02-02 17:30:07 -0800521 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
522 oldest_message_time) {
Austin Schuh6f3babe2020-01-26 20:34:50 -0800523 PushTimestampHeap(oldest_message_time, split_message_reader);
524 }
525 // Pushes SplitMessageReader onto the message heap. This should only be
526 // called when data is placed in the channel this class is merging for the
527 // reader.
528 void Update(
529 SplitMessageReader *split_message_reader,
Austin Schuhcde938c2020-02-02 17:30:07 -0800530 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
531 oldest_message_time) {
Austin Schuh6f3babe2020-01-26 20:34:50 -0800532 PushMessageHeap(oldest_message_time, split_message_reader);
533 }
534
Austin Schuhcde938c2020-02-02 17:30:07 -0800535 // Returns the oldest combined timestamp and data for this channel. If there
536 // isn't a matching piece of data, returns only the timestamp with no data.
537 // The caller can determine what the appropriate action is to recover.
Austin Schuh6f3babe2020-01-26 20:34:50 -0800538 std::tuple<DeliveryTimestamp, FlatbufferVector<MessageHeader>> PopOldest();
539
540 // Tracks if the channel merger has pushed this onto it's heap or not.
541 bool pushed() { return pushed_; }
542 // Sets if this has been pushed to the channel merger heap. Should only be
543 // called by the channel merger.
544 void set_pushed(bool pushed) { pushed_ = pushed; }
545
Austin Schuhcde938c2020-02-02 17:30:07 -0800546 // Returns a debug string with the heaps printed out.
547 std::string DebugString() const;
548
Austin Schuh8bd96322020-02-13 21:18:22 -0800549 // Returns true if we have timestamps.
550 bool has_timestamps() const { return has_timestamps_; }
551
552 // Records that one of the log files ran out of data. This should only be
553 // called by a SplitMessageReader.
554 void NoticeAtEnd();
555
Austin Schuh2f8fd752020-09-01 22:38:28 -0700556 aos::monotonic_clock::time_point channel_merger_time() {
557 if (has_timestamps_) {
558 return std::get<0>(timestamp_heap_[0]);
559 } else {
560 return std::get<0>(message_heap_[0]);
561 }
562 }
563
Austin Schuh6f3babe2020-01-26 20:34:50 -0800564 private:
565 // Pushes messages and timestamps to the corresponding heaps.
566 void PushMessageHeap(
Austin Schuhcde938c2020-02-02 17:30:07 -0800567 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
568 timestamp,
Austin Schuh6f3babe2020-01-26 20:34:50 -0800569 SplitMessageReader *split_message_reader);
570 void PushTimestampHeap(
Austin Schuhcde938c2020-02-02 17:30:07 -0800571 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
572 timestamp,
Austin Schuh6f3babe2020-01-26 20:34:50 -0800573 SplitMessageReader *split_message_reader);
574
575 // Pops a message from the message heap. This automatically triggers the
576 // split message reader to re-fetch any new data.
577 std::tuple<monotonic_clock::time_point, uint32_t,
578 FlatbufferVector<MessageHeader>>
579 PopMessageHeap();
Austin Schuhcde938c2020-02-02 17:30:07 -0800580
581 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
582 oldest_message() const;
583 std::tuple<monotonic_clock::time_point, uint32_t, const MessageHeader *>
584 oldest_timestamp() const;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800585 // Pops a message from the timestamp heap. This automatically triggers the
586 // split message reader to re-fetch any new data.
587 std::tuple<monotonic_clock::time_point, uint32_t,
588 FlatbufferVector<MessageHeader>>
589 PopTimestampHeap();
590
591 const Configuration *configuration_;
592
593 // If true, this is a forwarded channel and timestamps should be matched.
594 bool has_timestamps_ = false;
595
596 // Tracks if the ChannelMerger has pushed this onto it's queue.
597 bool pushed_ = false;
598
599 // The split message readers used for source data.
600 std::vector<SplitMessageReader *> split_message_readers_;
601
602 // The channel to merge.
603 int channel_index_;
604
605 // Our node.
606 int node_index_;
607
608 // Heaps for messages and timestamps.
609 std::vector<
610 std::tuple<monotonic_clock::time_point, uint32_t, SplitMessageReader *>>
611 message_heap_;
612 std::vector<
613 std::tuple<monotonic_clock::time_point, uint32_t, SplitMessageReader *>>
614 timestamp_heap_;
615
616 // Parent channel merger.
617 ChannelMerger *channel_merger_;
618};
619
620// This class handles constructing all the split message readers, channel
621// mergers, and combining the results.
622class ChannelMerger {
623 public:
624 // Builds a ChannelMerger around a set of log files. These are of the format:
625 // {
626 // {log1_part0, log1_part1, ...},
627 // {log2}
628 // }
629 // The inner vector is a list of log file chunks which form up a log file.
630 // The outer vector is a list of log files with subsets of the messages, or
631 // messages from different nodes.
632 ChannelMerger(const std::vector<std::vector<std::string>> &filenames);
633
634 // Returns the nodes that we know how to merge.
635 const std::vector<const Node *> nodes() const;
636 // Sets the node that we will return messages as. Returns true if the node
637 // has log files and will produce data. This can only be called once, and
638 // will likely corrupt state if called a second time.
639 bool SetNode(const Node *target_node);
640
641 // Everything else needs the node set before it works.
642
643 // Returns a timestamp for the oldest message in this group of logfiles.
Austin Schuh858c9f32020-08-31 16:56:12 -0700644 monotonic_clock::time_point OldestMessageTime() const;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800645 // Pops the oldest message.
646 std::tuple<TimestampMerger::DeliveryTimestamp, int,
647 FlatbufferVector<MessageHeader>>
648 PopOldest();
649
650 // Returns the config for this set of log files.
651 const Configuration *configuration() const {
652 return log_file_header()->configuration();
653 }
654
655 const LogFileHeader *log_file_header() const {
656 return &log_file_header_.message();
657 }
658
659 // Returns the start times for the configured node's log files.
Austin Schuhcde938c2020-02-02 17:30:07 -0800660 monotonic_clock::time_point monotonic_start_time() const {
Austin Schuh6f3babe2020-01-26 20:34:50 -0800661 return monotonic_clock::time_point(
662 std::chrono::nanoseconds(log_file_header()->monotonic_start_time()));
663 }
Austin Schuhcde938c2020-02-02 17:30:07 -0800664 realtime_clock::time_point realtime_start_time() const {
Austin Schuh6f3babe2020-01-26 20:34:50 -0800665 return realtime_clock::time_point(
666 std::chrono::nanoseconds(log_file_header()->realtime_start_time()));
667 }
668
669 // Returns the node set by SetNode above.
670 const Node *node() const { return node_; }
671
672 // Called by the TimestampMerger when new data is available with the provided
673 // timestamp and channel_index.
674 void Update(monotonic_clock::time_point timestamp, int channel_index) {
675 PushChannelHeap(timestamp, channel_index);
676 }
677
Austin Schuhcde938c2020-02-02 17:30:07 -0800678 // Returns a debug string with all the heaps in it. Generally only useful for
679 // debugging what went wrong.
680 std::string DebugString() const;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800681
Austin Schuh8bd96322020-02-13 21:18:22 -0800682 // Returns true if one of the log files has finished reading everything. When
683 // log file chunks are involved, this means that the last chunk in a log file
684 // has been read. It is acceptable to be missing data at this point in time.
685 bool at_end() const { return at_end_; }
686
687 // Marks that one of the log files is at the end. This should only be called
688 // by timestamp mergers.
689 void NoticeAtEnd() { at_end_ = true; }
690
Austin Schuhcde938c2020-02-02 17:30:07 -0800691 private:
Austin Schuh6f3babe2020-01-26 20:34:50 -0800692 // Pushes the timestamp for new data on the provided channel.
693 void PushChannelHeap(monotonic_clock::time_point timestamp,
694 int channel_index);
695
Austin Schuh2f8fd752020-09-01 22:38:28 -0700696 // CHECKs that channel_heap_ and timestamp_heap_ are valid heaps.
697 void VerifyHeaps();
698
Austin Schuh6f3babe2020-01-26 20:34:50 -0800699 // All the message readers.
700 std::vector<std::unique_ptr<SplitMessageReader>> split_message_readers_;
701
702 // The log header we are claiming to be.
Austin Schuh97789fc2020-08-01 14:42:45 -0700703 FlatbufferVector<LogFileHeader> log_file_header_;
Austin Schuh6f3babe2020-01-26 20:34:50 -0800704
705 // The timestamp mergers which combine data from the split message readers.
706 std::vector<TimestampMerger> timestamp_mergers_;
707
708 // A heap of the channel readers and timestamps for the oldest data in each.
Austin Schuh05b70472020-01-01 17:11:17 -0800709 std::vector<std::pair<monotonic_clock::time_point, int>> channel_heap_;
710
Austin Schuh6f3babe2020-01-26 20:34:50 -0800711 // Configured node.
712 const Node *node_;
713
Austin Schuh8bd96322020-02-13 21:18:22 -0800714 bool at_end_ = false;
715
Austin Schuh6f3babe2020-01-26 20:34:50 -0800716 // Cached copy of the list of nodes.
717 std::vector<const Node *> nodes_;
Austin Schuh2f8fd752020-09-01 22:38:28 -0700718
719 // Last time popped. Used to detect events being returned out of order.
720 monotonic_clock::time_point last_popped_time_ = monotonic_clock::min_time;
Austin Schuh05b70472020-01-01 17:11:17 -0800721};
Austin Schuha36c8902019-12-30 18:07:15 -0800722
Austin Schuhee711052020-08-24 16:06:09 -0700723// Returns the node name with a trailing space, or an empty string if we are on
724// a single node.
725std::string MaybeNodeName(const Node *);
726
Brian Silvermanf51499a2020-09-21 12:49:08 -0700727} // namespace aos::logger
Austin Schuha36c8902019-12-30 18:07:15 -0800728
729#endif // AOS_EVENTS_LOGGING_LOGFILE_UTILS_H_