|
| 1 | +use { |
| 2 | + crate::plugin::{ |
| 3 | + filter::encoder::{AccountEncoder, TransactionEncoder}, |
| 4 | + message::Message, |
| 5 | + }, |
| 6 | + rayon::{ThreadPool, ThreadPoolBuilder}, |
| 7 | + tokio::sync::{mpsc, oneshot}, |
| 8 | +}; |
| 9 | + |
| 10 | +pub struct ParallelEncoder { |
| 11 | + tx: mpsc::UnboundedSender<EncodeRequest>, |
| 12 | +} |
| 13 | + |
| 14 | +struct EncodeRequest { |
| 15 | + batch: Vec<(u64, Message)>, |
| 16 | + response: oneshot::Sender<Vec<(u64, Message)>>, |
| 17 | +} |
| 18 | + |
| 19 | +impl ParallelEncoder { |
| 20 | + pub fn new(num_threads: usize) -> (Self, std::thread::JoinHandle<()>) { |
| 21 | + let pool = ThreadPoolBuilder::new() |
| 22 | + .num_threads(num_threads) |
| 23 | + .thread_name(|i| format!("geyser-encoder-{i}")) |
| 24 | + .build() |
| 25 | + .expect("failed to create rayon pool"); |
| 26 | + |
| 27 | + let (tx, rx) = mpsc::unbounded_channel(); |
| 28 | + |
| 29 | + let handle = std::thread::Builder::new() |
| 30 | + .name("geyser-encoder-bridge".into()) |
| 31 | + .spawn(move || Self::bridge_loop(rx, pool)) |
| 32 | + .expect("failed to spawn encoder bridge"); |
| 33 | + |
| 34 | + (Self { tx }, handle) |
| 35 | + } |
| 36 | + |
| 37 | + fn bridge_loop(mut rx: mpsc::UnboundedReceiver<EncodeRequest>, pool: ThreadPool) { |
| 38 | + use rayon::prelude::*; |
| 39 | + |
| 40 | + while let Some(req) = rx.blocking_recv() { |
| 41 | + let EncodeRequest { |
| 42 | + mut batch, |
| 43 | + response, |
| 44 | + } = req; |
| 45 | + |
| 46 | + pool.install(|| { |
| 47 | + batch.par_iter_mut().for_each(|(_msgid, msg)| { |
| 48 | + Self::encode_message(msg); |
| 49 | + }); |
| 50 | + }); |
| 51 | + |
| 52 | + let _ = response.send(batch); |
| 53 | + } |
| 54 | + |
| 55 | + log::info!("exiting encoder bridge loop"); |
| 56 | + } |
| 57 | + |
| 58 | + fn encode_message(msg: &Message) { |
| 59 | + match msg { |
| 60 | + Message::Transaction(tx) => { |
| 61 | + if tx.transaction.pre_encoded.get().is_none() { |
| 62 | + TransactionEncoder::pre_encode(&tx.transaction); |
| 63 | + } |
| 64 | + } |
| 65 | + Message::Account(acc) => { |
| 66 | + if acc.account.pre_encoded.get().is_none() { |
| 67 | + AccountEncoder::pre_encode(&acc.account); |
| 68 | + } |
| 69 | + } |
| 70 | + _ => {} |
| 71 | + } |
| 72 | + } |
| 73 | + |
| 74 | + pub async fn encode(&self, batch: Vec<(u64, Message)>) -> Vec<(u64, Message)> { |
| 75 | + if batch.len() < 4 { |
| 76 | + return Self::encode_sync(batch); |
| 77 | + } |
| 78 | + |
| 79 | + let (tx, rx) = oneshot::channel(); |
| 80 | + |
| 81 | + // move batch, don't clone |
| 82 | + if self |
| 83 | + .tx |
| 84 | + .send(EncodeRequest { |
| 85 | + batch, |
| 86 | + response: tx, |
| 87 | + }) |
| 88 | + .is_err() |
| 89 | + { |
| 90 | + // channel closed - this shouldn't happen in normal operation |
| 91 | + panic!("encoder channel closed"); |
| 92 | + } |
| 93 | + |
| 94 | + rx.await.expect("encoder response failed") |
| 95 | + } |
| 96 | + |
| 97 | + fn encode_sync(mut batch: Vec<(u64, Message)>) -> Vec<(u64, Message)> { |
| 98 | + for (_msgid, msg) in &mut batch { |
| 99 | + Self::encode_message(msg); |
| 100 | + } |
| 101 | + batch |
| 102 | + } |
| 103 | +} |
| 104 | + |
| 105 | +#[cfg(test)] |
| 106 | +mod tests { |
| 107 | + use { |
| 108 | + super::*, |
| 109 | + crate::plugin::message::{ |
| 110 | + MessageAccount, MessageAccountInfo, MessageTransaction, MessageTransactionInfo, |
| 111 | + }, |
| 112 | + bytes::Bytes, |
| 113 | + prost_types::Timestamp, |
| 114 | + solana_pubkey::Pubkey, |
| 115 | + solana_signature::Signature, |
| 116 | + std::{ |
| 117 | + sync::{Arc, OnceLock}, |
| 118 | + time::SystemTime, |
| 119 | + }, |
| 120 | + }; |
| 121 | + |
| 122 | + fn create_test_transaction() -> Message { |
| 123 | + let tx_info = MessageTransactionInfo { |
| 124 | + signature: Signature::from([1u8; 64]), |
| 125 | + is_vote: false, |
| 126 | + transaction: Default::default(), |
| 127 | + meta: Default::default(), |
| 128 | + index: 0, |
| 129 | + account_keys: Default::default(), |
| 130 | + pre_encoded: OnceLock::new(), |
| 131 | + }; |
| 132 | + Message::Transaction(MessageTransaction { |
| 133 | + transaction: Arc::new(tx_info), |
| 134 | + slot: 100, |
| 135 | + created_at: Timestamp::from(SystemTime::now()), |
| 136 | + }) |
| 137 | + } |
| 138 | + |
| 139 | + fn create_test_account() -> Message { |
| 140 | + let acc_info = MessageAccountInfo { |
| 141 | + pubkey: Pubkey::new_unique(), |
| 142 | + lamports: 1000, |
| 143 | + owner: Pubkey::new_unique(), |
| 144 | + executable: false, |
| 145 | + rent_epoch: 0, |
| 146 | + data: Bytes::from(vec![1, 2, 3]), |
| 147 | + write_version: 1, |
| 148 | + txn_signature: None, |
| 149 | + pre_encoded: OnceLock::new(), |
| 150 | + }; |
| 151 | + Message::Account(MessageAccount { |
| 152 | + account: Arc::new(acc_info), |
| 153 | + slot: 100, |
| 154 | + is_startup: false, |
| 155 | + created_at: Timestamp::from(SystemTime::now()), |
| 156 | + }) |
| 157 | + } |
| 158 | + |
| 159 | + #[tokio::test] |
| 160 | + async fn test_parallel_encoder_transactions() { |
| 161 | + let (encoder, _handle) = ParallelEncoder::new(2); |
| 162 | + |
| 163 | + let batch: Vec<(u64, Message)> = (0..10).map(|i| (i, create_test_transaction())).collect(); |
| 164 | + |
| 165 | + let encoded = encoder.encode(batch).await; |
| 166 | + |
| 167 | + assert_eq!(encoded.len(), 10); |
| 168 | + for (_msgid, msg) in encoded { |
| 169 | + if let Message::Transaction(tx) = msg { |
| 170 | + assert!( |
| 171 | + tx.transaction.pre_encoded.get().is_some(), |
| 172 | + "transaction should be encoded" |
| 173 | + ); |
| 174 | + } |
| 175 | + } |
| 176 | + } |
| 177 | + |
| 178 | + #[tokio::test] |
| 179 | + async fn test_parallel_encoder_accounts() { |
| 180 | + let (encoder, _handle) = ParallelEncoder::new(2); |
| 181 | + |
| 182 | + let batch: Vec<(u64, Message)> = (0..10).map(|i| (i, create_test_account())).collect(); |
| 183 | + |
| 184 | + let encoded = encoder.encode(batch).await; |
| 185 | + |
| 186 | + assert_eq!(encoded.len(), 10); |
| 187 | + for (_msgid, msg) in encoded { |
| 188 | + if let Message::Account(acc) = msg { |
| 189 | + assert!( |
| 190 | + acc.account.pre_encoded.get().is_some(), |
| 191 | + "account should be encoded" |
| 192 | + ); |
| 193 | + } |
| 194 | + } |
| 195 | + } |
| 196 | + |
| 197 | + #[tokio::test] |
| 198 | + async fn test_small_batch_uses_sync() { |
| 199 | + let (encoder, _handle) = ParallelEncoder::new(2); |
| 200 | + |
| 201 | + // Small batch < 4 should use sync path |
| 202 | + let batch: Vec<(u64, Message)> = (0..2).map(|i| (i, create_test_transaction())).collect(); |
| 203 | + |
| 204 | + let encoded = encoder.encode(batch).await; |
| 205 | + |
| 206 | + assert_eq!(encoded.len(), 2); |
| 207 | + } |
| 208 | + |
| 209 | + #[tokio::test] |
| 210 | + async fn test_mixed_batch() { |
| 211 | + let (encoder, _handle) = ParallelEncoder::new(2); |
| 212 | + |
| 213 | + let mut batch: Vec<(u64, Message)> = Vec::new(); |
| 214 | + for i in 0..5 { |
| 215 | + batch.push((i * 2, create_test_transaction())); |
| 216 | + batch.push((i * 2 + 1, create_test_account())); |
| 217 | + } |
| 218 | + |
| 219 | + let encoded = encoder.encode(batch).await; |
| 220 | + |
| 221 | + assert_eq!(encoded.len(), 10); |
| 222 | + } |
| 223 | +} |
0 commit comments