GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Finally won first fight with the borrow checker for concurrent code

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2019-12-10 13:05:44 +00:00
parent 385e41a920
commit 2c986efcb1
3 changed files with 290 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+115 -6
View File
@@ -240,11 +240,88 @@ name = "fuchsia-zircon-sys"
version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-channel"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-core"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-executor"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-io"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-macro"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)",
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-sink"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-task"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-util"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"memchr 2.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
"pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)",
"proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)",
"proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
"slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "generic-array"
version = "0.12.3"
@@ -429,8 +506,9 @@ version = "0.1.0"
dependencies = [
"clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)",
"curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
"futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"sphinx 0.1.0",
"tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
"tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
@@ -443,11 +521,31 @@ name = "pin-project-lite"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "pin-utils"
version = "0.1.0-alpha.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "ppv-lite86"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "proc-macro-hack"
version = "0.5.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "proc-macro-nested"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "proc-macro2"
version = "1.0.6"
@@ -628,7 +726,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "syn"
version = "1.0.9"
version = "1.0.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -646,7 +744,7 @@ dependencies = [
[[package]]
name = "tokio"
version = "0.2.1"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"bytes 0.5.2 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -672,7 +770,7 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
@@ -770,7 +868,15 @@ dependencies = [
"checksum fuchsia-cprng 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a06f77d526c1a601b7c4cdd98f54b5eaabffc14d5f2f0296febdc7f357c6d3ba"
"checksum fuchsia-zircon 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82"
"checksum fuchsia-zircon-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7"
"checksum futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b6f16056ecbb57525ff698bb955162d0cd03bee84e6241c27ff75c08d8ca5987"
"checksum futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fcae98ca17d102fd8a3603727b9259fcf7fa4239b603d2142926189bc8999b86"
"checksum futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "79564c427afefab1dfb3298535b21eda083ef7935b4f0ecbfcb121f0aec10866"
"checksum futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1e274736563f686a837a0568b478bdabfeaec2dca794b5649b04e2fe1627c231"
"checksum futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e676577d229e70952ab25f3945795ba5b16d63ca794ca9d2c860e5595d20b5ff"
"checksum futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "52e7c56c15537adb4f76d0b7a76ad131cb4d2f4f32d3b0bcabcbe1c7c5e87764"
"checksum futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "171be33efae63c2d59e6dbba34186fe0d6394fb378069a76dfd80fdcffd43c16"
"checksum futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0bae52d6b29cf440e298856fec3965ee6fa71b06aa7495178615953fd669e5f9"
"checksum futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c0d66274fb76985d3c62c886d1da7ac4c0903a8c9f754e8fe0f35a6a6cc39e76"
"checksum generic-array 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c68f0274ae0e023facc3c97b2e00f076be70e254bc851d972503b328db79b2ec"
"checksum getrandom 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "e7db7ca94ed4cd01190ceee0d8a8052f08a247aa1b469a7f68c6a3b71afcf407"
"checksum hermit-abi 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "307c3c9f937f38e3534b1d6447ecf090cafcc9744e4a6360e8b037b2cf5af120"
@@ -793,7 +899,10 @@ dependencies = [
"checksum num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)" = "76dac5ed2a876980778b8b85f75a71b6cbf0db0b1232ee12f826bccb00d09d72"
"checksum opaque-debug 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2839e79665f131bdb5782e51f2c6c9599c133c6098982a54c794358bf432529c"
"checksum pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f0af6cbca0e6e3ce8692ee19fb8d734b641899e07b68eb73e9bbbd32f1703991"
"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587"
"checksum ppv-lite86 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "74490b50b9fbe561ac330df47c08f3f33073d2d00c150f719147d7c54522fa1b"
"checksum proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)" = "ecd45702f76d6d3c75a80564378ae228a85f0b59d2f3ed43c91b4a69eb2ebfc5"
"checksum proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "369a6ed065f249a159e06c45752c780bda2fb53c995718f9e484d08daa9eb42e"
"checksum proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9c9e470a8dc4aeae2dee2f335e8f533e2d4b347e1434e5671afc49b054592f27"
"checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe"
"checksum rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)" = "3ae1b169243eaf61759b8475a998f0a385e42042370f3a7dbaf35246eacc8412"
@@ -814,9 +923,9 @@ dependencies = [
"checksum strsim 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "8ea5119cdb4c55b55d432abb513a0429384878c15dde60cc77b1c99de1a95a6a"
"checksum subtle 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2d67a5a62ba6e01cb2192ff309324cb4875d0c451d55fe2319433abe7a05a8ee"
"checksum subtle 2.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7c65d530b10ccaeac294f349038a597e435b18fb456aadd0840a623f83b9e941"
"checksum syn 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)" = "f89693ae015201f8de93fd96bde2d065f8bfc3f97ce006d5bc9f900b97c0c7c0"
"checksum syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)" = "dff0acdb207ae2fe6d5976617f887eb1e35a2ba52c13c7234c790960cdad9238"
"checksum textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d326610f408c7a4eb6f51c37c330e496b08506c9457c9d34287ecc38809fb060"
"checksum tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "beeef686ef92a222de07e089f455d9f8478bbba9651718f9e4b276babe829082"
"checksum tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)" = "bcced6bb623d4bff3739c176c415f13c418f426395c169c9c3cd9a492c715b16"
"checksum tokio-macros 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d5795a71419535c6dcecc9b6ca95bdd3c2d6142f7e8343d7beb9923f129aa87e"
"checksum typenum 1.11.2 (registry+https://github.com/rust-lang/crates.io-index)" = "6d2783fe2d6b8c1101136184eb41be8b1ad379e4657050b8aaff0c79ee7575f9"
"checksum unicode-width 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)" = "caaa9d531767d1ff2150b9332433f32a24622147e5ebb1f26409d5da67afd479"
Cargo.toml
+2 -1
View File
@@ -8,6 +8,7 @@ edition = "2018"
[dependencies]
sphinx = { path = "../sphinx" }
tokio = { version = "0.2", features = ["full"] }
tokio = { version = "0.2.4", features = ["full"] }
futures = "0.3.1"
curve25519-dalek = "1.2.3"
clap = "2.33.0"
src/provider/mod.rs
+173 -34
View File
@@ -1,65 +1,204 @@
use std::borrow::Borrow;
use std::cell::RefCell;
use std::net::SocketAddr;
use std::path::{PathBuf, Path};
use std::pin::Pin;
use std::sync::Arc;
use std::sync::RwLock;
use curve25519_dalek::scalar::Scalar;
use futures::StreamExt;
use sphinx::{ProcessedPacket, SphinxPacket};
use sphinx::route::{DestinationAddressBytes, SURBIdentifier};
use tokio::prelude::*;
use tokio::runtime::Runtime;
use curve25519_dalek::scalar::Scalar;
use std::net::SocketAddr;
// TODO: this will probably need to be moved elsewhere I imagine
// DUPLICATE WITH MIXNODE CODE!!!
#[derive(Debug)]
pub enum MixProcessingError {
SphinxRecoveryError,
ReceivedForwardHopError,
InvalidPayload,
NonMatchingRecipient,
StoreFailure,
}
impl From<sphinx::ProcessingError> for MixProcessingError {
// for time being just have a single error instance for all possible results of sphinx::ProcessingError
fn from(_: sphinx::ProcessingError) -> Self {
use MixProcessingError::*;
SphinxRecoveryError
}
}
// ProcessingData defines all data required to correctly unwrap sphinx packets
// Do note that we're copying this struct around and hence the secret_key.
// It might, or might not be, what we want
#[derive(Debug, Clone)]
struct ProcessingData {
secret_key: Scalar,
store_dir: PathBuf,
}
impl ProcessingData {
fn new(secret_key: Scalar, store_dir: PathBuf) -> Self {
ProcessingData {
secret_key,
store_dir,
}
}
fn add_arc_rwlock(self) -> Arc<RwLock<Self>> {
Arc::new(RwLock::new(self))
}
}
struct StoreData {
client_address: DestinationAddressBytes,
client_surb_id: SURBIdentifier,
message: Vec<u8>,
}
struct PacketProcessor(());
impl PacketProcessor {
fn process_sphinx_data_packet(packet_data: &[u8], processing_data: &ProcessingData) -> Result<StoreData, MixProcessingError> {
let packet = SphinxPacket::from_bytes(packet_data.to_vec())?;
let (client_address, client_surb_id, payload) = match packet.process(processing_data.secret_key) {
ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload) => (client_address, surb_id, payload),
_ => return Err(MixProcessingError::ReceivedForwardHopError),
};
let (payload_destination, message) = payload.try_recover_destination_and_plaintext().ok_or_else(|| MixProcessingError::InvalidPayload)?;
if client_address != payload_destination {
return Err(MixProcessingError::NonMatchingRecipient);
}
Ok(StoreData {
client_address,
client_surb_id,
message,
})
}
fn store_processed_data(store_data: StoreData, store_dir: &PathBuf) -> Result<(), MixProcessingError> {
Ok(())
}
}
pub struct ServiceProvider {
network_address: SocketAddr,
secret_key: Scalar,
store_dir: PathBuf,
}
impl ServiceProvider {
pub fn new(network_address: SocketAddr, secret_key: Scalar) -> Self {
pub fn new(network_address: SocketAddr, secret_key: Scalar, store_dir: PathBuf) -> Self {
ServiceProvider {
network_address,
secret_key,
store_dir,
}
}
async fn process_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc<RwLock<ProcessingData>>) {
println!("look, we can read shared data here! {:?}", processing_data.read().unwrap());
// NOTE: processing_data is copied here!!
// let mut buf = [0u8; sphinx::PACKET_SIZE];
//
// // In a loop, read data from the socket and write the data back.
// loop {
// match socket.read(&mut buf).await {
// // socket closed
// Ok(n) if n == 0 => {
// println!("Remote connection closed.");
// return;
// }
// Ok(_) => {
//// let store_data = PacketProcessor::process_sphinx_data_packet(buf.as_ref(),
//// processing_data.read().unwrap().borrow()).unwrap();
// }
// Err(e) => {
// println!("failed to read from socket; err = {:?}", e);
// return;
// }
// };
//
// // Write the some data back
// if let Err(e) = socket.write_all(b"foomp").await {
// println!("failed to write reply to socket; err = {:?}", e);
// return;
// }
// }
}
pub fn start_listening(&self) -> Result<(), Box<dyn std::error::Error>> {
// Create the runtime, probably later move it to Provider struct itself?
// TODO: figure out the difference between Runtime and Handle
let mut rt = Runtime::new()?;
// let mut h = rt.handle();
// Spawn the root task
rt.block_on(async {
let my_address = "127.0.0.1:8081";
let mut listener = tokio::net::TcpListener::bind(my_address).await?;
println!("Starting Nym store-and-forward Provider on address {:?}", my_address);
println!("Waiting for input...");
let mut listener = tokio::net::TcpListener::bind(self.network_address).await?;
let processing_data = ProcessingData::new(self.secret_key, self.store_dir.clone()).add_arc_rwlock();
loop {
let (mut inbound, _) = listener.accept().await?;
let (socket, _) = listener.accept().await?;
let thread_processing_data = processing_data.clone();
tokio::spawn(async move {
let mut buf = [0; sphinx::PACKET_SIZE];
loop {
match inbound.read(&mut buf).await {
Ok(length) if length == 0 =>
{
println!("Remote connection closed.");
return;
}
Ok(_) => {
let packet = SphinxPacket::from_bytes(buf.to_vec()).unwrap();
let payload = match packet.process(Default::default()) {
ProcessedPacket::ProcessedPacketFinalHop(_, _, payload) => Some(payload),
_ => None,
}.unwrap();
let message = payload.get_content();
}
Err(e) => {
println!("failed to read from socket; err = {:?}", e);
return;
}
};
}
ServiceProvider::process_socket_connection(socket, thread_processing_data).await
});
}
// async move {
// let mut incoming = listener.incoming();
//
// while let Some(conn) = incoming.next().await {
// match conn {
// Err(e) => eprintln!("accept failed with error: {:?}", e),
// Ok(socket) => {
// let foomp2 = processing_data_foomp.clone();
// tokio::spawn(async move {
// ServiceProvider::process_socket_connection(socket, foomp2).await
//
//// ServiceProvider::process_socket_fixture(socket).await
// });
// }
// }
// }
//
// }.await;
//
// println!("Server went kaput");
// Ok(())
//// let processing_data = Arc::new(RwLock::new(ProcessingData::new(self.secret_key, self.store_dir.clone())));
// let processing_data = Arc::new(RwLock::new((ProcessingData::new(self.secret_key))));
//
// loop {
// let (socket, _) = listener.accept().await?;
//
//// tokio::
// tokio::spawn(async move {
//// processing_data.read();
//// let foo = processing_data.clone();
//// let foo = ProcessingData::new(self.secret_key.clone());
// // Process each socket concurrently.
//// ServiceProvider::process_socket_connection(socket, processing_data).await
// });
// }
})
}
}