Add quic port for accepting transactions

using quinn library streamer: Sign TLS cert with validator identity key Handle multiple incoming chunks
sakridge · Feb 4, 2022 · b50c617 · b50c617
1 parent 93789ca
commit b50c617
Show file tree

Hide file tree

Showing 10 changed files with 660 additions and 26 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/core/src/tpu.rs b/core/src/tpu.rs
@@ -26,6 +26,7 @@ use {
         cost_model::CostModel,
         vote_sender_types::{ReplayVoteReceiver, ReplayVoteSender},
     },
+    solana_sdk::signature::Keypair,
     std::{
         net::UdpSocket,
         sync::{atomic::AtomicBool, Arc, Mutex, RwLock},
@@ -40,6 +41,7 @@ pub struct TpuSockets {
     pub transaction_forwards: Vec<UdpSocket>,
     pub vote: Vec<UdpSocket>,
     pub broadcast: Vec<UdpSocket>,
+    pub transactions_quic: UdpSocket,
 }
 
 pub struct Tpu {
@@ -49,6 +51,7 @@ pub struct Tpu {
     banking_stage: BankingStage,
     cluster_info_vote_listener: ClusterInfoVoteListener,
     broadcast_stage: BroadcastStage,
+    tpu_quic_t: thread::JoinHandle<()>,
 }
 
 impl Tpu {
@@ -75,12 +78,14 @@ impl Tpu {
         tpu_coalesce_ms: u64,
         cluster_confirmed_slot_sender: GossipDuplicateConfirmedSlotsSender,
         cost_model: &Arc<RwLock<CostModel>>,
+        keypair: &Keypair,
     ) -> Self {
         let TpuSockets {
             transactions: transactions_sockets,
             transaction_forwards: tpu_forwards_sockets,
             vote: tpu_vote_sockets,
             broadcast: broadcast_sockets,
+            transactions_quic: transactions_quic_sockets,
         } = sockets;
 
         let (packet_sender, packet_receiver) = unbounded();
@@ -97,6 +102,15 @@ impl Tpu {
         );
         let (verified_sender, verified_receiver) = unbounded();
 
+        let tpu_quic_t = solana_streamer::quic::spawn_server(
+            transactions_quic_sockets,
+            keypair,
+            cluster_info.my_contact_info().tpu.ip(),
+            packet_sender,
+            exit.clone(),
+        )
+        .unwrap();
+
         let sigverify_stage = {
             let verifier = TransactionSigVerifier::default();
             SigVerifyStage::new(packet_receiver, verified_sender, verifier)
@@ -160,6 +174,7 @@ impl Tpu {
             banking_stage,
             cluster_info_vote_listener,
             broadcast_stage,
+            tpu_quic_t,
         }
     }
 
@@ -171,6 +186,7 @@ impl Tpu {
             self.cluster_info_vote_listener.join(),
             self.banking_stage.join(),
         ];
+        self.tpu_quic_t.join()?;
         let broadcast_result = self.broadcast_stage.join();
         for result in results {
             result?;

diff --git a/core/src/validator.rs b/core/src/validator.rs
@@ -531,8 +531,11 @@ impl Validator {
             }
         }
 
-        let mut cluster_info =
-            ClusterInfo::new(node.info.clone(), identity_keypair, socket_addr_space);
+        let mut cluster_info = ClusterInfo::new(
+            node.info.clone(),
+            identity_keypair.clone(),
+            socket_addr_space,
+        );
         cluster_info.set_contact_debug_interval(config.contact_debug_interval);
         cluster_info.set_entrypoints(cluster_entrypoints);
         cluster_info.restore_contact_info(ledger_path, config.contact_save_interval);
@@ -877,6 +880,7 @@ impl Validator {
                 transaction_forwards: node.sockets.tpu_forwards,
                 vote: node.sockets.tpu_vote,
                 broadcast: node.sockets.broadcast,
+                transactions_quic: node.sockets.tpu_quic,
             },
             &rpc_subscriptions,
             transaction_status_sender,
@@ -894,6 +898,7 @@ impl Validator {
             config.tpu_coalesce_ms,
             cluster_confirmed_slot_sender,
             &cost_model,
+            &identity_keypair,
         );
 
         datapoint_info!("validator-new", ("id", id.to_string(), String));