2018-05-29 10:18:12 -07:00
|
|
|
//! The `fetch_stage` batches input from a UDP socket and sends it to a channel.
|
|
|
|
|
2018-12-07 19:16:27 -08:00
|
|
|
use crate::service::Service;
|
2019-01-31 15:51:29 -08:00
|
|
|
use crate::streamer::{self, PacketReceiver, PacketSender};
|
2018-05-29 10:18:12 -07:00
|
|
|
use std::net::UdpSocket;
|
2019-03-04 19:02:03 -08:00
|
|
|
use std::sync::atomic::AtomicBool;
|
2018-05-29 10:18:12 -07:00
|
|
|
use std::sync::mpsc::channel;
|
2018-05-29 20:20:28 -07:00
|
|
|
use std::sync::Arc;
|
2018-07-03 21:14:08 -07:00
|
|
|
use std::thread::{self, JoinHandle};
|
2018-05-29 10:18:12 -07:00
|
|
|
|
|
|
|
pub struct FetchStage {
|
2018-07-03 21:14:08 -07:00
|
|
|
thread_hdls: Vec<JoinHandle<()>>,
|
2018-05-29 10:18:12 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
impl FetchStage {
|
2018-12-07 19:01:28 -08:00
|
|
|
#[allow(clippy::new_ret_no_self)]
|
2019-03-08 14:59:11 -08:00
|
|
|
pub fn new(
|
|
|
|
sockets: Vec<UdpSocket>,
|
2019-03-11 12:46:30 -07:00
|
|
|
tpu_via_blobs_sockets: Vec<UdpSocket>,
|
2019-03-08 14:59:11 -08:00
|
|
|
exit: &Arc<AtomicBool>,
|
|
|
|
) -> (Self, PacketReceiver) {
|
2019-01-31 15:51:29 -08:00
|
|
|
let (sender, receiver) = channel();
|
2019-03-08 14:59:11 -08:00
|
|
|
(
|
2019-03-11 12:46:30 -07:00
|
|
|
Self::new_with_sender(sockets, tpu_via_blobs_sockets, exit, &sender),
|
2019-03-08 14:59:11 -08:00
|
|
|
receiver,
|
|
|
|
)
|
2019-01-31 15:51:29 -08:00
|
|
|
}
|
|
|
|
pub fn new_with_sender(
|
|
|
|
sockets: Vec<UdpSocket>,
|
2019-03-11 12:46:30 -07:00
|
|
|
tpu_via_blobs_sockets: Vec<UdpSocket>,
|
2019-03-04 19:53:50 -08:00
|
|
|
exit: &Arc<AtomicBool>,
|
2019-01-31 15:51:29 -08:00
|
|
|
sender: &PacketSender,
|
|
|
|
) -> Self {
|
2018-09-06 14:13:40 -07:00
|
|
|
let tx_sockets = sockets.into_iter().map(Arc::new).collect();
|
2019-03-11 12:46:30 -07:00
|
|
|
let tpu_via_blobs_sockets = tpu_via_blobs_sockets.into_iter().map(Arc::new).collect();
|
|
|
|
Self::new_multi_socket(tx_sockets, tpu_via_blobs_sockets, exit, &sender)
|
2018-06-13 21:52:23 -07:00
|
|
|
}
|
2019-03-08 15:47:53 -08:00
|
|
|
|
2019-01-24 12:46:40 -08:00
|
|
|
fn new_multi_socket(
|
2018-08-28 16:32:40 -07:00
|
|
|
sockets: Vec<Arc<UdpSocket>>,
|
2019-03-11 12:46:30 -07:00
|
|
|
tpu_via_blobs_sockets: Vec<Arc<UdpSocket>>,
|
2019-03-04 19:53:50 -08:00
|
|
|
exit: &Arc<AtomicBool>,
|
2019-01-31 15:51:29 -08:00
|
|
|
sender: &PacketSender,
|
|
|
|
) -> Self {
|
2019-03-08 15:47:53 -08:00
|
|
|
let tpu_threads = sockets
|
|
|
|
.into_iter()
|
|
|
|
.map(|socket| streamer::receiver(socket, &exit, sender.clone(), "fetch-stage"));
|
|
|
|
|
2019-03-11 12:46:30 -07:00
|
|
|
let tpu_via_blobs_threads = tpu_via_blobs_sockets
|
2018-06-13 21:52:23 -07:00
|
|
|
.into_iter()
|
2019-03-08 15:47:53 -08:00
|
|
|
.map(|socket| streamer::blob_packet_receiver(socket, &exit, sender.clone()));
|
2018-05-29 10:18:12 -07:00
|
|
|
|
2019-03-11 12:46:30 -07:00
|
|
|
let thread_hdls: Vec<_> = tpu_threads.chain(tpu_via_blobs_threads).collect();
|
2019-03-04 19:02:03 -08:00
|
|
|
Self { thread_hdls }
|
2018-05-29 10:18:12 -07:00
|
|
|
}
|
|
|
|
}
|
2018-07-03 21:14:08 -07:00
|
|
|
|
|
|
|
impl Service for FetchStage {
|
2018-09-13 14:00:17 -07:00
|
|
|
type JoinReturnType = ();
|
2018-07-03 21:14:08 -07:00
|
|
|
|
|
|
|
fn join(self) -> thread::Result<()> {
|
2018-09-13 14:00:17 -07:00
|
|
|
for thread_hdl in self.thread_hdls {
|
2018-07-03 21:14:08 -07:00
|
|
|
thread_hdl.join()?;
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|