2018-03-29 11:20:54 -07:00
|
|
|
//! The AccountantSkel is a microservice that exposes the high-level
|
|
|
|
//! Accountant API to the network. Its message encoding is currently
|
|
|
|
//! in flux. Clients should AccountantStub to interact with it.
|
|
|
|
|
2018-02-28 09:07:54 -08:00
|
|
|
use accountant::Accountant;
|
2018-03-28 13:40:58 -07:00
|
|
|
use bincode::{deserialize, serialize};
|
2018-03-26 21:03:26 -07:00
|
|
|
use entry::Entry;
|
|
|
|
use hash::Hash;
|
2018-03-10 20:09:17 -08:00
|
|
|
use result::Result;
|
2018-03-26 21:03:26 -07:00
|
|
|
use serde_json;
|
|
|
|
use signature::PublicKey;
|
|
|
|
use std::default::Default;
|
2018-03-29 11:54:10 -07:00
|
|
|
use std::io::Write;
|
2018-03-29 12:09:21 -07:00
|
|
|
use std::net::{SocketAddr, UdpSocket};
|
2018-03-22 13:05:23 -07:00
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
2018-03-10 20:09:17 -08:00
|
|
|
use std::sync::mpsc::channel;
|
2018-03-26 21:03:26 -07:00
|
|
|
use std::sync::{Arc, Mutex};
|
2018-03-10 20:09:17 -08:00
|
|
|
use std::thread::{spawn, JoinHandle};
|
2018-03-26 21:03:26 -07:00
|
|
|
use std::time::Duration;
|
|
|
|
use streamer;
|
|
|
|
use transaction::Transaction;
|
2018-03-29 12:18:08 -07:00
|
|
|
use rayon::prelude::*;
|
2018-02-28 09:07:54 -08:00
|
|
|
|
2018-03-26 11:17:19 -07:00
|
|
|
pub struct AccountantSkel<W: Write + Send + 'static> {
|
2018-03-05 11:48:09 -08:00
|
|
|
pub acc: Accountant,
|
2018-03-20 22:15:44 -07:00
|
|
|
pub last_id: Hash,
|
2018-03-26 11:17:19 -07:00
|
|
|
writer: W,
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
|
|
|
|
2018-03-22 13:59:25 -07:00
|
|
|
#[cfg_attr(feature = "cargo-clippy", allow(large_enum_variant))]
|
2018-02-28 13:16:50 -08:00
|
|
|
#[derive(Serialize, Deserialize, Debug)]
|
2018-02-28 09:07:54 -08:00
|
|
|
pub enum Request {
|
2018-03-17 13:42:50 -07:00
|
|
|
Transaction(Transaction),
|
2018-03-06 10:03:41 -08:00
|
|
|
GetBalance { key: PublicKey },
|
|
|
|
GetId { is_last: bool },
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
|
|
|
|
2018-03-29 12:18:08 -07:00
|
|
|
impl Request {
|
|
|
|
/// Verify the request is valid.
|
|
|
|
pub fn verify(&self) -> bool {
|
|
|
|
match *self {
|
|
|
|
Request::Transaction(ref tr) => tr.verify(),
|
|
|
|
_ => true,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Parallel verfication of a batch of requests.
|
2018-03-29 12:09:21 -07:00
|
|
|
fn filter_valid_requests(reqs: Vec<(Request, SocketAddr)>) -> Vec<(Request, SocketAddr)> {
|
2018-03-29 12:18:08 -07:00
|
|
|
reqs.into_par_iter().filter({ |x| x.0.verify() }).collect()
|
2018-03-29 12:09:21 -07:00
|
|
|
}
|
|
|
|
|
2018-02-28 13:16:50 -08:00
|
|
|
#[derive(Serialize, Deserialize, Debug)]
|
2018-02-28 09:07:54 -08:00
|
|
|
pub enum Response {
|
2018-03-05 16:29:32 -08:00
|
|
|
Balance { key: PublicKey, val: Option<i64> },
|
2018-03-06 19:22:30 -08:00
|
|
|
Entries { entries: Vec<Entry> },
|
2018-03-06 16:36:45 -08:00
|
|
|
Id { id: Hash, is_last: bool },
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
|
|
|
|
2018-03-26 11:17:19 -07:00
|
|
|
impl<W: Write + Send + 'static> AccountantSkel<W> {
|
2018-03-29 11:20:54 -07:00
|
|
|
/// Create a new AccountantSkel that wraps the given Accountant.
|
2018-03-26 11:17:19 -07:00
|
|
|
pub fn new(acc: Accountant, w: W) -> Self {
|
2018-03-20 22:15:44 -07:00
|
|
|
let last_id = acc.first_id;
|
2018-03-21 14:43:39 -07:00
|
|
|
AccountantSkel {
|
|
|
|
acc,
|
|
|
|
last_id,
|
2018-03-26 11:17:19 -07:00
|
|
|
writer: w,
|
2018-03-21 14:43:39 -07:00
|
|
|
}
|
2018-03-20 22:15:44 -07:00
|
|
|
}
|
|
|
|
|
2018-03-29 11:20:54 -07:00
|
|
|
/// Process any Entry items that have been published by the Historian.
|
2018-03-26 11:17:19 -07:00
|
|
|
pub fn sync(&mut self) -> Hash {
|
2018-03-20 22:15:44 -07:00
|
|
|
while let Ok(entry) = self.acc.historian.receiver.try_recv() {
|
|
|
|
self.last_id = entry.id;
|
2018-03-28 13:40:58 -07:00
|
|
|
writeln!(self.writer, "{}", serde_json::to_string(&entry).unwrap()).unwrap();
|
2018-03-20 22:15:44 -07:00
|
|
|
}
|
|
|
|
self.last_id
|
2018-02-28 13:16:50 -08:00
|
|
|
}
|
|
|
|
|
2018-03-29 11:20:54 -07:00
|
|
|
/// Process Request items sent by clients.
|
2018-03-29 12:50:32 -07:00
|
|
|
pub fn log_verified_request(&mut self, msg: Request) -> Option<Response> {
|
2018-02-28 09:07:54 -08:00
|
|
|
match msg {
|
2018-03-06 10:43:53 -08:00
|
|
|
Request::Transaction(tr) => {
|
2018-03-29 12:50:32 -07:00
|
|
|
if let Err(err) = self.acc.log_verified_transaction(tr) {
|
2018-03-06 10:54:45 -08:00
|
|
|
eprintln!("Transaction error: {:?}", err);
|
2018-03-03 09:23:31 -08:00
|
|
|
}
|
2018-02-28 09:07:54 -08:00
|
|
|
None
|
|
|
|
}
|
|
|
|
Request::GetBalance { key } => {
|
2018-03-05 14:34:15 -08:00
|
|
|
let val = self.acc.get_balance(&key);
|
2018-02-28 09:07:54 -08:00
|
|
|
Some(Response::Balance { key, val })
|
|
|
|
}
|
2018-03-05 11:48:09 -08:00
|
|
|
Request::GetId { is_last } => Some(Response::Id {
|
|
|
|
id: if is_last {
|
2018-03-21 14:43:39 -07:00
|
|
|
self.sync()
|
2018-03-05 11:48:09 -08:00
|
|
|
} else {
|
|
|
|
self.acc.first_id
|
|
|
|
},
|
|
|
|
is_last,
|
|
|
|
}),
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
|
|
|
}
|
2018-03-27 13:45:04 -07:00
|
|
|
|
2018-03-10 20:09:17 -08:00
|
|
|
fn process(
|
2018-03-27 13:45:04 -07:00
|
|
|
obj: &Arc<Mutex<AccountantSkel<W>>>,
|
2018-03-10 20:09:17 -08:00
|
|
|
r_reader: &streamer::Receiver,
|
2018-03-24 23:46:25 -07:00
|
|
|
s_responder: &streamer::Responder,
|
|
|
|
packet_recycler: &streamer::PacketRecycler,
|
|
|
|
response_recycler: &streamer::ResponseRecycler,
|
2018-03-10 20:09:17 -08:00
|
|
|
) -> Result<()> {
|
|
|
|
let timer = Duration::new(1, 0);
|
|
|
|
let msgs = r_reader.recv_timeout(timer)?;
|
|
|
|
let msgs_ = msgs.clone();
|
2018-03-24 23:46:25 -07:00
|
|
|
let rsps = streamer::allocate(response_recycler);
|
2018-03-10 20:09:17 -08:00
|
|
|
let rsps_ = rsps.clone();
|
|
|
|
{
|
2018-03-29 12:09:21 -07:00
|
|
|
let mut reqs = vec![];
|
2018-03-22 13:15:29 -07:00
|
|
|
for packet in &msgs.read().unwrap().packets {
|
2018-03-29 11:55:41 -07:00
|
|
|
let rsp_addr = packet.meta.get_addr();
|
2018-03-24 18:01:40 -07:00
|
|
|
let sz = packet.meta.size;
|
2018-03-10 20:09:17 -08:00
|
|
|
let req = deserialize(&packet.data[0..sz])?;
|
2018-03-29 12:09:21 -07:00
|
|
|
reqs.push((req, rsp_addr));
|
|
|
|
}
|
|
|
|
let reqs = filter_valid_requests(reqs);
|
|
|
|
|
|
|
|
let mut num = 0;
|
|
|
|
let mut ursps = rsps.write().unwrap();
|
|
|
|
for (req, rsp_addr) in reqs {
|
2018-03-29 12:50:32 -07:00
|
|
|
if let Some(resp) = obj.lock().unwrap().log_verified_request(req) {
|
2018-03-25 08:06:33 -07:00
|
|
|
if ursps.responses.len() <= num {
|
2018-03-25 00:06:48 -07:00
|
|
|
ursps
|
|
|
|
.responses
|
2018-03-25 16:18:27 -07:00
|
|
|
.resize((num + 1) * 2, streamer::Response::default());
|
2018-03-24 23:46:25 -07:00
|
|
|
}
|
|
|
|
let rsp = &mut ursps.responses[num];
|
2018-03-10 20:09:17 -08:00
|
|
|
let v = serialize(&resp)?;
|
|
|
|
let len = v.len();
|
2018-03-24 23:46:25 -07:00
|
|
|
rsp.data[..len].copy_from_slice(&v);
|
2018-03-24 18:01:40 -07:00
|
|
|
rsp.meta.size = len;
|
2018-03-29 11:55:41 -07:00
|
|
|
rsp.meta.set_addr(&rsp_addr);
|
2018-03-10 20:09:17 -08:00
|
|
|
num += 1;
|
|
|
|
}
|
|
|
|
}
|
2018-03-25 08:06:33 -07:00
|
|
|
ursps.responses.resize(num, streamer::Response::default());
|
2018-03-10 20:09:17 -08:00
|
|
|
}
|
2018-03-24 23:46:25 -07:00
|
|
|
s_responder.send(rsps_)?;
|
2018-03-25 08:05:03 -07:00
|
|
|
streamer::recycle(packet_recycler, msgs_);
|
2018-03-10 20:09:17 -08:00
|
|
|
Ok(())
|
|
|
|
}
|
2018-02-28 09:07:54 -08:00
|
|
|
|
2018-03-29 11:20:54 -07:00
|
|
|
/// Create a UDP microservice that forwards messages the given AccountantSkel.
|
|
|
|
/// Set `exit` to shutdown its threads.
|
2018-03-10 20:09:17 -08:00
|
|
|
pub fn serve(
|
2018-03-26 11:17:19 -07:00
|
|
|
obj: Arc<Mutex<AccountantSkel<W>>>,
|
2018-03-10 20:09:17 -08:00
|
|
|
addr: &str,
|
2018-03-22 13:05:23 -07:00
|
|
|
exit: Arc<AtomicBool>,
|
2018-03-23 20:49:28 -07:00
|
|
|
) -> Result<Vec<JoinHandle<()>>> {
|
2018-03-10 20:09:17 -08:00
|
|
|
let read = UdpSocket::bind(addr)?;
|
|
|
|
// make sure we are on the same interface
|
|
|
|
let mut local = read.local_addr()?;
|
|
|
|
local.set_port(0);
|
|
|
|
let write = UdpSocket::bind(local)?;
|
2018-02-28 09:07:54 -08:00
|
|
|
|
2018-03-24 23:46:25 -07:00
|
|
|
let packet_recycler = Arc::new(Mutex::new(Vec::new()));
|
|
|
|
let response_recycler = Arc::new(Mutex::new(Vec::new()));
|
2018-03-10 20:09:17 -08:00
|
|
|
let (s_reader, r_reader) = channel();
|
2018-03-24 23:46:25 -07:00
|
|
|
let t_receiver = streamer::receiver(read, exit.clone(), packet_recycler.clone(), s_reader)?;
|
2018-02-28 13:16:50 -08:00
|
|
|
|
2018-03-24 23:46:25 -07:00
|
|
|
let (s_responder, r_responder) = channel();
|
2018-03-25 00:06:48 -07:00
|
|
|
let t_responder =
|
|
|
|
streamer::responder(write, exit.clone(), response_recycler.clone(), r_responder);
|
2018-03-10 20:09:17 -08:00
|
|
|
|
2018-03-27 13:45:04 -07:00
|
|
|
let skel = obj.clone();
|
|
|
|
let t_server = spawn(move || loop {
|
|
|
|
let e = AccountantSkel::process(
|
|
|
|
&skel,
|
|
|
|
&r_reader,
|
|
|
|
&s_responder,
|
|
|
|
&packet_recycler,
|
|
|
|
&response_recycler,
|
|
|
|
);
|
|
|
|
if e.is_err() && exit.load(Ordering::Relaxed) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
});
|
|
|
|
|
2018-03-28 19:13:10 -07:00
|
|
|
Ok(vec![t_receiver, t_responder, t_server])
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
|
|
|
}
|