integrating vortexor with sigverifier (#4424)

* integrating with sigverifier

sigverify integration -- move out sender creation

* changes metrics and thread name to particular to vortexor

* update comments

* made the channel bounded

Author: lijunwangs

Cargo.lock

@@ -267,7 +267,7 @@ impl BankingTracer {
         Self::channel(label, self.active_tracer.as_ref().cloned())
     }
 
-    fn create_channel_non_vote(&self) -> (BankingPacketSender, BankingPacketReceiver) {
+    pub fn create_channel_non_vote(&self) -> (BankingPacketSender, BankingPacketReceiver) {
         self.create_channel(ChannelLabel::NonVote)
     }
 

core/src/banking_trace.rs

@@ -35,6 +35,7 @@ rustls = { workspace = true }
 smallvec = { workspace = true }
 socket2 = { workspace = true }
 solana-clap-utils = { workspace = true }
+solana-core = { workspace = true }
 solana-measure = { workspace = true }
 solana-metrics = { workspace = true }
 solana-net-utils = { workspace = true }

vortexor/Cargo.toml

@@ -1,7 +1,8 @@
 use {
     clap::{value_t, value_t_or_exit},
-    crossbeam_channel::unbounded,
+    crossbeam_channel::bounded,
     solana_clap_utils::input_parsers::keypair_of,
+    solana_core::banking_trace::BankingTracer,
     solana_sdk::net::DEFAULT_TPU_COALESCE,
     solana_streamer::streamer::StakedNodes,
     solana_vortexor::{
@@ -14,6 +15,8 @@ use {
     },
 };
 
+const DEFAULT_CHANNEL_SIZE: usize = 100_000;
+
 pub fn main() {
     let default_args = DefaultArgs::default();
     let solana_version = solana_version::version!();
@@ -53,12 +56,22 @@ pub fn main() {
     let max_streams_per_ms = value_t_or_exit!(matches, "max_streams_per_ms", u64);
     let exit = Arc::new(AtomicBool::new(false));
     // To be linked with the Tpu sigverify and forwarder service
-    let (tpu_sender, _tpu_receiver) = unbounded();
-    let (tpu_fwd_sender, _tpu_fwd_receiver) = unbounded();
+    let (tpu_sender, tpu_receiver) = bounded(DEFAULT_CHANNEL_SIZE);
+    let (tpu_fwd_sender, _tpu_fwd_receiver) = bounded(DEFAULT_CHANNEL_SIZE);
 
     let tpu_sockets =
         Vortexor::create_tpu_sockets(bind_address, dynamic_port_range, num_quic_endpoints);
 
+    let (banking_tracer, _) = BankingTracer::new(
+        None, // Not interesed in banking tracing
+    )
+    .unwrap();
+
+    // The _non_vote_receiver will forward the verified transactions to its configured validator
+    let (non_vote_sender, _non_vote_receiver) = banking_tracer.create_channel_non_vote();
+
+    let sigverify_stage = Vortexor::create_sigverify_stage(tpu_receiver, non_vote_sender);
+
     // To be linked with StakedNodes service.
     let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
 
@@ -79,4 +92,5 @@ pub fn main() {
         exit,
     );
     vortexor.join().unwrap();
+    sigverify_stage.join().unwrap();
 }

vortexor/src/main.rs

@@ -1,5 +1,9 @@
 use {
-    crossbeam_channel::Sender,
+    crossbeam_channel::{Receiver, Sender},
+    solana_core::{
+        banking_trace::TracedSender, sigverify::TransactionSigVerifier,
+        sigverify_stage::SigVerifyStage,
+    },
     solana_net_utils::{bind_in_range_with_config, bind_more_with_config, SocketConfig},
     solana_perf::packet::PacketBatch,
     solana_sdk::{quic::NotifyKeyUpdate, signature::Keypair},
@@ -88,6 +92,19 @@ impl Vortexor {
         }
     }
 
+    pub fn create_sigverify_stage(
+        tpu_receiver: Receiver<solana_perf::packet::PacketBatch>,
+        non_vote_sender: TracedSender,
+    ) -> SigVerifyStage {
+        let verifier = TransactionSigVerifier::new(non_vote_sender);
+        SigVerifyStage::new(
+            tpu_receiver,
+            verifier,
+            "solSigVtxTpu",
+            "tpu-vortexor-verifier",
+        )
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub fn create_vortexor(
         tpu_sockets: TpuSockets,