Reapply "str-288: parallel encoding with Rayon threadpool and plugin lifecycle management (#661)"

This reverts commit 04c54df.

fix: attach parallel encoding to all broadcast_tx.send() paths

Author: Ozodimgba

Cargo.lock

@@ -54,6 +54,7 @@ prost = "0.14.0"
 prost-types = "0.14.0"
 prost_011 = { package = "prost", version = "0.11.9" }
 protoc-bin-vendored = "3.2.0"
+rayon = "1.11.0"
 serde = "1.0.145"
 serde_json = "1.0.86"
 smallvec = "1.15.1"

Cargo.toml

@@ -39,6 +39,7 @@ pin-project = { workspace = true }
 prometheus = { workspace = true }
 prost = { workspace = true }
 prost-types = { workspace = true }
+rayon = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }

yellowstone-grpc-geyser/Cargo.toml

@@ -213,6 +213,12 @@ pub struct ConfigGrpc {
         deserialize_with = "deserialize_int_str"
     )]
     pub replay_stored_slots: u64,
+    /// Number of threads for parallel encoding
+    #[serde(
+        default = "ConfigGrpc::encoder_threads_default",
+        deserialize_with = "deserialize_int_str"
+    )]
+    pub encoder_threads: usize,
     #[serde(default)]
     pub server_http2_adaptive_window: Option<bool>,
     #[serde(default, with = "humantime_serde")]
@@ -272,6 +278,10 @@ impl ConfigGrpc {
     const fn default_replay_stored_slots() -> u64 {
         0
     }
+
+    const fn encoder_threads_default() -> usize {
+        4
+    }
 }
 
 #[derive(Debug, Clone, Deserialize)]

yellowstone-grpc-geyser/src/config.rs

@@ -4,8 +4,9 @@ use {
         metered::MeteredLayer,
         metrics::{
             self, incr_grpc_method_call_count, set_subscriber_queue_size,
-            subscription_limit_exceeded_inc, DebugClientMessage,
+            subscription_limit_exceeded_inc, DebugClientMessage, GEYSER_BATCH_SIZE,
         },
+        parallel::ParallelEncoder,
         plugin::{
             filter::{
                 limits::FilterLimits,
@@ -516,6 +517,7 @@ impl GrpcService {
         is_reload: bool,
         service_cancellation_token: CancellationToken,
         task_tracker: TaskTracker,
+        parallel_encoder: ParallelEncoder,
     ) -> anyhow::Result<(
         Option<crossbeam_channel::Sender<Box<Message>>>,
         mpsc::UnboundedSender<Message>,
@@ -637,6 +639,7 @@ impl GrpcService {
                 replay_stored_slots_rx,
                 replay_first_available_slot,
                 config.replay_stored_slots,
+                parallel_encoder,
             )
             .await;
         });
@@ -679,6 +682,7 @@ impl GrpcService {
         replay_stored_slots_rx: Option<mpsc::Receiver<ReplayStoredSlotsRequest>>,
         replay_first_available_slot: Option<Arc<AtomicU64>>,
         replay_stored_slots: u64,
+        parallel_encoder: ParallelEncoder,
     ) {
         const PROCESSED_MESSAGES_MAX: usize = 31;
         const PROCESSED_MESSAGES_SLEEP: Duration = Duration::from_millis(10);
@@ -923,8 +927,10 @@ impl GrpcService {
 
                             // processed
                             processed_messages.push(message.clone());
+                            GEYSER_BATCH_SIZE.observe(processed_messages.len() as f64);
+                            let encoded = parallel_encoder.encode(processed_messages).await;
                             let _ =
-                                broadcast_tx.send((CommitmentLevel::Processed, processed_messages.into()));
+                                broadcast_tx.send((CommitmentLevel::Processed, encoded.into()));
                             processed_messages = Vec::with_capacity(PROCESSED_MESSAGES_MAX);
                             processed_sleep
                                 .as_mut()
@@ -962,8 +968,10 @@ impl GrpcService {
                                 || !confirmed_messages.is_empty()
                                 || !finalized_messages.is_empty()
                             {
+                                GEYSER_BATCH_SIZE.observe(processed_messages.len() as f64);
+                                let encoded = parallel_encoder.encode(processed_messages).await;
                                 let _ = broadcast_tx
-                                    .send((CommitmentLevel::Processed, processed_messages.into()));
+                                    .send((CommitmentLevel::Processed, encoded.into()));
                                 processed_messages = Vec::with_capacity(PROCESSED_MESSAGES_MAX);
                                 processed_sleep
                                     .as_mut()
@@ -984,7 +992,9 @@ impl GrpcService {
                 }
                 () = &mut processed_sleep => {
                     if !processed_messages.is_empty() {
-                        let _ = broadcast_tx.send((CommitmentLevel::Processed, processed_messages.into()));
+                        GEYSER_BATCH_SIZE.observe(processed_messages.len() as f64);
+                        let encoded = parallel_encoder.encode(processed_messages).await;
+                        let _ = broadcast_tx.send((CommitmentLevel::Processed, encoded.into()));
                         processed_messages = Vec::with_capacity(PROCESSED_MESSAGES_MAX);
                     }
                     processed_sleep.as_mut().reset(Instant::now() + PROCESSED_MESSAGES_SLEEP);

yellowstone-grpc-geyser/src/grpc.rs

@@ -2,6 +2,7 @@ pub mod config;
 pub mod grpc;
 pub mod metered;
 pub mod metrics;
+pub mod parallel;
 pub mod plugin;
 pub mod transport;
 pub(crate) mod util;

yellowstone-grpc-geyser/src/lib.rs

@@ -104,14 +104,6 @@ lazy_static::lazy_static! {
         &["subscriber_id"]
     ).unwrap();
 
-    static ref GRPC_SUBCRIBER_RX_LOAD: IntGaugeVec = IntGaugeVec::new(
-        Opts::new(
-            "grpc_subscriber_recv_bandwidth_load",
-            "Current Receiver rate of subscriber channel (in bytes per second)"
-        ),
-        &["subscriber_id"]
-    ).unwrap();
-
     static ref GRPC_CLIENT_DISCONNECTS: IntCounterVec = IntCounterVec::new(
         Opts::new(
             "grpc_client_disconnects_total",
@@ -146,56 +138,12 @@ lazy_static::lazy_static! {
         &["type"]
     ).unwrap();
 
-    static ref PRE_ENCODED_CACHE_SKIP: IntCounterVec = IntCounterVec::new(
-        Opts::new("yellowstone_grpc_pre_encoded_cache_skip", "Pre-encoded cache skips due to unempty data slices"),
-        &["type"]
-    ).unwrap();
-
-    static ref PRE_ENCODE_POPULATED_ON_ARRIVAL: IntCounterVec = IntCounterVec::new(
-    Opts::new("yellowstone_grpc_pre_encode_populated_on_arrival", "OnceLock already populated when subscriber encodes"),
-    &["type"]
-).unwrap();
-
-static ref PRE_ENCODE_EMPTY_ON_ARRIVAL: IntCounterVec = IntCounterVec::new(
-    Opts::new("yellowstone_grpc_pre_encode_empty_on_arrival", "OnceLock empty when subscriber encodes"),
-    &["type"]
-).unwrap();
-
-static ref PARALLEL_ENCODER_LATENCY_US: Histogram = Histogram::with_opts(
-    HistogramOpts::new(
-        "yellowstone_grpc_parallel_encoder_latency_us",
-        "Time spent in parallel encoder encode() in microseconds"
-    )
-    .buckets(vec![10.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0, 10000.0])
-).unwrap();
-
-static ref PARALLEL_ENCODER_QUEUE_DEPTH: IntGauge = IntGauge::new(
-    "yellowstone_grpc_parallel_encoder_queue_depth",
-    "Number of batches waiting to be encoded"
-).unwrap();
-
-static ref FAST_PATH_TIME_US: Histogram = Histogram::with_opts(
-    HistogramOpts::new(
-        "yellowstone_grpc_fast_path_time_us",
-        "Time spent in pre-encoded cache hit path in microseconds"
-    )
-    .buckets(vec![0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 25.0, 50.0])
-).unwrap();
-
-static ref FALLBACK_ENCODE_TIME_US: Histogram = Histogram::with_opts(
-    HistogramOpts::new(
-        "yellowstone_grpc_fallback_encode_time_us",
-        "Time spent in fallback encode path in microseconds"
-    )
-    .buckets(vec![1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0])
-).unwrap();
-
-static ref GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION: IntGaugeVec = IntGaugeVec::new(
-        Opts::new(
-            "grpc_concurrent_subscribe_per_tcp_connection",
-            "Current concurrent subscriptions per remote TCP peer socket address"
-        ),
-        &["remote_peer_sk_addr"]
+    static ref GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION: IntGaugeVec = IntGaugeVec::new(
+            Opts::new(
+                "grpc_concurrent_subscribe_per_tcp_connection",
+                "Current concurrent subscriptions per remote TCP peer socket address"
+            ),
+            &["remote_peer_sk_addr"]
     ).unwrap();
 
     static ref TOTAL_TRAFFIC_SENT: IntCounter = IntCounter::new(
@@ -374,19 +322,11 @@ impl PrometheusService {
             register!(GRPC_BYTES_SENT);
             register!(GEYSER_ACCOUNT_UPDATE_RECEIVED);
             register!(GRPC_SUBSCRIBER_SEND_BANDWIDTH_LOAD);
-            register!(GRPC_SUBCRIBER_RX_LOAD);
             register!(GRPC_SUBSCRIBER_QUEUE_SIZE);
             register!(GEYSER_BATCH_SIZE);
             register!(GRPC_CLIENT_DISCONNECTS);
             register!(PRE_ENCODED_CACHE_HIT);
             register!(PRE_ENCODED_CACHE_MISS);
-            register!(PRE_ENCODED_CACHE_SKIP);
-            register!(PRE_ENCODE_POPULATED_ON_ARRIVAL);
-            register!(PRE_ENCODE_EMPTY_ON_ARRIVAL);
-            register!(PARALLEL_ENCODER_LATENCY_US);
-            register!(PARALLEL_ENCODER_QUEUE_DEPTH);
-            register!(FAST_PATH_TIME_US);
-            register!(FALLBACK_ENCODE_TIME_US);
             register!(GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION);
             register!(TOTAL_TRAFFIC_SENT);
             register!(TRAFFIC_SENT_PER_REMOTE_IP);
@@ -599,12 +539,6 @@ pub fn set_subscriber_send_bandwidth_load<S: AsRef<str>>(subscriber_id: S, load:
         .set(load);
 }
 
-pub fn set_subscriber_recv_bandwidth_load<S: AsRef<str>>(subscriber_id: S, load: i64) {
-    GRPC_SUBCRIBER_RX_LOAD
-        .with_label_values(&[subscriber_id.as_ref()])
-        .set(load);
-}
-
 pub fn set_subscriber_queue_size<S: AsRef<str>>(subscriber_id: S, size: u64) {
     GRPC_SUBSCRIBER_QUEUE_SIZE
         .with_label_values(&[subscriber_id.as_ref()])
@@ -625,42 +559,6 @@ pub fn pre_encoded_cache_miss(msg_type: &str) {
     PRE_ENCODED_CACHE_MISS.with_label_values(&[msg_type]).inc();
 }
 
-pub fn pre_encoded_cache_skip(msg_type: &str) {
-    PRE_ENCODED_CACHE_SKIP.with_label_values(&[msg_type]).inc();
-}
-
-pub fn pre_encode_populated_on_arrival(msg_type: &str) {
-    PRE_ENCODE_POPULATED_ON_ARRIVAL
-        .with_label_values(&[msg_type])
-        .inc();
-}
-
-pub fn pre_encode_empty_on_arrival(msg_type: &str) {
-    PRE_ENCODE_EMPTY_ON_ARRIVAL
-        .with_label_values(&[msg_type])
-        .inc();
-}
-
-pub fn observe_parallel_encoder_latency_us(latency: f64) {
-    PARALLEL_ENCODER_LATENCY_US.observe(latency);
-}
-
-pub fn encoder_queue_depth_inc() {
-    PARALLEL_ENCODER_QUEUE_DEPTH.inc();
-}
-
-pub fn encoder_queue_depth_dec() {
-    PARALLEL_ENCODER_QUEUE_DEPTH.dec();
-}
-
-pub fn observe_fast_path_time_us(latency: f64) {
-    FAST_PATH_TIME_US.observe(latency);
-}
-
-pub fn observe_fallback_encode_time_us(latency: f64) {
-    FALLBACK_ENCODE_TIME_US.observe(latency);
-}
-
 pub fn incr_grpc_method_call_count<S: AsRef<str>>(method: S) {
     GRPC_METHOD_CALL_COUNT
         .with_label_values(&[method.as_ref()])
@@ -726,7 +624,6 @@ pub fn reset_metrics() {
     INVALID_FULL_BLOCKS.reset();
     GRPC_SUBSCRIBER_SEND_BANDWIDTH_LOAD.reset();
     GRPC_SUBSCRIBER_QUEUE_SIZE.reset();
-    GRPC_SUBCRIBER_RX_LOAD.reset();
 
     // Reset counter vectors (clears all label combinations)
     MISSED_STATUS_MESSAGE.reset();
@@ -740,6 +637,10 @@ pub fn reset_metrics() {
     GRPC_SUBSCRIPTION_LIMIT_EXCEEDED.reset();
     GRPC_METHOD_CALL_COUNT.reset();
 
+    // Pre-encoding
+    PRE_ENCODED_CACHE_HIT.reset();
+    PRE_ENCODED_CACHE_MISS.reset();
+
     // Note: VERSION and GEYSER_ACCOUNT_UPDATE_RECEIVED are intentionally not reset
     // - VERSION contains build info set once on startup
     // - GEYSER_ACCOUNT_UPDATE_RECEIVED is a Histogram which doesn't support reset()