From 31b575c25bbcfaf2a203c88e06cd87879d0237fa Mon Sep 17 00:00:00 2001
From: GroovieGermanikus <groovie@mango.markets>
Date: Fri, 17 May 2024 21:21:46 +0200
Subject: [PATCH] track recent delta and build histogram

---
 Cargo.lock                                    |  1 +
 yellowstone-grpc-geyser/Cargo.toml            |  1 +
 .../src/bin/yellowstone_mock_service.rs       |  6 ++--
 yellowstone-grpc-geyser/src/grpc.rs           | 31 ++++++++++++++++---
 yellowstone-grpc-geyser/src/lib.rs            |  1 +
 5 files changed, 34 insertions(+), 6 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 47d070e..cd7d45f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5312,6 +5312,7 @@ dependencies = [
  "git-version",
  "hostname",
  "hyper",
+ "itertools 0.10.5",
  "lazy_static",
  "log",
  "lz4_flex",
diff --git a/yellowstone-grpc-geyser/Cargo.toml b/yellowstone-grpc-geyser/Cargo.toml
index 6d25a7d..e757f56 100644
--- a/yellowstone-grpc-geyser/Cargo.toml
+++ b/yellowstone-grpc-geyser/Cargo.toml
@@ -48,6 +48,7 @@ tracing = "0.1.40"
 tracing-subscriber = { version = "0.2.15", features = ["env-filter"] }
 lz4_flex = { version = "0.11" }
 async-stream = "0.3.5"
+itertools = "0.10.5"
 
 [build-dependencies]
 anyhow = { workspace = true }
diff --git a/yellowstone-grpc-geyser/src/bin/yellowstone_mock_service.rs b/yellowstone-grpc-geyser/src/bin/yellowstone_mock_service.rs
index 98c107c..1c3b911 100644
--- a/yellowstone-grpc-geyser/src/bin/yellowstone_mock_service.rs
+++ b/yellowstone-grpc-geyser/src/bin/yellowstone_mock_service.rs
@@ -59,7 +59,9 @@ async fn mainnet_traffic(grpc_channel: UnboundedSender<Message>) {
         let sizes = vec![
             0, 8, 8, 165, 165, 165, 165, 11099, 11099, 11099, 11099, 11099, 11099,
         ];
-        const target_bytes_total: usize = 10_000_000;
+        // 10MB -> stream buffer size peaks at 30
+        // 30MB -> stream buffer size peaks at 10000th and more
+        const TARGET_BYTES_TOTAL: usize = 30_000_000;
         let mut bytes_total = 0;
 
         let mut requested_sizes: Vec<usize> = Vec::new();
@@ -67,7 +69,7 @@ async fn mainnet_traffic(grpc_channel: UnboundedSender<Message>) {
         for i in 0..99_999_999 {
             let data_size = sizes[i % sizes.len()];
 
-            if bytes_total + data_size > target_bytes_total {
+            if bytes_total + data_size > TARGET_BYTES_TOTAL {
                 break;
             }
 
diff --git a/yellowstone-grpc-geyser/src/grpc.rs b/yellowstone-grpc-geyser/src/grpc.rs
index 3c943ca..eeff6ab 100644
--- a/yellowstone-grpc-geyser/src/grpc.rs
+++ b/yellowstone-grpc-geyser/src/grpc.rs
@@ -1,3 +1,4 @@
+use std::collections::VecDeque;
 use crate::THROTTLE_ACCOUNT_LOGGING;
 use async_stream::__private::AsyncStream;
 use async_stream::stream;
@@ -5,7 +6,8 @@ use futures::{Stream, StreamExt, TryStreamExt};
 use log::{debug, warn};
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use std::time::SystemTime;
+use std::time::{SystemTime};
+use itertools::Itertools;
 use tokio::sync::mpsc::Receiver;
 use {
     crate::{
@@ -63,6 +65,7 @@ use {
         },
     },
 };
+use crate::histogram_stats_calculation::calculate_percentiles;
 
 #[derive(Debug, Clone)]
 pub struct MessageAccountInfo {
@@ -1351,10 +1354,30 @@ impl Geyser for GrpcService {
 
         let stream_tx_foo = stream_tx.clone();
         tokio::spawn(async move {
-            loop {
-                tokio::time::sleep(Duration::from_millis(20)).await;
 
-                info!("client #{id}: stream_tx fill: {}", stream_tx_foo.max_capacity() - stream_tx_foo.capacity());
+            const SIZE: usize = 1000;
+            let mut ring_buffer = VecDeque::with_capacity(SIZE);
+
+
+            let mut max_delta = 0;
+            for i in 0.. {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+
+                let delta = stream_tx_foo.max_capacity() - stream_tx_foo.capacity();
+
+                max_delta = max_delta.max(delta);
+                // info!("client #{id}: stream_tx fill: {}", delta);
+                ring_buffer.truncate(SIZE - 1);
+                ring_buffer.push_front(delta);
+
+                info!("client #{id}: stream_tx fill: {} max={}", delta, max_delta);
+                // info!("client #{id}: stream_tx fill: {:?}", ring_buffer);
+                if ring_buffer.len() == SIZE && i % 100 == 0 {
+                    let vec_float = ring_buffer.iter().sorted().map(|x| *x as f64).collect_vec();
+                    let percentiles = calculate_percentiles(&vec_float);
+                    info!("client #{id}: stream_tx fill percentiles: {}", percentiles);
+                }
+
             }
         });
 
diff --git a/yellowstone-grpc-geyser/src/lib.rs b/yellowstone-grpc-geyser/src/lib.rs
index 7a57f18..f301674 100644
--- a/yellowstone-grpc-geyser/src/lib.rs
+++ b/yellowstone-grpc-geyser/src/lib.rs
@@ -8,6 +8,7 @@ pub mod grpc;
 pub mod plugin;
 pub mod prom;
 pub mod version;
+pub mod histogram_stats_calculation;
 
 // log every X account write
 pub const THROTTLE_ACCOUNT_LOGGING: u64 = 50;