blockworks-foundation
/
solana-with-rpc-optimizations
mirror of https://github.com/blockworks-foundation/solana-with-rpc-optimizations.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Switch to UDP from TCP 

And remove all the sleep()'ing around.
This commit is contained in:
Greg Fitzgerald 2018-03-01 12:23:27 -07:00
parent 7111aa3b18
commit c9cc4b4369
5 changed files with 121 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/accountant.rs
View File

@ -2,7 +2,7 @@
//! event log to record transactions. Its users can deposit funds and
//! transfer funds to other users.
use log::{Event, PublicKey, Sha256Hash, Signature};
use log::{Entry, Event, PublicKey, Sha256Hash, Signature};
use historian::Historian;
use ring::signature::Ed25519KeyPair;
use std::sync::mpsc::{RecvError, SendError};
@ -51,7 +51,7 @@ impl Accountant {
}
}
pub fn sync(self: &mut Self) {
pub fn sync(self: &mut Self) -> Vec<Entry<u64>> {
let mut entries = vec![];
while let Ok(entry) = self.historian.receiver.try_recv() {
entries.push(entry);
@ -67,6 +67,8 @@ impl Accountant {
for e in &entries {
self.process_event(&e.event);
}
entries
}
pub fn deposit_signed(
@ -83,11 +85,11 @@ impl Accountant {
self: &Self,
n: u64,
keypair: &Ed25519KeyPair,
) -> Result<(), SendError<Event<u64>>> {
) -> Result<Signature, SendError<Event<u64>>> {
use log::{get_pubkey, sign_serialized};
let key = get_pubkey(keypair);
let sig = sign_serialized(&n, keypair);
self.deposit_signed(key, n, sig)
self.deposit_signed(key, n, sig).map(|_| sig)
}
pub fn transfer_signed(
@ -116,25 +118,41 @@ impl Accountant {
n: u64,
keypair: &Ed25519KeyPair,
to: PublicKey,
) -> Result<(), SendError<Event<u64>>> {
) -> Result<Signature, SendError<Event<u64>>> {
use log::{get_pubkey, sign_transaction_data};
let from = get_pubkey(keypair);
let sig = sign_transaction_data(&n, keypair, &to);
self.transfer_signed(from, to, n, sig)
self.transfer_signed(from, to, n, sig).map(|_| sig)
}
pub fn get_balance(self: &mut Self, pubkey: &PublicKey) -> Result<u64, RecvError> {
self.sync();
Ok(*self.balances.get(pubkey).unwrap_or(&0))
}
pub fn wait_on_signature(self: &mut Self, wait_sig: &Signature) {
use std::thread::sleep;
use std::time::Duration;
let mut entries = self.sync();
let mut found = false;
while !found {
found = entries.iter().any(|e| match e.event {
Event::Claim { sig, .. } => sig == *wait_sig,
Event::Transaction { sig, .. } => sig == *wait_sig,
_ => false,
});
if !found {
sleep(Duration::from_millis(30));
entries = self.sync();
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::thread::sleep;
use std::time::Duration;
use log::{generate_keypair, get_pubkey};
use historian::ExitReason;
@ -145,13 +163,13 @@ mod tests {
let alice_keypair = generate_keypair();
let bob_keypair = generate_keypair();
acc.deposit(10_000, &alice_keypair).unwrap();
acc.deposit(1_000, &bob_keypair).unwrap();
let sig = acc.deposit(1_000, &bob_keypair).unwrap();
acc.wait_on_signature(&sig);
sleep(Duration::from_millis(30));
let bob_pubkey = get_pubkey(&bob_keypair);
acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
let sig = acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
acc.wait_on_signature(&sig);
sleep(Duration::from_millis(30));
assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 1_500);
drop(acc.historian.sender);
@ -163,18 +181,20 @@ mod tests {
#[test]
fn test_invalid_transfer() {
use std::thread::sleep;
use std::time::Duration;
let zero = Sha256Hash::default();
let mut acc = Accountant::new(&zero, Some(2));
let alice_keypair = generate_keypair();
let bob_keypair = generate_keypair();
acc.deposit(10_000, &alice_keypair).unwrap();
acc.deposit(1_000, &bob_keypair).unwrap();
let sig = acc.deposit(1_000, &bob_keypair).unwrap();
acc.wait_on_signature(&sig);
sleep(Duration::from_millis(30));
let bob_pubkey = get_pubkey(&bob_keypair);
acc.transfer(10_001, &alice_keypair, bob_pubkey).unwrap();
sleep(Duration::from_millis(30));
let alice_pubkey = get_pubkey(&alice_keypair);
assert_eq!(acc.get_balance(&alice_pubkey).unwrap(), 10_000);
assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 1_000);
@ -192,10 +212,10 @@ mod tests {
let mut acc = Accountant::new(&zero, Some(2));
let keypair = generate_keypair();
acc.deposit(1, &keypair).unwrap();
acc.deposit(2, &keypair).unwrap();
let sig = acc.deposit(2, &keypair).unwrap();
acc.wait_on_signature(&sig);
let pubkey = get_pubkey(&keypair);
sleep(Duration::from_millis(30));
assert_eq!(acc.get_balance(&pubkey).unwrap(), 3);
drop(acc.historian.sender);
@ -211,13 +231,12 @@ mod tests {
let mut acc = Accountant::new(&zero, Some(2));
let alice_keypair = generate_keypair();
let bob_keypair = generate_keypair();
acc.deposit(10_000, &alice_keypair).unwrap();
let sig = acc.deposit(10_000, &alice_keypair).unwrap();
acc.wait_on_signature(&sig);
sleep(Duration::from_millis(30));
let bob_pubkey = get_pubkey(&bob_keypair);
acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
sleep(Duration::from_millis(30));
let sig = acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
acc.wait_on_signature(&sig);
assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 500);
drop(acc.historian.sender);

22
src/accountant_skel.rs
View File

@ -23,11 +23,15 @@ pub enum Request {
GetBalance {
key: PublicKey,
},
Wait {
sig: Signature,
},
}
#[derive(Serialize, Deserialize, Debug)]
pub enum Response {
Balance { key: PublicKey, val: u64 },
Confirmed { sig: Signature },
}
impl AccountantSkel {
@ -49,26 +53,28 @@ impl AccountantSkel {
let val = self.obj.get_balance(&key).unwrap();
Some(Response::Balance { key, val })
}
Request::Wait { sig } => {
self.obj.wait_on_signature(&sig);
Some(Response::Confirmed { sig })
}
}
}
/// TCP Server that forwards messages to Accountant methods.
/// UDP Server that forwards messages to Accountant methods.
pub fn serve(self: &mut Self, addr: &str) -> io::Result<()> {
use std::net::TcpListener;
use std::io::{Read, Write};
use std::net::UdpSocket;
use bincode::{deserialize, serialize};
let listener = TcpListener::bind(addr)?;
let socket = UdpSocket::bind(addr)?;
let mut buf = vec![0u8; 1024];
loop {
//println!("skel: Waiting for incoming connections...");
let (mut stream, _from_addr) = listener.accept()?;
let _sz = stream.read(&mut buf)?;
//println!("skel: Waiting for incoming packets...");
let (_sz, src) = socket.recv_from(&mut buf)?;
// TODO: Return a descriptive error message if deserialization fails.
let req = deserialize(&buf).expect("deserialize request");
if let Some(resp) = self.process_request(req) {
stream.write(&serialize(&resp).expect("serialize response"))?;
socket.send_to(&serialize(&resp).expect("serialize response"), &src)?;
}
}
}

63
src/accountant_stub.rs
View File

@ -2,9 +2,8 @@
//! event log to record transactions. Its users can deposit funds and
//! transfer funds to other users.
use std::net::TcpStream;
use std::net::UdpSocket;
use std::io;
use std::io::{Read, Write};
use bincode::{deserialize, serialize};
use log::{PublicKey, Signature};
use ring::signature::Ed25519KeyPair;
@ -12,12 +11,14 @@ use accountant_skel::{Request, Response};
pub struct AccountantStub {
pub addr: String,
pub socket: UdpSocket,
}
impl AccountantStub {
pub fn new(addr: &str) -> Self {
pub fn new(addr: &str, socket: UdpSocket) -> Self {
AccountantStub {
addr: addr.to_string(),
socket,
}
}
@ -29,15 +30,14 @@ impl AccountantStub {
) -> io::Result<usize> {
let req = Request::Deposit { key, val, sig };
let data = serialize(&req).unwrap();
let mut stream = TcpStream::connect(&self.addr)?;
stream.write(&data)
self.socket.send_to(&data, &self.addr)
}
pub fn deposit(self: &mut Self, n: u64, keypair: &Ed25519KeyPair) -> io::Result<usize> {
pub fn deposit(self: &mut Self, n: u64, keypair: &Ed25519KeyPair) -> io::Result<Signature> {
use log::{get_pubkey, sign_serialized};
let key = get_pubkey(keypair);
let sig = sign_serialized(&n, keypair);
self.deposit_signed(key, n, sig)
self.deposit_signed(key, n, sig).map(|_| sig)
}
pub fn transfer_signed(
@ -49,8 +49,7 @@ impl AccountantStub {
) -> io::Result<usize> {
let req = Request::Transfer { from, to, val, sig };
let data = serialize(&req).unwrap();
let mut stream = TcpStream::connect(&self.addr)?;
stream.write(&data)
self.socket.send_to(&data, &self.addr)
}
pub fn transfer(
@ -58,24 +57,39 @@ impl AccountantStub {
n: u64,
keypair: &Ed25519KeyPair,
to: PublicKey,
) -> io::Result<usize> {
) -> io::Result<Signature> {
use log::{get_pubkey, sign_transaction_data};
let from = get_pubkey(keypair);
let sig = sign_transaction_data(&n, keypair, &to);
self.transfer_signed(from, to, n, sig)
self.transfer_signed(from, to, n, sig).map(|_| sig)
}
pub fn get_balance(self: &mut Self, pubkey: &PublicKey) -> io::Result<u64> {
let mut stream = TcpStream::connect(&self.addr)?;
let req = Request::GetBalance { key: *pubkey };
let data = serialize(&req).expect("serialize GetBalance");
stream.write(&data)?;
self.socket.send_to(&data, &self.addr)?;
let mut buf = vec![0u8; 1024];
stream.read(&mut buf)?;
self.socket.recv_from(&mut buf)?;
let resp = deserialize(&buf).expect("deserialize balance");
let Response::Balance { key, val } = resp;
assert_eq!(key, *pubkey);
Ok(val)
if let Response::Balance { key, val } = resp {
assert_eq!(key, *pubkey);
return Ok(val);
}
Ok(0)
}
pub fn wait_on_signature(self: &mut Self, wait_sig: &Signature) -> io::Result<()> {
let req = Request::Wait { sig: *wait_sig };
let data = serialize(&req).unwrap();
self.socket.send_to(&data, &self.addr).map(|_| ())?;
let mut buf = vec![0u8; 1024];
self.socket.recv_from(&mut buf)?;
let resp = deserialize(&buf).expect("deserialize signature");
if let Response::Confirmed { sig } = resp {
assert_eq!(sig, *wait_sig);
}
Ok(())
}
}
@ -90,7 +104,8 @@ mod tests {
#[test]
fn test_accountant_stub() {
let addr = "127.0.0.1:8000";
let addr = "127.0.0.1:9000";
let send_addr = "127.0.0.1:9001";
spawn(move || {
let zero = Sha256Hash::default();
let acc = Accountant::new(&zero, None);
@ -100,17 +115,17 @@ mod tests {
sleep(Duration::from_millis(30));
let mut acc = AccountantStub::new(addr);
let socket = UdpSocket::bind(send_addr).unwrap();
let mut acc = AccountantStub::new(addr, socket);
let alice_keypair = generate_keypair();
let bob_keypair = generate_keypair();
acc.deposit(10_000, &alice_keypair).unwrap();
acc.deposit(1_000, &bob_keypair).unwrap();
let sig = acc.deposit(1_000, &bob_keypair).unwrap();
acc.wait_on_signature(&sig).unwrap();
sleep(Duration::from_millis(30));
let bob_pubkey = get_pubkey(&bob_keypair);
acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
sleep(Duration::from_millis(300));
let sig = acc.transfer(500, &alice_keypair, bob_pubkey).unwrap();
acc.wait_on_signature(&sig).unwrap();
assert_eq!(acc.get_balance(&bob_pubkey).unwrap(), 1_500);
}
}

46
src/bin/client-demo.rs
View File

@ -2,37 +2,41 @@ extern crate silk;
fn main() {
use silk::accountant_stub::AccountantStub;
use std::thread::sleep;
use std::time::Duration;
use std::time::Instant;
use std::net::UdpSocket;
use silk::log::{generate_keypair, get_pubkey};
let addr = "127.0.0.1:8000";
let mut acc = AccountantStub::new(addr);
let send_addr = "127.0.0.1:8001";
let socket = UdpSocket::bind(send_addr).unwrap();
let mut acc = AccountantStub::new(addr, socket);
let alice_keypair = generate_keypair();
let bob_keypair = generate_keypair();
let txs = 10_000;
println!("Depositing {} units in Alice's account...", txs);
acc.deposit(txs, &alice_keypair).unwrap();
//acc.deposit(1_000, &bob_keypair).unwrap();
println!("Done.");
sleep(Duration::from_millis(30));
let alice_pubkey = get_pubkey(&alice_keypair);
let bob_pubkey = get_pubkey(&bob_keypair);
println!("Transferring 1 unit {} times...", txs);
for _ in 0..txs {
acc.transfer(1, &alice_keypair, bob_pubkey).unwrap();
}
let txs = 10_000;
println!("Depositing {} units in Alice's account...", txs);
let sig = acc.deposit(txs, &alice_keypair).unwrap();
acc.wait_on_signature(&sig).unwrap();
assert_eq!(acc.get_balance(&alice_pubkey).unwrap(), txs);
println!("Done.");
sleep(Duration::from_millis(20));
let mut alice_val = acc.get_balance(&alice_pubkey).unwrap();
while alice_val > 0 {
println!("Checking on Alice's Balance {}", alice_val);
sleep(Duration::from_millis(20));
alice_val = acc.get_balance(&alice_pubkey).unwrap();
println!("Transferring 1 unit {} times...", txs);
let now = Instant::now();
let mut sig = sig;
for i in 0..txs {
sig = acc.transfer(1, &alice_keypair, bob_pubkey).unwrap();
if i % 200 == 1 {
acc.wait_on_signature(&sig).unwrap();
}
}
println!("Done. Checking balances.");
println!("Waiting for last transaction to be confirmed...",);
acc.wait_on_signature(&sig).unwrap();
let duration = now.elapsed();
let ns = duration.as_secs() * 1_000_000_000 + duration.subsec_nanos() as u64;
let tps = (txs * 1_000_000_000) as f64 / ns as f64;
println!("Done. {} tps!", tps);
println!(
"Alice's Final Balance {}",
acc.get_balance(&alice_pubkey).unwrap()

4
src/historian.rs
View File

@ -116,8 +116,8 @@ pub fn create_logger<T: 'static + Serialize + Clone + Debug + Send>(
impl<T: 'static + Serialize + Clone + Debug + Send> Historian<T> {
pub fn new(start_hash: &Sha256Hash, ms_per_tick: Option<u64>) -> Self {
use std::sync::mpsc::sync_channel;
let (sender, event_receiver) = sync_channel(4000);
let (entry_sender, receiver) = sync_channel(4000);
let (sender, event_receiver) = sync_channel(12000);
let (entry_sender, receiver) = sync_channel(12000);
let thread_hdl = create_logger(*start_hash, ms_per_tick, event_receiver, entry_sender);
Historian {
sender,