Austin Schuh | b06f03b | 2021-02-17 22:00:37 -0800 | [diff] [blame] | 1 | #ifndef AOS_EVENTS_LOGGING_LOG_READER_H_ |
| 2 | #define AOS_EVENTS_LOGGING_LOG_READER_H_ |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 3 | |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 4 | #include <chrono> |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 5 | #include <deque> |
Austin Schuh | 05b7047 | 2020-01-01 17:11:17 -0800 | [diff] [blame] | 6 | #include <string_view> |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 7 | #include <tuple> |
Austin Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 8 | #include <vector> |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 9 | |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 10 | #include "aos/events/event_loop.h" |
Austin Schuh | f6f9bf3 | 2020-10-11 14:37:43 -0700 | [diff] [blame] | 11 | #include "aos/events/logging/logfile_sorting.h" |
Austin Schuh | a36c890 | 2019-12-30 18:07:15 -0800 | [diff] [blame] | 12 | #include "aos/events/logging/logfile_utils.h" |
James Kuszmaul | 38735e8 | 2019-12-07 16:42:06 -0800 | [diff] [blame] | 13 | #include "aos/events/logging/logger_generated.h" |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 14 | #include "aos/events/shm_event_loop.h" |
Austin Schuh | 9254752 | 2019-12-28 14:33:43 -0800 | [diff] [blame] | 15 | #include "aos/events/simulated_event_loop.h" |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 16 | #include "aos/network/message_bridge_server_generated.h" |
Austin Schuh | 0ca1fd3 | 2020-12-18 22:53:05 -0800 | [diff] [blame] | 17 | #include "aos/network/multinode_timestamp_filter.h" |
Austin Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 18 | #include "aos/network/remote_message_generated.h" |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 19 | #include "aos/network/timestamp_filter.h" |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 20 | #include "aos/time/time.h" |
Austin Schuh | 4385b14 | 2021-03-14 21:31:13 -0700 | [diff] [blame] | 21 | #include "aos/uuid.h" |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 22 | #include "flatbuffers/flatbuffers.h" |
| 23 | |
| 24 | namespace aos { |
| 25 | namespace logger { |
| 26 | |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 27 | class EventNotifier; |
| 28 | |
Austin Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 29 | // 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 Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 57 | // Replays all the channels in the logfile to the event loop. |
| 58 | class LogReader { |
| 59 | public: |
James Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 60 | // 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 Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 64 | // |
Austin Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 65 | // The single file constructor calls SortParts internally. |
James Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 66 | LogReader(std::string_view filename, |
| 67 | const Configuration *replay_configuration = nullptr); |
Austin Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 68 | LogReader(std::vector<LogFile> log_files, |
Austin Schuh | 11d4373 | 2020-09-21 17:28:30 -0700 | [diff] [blame] | 69 | const Configuration *replay_configuration = nullptr); |
James Kuszmaul | 7daef36 | 2019-12-31 18:28:17 -0800 | [diff] [blame] | 70 | ~LogReader(); |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 71 | |
Austin Schuh | 6331ef9 | 2020-01-07 18:28:09 -0800 | [diff] [blame] | 72 | // 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 Kuszmaul | 84ff3e5 | 2020-01-03 19:48:53 -0800 | [diff] [blame] | 78 | void Register(SimulatedEventLoopFactory *event_loop_factory); |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 79 | |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 80 | // 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 Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 84 | // 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 Schuh | 6331ef9 | 2020-01-07 18:28:09 -0800 | [diff] [blame] | 93 | // Creates an SimulatedEventLoopFactory accessible via event_loop_factory(), |
| 94 | // and then calls Register. |
| 95 | void Register(); |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 96 | |
Austin Schuh | 6331ef9 | 2020-01-07 18:28:09 -0800 | [diff] [blame] | 97 | // Registers callbacks for all the events after the log file starts. This is |
| 98 | // only useful when replaying live. |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 99 | void Register(EventLoop *event_loop); |
Austin Schuh | 6331ef9 | 2020-01-07 18:28:09 -0800 | [diff] [blame] | 100 | |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 101 | // 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 Kuszmaul | 84ff3e5 | 2020-01-03 19:48:53 -0800 | [diff] [blame] | 108 | // 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 Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 112 | void Deregister(); |
| 113 | |
Austin Schuh | 0c29701 | 2020-09-16 18:41:59 -0700 | [diff] [blame] | 114 | // 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 Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 119 | const Configuration *logged_configuration() const; |
Austin Schuh | 11d4373 | 2020-09-21 17:28:30 -0700 | [diff] [blame] | 120 | // 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 Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 125 | // The pointer is invalidated whenever RemapLoggedChannel is called. |
Austin Schuh | 15649d6 | 2019-12-28 16:36:38 -0800 | [diff] [blame] | 126 | const Configuration *configuration() const; |
| 127 | |
Austin Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 128 | // Returns the nodes that this log file was created on. This is a list of |
Austin Schuh | 0767662 | 2021-01-21 18:59:17 -0800 | [diff] [blame] | 129 | // pointers to a node in the nodes() list inside logged_configuration(). |
| 130 | std::vector<const Node *> LoggedNodes() const; |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 131 | |
| 132 | // Returns the starting timestamp for the log file. |
Austin Schuh | 11d4373 | 2020-09-21 17:28:30 -0700 | [diff] [blame] | 133 | 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 Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 137 | |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 138 | // 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 Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 146 | // 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 Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 151 | std::string_view add_prefix = "/original", |
| 152 | std::string_view new_type = ""); |
James Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 153 | template <typename T> |
| 154 | void RemapLoggedChannel(std::string_view name, |
Austin Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 155 | std::string_view add_prefix = "/original", |
| 156 | std::string_view new_type = "") { |
| 157 | RemapLoggedChannel(name, T::GetFullyQualifiedName(), add_prefix, new_type); |
James Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 158 | } |
| 159 | |
Austin Schuh | 01b4c35 | 2020-09-21 23:09:39 -0700 | [diff] [blame] | 160 | // 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 Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 163 | // 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 Schuh | 01b4c35 | 2020-09-21 23:09:39 -0700 | [diff] [blame] | 165 | // |
| 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 Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 171 | std::string_view add_prefix = "/original", |
| 172 | std::string_view new_type = ""); |
Brian Silverman | de9f3ff | 2020-04-28 16:56:58 -0700 | [diff] [blame] | 173 | template <typename T> |
Austin Schuh | 01b4c35 | 2020-09-21 23:09:39 -0700 | [diff] [blame] | 174 | void RemapLoggedChannel(std::string_view name, const Node *node, |
Austin Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 175 | std::string_view add_prefix = "/original", |
| 176 | std::string_view new_type = "") { |
| 177 | RemapLoggedChannel(name, T::GetFullyQualifiedName(), node, add_prefix, |
| 178 | new_type); |
Austin Schuh | 01b4c35 | 2020-09-21 23:09:39 -0700 | [diff] [blame] | 179 | } |
| 180 | |
| 181 | template <typename T> |
| 182 | bool HasChannel(std::string_view name, const Node *node = nullptr) { |
Austin Schuh | 0ca51f3 | 2020-12-25 21:51:45 -0800 | [diff] [blame] | 183 | return configuration::GetChannel(logged_configuration(), name, |
Austin Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 184 | T::GetFullyQualifiedName(), "", node, |
| 185 | true) != nullptr; |
Brian Silverman | de9f3ff | 2020-04-28 16:56:58 -0700 | [diff] [blame] | 186 | } |
| 187 | |
Austin Schuh | 8252906 | 2021-12-08 12:09:52 -0800 | [diff] [blame] | 188 | 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 Kuszmaul | 4f106fb | 2021-01-05 20:53:02 -0800 | [diff] [blame] | 196 | // 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 Schuh | 5ee5687 | 2021-01-30 16:53:34 -0800 | [diff] [blame] | 199 | const Channel *channel = |
| 200 | configuration::GetChannel(logged_configuration(), name, |
| 201 | T::GetFullyQualifiedName(), "", node, true); |
James Kuszmaul | 4f106fb | 2021-01-05 20:53:02 -0800 | [diff] [blame] | 202 | if (channel == nullptr) return false; |
| 203 | return channel->logger() != LoggerConfig::NOT_LOGGED; |
| 204 | } |
| 205 | |
Austin Schuh | 1c22735 | 2021-09-17 12:53:54 -0700 | [diff] [blame] | 206 | // Returns a list of all the original channels from remapping. |
| 207 | std::vector<const Channel *> RemappedChannels() const; |
| 208 | |
James Kuszmaul | 84ff3e5 | 2020-01-03 19:48:53 -0800 | [diff] [blame] | 209 | SimulatedEventLoopFactory *event_loop_factory() { |
| 210 | return event_loop_factory_; |
| 211 | } |
| 212 | |
Austin Schuh | 0ca51f3 | 2020-12-25 21:51:45 -0800 | [diff] [blame] | 213 | std::string_view name() const { return log_files_[0].name; } |
Austin Schuh | 0c29701 | 2020-09-16 18:41:59 -0700 | [diff] [blame] | 214 | |
James Kuszmaul | 71a8193 | 2020-12-15 21:08:01 -0800 | [diff] [blame] | 215 | // 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 Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 221 | private: |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 222 | 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 Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 227 | const Channel *channel); |
| 228 | |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 229 | // Queues at least max_out_of_order_duration_ messages into channels_. |
| 230 | void QueueMessages(); |
James Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 231 | // Handle constructing a configuration with all the additional remapped |
| 232 | // channels from calls to RemapLoggedChannel. |
| 233 | void MakeRemappedConfig(); |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 234 | |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 235 | // 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 Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 242 | const std::vector<LogFile> log_files_; |
Austin Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 243 | |
Austin Schuh | 969cd60 | 2021-01-03 00:09:45 -0800 | [diff] [blame] | 244 | // Class to manage sending RemoteMessages on the provided node after the |
| 245 | // correct delay. |
Austin Schuh | 5ee5687 | 2021-01-30 16:53:34 -0800 | [diff] [blame] | 246 | class RemoteMessageSender { |
Austin Schuh | 969cd60 | 2021-01-03 00:09:45 -0800 | [diff] [blame] | 247 | 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 257 | BootTimestamp monotonic_timestamp_time, size_t source_boot_count); |
Austin Schuh | 969cd60 | 2021-01-03 00:09:45 -0800 | [diff] [blame] | 258 | |
| 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 Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 287 | // State per node. |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 288 | class State { |
| 289 | public: |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 290 | State(std::unique_ptr<TimestampMapper> timestamp_mapper, |
| 291 | message_bridge::MultiNodeNoncausalOffsetEstimator *multinode_filters, |
| 292 | const Node *node); |
Austin Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 293 | |
| 294 | // Connects up the timestamp mappers. |
| 295 | void AddPeer(State *peer); |
Austin Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 296 | |
Austin Schuh | e639ea1 | 2021-01-25 13:00:22 -0800 | [diff] [blame] | 297 | TimestampMapper *timestamp_mapper() { return timestamp_mapper_.get(); } |
| 298 | |
Austin Schuh | dda74ec | 2021-01-03 19:30:37 -0800 | [diff] [blame] | 299 | // Returns the next sorted message with all the timestamps extracted and |
| 300 | // matched. |
| 301 | TimestampedMessage PopOldest(); |
Austin Schuh | 188eabe | 2020-12-29 23:41:13 -0800 | [diff] [blame] | 302 | |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 303 | // Returns the monotonic time of the oldest message. |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 304 | BootTimestamp OldestMessageTime(); |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 305 | |
| 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 Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 309 | 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 320 | return node_event_loop_factory_->boot_count(); |
| 321 | } |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 322 | |
| 323 | // Primes the queues inside State. Should be called before calling |
| 324 | // OldestMessageTime. |
| 325 | void SeedSortedMessages(); |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 326 | |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 327 | 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 Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 333 | NotifyLogfileStart(); |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 334 | return; |
| 335 | } |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 336 | 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 340 | } |
| 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 354 | // Returns the starting time for this node. |
Austin Schuh | 2dc8c7d | 2021-07-01 17:41:28 -0700 | [diff] [blame] | 355 | 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 359 | } |
Austin Schuh | 2dc8c7d | 2021-07-01 17:41:28 -0700 | [diff] [blame] | 360 | 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 364 | } |
| 365 | |
| 366 | // Sets the node event loop factory for replaying into a |
| 367 | // SimulatedEventLoopFactory. Returns the EventLoop to use. |
Austin Schuh | 60e7794 | 2022-05-16 17:48:24 -0700 | [diff] [blame] | 368 | void SetNodeEventLoopFactory(NodeEventLoopFactory *node_event_loop_factory, |
| 369 | SimulatedEventLoopFactory *event_loop_factory); |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 370 | |
| 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 375 | 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 382 | // 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 Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 392 | // Returns the MessageHeader sender to log delivery timestamps to for the |
| 393 | // provided remote node. |
Austin Schuh | 61e973f | 2021-02-21 21:43:56 -0800 | [diff] [blame] | 394 | RemoteMessageSender *RemoteTimestampSender(const Channel *channel, |
| 395 | const Connection *connection); |
Austin Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 396 | |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 397 | // 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 Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 401 | CHECK(node_event_loop_factory_); |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 402 | return node_event_loop_factory_->ToDistributedClock(time); |
| 403 | } |
| 404 | |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 405 | // Returns the current time on the remote node which sends messages on |
| 406 | // channel_index. |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 407 | 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 412 | } |
| 413 | |
Austin Schuh | 5ee5687 | 2021-01-30 16:53:34 -0800 | [diff] [blame] | 414 | // Returns the start time of the remote for the provided channel. |
| 415 | monotonic_clock::time_point monotonic_remote_start_time( |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 416 | size_t boot_count, size_t channel_index) { |
Austin Schuh | 2dc8c7d | 2021-07-01 17:41:28 -0700 | [diff] [blame] | 417 | return channel_source_state_[channel_index]->monotonic_start_time( |
| 418 | boot_count); |
Austin Schuh | 5ee5687 | 2021-01-30 16:53:34 -0800 | [diff] [blame] | 419 | } |
| 420 | |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 421 | void DestroyEventLoop() { event_loop_unique_ptr_.reset(); } |
| 422 | |
| 423 | EventLoop *MakeEventLoop() { |
| 424 | CHECK(!event_loop_unique_ptr_); |
James Kuszmaul | 890c249 | 2022-04-06 14:59:31 -0700 | [diff] [blame] | 425 | // 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 430 | return event_loop_unique_ptr_.get(); |
| 431 | } |
| 432 | |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 433 | distributed_clock::time_point RemoteToDistributedClock( |
| 434 | size_t channel_index, monotonic_clock::time_point time) { |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 435 | CHECK(node_event_loop_factory_); |
Austin Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 436 | return channel_source_state_[channel_index] |
| 437 | ->node_event_loop_factory_->ToDistributedClock(time); |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 438 | } |
| 439 | |
| 440 | const Node *remote_node(size_t channel_index) { |
Austin Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 441 | return channel_source_state_[channel_index] |
| 442 | ->node_event_loop_factory_->node(); |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 443 | } |
| 444 | |
Stephan Pleines | 559fa6c | 2022-01-06 17:23:51 -0800 | [diff] [blame] | 445 | monotonic_clock::time_point monotonic_now() const { |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 446 | return event_loop_->monotonic_now(); |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 447 | } |
| 448 | |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 449 | // Sets the number of channels. |
| 450 | void SetChannelCount(size_t count); |
| 451 | |
| 452 | // Sets the sender, filter, and target factory for a channel. |
Austin Schuh | 969cd60 | 2021-01-03 00:09:45 -0800 | [diff] [blame] | 453 | void SetChannel(size_t logged_channel_index, size_t factory_channel_index, |
| 454 | std::unique_ptr<RawSender> sender, |
| 455 | message_bridge::NoncausalOffsetEstimator *filter, |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 456 | 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 463 | |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 464 | // 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 476 | // Unregisters everything so we can destory the event loop. |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 477 | // TODO(austin): Is this needed? OnShutdown should be able to serve this |
| 478 | // need. |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 479 | 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 484 | 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 492 | } |
| 493 | |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 494 | // 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 506 | // Sets the next wakeup time on the replay callback. |
| 507 | void Setup(monotonic_clock::time_point next_time) { |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 508 | timer_handler_->Setup(next_time + clock_offset()); |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 509 | } |
| 510 | |
| 511 | // Sends a buffer on the provided channel index. |
Austin Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 512 | bool Send(const TimestampedMessage ×tamped_message); |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 513 | |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 514 | void SetClockOffset(); |
| 515 | std::chrono::nanoseconds clock_offset() const { return clock_offset_; } |
| 516 | |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 517 | // Returns a debug string for the channel merger. |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 518 | std::string DebugString() const { |
Austin Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 519 | if (!timestamp_mapper_) { |
Austin Schuh | e639ea1 | 2021-01-25 13:00:22 -0800 | [diff] [blame] | 520 | return ""; |
Austin Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 521 | } |
Austin Schuh | e639ea1 | 2021-01-25 13:00:22 -0800 | [diff] [blame] | 522 | return timestamp_mapper_->DebugString(); |
Austin Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 523 | } |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 524 | |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 525 | void ClearRemoteTimestampSenders() { |
| 526 | channel_timestamp_loggers_.clear(); |
| 527 | timestamp_loggers_.clear(); |
| 528 | } |
| 529 | |
Austin Schuh | bd5f74a | 2021-11-11 20:55:38 -0800 | [diff] [blame] | 530 | 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 Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 546 | private: |
| 547 | // Log file. |
Austin Schuh | 287d43d | 2020-12-04 20:19:33 -0800 | [diff] [blame] | 548 | std::unique_ptr<TimestampMapper> timestamp_mapper_; |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 549 | |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 550 | // Senders. |
| 551 | std::vector<std::unique_ptr<RawSender>> channels_; |
Austin Schuh | 969cd60 | 2021-01-03 00:09:45 -0800 | [diff] [blame] | 552 | std::vector<RemoteMessageSender *> remote_timestamp_senders_; |
Austin Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 553 | // 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 Schuh | 9942bae | 2021-01-07 22:06:44 -0800 | [diff] [blame] | 557 | 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 Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 567 | monotonic_clock::min_time; |
Austin Schuh | 9942bae | 2021-01-07 22:06:44 -0800 | [diff] [blame] | 568 | uint32_t starting_queue_index = 0xffffffff; |
Austin Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 569 | |
Austin Schuh | 9942bae | 2021-01-07 22:06:44 -0800 | [diff] [blame] | 570 | // 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 Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 577 | 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 Schuh | 9942bae | 2021-01-07 22:06:44 -0800 | [diff] [blame] | 582 | // initially sends the message. Compress using ranges and offsets. |
| 583 | std::vector<std::unique_ptr<std::vector<ContiguousSentTimestamp>>> |
| 584 | queue_index_map_; |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 585 | |
| 586 | // Factory (if we are in sim) that this loop was created on. |
| 587 | NodeEventLoopFactory *node_event_loop_factory_ = nullptr; |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 588 | SimulatedEventLoopFactory *event_loop_factory_ = nullptr; |
| 589 | |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 590 | std::unique_ptr<EventLoop> event_loop_unique_ptr_; |
| 591 | // Event loop. |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 592 | const Node *node_ = nullptr; |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 593 | EventLoop *event_loop_ = nullptr; |
| 594 | // And timer used to send messages. |
Austin Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 595 | TimerHandler *timer_handler_ = nullptr; |
| 596 | TimerHandler *startup_timer_ = nullptr; |
Austin Schuh | 858c9f3 | 2020-08-31 16:56:12 -0700 | [diff] [blame] | 597 | |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 598 | std::unique_ptr<EventNotifier> start_event_notifier_; |
| 599 | std::unique_ptr<EventNotifier> end_event_notifier_; |
| 600 | |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 601 | // 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 Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 605 | std::vector<message_bridge::NoncausalOffsetEstimator *> filters_; |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 606 | message_bridge::MultiNodeNoncausalOffsetEstimator *multinode_filters_; |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 607 | |
| 608 | // List of NodeEventLoopFactorys (or nullptr if it isn't a forwarded |
| 609 | // channel) which correspond to the originating node. |
Austin Schuh | 8d7e0bb | 2020-10-02 17:57:00 -0700 | [diff] [blame] | 610 | std::vector<State *> channel_source_state_; |
| 611 | |
Austin Schuh | 61e973f | 2021-02-21 21:43:56 -0800 | [diff] [blame] | 612 | // 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 622 | |
James Kuszmaul | 0963242 | 2022-05-25 15:56:19 -0700 | [diff] [blame^] | 623 | // 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 Schuh | 58646e2 | 2021-08-23 23:51:46 -0700 | [diff] [blame] | 628 | 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 Schuh | bd5f74a | 2021-11-11 20:55:38 -0800 | [diff] [blame] | 633 | |
| 634 | bool found_last_message_ = false; |
| 635 | std::vector<bool> last_message_; |
Austin Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 636 | }; |
| 637 | |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 638 | // 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 Schuh | 2f8fd75 | 2020-09-01 22:38:28 -0700 | [diff] [blame] | 644 | message_bridge::NoncausalOffsetEstimator *GetFilter(const Node *node_a, |
| 645 | const Node *node_b); |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 646 | |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 647 | // List of filters for a connection. The pointer to the first node will be |
| 648 | // less than the second node. |
Austin Schuh | 0ca1fd3 | 2020-12-18 22:53:05 -0800 | [diff] [blame] | 649 | std::unique_ptr<message_bridge::MultiNodeNoncausalOffsetEstimator> filters_; |
Austin Schuh | 8bd9632 | 2020-02-13 21:18:22 -0800 | [diff] [blame] | 650 | |
James Kuszmaul | 84ff3e5 | 2020-01-03 19:48:53 -0800 | [diff] [blame] | 651 | std::unique_ptr<FlatbufferDetachedBuffer<Configuration>> |
| 652 | remapped_configuration_buffer_; |
| 653 | |
James Kuszmaul | 84ff3e5 | 2020-01-03 19:48:53 -0800 | [diff] [blame] | 654 | std::unique_ptr<SimulatedEventLoopFactory> event_loop_factory_unique_ptr_; |
| 655 | SimulatedEventLoopFactory *event_loop_factory_ = nullptr; |
James Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 656 | |
| 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 Schuh | 0de30f3 | 2020-12-06 12:44:28 -0800 | [diff] [blame] | 660 | struct RemappedChannel { |
| 661 | std::string remapped_name; |
| 662 | std::string new_type; |
| 663 | }; |
| 664 | std::map<size_t, RemappedChannel> remapped_channels_; |
Austin Schuh | 01b4c35 | 2020-09-21 23:09:39 -0700 | [diff] [blame] | 665 | std::vector<MapT> maps_; |
James Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 666 | |
Austin Schuh | 6f3babe | 2020-01-26 20:34:50 -0800 | [diff] [blame] | 667 | // 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 Kuszmaul | c7bbb3e | 2020-01-03 20:01:00 -0800 | [diff] [blame] | 671 | const Configuration *remapped_configuration_ = nullptr; |
| 672 | const Configuration *replay_configuration_ = nullptr; |
Austin Schuh | cde938c | 2020-02-02 17:30:07 -0800 | [diff] [blame] | 673 | |
| 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 Kuszmaul | 71a8193 | 2020-12-15 21:08:01 -0800 | [diff] [blame] | 678 | |
| 679 | // Whether to exit the SimulatedEventLoop when we finish reading the logs. |
| 680 | bool exit_on_finish_ = true; |
Austin Schuh | e33c08d | 2022-02-03 18:15:21 -0800 | [diff] [blame] | 681 | |
| 682 | realtime_clock::time_point start_time_ = realtime_clock::min_time; |
| 683 | realtime_clock::time_point end_time_ = realtime_clock::max_time; |
Austin Schuh | e309d2a | 2019-11-29 13:25:21 -0800 | [diff] [blame] | 684 | }; |
| 685 | |
| 686 | } // namespace logger |
| 687 | } // namespace aos |
| 688 | |
Austin Schuh | b06f03b | 2021-02-17 22:00:37 -0800 | [diff] [blame] | 689 | #endif // AOS_EVENTS_LOGGING_LOG_READER_H_ |