Add beginning of relayer (solana-labs#1)

Cargo.toml

@@ -3,5 +3,6 @@ members = [
     "core",
     "jito-protos",
     "transaction-relayer",
-    "rpc"
+    "rpc",
+    "relayer"
 ]

core/src/fetch_stage.rs

@@ -1,22 +1,20 @@
 //! The `fetch_stage` batches input from a UDP socket and sends it to a channel.
 
-use crossbeam_channel::RecvError;
-use {
-    crossbeam_channel::RecvTimeoutError,
-    solana_perf::{packet::PacketBatchRecycler, recycler::Recycler},
-    solana_sdk::packet::{Packet, PacketFlags},
-    solana_streamer::streamer::{
-        self, PacketBatchReceiver, PacketBatchSender, StreamerReceiveStats,
-    },
-    std::{
-        net::UdpSocket,
-        sync::{
-            atomic::{AtomicBool, Ordering},
-            Arc,
-        },
-        thread::{self, sleep, Builder, JoinHandle},
-        time::Duration,
+use std::{
+    net::UdpSocket,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
     },
+    thread::{self, sleep, Builder, JoinHandle},
+    time::Duration,
+};
+
+use crossbeam_channel::{RecvError, RecvTimeoutError};
+use solana_perf::{packet::PacketBatchRecycler, recycler::Recycler};
+use solana_sdk::packet::{Packet, PacketFlags};
+use solana_streamer::streamer::{
+    self, PacketBatchReceiver, PacketBatchSender, StreamerReceiveStats,
 };
 
 #[derive(Debug, thiserror::Error)]

core/src/staked_nodes_updater_service.rs

@@ -1,19 +1,18 @@
-use jito_rpc::load_balancer::LoadBalancer;
-use log::{error, info};
-use solana_client::client_error;
-use solana_client::rpc_response::RpcContactInfo;
-use std::sync::Mutex;
 use std::{
     collections::HashMap,
     net::IpAddr,
     sync::{
         atomic::{AtomicBool, Ordering},
-        Arc, RwLock,
+        Arc, Mutex, RwLock,
     },
     thread::{self, sleep, Builder, JoinHandle},
     time::{Duration, Instant},
 };
 
+use jito_rpc::load_balancer::LoadBalancer;
+use log::error;
+use solana_client::{client_error, rpc_response::RpcContactInfo};
+
 const IP_TO_STAKE_REFRESH_DURATION: Duration = Duration::from_secs(5);
 
 pub struct StakedNodesUpdaterService {
@@ -81,6 +80,7 @@ impl StakedNodesUpdaterService {
                     }
                 })
                 .collect();
+            *last_stakes = Instant::now();
             Ok(true)
         } else {
             sleep(Duration::from_millis(1));

core/src/tpu.rs

@@ -1,24 +1,26 @@
 //! The `tpu` module implements the Transaction Processing Unit, a
 //! multi-stage transaction processing pipeline in software.
 
-use crate::fetch_stage::FetchStage;
-use crate::staked_nodes_updater_service::StakedNodesUpdaterService;
+use std::{
+    collections::HashMap,
+    net::{IpAddr, UdpSocket},
+    sync::{atomic::AtomicBool, Arc, Mutex, RwLock},
+    thread,
+    thread::JoinHandle,
+};
+
 use crossbeam_channel::{unbounded, Receiver};
 use jito_rpc::load_balancer::LoadBalancer;
-use solana_core::banking_stage::BankingPacketBatch;
-use solana_core::find_packet_sender_stake_stage::FindPacketSenderStakeStage;
-use solana_core::sigverify::TransactionSigVerifier;
-use solana_core::sigverify_stage::SigVerifyStage;
+use solana_core::{
+    banking_stage::BankingPacketBatch, find_packet_sender_stake_stage::FindPacketSenderStakeStage,
+    sigverify::TransactionSigVerifier, sigverify_stage::SigVerifyStage,
+};
 use solana_sdk::signature::Keypair;
 use solana_streamer::quic::{
     spawn_server, StreamStats, MAX_STAKED_CONNECTIONS, MAX_UNSTAKED_CONNECTIONS,
 };
-use std::collections::HashMap;
-use std::net::{IpAddr, UdpSocket};
-use std::sync::atomic::AtomicBool;
-use std::sync::{Arc, Mutex, RwLock};
-use std::thread;
-use std::thread::JoinHandle;
+
+use crate::{fetch_stage::FetchStage, staked_nodes_updater_service::StakedNodesUpdaterService};
 
 pub const DEFAULT_TPU_COALESCE_MS: u64 = 5;
 

jito-protos/src/lib.rs

@@ -10,6 +10,10 @@ pub mod packet {
     tonic::include_proto!("packet");
 }
 
+pub mod relayer {
+    tonic::include_proto!("relayer");
+}
+
 pub mod searcher {
     tonic::include_proto!("searcher");
 }

relayer/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "jito-relayer"
+version = "0.1.0"
+description = "Relayer GRPC server"
+authors = ["Jito Team <team@jito.wtf>"]
+homepage = "https://jito.wtf/"
+edition = "2021"
+publish = false
+
+[dependencies]
+crossbeam-channel = "0.5.4"
+jito-protos = { path = "../jito-protos" }
+log = "0.4.17"
+solana-sdk = "1.10.24"
+solana-client = "1.10.24"
+solana-core = "1.10.24"
+tokio-stream = "0.1.9"
+tokio = "1.14.1"
+tonic = "0.7.2"
+uuid = { version = "1.1.2", features = ["v4"] }

relayer/src/lib.rs

@@ -0,0 +1,3 @@
+mod manager;
+pub mod relayer;
+mod router;

relayer/src/manager.rs

@@ -0,0 +1 @@
+

relayer/src/relayer.rs

@@ -0,0 +1,68 @@
+use std::time::Duration;
+
+use crossbeam_channel::Receiver;
+use jito_protos::relayer::{
+    relayer_service_server::RelayerService, HeartbeatResponse, HeartbeatSubscriptionRequest,
+    PacketSubscriptionRequest, PacketSubscriptionResponse,
+};
+use log::error;
+use solana_core::banking_stage::BankingPacketBatch;
+use solana_sdk::clock::Slot;
+use tokio::{sync::mpsc::channel, time::sleep};
+use tokio_stream::wrappers::ReceiverStream;
+use tonic::{Request, Response, Status};
+
+use crate::router::Router;
+
+pub struct Relayer {
+    router: Router,
+}
+
+impl Relayer {
+    pub fn new(
+        slot_receiver: Receiver<Slot>,
+        packet_receiver: Receiver<BankingPacketBatch>,
+    ) -> Relayer {
+        let router = Router::new(slot_receiver, packet_receiver);
+        Relayer { router }
+    }
+}
+
+#[tonic::async_trait]
+impl RelayerService for Relayer {
+    type SubscribeHeartbeatStream = ReceiverStream<Result<HeartbeatResponse, Status>>;
+
+    async fn subscribe_heartbeat(
+        &self,
+        _request: Request<HeartbeatSubscriptionRequest>,
+    ) -> Result<Response<Self::SubscribeHeartbeatStream>, Status> {
+        let (sender, receiver) = channel(2);
+
+        tokio::spawn(async move {
+            if let Err(e) = sender.send(Ok(HeartbeatResponse::default())).await {
+                error!("subscribe_heartbeat error sending response: {:?}", e);
+            }
+            sleep(Duration::from_millis(500)).await;
+        });
+
+        Ok(Response::new(ReceiverStream::new(receiver)))
+    }
+
+    type SubscribePacketsStream = ReceiverStream<Result<PacketSubscriptionResponse, Status>>;
+
+    async fn subscribe_packets(
+        &self,
+        _request: Request<PacketSubscriptionRequest>,
+    ) -> Result<Response<Self::SubscribePacketsStream>, Status> {
+        let (sender, receiver) = channel(100);
+
+        tokio::spawn(async move {
+            if let Err(e) = sender.send(Ok(PacketSubscriptionResponse::default())).await {
+                error!("subscribe_packets error sending response: {:?}", e);
+            }
+            sleep(Duration::from_millis(500)).await;
+        });
+
+        Ok(Response::new(ReceiverStream::new(receiver)))
+    }
+}

relayer/src/router.rs

@@ -0,0 +1,20 @@
+use crossbeam_channel::Receiver;
+use solana_core::banking_stage::BankingPacketBatch;
+use solana_sdk::clock::Slot;
+
+pub struct Router {
+    slot_receiver: Receiver<Slot>,
+    packet_receiver: Receiver<BankingPacketBatch>,
+}
+
+impl Router {
+    pub fn new(
+        slot_receiver: Receiver<Slot>,
+        packet_receiver: Receiver<BankingPacketBatch>,
+    ) -> Router {
+        Router {
+            slot_receiver,
+            packet_receiver,
+        }
+    }
+}

rpc/src/load_balancer.rs

@@ -1,15 +1,21 @@
-use crossbeam_channel::RecvTimeoutError;
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Mutex,
+    },
+    thread,
+    thread::{sleep, Builder, JoinHandle},
+    time::Duration,
+};
+
+use crossbeam_channel::{unbounded, Receiver, RecvTimeoutError, Sender};
 use log::{error, info};
-use solana_client::pubsub_client::PubsubClient;
-use solana_client::rpc_client::RpcClient;
-use solana_sdk::clock::Slot;
-use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
-use std::collections::HashMap;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex};
-use std::thread;
-use std::thread::{sleep, Builder, JoinHandle};
-use std::time::Duration;
+use solana_client::{pubsub_client::PubsubClient, rpc_client::RpcClient};
+use solana_sdk::{
+    clock::Slot,
+    commitment_config::{CommitmentConfig, CommitmentLevel},
+};
 
 pub struct LoadBalancer {
     // (http, websocket)
@@ -19,7 +25,10 @@ pub struct LoadBalancer {
 }
 
 impl LoadBalancer {
-    pub fn new(servers: &[(String, String)], exit: &Arc<AtomicBool>) -> LoadBalancer {
+    pub fn new(
+        servers: &[(String, String)],
+        exit: &Arc<AtomicBool>,
+    ) -> (LoadBalancer, Receiver<Slot>) {
         let server_to_slot = HashMap::from_iter(servers.iter().map(|(_, ws)| (ws.clone(), 0)));
         let server_to_slot = Arc::new(Mutex::new(server_to_slot));
 
@@ -40,31 +49,41 @@ impl LoadBalancer {
         }));
         let server_to_rpc_client = Arc::new(Mutex::new(server_to_rpc_client));
 
+        let (slot_sender, slot_receiver) = unbounded();
         let subscription_threads = Self::start_subscription_threads(
             servers.to_vec(),
             &server_to_slot,
             &server_to_rpc_client,
             exit,
+            slot_sender,
         );
-        LoadBalancer {
-            server_to_slot,
-            server_to_rpc_client,
-            subscription_threads,
-        }
+        (
+            LoadBalancer {
+                server_to_slot,
+                server_to_rpc_client,
+                subscription_threads,
+            },
+            slot_receiver,
+        )
     }
 
     fn start_subscription_threads(
         servers: Vec<(String, String)>,
         server_to_slot: &Arc<Mutex<HashMap<String, Slot>>>,
         _server_to_rpc_client: &Arc<Mutex<HashMap<String, Arc<RpcClient>>>>,
         exit: &Arc<AtomicBool>,
+        slot_sender: Sender<Slot>,
     ) -> Vec<JoinHandle<()>> {
+        let highest_slot = Arc::new(Mutex::new(Slot::default()));
+
         servers
             .iter()
             .map(|(_, websocket_url)| {
                 let exit = exit.clone();
                 let websocket_url = websocket_url.clone();
                 let server_to_slot = server_to_slot.clone();
+                let slot_sender = slot_sender.clone();
+                let highest_slot = highest_slot.clone();
 
                 Builder::new()
                     .name(format_args!("rpc-thread({})", websocket_url).to_string())
@@ -85,6 +104,18 @@ impl LoadBalancer {
                                                     .lock()
                                                     .unwrap()
                                                     .insert(websocket_url.clone(), slot.slot);
+                                                {
+                                                    let mut highest_slot_l =
+                                                        highest_slot.lock().unwrap();
+                                                    if slot.slot > *highest_slot_l {
+                                                        *highest_slot_l = slot.slot;
+                                                        if let Err(e) = slot_sender.send(slot.slot)
+                                                        {
+                                                            error!("error sending slot: {}", e);
+                                                            break;
+                                                        }
+                                                    }
+                                                }
                                             }
                                             Err(RecvTimeoutError::Timeout) => {}
                                             Err(RecvTimeoutError::Disconnected) => {

rustfmt.toml

@@ -0,0 +1,4 @@
+edition = "2018"  # required by rust-analyzer
+imports_granularity="Crate"
+format_code_in_doc_comments = true
+group_imports = "StdExternalCrate"

transaction-relayer/Cargo.toml

@@ -16,6 +16,7 @@ itertools = "0.10.3"
 jito-core = { path = "../core" }
 jito-protos = { path = "../jito-protos" }
 jito-rpc = { path = "../rpc" }
+jito-relayer = { path = "../relayer" }
 solana-metrics = "1.10.24"
 solana-net-utils = "1.10.24"
 solana-perf = "1.10.24"