2018-03-30 13:10:27 -07:00
|
|
|
//! The `accountant_skel` module is a microservice that exposes the high-level
|
2018-03-29 11:20:54 -07:00
|
|
|
//! Accountant API to the network. Its message encoding is currently
|
2018-03-30 10:43:38 -07:00
|
|
|
//! in flux. Clients should use AccountantStub to interact with it.
|
2018-03-29 11:20:54 -07:00
|
|
|
|
2018-02-28 09:07:54 -08:00
|
|
|
use accountant::Accountant;
|
2018-03-28 13:40:58 -07:00
|
|
|
use bincode::{deserialize, serialize};
|
2018-04-05 21:39:07 -07:00
|
|
|
use ecdsa;
|
2018-03-26 21:03:26 -07:00
|
|
|
use entry::Entry;
|
2018-04-02 13:41:07 -07:00
|
|
|
use event::Event;
|
2018-03-26 21:03:26 -07:00
|
|
|
use hash::Hash;
|
2018-04-02 20:15:21 -07:00
|
|
|
use historian::Historian;
|
2018-03-26 21:07:11 -07:00
|
|
|
use packet;
|
|
|
|
use packet::SharedPackets;
|
2018-04-02 20:15:21 -07:00
|
|
|
use rayon::prelude::*;
|
|
|
|
use recorder::Signal;
|
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;
|
2018-03-26 21:07:11 -07:00
|
|
|
use std::cmp::max;
|
|
|
|
use std::collections::VecDeque;
|
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-04-11 10:17:00 -07:00
|
|
|
use std::sync::mpsc::{channel, Receiver, Sender};
|
2018-03-26 21:07:11 -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-02-28 09:07:54 -08:00
|
|
|
|
2018-03-26 11:17:19 -07:00
|
|
|
pub struct AccountantSkel<W: Write + Send + 'static> {
|
2018-04-02 08:30:10 -07:00
|
|
|
acc: Accountant,
|
|
|
|
last_id: Hash,
|
2018-03-26 11:17:19 -07:00
|
|
|
writer: W,
|
2018-04-02 13:41:07 -07:00
|
|
|
historian: Historian,
|
2018-04-17 14:31:52 -07:00
|
|
|
entry_info_subscribers: Vec<SocketAddr>,
|
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-04-11 17:30:53 -07:00
|
|
|
#[derive(Serialize, Deserialize, Debug, Clone)]
|
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 },
|
2018-04-02 08:30:10 -07:00
|
|
|
GetLastId,
|
2018-04-17 14:31:52 -07:00
|
|
|
Subscribe { subscriptions: Vec<Subscription> },
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Serialize, Deserialize, Debug, Clone)]
|
|
|
|
pub enum Subscription {
|
|
|
|
EntryInfo,
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Serialize, Deserialize, Debug, Clone)]
|
|
|
|
pub struct EntryInfo {
|
2018-04-17 15:14:53 -07:00
|
|
|
pub id: Hash,
|
|
|
|
pub num_hashes: u64,
|
|
|
|
pub num_events: u64,
|
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 {
|
2018-03-26 21:07:11 -07:00
|
|
|
Request::Transaction(ref tr) => tr.verify_plan(),
|
2018-03-29 12:18:08 -07:00
|
|
|
_ => true,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
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-04-17 15:14:53 -07:00
|
|
|
EntryInfo(EntryInfo),
|
2018-04-02 08:30:10 -07:00
|
|
|
LastId { id: Hash },
|
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-04-02 13:41:07 -07:00
|
|
|
pub fn new(acc: Accountant, last_id: Hash, writer: W, historian: Historian) -> Self {
|
2018-03-21 14:43:39 -07:00
|
|
|
AccountantSkel {
|
|
|
|
acc,
|
|
|
|
last_id,
|
2018-04-02 08:30:10 -07:00
|
|
|
writer,
|
2018-04-02 13:41:07 -07:00
|
|
|
historian,
|
2018-04-17 14:31:52 -07:00
|
|
|
entry_info_subscribers: vec![],
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn notify_entry_info_subscribers(&mut self, entry: &Entry) {
|
|
|
|
// TODO: No need to bind().
|
|
|
|
let socket = UdpSocket::bind("127.0.0.1:0").expect("bind");
|
|
|
|
|
|
|
|
for addr in &self.entry_info_subscribers {
|
|
|
|
let entry_info = EntryInfo {
|
|
|
|
id: entry.id,
|
|
|
|
num_hashes: entry.num_hashes,
|
|
|
|
num_events: entry.events.len() as u64,
|
|
|
|
};
|
2018-04-17 15:14:53 -07:00
|
|
|
let data = serialize(&Response::EntryInfo(entry_info)).expect("serialize EntryInfo");
|
2018-04-17 14:31:52 -07:00
|
|
|
let _res = socket.send_to(&data, addr);
|
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-04-02 13:41:07 -07:00
|
|
|
while let Ok(entry) = self.historian.receiver.try_recv() {
|
2018-03-20 22:15:44 -07:00
|
|
|
self.last_id = entry.id;
|
2018-04-05 08:53:58 -07:00
|
|
|
self.acc.register_entry_id(&self.last_id);
|
2018-03-28 13:40:58 -07:00
|
|
|
writeln!(self.writer, "{}", serde_json::to_string(&entry).unwrap()).unwrap();
|
2018-04-17 14:31:52 -07:00
|
|
|
self.notify_entry_info_subscribers(&entry);
|
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-04-11 10:17:00 -07:00
|
|
|
pub fn process_request(
|
|
|
|
&mut self,
|
|
|
|
msg: Request,
|
|
|
|
rsp_addr: SocketAddr,
|
|
|
|
) -> Option<(Response, SocketAddr)> {
|
2018-02-28 09:07:54 -08:00
|
|
|
match msg {
|
|
|
|
Request::GetBalance { key } => {
|
2018-03-05 14:34:15 -08:00
|
|
|
let val = self.acc.get_balance(&key);
|
2018-04-11 10:17:00 -07:00
|
|
|
Some((Response::Balance { key, val }, rsp_addr))
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
2018-04-11 10:17:00 -07:00
|
|
|
Request::GetLastId => Some((Response::LastId { id: self.sync() }, rsp_addr)),
|
|
|
|
Request::Transaction(_) => unreachable!(),
|
2018-04-17 14:31:52 -07:00
|
|
|
Request::Subscribe { subscriptions } => {
|
|
|
|
for subscription in subscriptions {
|
|
|
|
match subscription {
|
|
|
|
Subscription::EntryInfo => self.entry_info_subscribers.push(rsp_addr),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
None
|
|
|
|
}
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
|
|
|
}
|
2018-03-27 13:45:04 -07:00
|
|
|
|
2018-04-11 08:02:33 -07:00
|
|
|
fn recv_batch(recvr: &streamer::PacketReceiver) -> Result<Vec<SharedPackets>> {
|
2018-03-26 21:07:11 -07:00
|
|
|
let timer = Duration::new(1, 0);
|
|
|
|
let msgs = recvr.recv_timeout(timer)?;
|
|
|
|
trace!("got msgs");
|
2018-04-11 08:02:33 -07:00
|
|
|
let mut batch = vec![msgs];
|
2018-03-26 21:07:11 -07:00
|
|
|
while let Ok(more) = recvr.try_recv() {
|
|
|
|
trace!("got more msgs");
|
2018-04-11 08:02:33 -07:00
|
|
|
batch.push(more);
|
2018-03-26 21:07:11 -07:00
|
|
|
}
|
2018-04-11 08:02:33 -07:00
|
|
|
info!("batch len {}", batch.len());
|
|
|
|
Ok(batch)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn verify_batch(batch: Vec<SharedPackets>) -> Vec<Vec<(SharedPackets, Vec<u8>)>> {
|
|
|
|
let chunk_size = max(1, (batch.len() + 3) / 4);
|
|
|
|
let batches: Vec<_> = batch.chunks(chunk_size).map(|x| x.to_vec()).collect();
|
|
|
|
batches
|
2018-03-26 21:07:11 -07:00
|
|
|
.into_par_iter()
|
2018-04-11 08:02:33 -07:00
|
|
|
.map(|batch| {
|
|
|
|
let r = ecdsa::ed25519_verify(&batch);
|
|
|
|
batch.into_iter().zip(r).collect()
|
|
|
|
})
|
|
|
|
.collect()
|
|
|
|
}
|
|
|
|
|
|
|
|
fn verifier(
|
|
|
|
recvr: &streamer::PacketReceiver,
|
|
|
|
sendr: &Sender<Vec<(SharedPackets, Vec<u8>)>>,
|
|
|
|
) -> Result<()> {
|
|
|
|
let batch = Self::recv_batch(recvr)?;
|
|
|
|
let verified_batches = Self::verify_batch(batch);
|
|
|
|
for xs in verified_batches {
|
2018-04-10 20:12:59 -07:00
|
|
|
sendr.send(xs)?;
|
2018-03-26 21:07:11 -07:00
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn deserialize_packets(p: &packet::Packets) -> Vec<Option<(Request, SocketAddr)>> {
|
2018-04-06 14:52:58 -07:00
|
|
|
p.packets
|
|
|
|
.par_iter()
|
|
|
|
.map(|x| {
|
|
|
|
deserialize(&x.data[0..x.meta.size])
|
|
|
|
.map(|req| (req, x.meta.addr()))
|
|
|
|
.ok()
|
|
|
|
})
|
|
|
|
.collect()
|
2018-03-26 21:07:11 -07:00
|
|
|
}
|
|
|
|
|
2018-04-11 10:17:00 -07:00
|
|
|
/// Split Request list into verified transactions and the rest
|
|
|
|
fn partition_requests(
|
|
|
|
req_vers: Vec<(Request, SocketAddr, u8)>,
|
|
|
|
) -> (Vec<Transaction>, Vec<(Request, SocketAddr)>) {
|
|
|
|
let mut trs = vec![];
|
|
|
|
let mut reqs = vec![];
|
|
|
|
for (msg, rsp_addr, verify) in req_vers {
|
|
|
|
match msg {
|
|
|
|
Request::Transaction(tr) => {
|
|
|
|
if verify != 0 {
|
|
|
|
trs.push(tr);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
_ => reqs.push((msg, rsp_addr)),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
(trs, reqs)
|
|
|
|
}
|
|
|
|
|
2018-04-06 15:34:52 -07:00
|
|
|
fn process_packets(
|
2018-04-11 10:17:00 -07:00
|
|
|
&mut self,
|
2018-04-10 20:43:53 -07:00
|
|
|
req_vers: Vec<(Request, SocketAddr, u8)>,
|
2018-04-11 10:17:00 -07:00
|
|
|
) -> Result<Vec<(Response, SocketAddr)>> {
|
|
|
|
let (trs, reqs) = Self::partition_requests(req_vers);
|
|
|
|
|
|
|
|
// Process the transactions in parallel and then log the successful ones.
|
|
|
|
for result in self.acc.process_verified_transactions(trs) {
|
|
|
|
if let Ok(tr) = result {
|
|
|
|
self.historian
|
|
|
|
.sender
|
|
|
|
.send(Signal::Event(Event::Transaction(tr)))?;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-04-12 19:53:34 -07:00
|
|
|
// Let validators know they should not attempt to process additional
|
|
|
|
// transactions in parallel.
|
|
|
|
self.historian.sender.send(Signal::Tick)?;
|
|
|
|
|
2018-04-11 10:17:00 -07:00
|
|
|
// Process the remaining requests serially.
|
|
|
|
let rsps = reqs.into_iter()
|
|
|
|
.filter_map(|(req, rsp_addr)| self.process_request(req, rsp_addr))
|
|
|
|
.collect();
|
|
|
|
|
|
|
|
Ok(rsps)
|
2018-04-06 15:34:52 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
fn serialize_response(
|
2018-04-06 15:21:20 -07:00
|
|
|
resp: Response,
|
|
|
|
rsp_addr: SocketAddr,
|
2018-04-06 15:34:52 -07:00
|
|
|
blob_recycler: &packet::BlobRecycler,
|
2018-04-06 15:21:20 -07:00
|
|
|
) -> Result<packet::SharedBlob> {
|
|
|
|
let blob = blob_recycler.allocate();
|
|
|
|
{
|
|
|
|
let mut b = blob.write().unwrap();
|
|
|
|
let v = serialize(&resp)?;
|
|
|
|
let len = v.len();
|
|
|
|
b.data[..len].copy_from_slice(&v);
|
|
|
|
b.meta.size = len;
|
|
|
|
b.meta.set_addr(&rsp_addr);
|
|
|
|
}
|
|
|
|
Ok(blob)
|
|
|
|
}
|
|
|
|
|
2018-04-06 15:34:52 -07:00
|
|
|
fn serialize_responses(
|
|
|
|
rsps: Vec<(Response, SocketAddr)>,
|
2018-04-06 15:12:13 -07:00
|
|
|
blob_recycler: &packet::BlobRecycler,
|
|
|
|
) -> Result<VecDeque<packet::SharedBlob>> {
|
2018-04-06 15:34:52 -07:00
|
|
|
let mut blobs = VecDeque::new();
|
|
|
|
for (resp, rsp_addr) in rsps {
|
|
|
|
blobs.push_back(Self::serialize_response(resp, rsp_addr, blob_recycler)?);
|
2018-04-06 15:12:13 -07:00
|
|
|
}
|
2018-04-06 15:34:52 -07:00
|
|
|
Ok(blobs)
|
2018-04-06 15:12:13 -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-04-10 20:12:59 -07:00
|
|
|
verified_receiver: &Receiver<Vec<(SharedPackets, Vec<u8>)>>,
|
2018-04-02 19:32:58 -07:00
|
|
|
blob_sender: &streamer::BlobSender,
|
|
|
|
packet_recycler: &packet::PacketRecycler,
|
|
|
|
blob_recycler: &packet::BlobRecycler,
|
2018-03-10 20:09:17 -08:00
|
|
|
) -> Result<()> {
|
|
|
|
let timer = Duration::new(1, 0);
|
2018-04-10 20:12:59 -07:00
|
|
|
let mms = verified_receiver.recv_timeout(timer)?;
|
|
|
|
for (msgs, vers) in mms {
|
2018-04-06 15:12:13 -07:00
|
|
|
let reqs = Self::deserialize_packets(&msgs.read().unwrap());
|
2018-04-10 20:43:53 -07:00
|
|
|
let req_vers = reqs.into_iter()
|
|
|
|
.zip(vers)
|
|
|
|
.filter_map(|(req, ver)| req.map(|(msg, addr)| (msg, addr, ver)))
|
|
|
|
.filter(|x| x.0.verify())
|
|
|
|
.collect();
|
2018-04-11 10:17:00 -07:00
|
|
|
let rsps = obj.lock().unwrap().process_packets(req_vers)?;
|
2018-04-06 15:34:52 -07:00
|
|
|
let blobs = Self::serialize_responses(rsps, blob_recycler)?;
|
2018-04-06 15:21:20 -07:00
|
|
|
if !blobs.is_empty() {
|
2018-03-26 21:07:11 -07:00
|
|
|
//don't wake up the other side if there is nothing
|
2018-04-06 15:21:20 -07:00
|
|
|
blob_sender.send(blobs)?;
|
2018-03-26 21:07:11 -07:00
|
|
|
}
|
2018-04-06 14:58:11 -07:00
|
|
|
packet_recycler.recycle(msgs);
|
2018-04-17 14:31:52 -07:00
|
|
|
|
|
|
|
// Write new entries to the ledger and notify subscribers.
|
|
|
|
obj.lock().unwrap().sync();
|
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-04-02 19:32:58 -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-04-02 19:32:58 -07:00
|
|
|
let packet_recycler = packet::PacketRecycler::default();
|
|
|
|
let blob_recycler = packet::BlobRecycler::default();
|
|
|
|
let (packet_sender, packet_receiver) = channel();
|
|
|
|
let t_receiver =
|
|
|
|
streamer::receiver(read, exit.clone(), packet_recycler.clone(), packet_sender)?;
|
|
|
|
let (blob_sender, blob_receiver) = channel();
|
2018-03-25 00:06:48 -07:00
|
|
|
let t_responder =
|
2018-04-02 19:32:58 -07:00
|
|
|
streamer::responder(write, exit.clone(), blob_recycler.clone(), blob_receiver);
|
2018-03-26 21:07:11 -07:00
|
|
|
let (verified_sender, verified_receiver) = channel();
|
|
|
|
|
|
|
|
let exit_ = exit.clone();
|
|
|
|
let t_verifier = spawn(move || loop {
|
|
|
|
let e = Self::verifier(&packet_receiver, &verified_sender);
|
|
|
|
if e.is_err() && exit_.load(Ordering::Relaxed) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
});
|
|
|
|
|
2018-04-02 19:32:58 -07:00
|
|
|
let skel = obj.clone();
|
2018-03-27 13:45:04 -07:00
|
|
|
let t_server = spawn(move || loop {
|
|
|
|
let e = AccountantSkel::process(
|
2018-04-02 19:32:58 -07:00
|
|
|
&skel,
|
2018-03-26 21:07:11 -07:00
|
|
|
&verified_receiver,
|
2018-04-02 19:32:58 -07:00
|
|
|
&blob_sender,
|
2018-03-27 13:45:04 -07:00
|
|
|
&packet_recycler,
|
2018-04-02 19:32:58 -07:00
|
|
|
&blob_recycler,
|
2018-03-27 13:45:04 -07:00
|
|
|
);
|
|
|
|
if e.is_err() && exit.load(Ordering::Relaxed) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
});
|
2018-03-26 21:07:11 -07:00
|
|
|
Ok(vec![t_receiver, t_responder, t_server, t_verifier])
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-04-11 17:30:53 -07:00
|
|
|
#[cfg(test)]
|
|
|
|
pub fn to_packets(r: &packet::PacketRecycler, reqs: Vec<Request>) -> Vec<SharedPackets> {
|
|
|
|
let mut out = vec![];
|
|
|
|
for rrs in reqs.chunks(packet::NUM_PACKETS) {
|
|
|
|
let p = r.allocate();
|
|
|
|
p.write()
|
|
|
|
.unwrap()
|
|
|
|
.packets
|
|
|
|
.resize(rrs.len(), Default::default());
|
|
|
|
for (i, o) in rrs.iter().zip(p.write().unwrap().packets.iter_mut()) {
|
|
|
|
let v = serialize(&i).expect("serialize request");
|
|
|
|
let len = v.len();
|
|
|
|
o.data[..len].copy_from_slice(&v);
|
|
|
|
o.meta.size = len;
|
|
|
|
}
|
|
|
|
out.push(p);
|
|
|
|
}
|
|
|
|
return out;
|
|
|
|
}
|
|
|
|
|
2018-03-26 21:07:11 -07:00
|
|
|
#[cfg(test)]
|
|
|
|
mod tests {
|
2018-04-11 17:30:53 -07:00
|
|
|
use accountant_skel::{to_packets, Request};
|
2018-03-26 21:07:11 -07:00
|
|
|
use bincode::serialize;
|
|
|
|
use ecdsa;
|
2018-04-11 17:30:53 -07:00
|
|
|
use packet::{PacketRecycler, NUM_PACKETS};
|
2018-03-26 21:07:11 -07:00
|
|
|
use transaction::{memfind, test_tx};
|
2018-04-11 14:44:50 -07:00
|
|
|
|
|
|
|
use accountant::Accountant;
|
2018-04-12 16:04:11 -07:00
|
|
|
use accountant_skel::AccountantSkel;
|
|
|
|
use accountant_stub::AccountantStub;
|
2018-04-12 19:53:34 -07:00
|
|
|
use entry::Entry;
|
2018-04-24 10:15:08 -07:00
|
|
|
use futures::Future;
|
2018-04-11 14:44:50 -07:00
|
|
|
use historian::Historian;
|
2018-04-12 16:04:11 -07:00
|
|
|
use mint::Mint;
|
|
|
|
use plan::Plan;
|
2018-04-12 19:53:34 -07:00
|
|
|
use recorder::Signal;
|
2018-04-12 16:04:11 -07:00
|
|
|
use signature::{KeyPair, KeyPairUtil};
|
|
|
|
use std::io::sink;
|
2018-04-12 19:53:34 -07:00
|
|
|
use std::net::{SocketAddr, UdpSocket};
|
2018-04-11 14:44:50 -07:00
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
2018-04-12 16:04:11 -07:00
|
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use std::thread::sleep;
|
2018-04-11 14:44:50 -07:00
|
|
|
use std::time::Duration;
|
|
|
|
use transaction::Transaction;
|
|
|
|
|
2018-03-26 21:07:11 -07:00
|
|
|
#[test]
|
|
|
|
fn test_layout() {
|
|
|
|
let tr = test_tx();
|
|
|
|
let tx = serialize(&tr).unwrap();
|
|
|
|
let packet = serialize(&Request::Transaction(tr)).unwrap();
|
|
|
|
assert_matches!(memfind(&packet, &tx), Some(ecdsa::TX_OFFSET));
|
2018-04-11 17:30:53 -07:00
|
|
|
assert_matches!(memfind(&packet, &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]), None);
|
|
|
|
}
|
|
|
|
#[test]
|
|
|
|
fn test_to_packets() {
|
|
|
|
let tr = Request::Transaction(test_tx());
|
|
|
|
let re = PacketRecycler::default();
|
|
|
|
let rv = to_packets(&re, vec![tr.clone(); 1]);
|
|
|
|
assert_eq!(rv.len(), 1);
|
|
|
|
assert_eq!(rv[0].read().unwrap().packets.len(), 1);
|
|
|
|
|
|
|
|
let rv = to_packets(&re, vec![tr.clone(); NUM_PACKETS]);
|
|
|
|
assert_eq!(rv.len(), 1);
|
|
|
|
assert_eq!(rv[0].read().unwrap().packets.len(), NUM_PACKETS);
|
|
|
|
|
|
|
|
let rv = to_packets(&re, vec![tr.clone(); NUM_PACKETS + 1]);
|
|
|
|
assert_eq!(rv.len(), 2);
|
|
|
|
assert_eq!(rv[0].read().unwrap().packets.len(), NUM_PACKETS);
|
|
|
|
assert_eq!(rv[1].read().unwrap().packets.len(), 1);
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
2018-04-11 14:44:50 -07:00
|
|
|
|
2018-04-12 19:53:34 -07:00
|
|
|
#[test]
|
|
|
|
fn test_accounting_sequential_consistency() {
|
|
|
|
// In this attack we'll demonstrate that a verifier can interpret the ledger
|
|
|
|
// differently if either the server doesn't signal the ledger to add an
|
|
|
|
// Entry OR if the verifier tries to parallelize across multiple Entries.
|
|
|
|
let mint = Mint::new(2);
|
|
|
|
let acc = Accountant::new(&mint);
|
|
|
|
let rsp_addr: SocketAddr = "0.0.0.0:0".parse().expect("socket address");
|
|
|
|
let historian = Historian::new(&mint.last_id(), None);
|
|
|
|
let mut skel = AccountantSkel::new(acc, mint.last_id(), sink(), historian);
|
|
|
|
|
|
|
|
// Process a batch that includes a transaction that receives two tokens.
|
|
|
|
let alice = KeyPair::new();
|
|
|
|
let tr = Transaction::new(&mint.keypair(), alice.pubkey(), 2, mint.last_id());
|
|
|
|
let req_vers = vec![(Request::Transaction(tr), rsp_addr, 1_u8)];
|
|
|
|
assert!(skel.process_packets(req_vers).is_ok());
|
|
|
|
|
|
|
|
// Process a second batch that spends one of those tokens.
|
|
|
|
let tr = Transaction::new(&alice, mint.pubkey(), 1, mint.last_id());
|
|
|
|
let req_vers = vec![(Request::Transaction(tr), rsp_addr, 1_u8)];
|
|
|
|
assert!(skel.process_packets(req_vers).is_ok());
|
|
|
|
|
|
|
|
// Collect the ledger and feed it to a new accountant.
|
|
|
|
skel.historian.sender.send(Signal::Tick).unwrap();
|
|
|
|
drop(skel.historian.sender);
|
|
|
|
let entries: Vec<Entry> = skel.historian.receiver.iter().collect();
|
|
|
|
|
|
|
|
// Assert the user holds one token, not two. If the server only output one
|
|
|
|
// entry, then the second transaction will be rejected, because it drives
|
|
|
|
// the account balance below zero before the credit is added.
|
|
|
|
let acc = Accountant::new(&mint);
|
|
|
|
for entry in entries {
|
|
|
|
acc.process_verified_events(entry.events).unwrap();
|
|
|
|
}
|
|
|
|
assert_eq!(acc.get_balance(&alice.pubkey()), Some(1));
|
|
|
|
}
|
|
|
|
|
2018-04-11 14:44:50 -07:00
|
|
|
#[test]
|
|
|
|
fn test_accountant_bad_sig() {
|
|
|
|
let serve_port = 9002;
|
|
|
|
let send_port = 9003;
|
|
|
|
let addr = format!("127.0.0.1:{}", serve_port);
|
|
|
|
let send_addr = format!("127.0.0.1:{}", send_port);
|
|
|
|
let alice = Mint::new(10_000);
|
|
|
|
let acc = Accountant::new(&alice);
|
|
|
|
let bob_pubkey = KeyPair::new().pubkey();
|
|
|
|
let exit = Arc::new(AtomicBool::new(false));
|
|
|
|
let historian = Historian::new(&alice.last_id(), Some(30));
|
|
|
|
let acc = Arc::new(Mutex::new(AccountantSkel::new(
|
|
|
|
acc,
|
|
|
|
alice.last_id(),
|
|
|
|
sink(),
|
|
|
|
historian,
|
|
|
|
)));
|
|
|
|
let _threads = AccountantSkel::serve(&acc, &addr, exit.clone()).unwrap();
|
|
|
|
sleep(Duration::from_millis(300));
|
|
|
|
|
|
|
|
let socket = UdpSocket::bind(send_addr).unwrap();
|
|
|
|
socket.set_read_timeout(Some(Duration::new(5, 0))).unwrap();
|
|
|
|
|
|
|
|
let acc = AccountantStub::new(&addr, socket);
|
2018-04-24 10:15:08 -07:00
|
|
|
let last_id = acc.get_last_id().wait().unwrap();
|
2018-04-11 14:44:50 -07:00
|
|
|
|
|
|
|
let tr = Transaction::new(&alice.keypair(), bob_pubkey, 500, last_id);
|
|
|
|
|
|
|
|
let _sig = acc.transfer_signed(tr).unwrap();
|
|
|
|
|
2018-04-24 10:15:08 -07:00
|
|
|
let last_id = acc.get_last_id().wait().unwrap();
|
2018-04-11 14:44:50 -07:00
|
|
|
|
|
|
|
let mut tr2 = Transaction::new(&alice.keypair(), bob_pubkey, 501, last_id);
|
|
|
|
tr2.data.tokens = 502;
|
|
|
|
tr2.data.plan = Plan::new_payment(502, bob_pubkey);
|
|
|
|
let _sig = acc.transfer_signed(tr2).unwrap();
|
|
|
|
|
2018-04-24 10:15:08 -07:00
|
|
|
assert_eq!(acc.get_balance(&bob_pubkey).wait().unwrap(), 500);
|
2018-04-11 14:44:50 -07:00
|
|
|
exit.store(true, Ordering::Relaxed);
|
|
|
|
}
|
|
|
|
|
2018-02-28 09:07:54 -08:00
|
|
|
}
|
2018-04-11 13:05:29 -07:00
|
|
|
|
|
|
|
#[cfg(all(feature = "unstable", test))]
|
|
|
|
mod bench {
|
|
|
|
extern crate test;
|
|
|
|
use self::test::Bencher;
|
|
|
|
use accountant::{Accountant, MAX_ENTRY_IDS};
|
2018-04-11 19:24:14 -07:00
|
|
|
use accountant_skel::*;
|
|
|
|
use bincode::serialize;
|
|
|
|
use hash::hash;
|
2018-04-11 13:05:29 -07:00
|
|
|
use mint::Mint;
|
2018-04-11 19:24:14 -07:00
|
|
|
use signature::{KeyPair, KeyPairUtil};
|
2018-04-11 13:05:29 -07:00
|
|
|
use std::collections::HashSet;
|
|
|
|
use std::io::sink;
|
|
|
|
use std::time::Instant;
|
2018-04-11 19:24:14 -07:00
|
|
|
use transaction::Transaction;
|
2018-04-11 13:05:29 -07:00
|
|
|
|
|
|
|
#[bench]
|
|
|
|
fn process_packets_bench(_bencher: &mut Bencher) {
|
|
|
|
let mint = Mint::new(100_000_000);
|
|
|
|
let acc = Accountant::new(&mint);
|
|
|
|
let rsp_addr: SocketAddr = "0.0.0.0:0".parse().expect("socket address");
|
|
|
|
// Create transactions between unrelated parties.
|
|
|
|
let txs = 100_000;
|
|
|
|
let last_ids: Mutex<HashSet<Hash>> = Mutex::new(HashSet::new());
|
|
|
|
let transactions: Vec<_> = (0..txs)
|
|
|
|
.into_par_iter()
|
|
|
|
.map(|i| {
|
|
|
|
// Seed the 'to' account and a cell for its signature.
|
|
|
|
let dummy_id = i % (MAX_ENTRY_IDS as i32);
|
|
|
|
let last_id = hash(&serialize(&dummy_id).unwrap()); // Semi-unique hash
|
|
|
|
{
|
|
|
|
let mut last_ids = last_ids.lock().unwrap();
|
|
|
|
if !last_ids.contains(&last_id) {
|
|
|
|
last_ids.insert(last_id);
|
|
|
|
acc.register_entry_id(&last_id);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Seed the 'from' account.
|
|
|
|
let rando0 = KeyPair::new();
|
|
|
|
let tr = Transaction::new(&mint.keypair(), rando0.pubkey(), 1_000, last_id);
|
|
|
|
acc.process_verified_transaction(&tr).unwrap();
|
|
|
|
|
|
|
|
let rando1 = KeyPair::new();
|
|
|
|
let tr = Transaction::new(&rando0, rando1.pubkey(), 2, last_id);
|
|
|
|
acc.process_verified_transaction(&tr).unwrap();
|
|
|
|
|
|
|
|
// Finally, return a transaction that's unique
|
|
|
|
Transaction::new(&rando0, rando1.pubkey(), 1, last_id)
|
|
|
|
})
|
|
|
|
.collect();
|
|
|
|
|
|
|
|
let req_vers = transactions
|
|
|
|
.into_iter()
|
|
|
|
.map(|tr| (Request::Transaction(tr), rsp_addr, 1_u8))
|
|
|
|
.collect();
|
|
|
|
|
|
|
|
let historian = Historian::new(&mint.last_id(), None);
|
|
|
|
let mut skel = AccountantSkel::new(acc, mint.last_id(), sink(), historian);
|
|
|
|
|
|
|
|
let now = Instant::now();
|
|
|
|
assert!(skel.process_packets(req_vers).is_ok());
|
|
|
|
let duration = now.elapsed();
|
|
|
|
let sec = duration.as_secs() as f64 + duration.subsec_nanos() as f64 / 1_000_000_000.0;
|
|
|
|
let tps = txs as f64 / sec;
|
|
|
|
|
|
|
|
// Ensure that all transactions were successfully logged.
|
|
|
|
drop(skel.historian.sender);
|
|
|
|
let entries: Vec<Entry> = skel.historian.receiver.iter().collect();
|
|
|
|
assert_eq!(entries.len(), 1);
|
|
|
|
assert_eq!(entries[0].events.len(), txs as usize);
|
|
|
|
|
|
|
|
println!("{} tps", tps);
|
|
|
|
}
|
|
|
|
}
|