2018-09-21 15:32:15 -07:00
|
|
|
use blob_fetch_stage::BlobFetchStage;
|
|
|
|
use crdt::{Crdt, Node, NodeInfo};
|
|
|
|
use ncp::Ncp;
|
|
|
|
use service::Service;
|
|
|
|
use std::net::SocketAddr;
|
|
|
|
use std::net::UdpSocket;
|
|
|
|
use std::sync::atomic::AtomicBool;
|
|
|
|
use std::sync::mpsc::channel;
|
|
|
|
use std::sync::{Arc, RwLock};
|
|
|
|
use std::thread::JoinHandle;
|
|
|
|
use store_ledger_stage::StoreLedgerStage;
|
|
|
|
use streamer::BlobReceiver;
|
|
|
|
use window;
|
|
|
|
use window_service::window_service;
|
|
|
|
|
|
|
|
pub struct Replicator {
|
|
|
|
ncp: Ncp,
|
|
|
|
fetch_stage: BlobFetchStage,
|
|
|
|
store_ledger_stage: StoreLedgerStage,
|
|
|
|
t_window: JoinHandle<()>,
|
|
|
|
pub retransmit_receiver: BlobReceiver,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Replicator {
|
|
|
|
pub fn new(
|
|
|
|
entry_height: u64,
|
|
|
|
exit: &Arc<AtomicBool>,
|
|
|
|
ledger_path: Option<&str>,
|
|
|
|
node: Node,
|
|
|
|
network_addr: Option<SocketAddr>,
|
|
|
|
) -> Replicator {
|
|
|
|
let window = window::new_window_from_entries(&[], entry_height, &node.info);
|
|
|
|
let shared_window = Arc::new(RwLock::new(window));
|
|
|
|
|
|
|
|
let crdt = Arc::new(RwLock::new(Crdt::new(node.info).expect("Crdt::new")));
|
|
|
|
|
|
|
|
let leader_info = network_addr.map(|i| NodeInfo::new_entry_point(&i));
|
|
|
|
|
|
|
|
if let Some(leader_info) = leader_info.as_ref() {
|
|
|
|
crdt.write().unwrap().insert(leader_info);
|
|
|
|
} else {
|
|
|
|
panic!("No leader info!");
|
|
|
|
}
|
|
|
|
|
|
|
|
let repair_socket = Arc::new(node.sockets.repair);
|
|
|
|
let mut blob_sockets: Vec<Arc<UdpSocket>> =
|
|
|
|
node.sockets.replicate.into_iter().map(Arc::new).collect();
|
|
|
|
blob_sockets.push(repair_socket.clone());
|
|
|
|
let (fetch_stage, blob_fetch_receiver) =
|
|
|
|
BlobFetchStage::new_multi_socket(blob_sockets, exit.clone());
|
|
|
|
|
2018-09-21 16:50:58 -07:00
|
|
|
let (entry_window_sender, entry_window_receiver) = channel();
|
2018-09-21 15:32:15 -07:00
|
|
|
// todo: pull blobs off the retransmit_receiver and recycle them?
|
|
|
|
let (retransmit_sender, retransmit_receiver) = channel();
|
|
|
|
let t_window = window_service(
|
|
|
|
crdt.clone(),
|
|
|
|
shared_window.clone(),
|
|
|
|
entry_height,
|
|
|
|
blob_fetch_receiver,
|
2018-09-21 16:50:58 -07:00
|
|
|
entry_window_sender,
|
2018-09-21 15:32:15 -07:00
|
|
|
retransmit_sender,
|
|
|
|
repair_socket,
|
|
|
|
);
|
|
|
|
|
2018-09-21 16:50:58 -07:00
|
|
|
let store_ledger_stage = StoreLedgerStage::new(entry_window_receiver, ledger_path);
|
2018-09-21 15:32:15 -07:00
|
|
|
|
|
|
|
let ncp = Ncp::new(
|
|
|
|
&crdt,
|
|
|
|
shared_window.clone(),
|
|
|
|
ledger_path,
|
|
|
|
node.sockets.gossip,
|
|
|
|
exit.clone(),
|
|
|
|
);
|
|
|
|
|
|
|
|
Replicator {
|
|
|
|
ncp,
|
|
|
|
fetch_stage,
|
|
|
|
store_ledger_stage,
|
|
|
|
t_window,
|
|
|
|
retransmit_receiver,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn join(self) {
|
|
|
|
self.ncp.join().unwrap();
|
|
|
|
self.t_window.join().unwrap();
|
|
|
|
self.fetch_stage.join().unwrap();
|
|
|
|
self.store_ledger_stage.join().unwrap();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
|
|
mod tests {
|
|
|
|
use client::mk_client;
|
|
|
|
use crdt::Node;
|
|
|
|
use fullnode::Fullnode;
|
|
|
|
use ledger::{genesis, read_ledger};
|
|
|
|
use logger;
|
|
|
|
use replicator::Replicator;
|
|
|
|
use signature::{Keypair, KeypairUtil};
|
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
|
|
use std::sync::Arc;
|
|
|
|
use std::thread::sleep;
|
|
|
|
use std::time::Duration;
|
|
|
|
|
|
|
|
#[test]
|
|
|
|
fn test_replicator_startup() {
|
|
|
|
logger::setup();
|
|
|
|
info!("starting replicator test");
|
|
|
|
let entry_height = 0;
|
|
|
|
let replicator_ledger_path = "replicator_test_replicator_ledger";
|
|
|
|
|
|
|
|
let exit = Arc::new(AtomicBool::new(false));
|
|
|
|
|
|
|
|
let leader_ledger_path = "replicator_test_leader_ledger";
|
|
|
|
let (mint, leader_ledger_path) = genesis(leader_ledger_path, 100);
|
|
|
|
|
|
|
|
info!("starting leader node");
|
|
|
|
let leader_keypair = Keypair::new();
|
|
|
|
let leader_node = Node::new_localhost_with_pubkey(leader_keypair.pubkey());
|
|
|
|
let network_addr = leader_node.sockets.gossip.local_addr().unwrap();
|
|
|
|
let leader_info = leader_node.info.clone();
|
|
|
|
let leader_rotation_interval = 20;
|
|
|
|
let leader = Fullnode::new(
|
|
|
|
leader_node,
|
|
|
|
&leader_ledger_path,
|
|
|
|
leader_keypair,
|
|
|
|
None,
|
|
|
|
false,
|
|
|
|
Some(leader_rotation_interval),
|
|
|
|
);
|
|
|
|
|
|
|
|
let mut leader_client = mk_client(&leader_info);
|
|
|
|
|
|
|
|
let bob = Keypair::new();
|
|
|
|
|
|
|
|
let last_id = leader_client.get_last_id();
|
|
|
|
leader_client
|
|
|
|
.transfer(1, &mint.keypair(), bob.pubkey(), &last_id)
|
|
|
|
.unwrap();
|
|
|
|
|
|
|
|
let replicator_keypair = Keypair::new();
|
|
|
|
|
|
|
|
info!("starting replicator node");
|
|
|
|
let replicator_node = Node::new_localhost_with_pubkey(replicator_keypair.pubkey());
|
|
|
|
let replicator = Replicator::new(
|
|
|
|
entry_height,
|
|
|
|
&exit,
|
|
|
|
Some(replicator_ledger_path),
|
|
|
|
replicator_node,
|
|
|
|
Some(network_addr),
|
|
|
|
);
|
|
|
|
|
|
|
|
let mut num_entries = 0;
|
|
|
|
for _ in 0..10 {
|
|
|
|
match read_ledger(replicator_ledger_path, true) {
|
|
|
|
Ok(entries) => {
|
|
|
|
for _ in entries {
|
|
|
|
num_entries += 1;
|
|
|
|
}
|
|
|
|
info!("{} entries", num_entries);
|
|
|
|
if num_entries > 0 {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
Err(e) => {
|
|
|
|
info!("error reading ledger: {:?}", e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
sleep(Duration::new(1, 0));
|
|
|
|
let last_id = leader_client.get_last_id();
|
|
|
|
leader_client
|
|
|
|
.transfer(1, &mint.keypair(), bob.pubkey(), &last_id)
|
|
|
|
.unwrap();
|
|
|
|
}
|
|
|
|
assert!(num_entries > 0);
|
|
|
|
exit.store(true, Ordering::Relaxed);
|
|
|
|
replicator.join();
|
|
|
|
leader.exit();
|
|
|
|
}
|
|
|
|
}
|