Add session metrics logging for statesync server

maxkozlovsky · maxkozlovsky · commit 47e9fb48a36b · 2025-11-07T11:33:26.000-05:00
Added per-request logging of disk I/O statistics, upsert counts, and
throughput metrics to enable performance monitoring and debugging of
statesync server operations.

Fix the read counter in async io to use 64 bit, 32 bit was prone to
overflow.

Add counter for number of bytes read in async io.

This code was generated using Claude Sonnet 4.5
diff --git a/category/async/io.cpp b/category/async/io.cpp
@@ -302,12 +302,13 @@ AsyncIO::~AsyncIO()
     ::close(fds_.msgwrite);
 }
 
-void AsyncIO::account_read_()
+void AsyncIO::account_read_(size_t size)
 {
     if (++records_.inflight_rd > records_.max_inflight_rd) {
         records_.max_inflight_rd = records_.inflight_rd;
     }
     ++records_.nreads;
+    records_.bytes_read += size;
 }
 
 void AsyncIO::submit_request_sqe_(
@@ -520,7 +521,7 @@ size_t AsyncIO::poll_uring_(bool blocking, unsigned poll_rings_mask)
                 auto const &read = concurrent_read_ios_pending_.front();
                 submit_request_sqe_(
                     read.buffer, read.offset, read.op, read.op->io_priority());
-                account_read_();
+                account_read_(read.buffer.size());
                 concurrent_read_ios_pending_.pop_front();
             }
         }
diff --git a/category/async/io.hpp b/category/async/io.hpp
@@ -53,9 +53,10 @@ struct IORecord
     unsigned max_inflight_rd_scatter{0};
     unsigned max_inflight_wr{0};
 
-    unsigned nreads{0};
+    uint64_t nreads{0};
     // Reads and scatter reads which got a EAGAIN and were retried
     unsigned reads_retried{0};
+    uint64_t bytes_read{0};
 };
 
 class AsyncIO final
@@ -129,7 +130,7 @@ class AsyncIO final
         void *uring_data, enum erased_connected_operation::io_priority prio);
     void submit_request_(timed_invocation_state *state, void *uring_data);
 
-    void account_read_();
+    void account_read_(size_t size);
 
     void poll_uring_while_submission_queue_full_();
     size_t poll_uring_(bool blocking, unsigned poll_rings_mask);
@@ -235,6 +236,16 @@ class AsyncIO final
         return records_.inflight_ts.load(std::memory_order_relaxed);
     }
 
+    uint64_t total_reads_submitted() const noexcept
+    {
+        return records_.nreads;
+    }
+
+    uint64_t total_bytes_read() const noexcept
+    {
+        return records_.bytes_read;
+    }
+
     unsigned concurrent_read_io_limit() const noexcept
     {
         return concurrent_read_io_limit_;
@@ -338,6 +349,8 @@ class AsyncIO final
         records_.max_inflight_rd_scatter = 0;
         records_.max_inflight_wr = 0;
         records_.nreads = 0;
+        records_.bytes_read = 0;
+        records_.reads_retried = 0;
     }
 
     size_t submit_read_request(
@@ -355,8 +368,9 @@ class AsyncIO final
             return size_t(-1); // we never complete immediately
         }
 
+        auto const size = buffer.size();
         submit_request_(buffer, offset, uring_data, uring_data->io_priority());
-        account_read_();
+        account_read_(size);
         return size_t(-1); // we never complete immediately
     }
 
@@ -659,7 +673,7 @@ class AsyncIO final
 using erased_connected_operation_ptr =
     AsyncIO::erased_connected_operation_unique_ptr_type;
 
-static_assert(sizeof(AsyncIO) == 272);
+static_assert(sizeof(AsyncIO) == 296);
 static_assert(alignof(AsyncIO) == 8);
 
 namespace detail
diff --git a/category/statesync/statesync_server.cpp b/category/statesync/statesync_server.cpp
@@ -13,6 +13,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+#include <category/async/io.hpp>
 #include <category/core/assert.h>
 #include <category/core/basic_formatter.hpp>
 #include <category/core/byte_string.hpp>
@@ -116,7 +117,8 @@ byte_string from_prefix(uint64_t const prefix, size_t const n_bytes)
 
 bool send_deletion(
     monad_statesync_server *const sync, monad_sync_request const &rq,
-    monad_statesync_server_context &ctx)
+    monad_statesync_server_context &ctx, uint64_t *const num_upserts,
+    uint64_t *const upsert_bytes)
 {
     MONAD_ASSERT(
         rq.old_target <= rq.target || rq.old_target == INVALID_BLOCK_NUM);
@@ -125,8 +127,10 @@ bool send_deletion(
         return true;
     }
 
-    auto const fn = [sync, prefix = from_prefix(rq.prefix, rq.prefix_bytes)](
-                        Deletion const &deletion) {
+    auto const fn = [sync,
+                     prefix = from_prefix(rq.prefix, rq.prefix_bytes),
+                     num_upserts,
+                     upsert_bytes](Deletion const &deletion) {
         auto const &[addr, key] = deletion;
         auto const hash = keccak256(addr.bytes);
         byte_string_view const view{hash.bytes, sizeof(hash.bytes)};
@@ -141,6 +145,8 @@ bool send_deletion(
                 sizeof(addr),
                 nullptr,
                 0);
+            ++(*num_upserts);
+            *upsert_bytes += sizeof(addr);
         }
         else {
             auto const skey = rlp::encode_bytes32_compact(key.value());
@@ -151,6 +157,8 @@ bool send_deletion(
                 sizeof(addr),
                 skey.data(),
                 skey.size());
+            ++(*num_upserts);
+            *upsert_bytes += sizeof(addr) + skey.size();
         }
     };
 
@@ -165,6 +173,17 @@ bool send_deletion(
 bool statesync_server_handle_request(
     monad_statesync_server *const sync, monad_sync_request const rq)
 {
+    uint64_t disk_ios_start = 0;
+    uint64_t disk_bytes_start = 0;
+    auto const *io = monad::async::AsyncIO::thread_instance();
+    if (io != nullptr) {
+        disk_ios_start = io->total_reads_submitted();
+        disk_bytes_start = io->total_bytes_read();
+    }
+
+    uint64_t num_upserts = 0;
+    uint64_t upsert_bytes = 0;
+
     struct Traverse final : public TraverseMachine
     {
         unsigned char nibble;
@@ -174,16 +193,21 @@ bool statesync_server_handle_request(
         NibblesView prefix;
         uint64_t from;
         uint64_t until;
+        uint64_t *num_upserts;
+        uint64_t *upsert_bytes;
 
         Traverse(
             monad_statesync_server *const sync, NibblesView const prefix,
-            uint64_t const from, uint64_t const until)
+            uint64_t const from, uint64_t const until,
+            uint64_t *const num_upserts, uint64_t *const upsert_bytes)
             : nibble{INVALID_BRANCH}
             , depth{0}
             , sync{sync}
             , prefix{prefix}
             , from{from}
             , until{until}
+            , num_upserts{num_upserts}
+            , upsert_bytes{upsert_bytes}
         {
         }
 
@@ -235,13 +259,11 @@ bool statesync_server_handle_request(
                                              unsigned char const *const v1 =
                                                  nullptr,
                                              uint64_t const size1 = 0) {
+                    uint64_t const size2 = node.value().size();
                     sync->statesync_server_send_upsert(
-                        sync->net,
-                        type,
-                        v1,
-                        size1,
-                        node.value().data(),
-                        node.value().size());
+                        sync->net, type, v1, size1, node.value().data(), size2);
+                    ++(*num_upserts);
+                    *upsert_bytes += size1 + size2;
                 };
 
                 if (nibble == CODE_NIBBLE) {
@@ -332,9 +354,11 @@ bool statesync_server_handle_request(
             val.size(),
             nullptr,
             0);
+        ++num_upserts;
+        upsert_bytes += val.size();
     }
 
-    if (!send_deletion(sync, rq, *ctx)) {
+    if (!send_deletion(sync, rq, *ctx, &num_upserts, &upsert_bytes)) {
         return false;
     }
 
@@ -350,12 +374,25 @@ bool statesync_server_handle_request(
     }
 
     [[maybe_unused]] auto const begin = std::chrono::steady_clock::now();
-    Traverse traverse(sync, NibblesView{bytes}, rq.from, rq.until);
+    Traverse traverse(
+        sync,
+        NibblesView{bytes},
+        rq.from,
+        rq.until,
+        &num_upserts,
+        &upsert_bytes);
     if (!db.traverse(finalized_root, traverse, rq.target)) {
         return false;
     }
     [[maybe_unused]] auto const end = std::chrono::steady_clock::now();
 
+    uint64_t disk_ios_submitted = 0;
+    uint64_t disk_bytes_total = 0;
+    if (io != nullptr) {
+        disk_ios_submitted = io->total_reads_submitted() - disk_ios_start;
+        disk_bytes_total = io->total_bytes_read() - disk_bytes_start;
+    }
+
     LOG_INFO(
         "processed request prefix={} prefix_bytes={} target={} from={} "
         "until={} "
@@ -369,6 +406,33 @@ bool statesync_server_handle_request(
         std::chrono::duration_cast<std::chrono::microseconds>(end - start),
         std::chrono::duration_cast<std::chrono::microseconds>(end - begin));
 
+    auto const elapsed_seconds =
+        std::chrono::duration_cast<std::chrono::duration<double>>(end - start)
+            .count();
+    double disk_ios_per_sec = 0.0;
+    double disk_bytes_per_sec = 0.0;
+    double upsert_bytes_per_sec = 0.0;
+    if (elapsed_seconds > 0.0) {
+        disk_ios_per_sec =
+            static_cast<double>(disk_ios_submitted) / elapsed_seconds;
+        disk_bytes_per_sec =
+            static_cast<double>(disk_bytes_total) / elapsed_seconds;
+        upsert_bytes_per_sec =
+            static_cast<double>(upsert_bytes) / elapsed_seconds;
+    }
+
+    LOG_INFO(
+        "session metrics: disk_ios={} disk_bytes={} num_upserts={} "
+        "upsert_bytes={} | disk_ios/s={:.1f} disk_bytes/s={:.1f} "
+        "upsert_bytes/s={:.1f}",
+        disk_ios_submitted,
+        disk_bytes_total,
+        num_upserts,
+        upsert_bytes,
+        disk_ios_per_sec,
+        disk_bytes_per_sec,
+        upsert_bytes_per_sec);
+
     return true;
 }
 
diff --git a/category/statesync/test/test_statesync.cpp b/category/statesync/test/test_statesync.cpp
@@ -1047,6 +1047,7 @@ TEST_F(StateSyncFixture, benchmark)
     handle_target(cctx, hdr);
     run();
     EXPECT_TRUE(monad_statesync_client_finalize(cctx));
+    quill::flush();
 }
 
 TEST(Deletions, history_length)

Original file line number	Diff line number	Diff line change
`@@ -302,12 +302,13 @@ AsyncIO::~AsyncIO()`
`302`	`302`	`::close(fds_.msgwrite);`
`303`	`303`	`}`
`304`	`304`
`305`		`-void AsyncIO::account_read_()`
	`305`	`+void AsyncIO::account_read_(size_t size)`
`306`	`306`	`{`
`307`	`307`	`if (++records_.inflight_rd > records_.max_inflight_rd) {`
`308`	`308`	`records_.max_inflight_rd = records_.inflight_rd;`
`309`	`309`	`}`
`310`	`310`	`++records_.nreads;`
	`311`	`+ records_.bytes_read += size;`
`311`	`312`	`}`
`312`	`313`
`313`	`314`	`void AsyncIO::submit_request_sqe_(`
`@@ -520,7 +521,7 @@ size_t AsyncIO::poll_uring_(bool blocking, unsigned poll_rings_mask)`
`520`	`521`	`auto const &read = concurrent_read_ios_pending_.front();`
`521`	`522`	`submit_request_sqe_(`
`522`	`523`	`read.buffer, read.offset, read.op, read.op->io_priority());`
`523`		`- account_read_();`
	`524`	`+ account_read_(read.buffer.size());`
`524`	`525`	`concurrent_read_ios_pending_.pop_front();`
`525`	`526`	`}`
`526`	`527`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1047,6 +1047,7 @@ TEST_F(StateSyncFixture, benchmark)`
`1047`	`1047`	`handle_target(cctx, hdr);`
`1048`	`1048`	`run();`
`1049`	`1049`	`EXPECT_TRUE(monad_statesync_client_finalize(cctx));`
	`1050`	`+ quill::flush();`
`1050`	`1051`	`}`
`1051`	`1052`
`1052`	`1053`	`TEST(Deletions, history_length)`