reactor
/
ptth
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Compare commits

base: reactor:05ba4822f831cba1e3e15798722da766a8f0f46a
Branches Tags
reactor:main
..
compare: reactor:f867f32927f3998f282f4b7591215b9516a83ac8
Branches Tags
reactor:main

No commits in common. "05ba4822f831cba1e3e15798722da766a8f0f46a" and "f867f32927f3998f282f4b7591215b9516a83ac8" have entirely different histories.
05ba4822f8 ... f867f32927

7 changed files with 67 additions and 220 deletions
Show Stats Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -1328,7 +1328,6 @@ dependencies = [
"anyhow",
"base64",
"futures-util",
"hyper",
"quinn",
"rcgen",
"rmp-serde",

24
docs/how-to/test-ptth-quic.md
View File

@ -1,24 +0,0 @@
# How-to: Test PTTH_QUIC
## Initial setup
- Open 3 terminals in `prototypes/quic_demo`
- Use `export RUST_LOG=quic_demo_relay_server=debug` to enable debug logging
for the terminal that will run the relay server (P3)
- Use `export RUST_LOG=quic_demo_end_server=debug` for the terminal that
will run the end server (P4)
- Use `export RUST_LOG=quic_demo_client=debug` for the terminal that
will run the client (P2)
When the relay server is running, use curl to get the list of connected
end servers: `curl 127.0.0.1:4004`
## Test loop - Happy path
- Start a relay `cargo run --bin quic_demo_relay_server`
- Verify that the relay has no end servers connected
- Start an end server `cargo run --bin quic_demo_end_server -- --debug-echo`
- Verify that the end server connected
- Start a client `cargo run --bin quic_demo_client`
- Connect to the client and verify that the debug echo server is running
`nc 127.0.0.1 30381`

1
prototypes/quic_demo/Cargo.toml
View File

@ -12,7 +12,6 @@ anyhow = "1.0.38"
base64 = "0.13.0"
# fltk = "1.1.1"
futures-util = "0.3.9"
hyper = { version = "0.14.4", features = ["http1", "server", "stream", "tcp"] }
quinn = "0.7.2"
rcgen = "0.8.11"
rmp-serde = "0.15.5"

52
prototypes/quic_demo/src/bin/quic_demo_client.rs
View File

@ -46,45 +46,37 @@ async fn main () -> anyhow::Result <()> {
let server_tcp_port = opt.server_tcp_port.unwrap_or (30382);
let listener = TcpListener::bind (("127.0.0.1", client_tcp_port)).await?;
debug! ("Accepting local TCP connections from P1");
// End of per-port stuff
// Beginning of per-connection stuff
let task_tcp_server = tokio::spawn (async move {
loop {
let (tcp_socket, _) = listener.accept ().await?;
let connection = connection.clone ();
let server_id = server_id.clone ();
loop {
let (tcp_socket, _) = listener.accept ().await?;
let connection = connection.clone ();
let server_id = server_id.clone ();
tokio::spawn (async move {
let (local_recv, local_send) = tcp_socket.into_split ();
tokio::spawn (async move {
let (local_recv, local_send) = tcp_socket.into_split ();
debug! ("Starting PTTH connection");
debug! ("Starting PTTH connection");
let (relay_send, relay_recv) = protocol::p2_connect_to_p5 (&connection, &server_id, server_tcp_port).await?;
let (relay_send, relay_recv) = protocol::p2_connect_to_p5 (&connection, &server_id, server_tcp_port).await?;
trace! ("Relaying bytes...");
trace! ("Relaying bytes...");
let ptth_conn = quic_demo::connection::NewConnection {
local_send,
local_recv,
relay_send,
relay_recv,
}.build ();
let ptth_conn = quic_demo::connection::NewConnection {
local_send,
local_recv,
relay_send,
relay_recv,
}.build ();
ptth_conn.wait_for_close ().await?;
ptth_conn.wait_for_close ().await?;
debug! ("Ended PTTH connection");
debug! ("Ended PTTH connection");
Ok::<_, anyhow::Error> (())
});
}
Ok::<_, anyhow::Error> (())
});
debug! ("Accepting local TCP connections from P1");
task_tcp_server.await??;
Ok (())
Ok::<_, anyhow::Error> (())
});
}
}

42
prototypes/quic_demo/src/bin/quic_demo_end_server.rs
View File

@ -10,41 +10,37 @@ struct Opt {
relay_addr: Option <String>,
#[structopt (long)]
server_id: Option <PeerId>,
#[structopt (long)]
debug_echo: bool,
}
#[tokio::main]
async fn main () -> anyhow::Result <()> {
tracing_subscriber::fmt::init ();
let opt = Arc::new (Opt::from_args ());
let opt = Opt::from_args ();
let server_cert = tokio::fs::read ("quic_server.crt").await?;
let relay_addr = opt.relay_addr.clone ().unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
let relay_addr = opt.relay_addr.unwrap_or_else (|| String::from ("127.0.0.1:30380")).parse ()?;
let endpoint = make_client_endpoint ("0.0.0.0:0".parse ()?, &[&server_cert])?;
trace! ("Connecting to relay server");
let server_id = opt.server_id.clone ().unwrap_or_else (|| "bogus_server".to_string ());
let server_id = opt.server_id.unwrap_or_else (|| "bogus_server".to_string ());
let quinn::NewConnection {
mut bi_streams,
..
} = protocol::p4_connect_to_p3 (&endpoint, &relay_addr, &server_id).await?;
debug! ("Connected to relay server");
trace! ("Accepting bi streams from P3");
loop {
let (relay_send, relay_recv) = bi_streams.next ().await.ok_or_else (|| anyhow::anyhow! ("Relay server didn't open a bi stream"))??;
tokio::spawn (handle_bi_stream (Arc::clone (&opt), relay_send, relay_recv));
tokio::spawn (handle_bi_stream (relay_send, relay_recv));
}
}
async fn handle_bi_stream (
opt: Arc <Opt>,
relay_send: quinn::SendStream,
mut relay_recv: quinn::RecvStream,
) -> anyhow::Result <()>
@ -53,14 +49,13 @@ async fn handle_bi_stream (
protocol::P3ToP4Stream::NewPtthConnection {
client_id,
..
} => handle_new_ptth_connection (opt, relay_send, relay_recv, client_id).await?,
} => handle_new_ptth_connection (relay_send, relay_recv, client_id).await?,
}
Ok (())
}
async fn handle_new_ptth_connection (
opt: Arc <Opt>,
mut relay_send: quinn::SendStream,
mut relay_recv: quinn::RecvStream,
_client_id: String,
@ -77,26 +72,19 @@ async fn handle_new_ptth_connection (
debug! ("Started PTTH connection");
if opt.debug_echo {
relay_send.write (b"Connected to P4=P5 debug echo server\n").await?;
debug! ("Relaying bytes using internal debug echo server (P4=P5)");
tokio::io::copy (&mut relay_recv, &mut relay_send).await?;
}
else {
let stream = TcpStream::connect (("127.0.0.1", p4_to_p5_req.port)).await?;
let (local_recv, local_send) = stream.into_split ();
let stream = TcpStream::connect (("127.0.0.1", p4_to_p5_req.port)).await?;
let (local_recv, local_send) = stream.into_split ();
trace! ("Relaying bytes...");
trace! ("Relaying bytes...");
let ptth_conn = quic_demo::connection::NewConnection {
local_send,
local_recv,
relay_send,
relay_recv,
}.build ();
let ptth_conn = quic_demo::connection::NewConnection {
local_send,
local_recv,
relay_send,
relay_recv,
}.build ();
ptth_conn.wait_for_close ().await?;
}
ptth_conn.wait_for_close ().await?;
Ok (())
}

135
prototypes/quic_demo/src/bin/quic_demo_relay_server.rs
View File

@ -1,14 +1,3 @@
use hyper::{
Body,
Request,
Response,
Server,
service::{
make_service_fn,
service_fn,
},
StatusCode,
};
use structopt::StructOpt;
use quic_demo::prelude::*;
@ -31,120 +20,28 @@ async fn main () -> anyhow::Result <()> {
println! ("Base64 cert: {}", base64::encode (&server_cert));
tokio::fs::write ("quic_server.crt", &server_cert).await?;
let relay_state = Arc::new (RelayState::default ());
let relay_state = RelayState::default ();
let relay_state = Arc::new (relay_state);
let make_svc = {
while let Some (conn) = incoming.next ().await {
let relay_state = Arc::clone (&relay_state);
make_service_fn (move |_conn| {
let relay_state = Arc::clone (&relay_state);
async move {
Ok::<_, String> (service_fn (move |req| {
let relay_state = Arc::clone (&relay_state);
handle_http (req, relay_state)
}))
}
})
};
let http_addr = SocketAddr::from (([0, 0, 0, 0], 4004));
let http_server = Server::bind (&http_addr);
let tcp_port = 30382;
let tcp_listener = TcpListener::bind (("127.0.0.1", tcp_port)).await?;
let task_quic_server = {
let relay_state = Arc::clone (&relay_state);
// Each new peer QUIC connection gets its own task
tokio::spawn (async move {
while let Some (conn) = incoming.next ().await {
let relay_state = Arc::clone (&relay_state);
let active = relay_state.stats.quic.connect ();
debug! ("QUIC connections: {}", active);
// Each new peer QUIC connection gets its own task
tokio::spawn (async move {
let active = relay_state.stats.quic.connect ();
debug! ("QUIC connections: {}", active);
match handle_quic_connection (Arc::clone (&relay_state), conn).await {
Ok (_) => (),
Err (e) => warn! ("handle_quic_connection {:?}", e),
}
let active = relay_state.stats.quic.disconnect ();
debug! ("QUIC connections: {}", active);
});
match handle_quic_connection (Arc::clone (&relay_state), conn).await {
Ok (_) => (),
Err (e) => warn! ("handle_quic_connection {:?}", e),
}
Ok::<_, anyhow::Error> (())
})
};
let task_http_server = tokio::spawn (async move {
http_server.serve (make_svc).await?;
Ok::<_, anyhow::Error> (())
});
let task_tcp_server = {
let relay_state = Arc::clone (&relay_state);
tokio::spawn (async move {
loop {
let (tcp_socket, _) = tcp_listener.accept ().await?;
let server_id = "bogus_server".to_string ();
let relay_state = Arc::clone (&relay_state);
tokio::spawn (async move {
let (client_recv, client_send) = tcp_socket.into_split ();
debug! ("Accepted direct TCP connection P1 --> P3");
let p4_server_proxies = relay_state.p4_server_proxies.lock ().await;
let p4 = match p4_server_proxies.get ("bogus_server") {
Some (x) => x,
None => bail! ("That server isn't connected"),
};
unimplemented! ();
/*
p4.req_channel.send (RequestP2ToP4 {
client_send,
client_recv,
client_id: "bogus_client".to_string (),
}).await.map_err (|_| anyhow::anyhow! ("Can't send request to P4 server"))?;
*/
Ok (())
});
}
Ok::<_, anyhow::Error> (())
})
};
debug! ("Serving HTTP on {:?}", http_addr);
task_quic_server.await??;
task_http_server.await??;
task_tcp_server.await??;
Ok (())
}
async fn handle_http (_req: Request <Body>, relay_state: Arc <RelayState>)
-> anyhow::Result <Response <Body>>
{
let debug_string;
{
let p4_server_proxies = relay_state.p4_server_proxies.lock ().await;
debug_string = format! ("{:#?}\n", p4_server_proxies.keys ().collect::<Vec<_>> ());
let active = relay_state.stats.quic.disconnect ();
debug! ("QUIC connections: {}", active);
});
}
let resp = Response::builder ()
.status (StatusCode::OK)
.header ("content-type", "text/plain")
.body (Body::from (debug_string))?;
Ok (resp)
Ok (())
}
#[derive (Default)]
@ -199,8 +96,8 @@ struct RequestP2ToP4 {
client_id: String,
}
struct PtthNewConnection <ClientSend: AsyncWrite + Unpin> {
client_send: ClientSend,
struct PtthNewConnection {
client_send: quinn::SendStream,
client_recv: quinn::RecvStream,
server_send: quinn::SendStream,
server_recv: quinn::RecvStream,
@ -211,7 +108,7 @@ struct PtthConnection {
downlink_task: JoinHandle <anyhow::Result <()>>,
}
impl <ClientSend: AsyncWrite + Unpin> PtthNewConnection <ClientSend> {
impl PtthNewConnection {
fn build (self) -> PtthConnection {
let Self {
mut client_send,

4
prototypes/quic_demo/src/prelude.rs
View File

@ -1,6 +1,5 @@
pub use std::{
collections::*,
net::SocketAddr,
sync::{
Arc,
atomic::{
@ -18,12 +17,9 @@ pub use anyhow::{
pub use futures_util::StreamExt;
pub use tokio::{
io::{
AsyncRead,
AsyncReadExt,
AsyncWrite,
AsyncWriteExt,
},
net::TcpListener,
sync::{
Mutex,
mpsc,