Binary event tracer

CMakeLists.txt

@@ -688,6 +688,7 @@ add_library (seastar
   include/seastar/util/closeable.hh
   include/seastar/util/source_location-compat.hh
   include/seastar/util/short_streams.hh
+  include/seastar/util/trace.hh
   include/seastar/websocket/server.hh
   src/core/alien.cc
   src/core/file.cc
@@ -777,6 +778,7 @@ add_library (seastar
   src/util/read_first_line.cc
   src/util/tmp_file.cc
   src/util/short_streams.cc
+  src/util/trace.cc
   src/websocket/server.cc
   )
 

apps/CMakeLists.txt

@@ -54,3 +54,4 @@ add_subdirectory (rpc_tester)
 add_subdirectory (iotune)
 add_subdirectory (memcached)
 add_subdirectory (seawreck)
+add_subdirectory (trace)

apps/io_tester/io_tester.cc

@@ -32,6 +32,7 @@
 #include <seastar/core/loop.hh>
 #include <seastar/core/with_scheduling_group.hh>
 #include <seastar/core/metrics_api.hh>
+#include <seastar/core/io_queue.hh>
 #include <seastar/core/io_intent.hh>
 #include <seastar/util/later.hh>
 #include <chrono>
@@ -1114,6 +1115,7 @@ int main(int ac, char** av) {
         ("duration", bpo::value<unsigned>()->default_value(10), "for how long (in seconds) to run the test")
         ("conf", bpo::value<sstring>()->default_value("./conf.yaml"), "YAML file containing benchmark specification")
         ("keep-files", bpo::value<bool>()->default_value(false), "keep test files, next run may re-use them")
+        ("trace", bpo::value<bool>()->default_value(false), "start reactor tracing")
     ;
 
     distributed<context> ctx;
@@ -1136,6 +1138,7 @@ int main(int ac, char** av) {
             }
 
             keep_files = opts["keep-files"].as<bool>();
+            bool trace = opts["trace"].as<bool>();
             auto& duration = opts["duration"].as<unsigned>();
             auto& yaml = opts["conf"].as<sstring>();
             YAML::Node doc = YAML::LoadFile(yaml);
@@ -1181,6 +1184,13 @@ int main(int ac, char** av) {
             ctx.invoke_on_all([] (auto& c) {
                 return c.start();
             }).get();
+            if (trace) {
+                std::cout << "Starting tracing..." << std::endl;
+                auto st = file_stat(storage).get();
+                smp::invoke_on_all([st, duration] {
+                    return engine().get_io_queue(st.device_id).start_tracing(std::chrono::seconds(duration), 1ull << 30);
+                }).get();
+            }
             std::cout << "Starting evaluation..." << std::endl;
             ctx.invoke_on_all([] (auto& c) {
                 return c.issue_requests();

apps/trace/CMakeLists.txt

@@ -0,0 +1,24 @@
+#
+# This file is open source software, licensed to you under the terms
+# of the Apache License, Version 2.0 (the "License").  See the NOTICE file
+# distributed with this work for additional information regarding copyright
+# ownership.  You may not use this file except in compliance with the License.
+#
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+#
+# Copyright (C) 2024 Scylladb, Ltd.
+#
+
+seastar_add_app (trace_decode
+  SOURCES decode.cc)

apps/trace/decode.cc

@@ -0,0 +1,220 @@
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership.  You may not use this file except in compliance with the License.
+ *
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright (C) 2024 ScyllaDB
+ */
+#include <seastar/core/app-template.hh>
+#include <seastar/core/thread.hh>
+#include <seastar/core/reactor.hh>
+#include <seastar/core/fstream.hh>
+#include <seastar/util/trace.hh>
+#include <seastar/core/byteorder.hh>
+#include <seastar/core/internal/io_trace.hh>
+#include <seastar/core/internal/fq_trace.hh>
+
+using namespace seastar;
+
+size_t event_body_size(internal::trace_event ev) {
+    switch (ev) {
+    case internal::trace_event::TICK:
+        return 0;
+    case internal::trace_event::IO_POLL:
+        return internal::event_tracer<internal::trace_event::IO_POLL>::size();
+    case internal::trace_event::IO_QUEUE:
+        return internal::event_tracer<internal::trace_event::IO_QUEUE>::size();
+    case internal::trace_event::IO_DISPATCH:
+        return internal::event_tracer<internal::trace_event::IO_DISPATCH>::size();
+    case internal::trace_event::IO_COMPLETE:
+        return internal::event_tracer<internal::trace_event::IO_COMPLETE>::size();
+    case internal::trace_event::FQ_WAIT_CAPACITY:
+        return internal::event_tracer<internal::trace_event::FQ_WAIT_CAPACITY>::size();
+    case internal::trace_event::BUFFER_HEAD:
+    case internal::trace_event::OPENING:
+    case internal::trace_event::T800:
+        break;
+    }
+    fmt::print("Invalid event met ({})\n", static_cast<int>(ev));
+    throw std::runtime_error("invalid file content");
+    return 0;
+}
+
+template <internal::trace_event Ev>
+std::string parse(temporary_buffer<char> body);
+
+template<>
+std::string parse<internal::trace_event::TICK>(temporary_buffer<char> body) {
+    return "TICK";
+}
+
+template<>
+std::string parse<internal::trace_event::IO_POLL>(temporary_buffer<char> body) {
+    return "IO_POLL";
+}
+
+template<>
+std::string parse<internal::trace_event::IO_QUEUE>(temporary_buffer<char> body) {
+    const char* b = body.get();
+    auto rw = read_le<uint8_t>(b + 0);
+    auto cid = read_le<uint8_t>(b + 1);
+    auto len = read_le<uint16_t>(b + 2);
+    auto rq = read_le<uint32_t>(b + 4);
+    return format("IO Q {:04x} {} class {} {}", rq, rw == 0 ? "w" : "r", cid, len << 9);
+}
+
+template<>
+std::string parse<internal::trace_event::IO_DISPATCH>(temporary_buffer<char> body) {
+    const char* b = body.get();
+    auto rq = read_le<uint32_t>(b);
+    return format("IO D {:04x}", rq);
+}
+
+template<>
+std::string parse<internal::trace_event::IO_COMPLETE>(temporary_buffer<char> body) {
+    const char* b = body.get();
+    auto rq = read_le<uint32_t>(b);
+    return format("IO C {:04x}", rq);
+}
+
+template<>
+std::string parse<internal::trace_event::FQ_WAIT_CAPACITY>(temporary_buffer<char> body) {
+    return "FQ WAIT";
+}
+
+std::string parse_event(internal::trace_event ev, temporary_buffer<char> body) {
+    switch (ev) {
+    case internal::trace_event::TICK:
+        return parse<internal::trace_event::TICK>(std::move(body));
+    case internal::trace_event::IO_POLL:
+        return parse<internal::trace_event::IO_POLL>(std::move(body));
+    case internal::trace_event::IO_QUEUE:
+        return parse<internal::trace_event::IO_QUEUE>(std::move(body));
+    case internal::trace_event::IO_DISPATCH:
+        return parse<internal::trace_event::IO_DISPATCH>(std::move(body));
+    case internal::trace_event::IO_COMPLETE:
+        return parse<internal::trace_event::IO_COMPLETE>(std::move(body));
+    case internal::trace_event::FQ_WAIT_CAPACITY:
+        return parse<internal::trace_event::FQ_WAIT_CAPACITY>(std::move(body));
+    case internal::trace_event::BUFFER_HEAD:
+    case internal::trace_event::OPENING:
+    case internal::trace_event::T800:
+        break;
+    }
+    return "";
+}
+
+std::string format_ts(uint64_t timestamp) {
+    // timestamps are in usec
+    return format("{:03d}.{:06d}", timestamp/1000000, timestamp%1000000);
+}
+
+int main(int ac, char** av) {
+    namespace bpo = boost::program_options;
+
+    app_template app;
+    auto opt_add = app.add_options();
+    opt_add
+        ("file", bpo::value<sstring>()->default_value("trace.0.bin"), "file to decode")
+    ;
+
+    return app.run(ac, av, [&] {
+        return seastar::async([&] {
+            auto& opts = app.configuration();
+            auto& fname = opts["file"].as<sstring>();
+            auto f = open_file_dma(fname, open_flags::ro).get();
+            auto in = make_file_input_stream(std::move(f));
+            size_t consumed = 0;
+            uint64_t timestamp = 0;
+
+            while (true) {
+                auto tb = in.read_exactly(sizeof(uint8_t)).get();
+                if (!tb) {
+                    break;
+                }
+                const char* buf = tb.get();
+                auto ev = static_cast<internal::trace_event>(*(uint8_t*)buf);
+
+                if (ev == internal::trace_event::OPENING) {
+                    tb = in.read_exactly(sizeof(uint8_t) + sizeof(uint16_t)).get();
+                    if (!tb) {
+                        fmt::print("corrupted buffer -- no opening body\n");
+                        break;
+                    }
+                    buf = tb.get();
+                    auto shard = read_le<uint8_t>(buf);
+                    auto size = read_le<uint16_t>(buf + 1);
+
+                    tb = in.read_exactly(size).get();
+                    if (!tb) {
+                        fmt::print("corrupted buffer -- no opening payload (size={})\n", size);
+                        break;
+                    }
+                    buf = tb.get();
+                    fmt::print("{:08x}:- OPENING: shard={} text={}\n", consumed, shard, std::string(buf, size));
+                    consumed += sizeof(uint8_t) + sizeof(uint8_t) + sizeof(uint16_t) + size;
+                    continue;
+                }
+
+                if (ev == internal::trace_event::BUFFER_HEAD) {
+                    tb = in.read_exactly(2 * sizeof(uint32_t)).get();
+                    if (!tb) {
+                        fmt::print("corrupted buffer -- no buffer-head body\n");
+                        break;
+                    }
+                    buf = tb.get();
+                    auto ts = read_le<uint32_t>(buf);
+                    timestamp += ts;
+                    auto skip = read_le<uint32_t>(buf + sizeof(uint32_t));
+                    fmt::print("{:08x}:{} --- buffer (skipped {}) ---\n", consumed, format_ts(timestamp), skip);
+                    consumed = sizeof(uint8_t) + 2 * sizeof(uint32_t);
+                    continue;
+                }
+
+                if (ev == internal::trace_event::T800) {
+                    auto rem = internal::tracer::buffer_size - consumed - sizeof(uint8_t);
+                    fmt::print("{:08x}:- REM -- skip {} bytes\n", consumed, rem);
+                    if (rem > 0) {
+                        in.read_exactly(rem).get();
+                    }
+                    consumed = 0;
+                    continue;
+                }
+
+                tb = in.read_exactly(sizeof(uint16_t)).get();
+                if (!tb) {
+                    fmt::print("corrupted buffer -- no timestamp\n");
+                    break;
+                }
+                buf = tb.get();
+                auto ts = read_le<uint16_t>(buf);
+                timestamp += ts;
+
+                auto sz = event_body_size(ev);
+                if (sz > 0) {
+                    tb = in.read_exactly(sz).get();
+                    if (!tb) {
+                        fmt::print("corrupted file -- no event body\n");
+                    }
+                } else {
+                    tb = temporary_buffer<char>();
+                }
+                fmt::print("{:08x}:{} {}\n", consumed, format_ts(timestamp), parse_event(ev, std::move(tb)));
+                consumed += sz + sizeof(uint8_t) + sizeof(uint16_t);
+            }
+        });
+    });
+}

include/seastar/core/fair_queue.hh

@@ -39,6 +39,10 @@ namespace bi = boost::intrusive;
 
 namespace seastar {
 
+namespace internal {
+class tracer;
+}
+
 /// \brief describes a request that passes through the \ref fair_queue.
 ///
 /// A ticket is specified by a \c weight and a \c size. For example, one can specify a request of \c weight
@@ -350,6 +354,7 @@ private:
     };
 
     std::optional<pending> _pending;
+    internal::tracer& _tracer;
 
     void push_priority_class(priority_class_data& pc) noexcept;
     void push_priority_class_from_idle(priority_class_data& pc) noexcept;
@@ -364,7 +369,7 @@ public:
     /// Constructs a fair queue with configuration parameters \c cfg.
     ///
     /// \param cfg an instance of the class \ref config
-    explicit fair_queue(fair_group& shared, config cfg);
+    explicit fair_queue(fair_group& shared, internal::tracer&, config cfg);
     fair_queue(fair_queue&&) = delete;
     ~fair_queue();
 

include/seastar/core/internal/fq_trace.hh

@@ -0,0 +1,37 @@
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership.  You may not use this file except in compliance with the License.
+ *
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*
+ * Copyright (C) 2024 ScyllaDB.
+ */
+
+
+#pragma once
+#include <seastar/util/trace.hh>
+
+namespace seastar {
+namespace internal {
+
+template<>
+struct event_tracer<trace_event::FQ_WAIT_CAPACITY> {
+    static int size() noexcept { return 0; }
+    static void put(char* buf) noexcept { }
+};
+
+} // internal namespace
+} // seastar namespace