Adds graceful shutdown for all tasks, rdsys stream inconsistent shutdown
This commit is contained in:
onyinyang
parent
9616b95b5d
commit
9e78a90541
|
|
@ -18,7 +18,7 @@ use lox::{
|
|||
};
|
||||
use lox::{BridgeAuth, BridgeDb, OPENINV_LENGTH};
|
||||
use rand::RngCore;
|
||||
use rdsys_backend::{proto::ResourceDiff, start_stream, ResourceStream};
|
||||
use rdsys_backend::{proto::ResourceDiff, start_stream};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::{
|
||||
convert::Infallible,
|
||||
|
|
@ -29,13 +29,12 @@ use std::{
|
|||
sync::{Arc, Mutex},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use serde_json;
|
||||
use serde_with::serde_as;
|
||||
use tokio::{
|
||||
spawn,
|
||||
sync::{mpsc, oneshot},
|
||||
time::sleep, task::yield_now,
|
||||
sync::{broadcast, mpsc, oneshot},
|
||||
time::sleep, signal,
|
||||
};
|
||||
|
||||
#[serde_as]
|
||||
|
|
@ -373,9 +372,21 @@ async fn shutdown_signal() {
|
|||
tokio::signal::ctrl_c()
|
||||
.await
|
||||
.expect("failed to listen for ctrl+c signal");
|
||||
println!("Shut down Lox Server");
|
||||
}
|
||||
|
||||
async fn rdsys_sender(rstream: ResourceStream, tx: Sender<ResourceDiff>) {
|
||||
async fn rdsys_stream(rtype: ResourceInfo, tx: Sender<ResourceDiff>, mut kill: broadcast::Receiver<()>) {
|
||||
tokio:: select! {
|
||||
start_rdsys_stream = rdsys_sender(rtype, tx) => start_rdsys_stream ,
|
||||
_ = kill.recv() => {println!("Shut down rdsys stream"); return},
|
||||
}
|
||||
}
|
||||
|
||||
async fn rdsys_sender(rtype: ResourceInfo, tx: Sender<ResourceDiff>) {
|
||||
let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types)
|
||||
.await
|
||||
.expect("rdsys stream initialization failed. Start rdsys or check config.json");
|
||||
sleep(Duration::from_millis(1)).await;
|
||||
for diff in rstream {
|
||||
println!("Received diff: {:?}", diff); //send this through a channel
|
||||
tx.send(diff).await.unwrap();
|
||||
|
|
@ -383,6 +394,13 @@ async fn rdsys_sender(rstream: ResourceStream, tx: Sender<ResourceDiff>) {
|
|||
}
|
||||
}
|
||||
|
||||
async fn rdsys_bridge_parser(rdsys_tx: mpsc::Sender<Command>, rx: Receiver<ResourceDiff>, mut kill: broadcast::Receiver<()>) {
|
||||
tokio:: select! {
|
||||
start_bridge_parser = parse_bridges(rdsys_tx, rx) => start_bridge_parser ,
|
||||
_ = kill.recv() => {println!("Shut down bridge_parser"); return},
|
||||
}
|
||||
}
|
||||
|
||||
async fn parse_bridges(rdsys_tx: mpsc::Sender<Command>, rx: Receiver<ResourceDiff>) {
|
||||
loop {
|
||||
let resourcediff = rx.recv().await.unwrap();
|
||||
|
|
@ -394,6 +412,131 @@ async fn parse_bridges(rdsys_tx: mpsc::Sender<Command>, rx: Receiver<ResourceDif
|
|||
}
|
||||
}
|
||||
|
||||
async fn create_context_manager(context_rx: mpsc::Receiver<Command>, mut kill: broadcast::Receiver<()>) {
|
||||
tokio:: select! {
|
||||
create_context = context_manager(context_rx) => create_context,
|
||||
_ = kill.recv() => {println!("Shut down context_manager"); return},
|
||||
}
|
||||
}
|
||||
|
||||
async fn context_manager(mut context_rx: mpsc::Receiver<Command>) {
|
||||
// pass in distribution of open invite vs. hot spare buckets?
|
||||
|
||||
let bridgedb = BridgeDb::new();
|
||||
let lox_auth = BridgeAuth::new(bridgedb.pubkey);
|
||||
|
||||
let context = LoxServerContext {
|
||||
db: Arc::new(Mutex::new(bridgedb)),
|
||||
ba: Arc::new(Mutex::new(lox_auth)),
|
||||
};
|
||||
|
||||
while let Some(cmd) = context_rx.recv().await {
|
||||
use Command::*;
|
||||
|
||||
match cmd {
|
||||
Rdsys { resourcediff } => {
|
||||
for new_resource in resourcediff.new {
|
||||
for pt in new_resource {
|
||||
println!("A NEW RESOURCE: {:?}", pt);
|
||||
let mut bucket = [
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
];
|
||||
let mut count = 0;
|
||||
for resource in pt.1 {
|
||||
let mut ip_bytes: [u8; 16] = [0; 16];
|
||||
ip_bytes[..resource.address.len()]
|
||||
.copy_from_slice(resource.address.as_bytes());
|
||||
let infostr: String = format!(
|
||||
"type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}",
|
||||
resource.r#type,
|
||||
resource.blocked_in,
|
||||
resource.protocol,
|
||||
resource.fingerprint,
|
||||
resource.or_addresses,
|
||||
resource.distribution,
|
||||
resource.flags,
|
||||
resource.params,
|
||||
);
|
||||
let mut info_bytes: [u8; BRIDGE_BYTES - 18] =
|
||||
[0; BRIDGE_BYTES - 18];
|
||||
|
||||
info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
|
||||
let bridgeline = BridgeLine {
|
||||
addr: ip_bytes,
|
||||
port: resource.port,
|
||||
info: info_bytes,
|
||||
};
|
||||
|
||||
println!("Now it's a bridgeline: {:?}", bridgeline);
|
||||
if count < 2 {
|
||||
bucket[count] = bridgeline;
|
||||
count += 1;
|
||||
} else {
|
||||
context.add_openinv_bucket(bucket);
|
||||
count = 0;
|
||||
bucket = [
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for changed_resource in resourcediff.changed {
|
||||
println!("A NEW CHANGED RESOURCE: {:?}", changed_resource);
|
||||
}
|
||||
for gone_resource in resourcediff.gone {
|
||||
for pt in gone_resource {
|
||||
println!("A NEW GONE RESOURCE: {:?}", pt);
|
||||
for resource in pt.1 {
|
||||
let mut ip_bytes: [u8; 16] = [0; 16];
|
||||
ip_bytes[..resource.address.len()]
|
||||
.copy_from_slice(resource.address.as_bytes());
|
||||
let infostr: String = format!(
|
||||
"type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}",
|
||||
resource.r#type,
|
||||
resource.blocked_in,
|
||||
resource.protocol,
|
||||
resource.fingerprint,
|
||||
resource.or_addresses,
|
||||
resource.distribution,
|
||||
resource.flags,
|
||||
resource.params,
|
||||
);
|
||||
let mut info_bytes: [u8; BRIDGE_BYTES - 18] =
|
||||
[0; BRIDGE_BYTES - 18];
|
||||
|
||||
info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
|
||||
let bridgeline = BridgeLine {
|
||||
addr: ip_bytes,
|
||||
port: resource.port,
|
||||
info: info_bytes,
|
||||
};
|
||||
|
||||
println!("Now it's a bridgeline: {:?}", bridgeline);
|
||||
context.add_unreachable(bridgeline);
|
||||
}
|
||||
}
|
||||
}
|
||||
context.encrypt_table();
|
||||
sleep(Duration::from_millis(1)).await;
|
||||
}
|
||||
Request { req, sender } => {
|
||||
let response = handle(context.clone(), req).await;
|
||||
if let Err(e) = sender.send(response) {
|
||||
eprintln!("Server Response Error: {:?}", e);
|
||||
};
|
||||
sleep(Duration::from_millis(1)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum Command {
|
||||
Rdsys {
|
||||
|
|
@ -405,7 +548,6 @@ enum Command {
|
|||
},
|
||||
}
|
||||
|
||||
// Run with cargo run -- config.json
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
let args: Vec<String> = env::args().collect();
|
||||
|
|
@ -414,133 +556,38 @@ async fn main() {
|
|||
// Read the JSON contents of the file as a ResourceInfo
|
||||
let rtype: ResourceInfo = serde_json::from_reader(reader).unwrap();
|
||||
|
||||
let (rdsys_tx, mut context_rx) = mpsc::channel(32);
|
||||
let (rdsys_tx, context_rx) = mpsc::channel(32);
|
||||
let request_tx = rdsys_tx.clone();
|
||||
|
||||
let context_manager = spawn(async move {
|
||||
// pass in distribution of open invite vs. hot spare buckets?
|
||||
let bridgedb = BridgeDb::new();
|
||||
let lox_auth = BridgeAuth::new(bridgedb.pubkey);
|
||||
|
||||
let context = LoxServerContext {
|
||||
db: Arc::new(Mutex::new(bridgedb)),
|
||||
ba: Arc::new(Mutex::new(lox_auth)),
|
||||
};
|
||||
|
||||
while let Some(cmd) = context_rx.recv().await {
|
||||
use Command::*;
|
||||
// create the shutdown broadcast channel
|
||||
let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16);
|
||||
let kill_stream = shutdown_tx.subscribe();
|
||||
let kill_parser = shutdown_tx.subscribe();
|
||||
let kill_context= shutdown_tx.subscribe();
|
||||
|
||||
match cmd {
|
||||
Rdsys { resourcediff } => {
|
||||
for new_resource in resourcediff.new {
|
||||
for pt in new_resource {
|
||||
println!("A NEW RESOURCE: {:?}", pt);
|
||||
let mut bucket = [
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
];
|
||||
let mut count = 0;
|
||||
for resource in pt.1 {
|
||||
let mut ip_bytes: [u8; 16] = [0; 16];
|
||||
ip_bytes[..resource.address.len()]
|
||||
.copy_from_slice(resource.address.as_bytes());
|
||||
let infostr: String = format!(
|
||||
"type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}",
|
||||
resource.r#type,
|
||||
resource.blocked_in,
|
||||
resource.protocol,
|
||||
resource.fingerprint,
|
||||
resource.or_addresses,
|
||||
resource.distribution,
|
||||
resource.flags,
|
||||
resource.params,
|
||||
);
|
||||
let mut info_bytes: [u8; BRIDGE_BYTES - 18] =
|
||||
[0; BRIDGE_BYTES - 18];
|
||||
// Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx
|
||||
let shutdown_handler = spawn(async move {
|
||||
tokio::select! {
|
||||
_ = signal::ctrl_c() => {
|
||||
drop(shutdown_tx);
|
||||
println!("Kill Sent, all threads should shutdown.");
|
||||
|
||||
info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
|
||||
let bridgeline = BridgeLine {
|
||||
addr: ip_bytes,
|
||||
port: resource.port,
|
||||
info: info_bytes,
|
||||
};
|
||||
|
||||
println!("Now it's a bridgeline: {:?}", bridgeline);
|
||||
if count < 2 {
|
||||
bucket[count] = bridgeline;
|
||||
count += 1;
|
||||
} else {
|
||||
context.add_openinv_bucket(bucket);
|
||||
count = 0;
|
||||
bucket = [
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
BridgeLine::default(),
|
||||
];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for changed_resource in resourcediff.changed {
|
||||
println!("A NEW CHANGED RESOURCE: {:?}", changed_resource);
|
||||
}
|
||||
for gone_resource in resourcediff.gone {
|
||||
for pt in gone_resource {
|
||||
println!("A NEW GONE RESOURCE: {:?}", pt);
|
||||
for resource in pt.1 {
|
||||
let mut ip_bytes: [u8; 16] = [0; 16];
|
||||
ip_bytes[..resource.address.len()]
|
||||
.copy_from_slice(resource.address.as_bytes());
|
||||
let infostr: String = format!(
|
||||
"type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}",
|
||||
resource.r#type,
|
||||
resource.blocked_in,
|
||||
resource.protocol,
|
||||
resource.fingerprint,
|
||||
resource.or_addresses,
|
||||
resource.distribution,
|
||||
resource.flags,
|
||||
resource.params,
|
||||
);
|
||||
let mut info_bytes: [u8; BRIDGE_BYTES - 18] =
|
||||
[0; BRIDGE_BYTES - 18];
|
||||
|
||||
info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes());
|
||||
let bridgeline = BridgeLine {
|
||||
addr: ip_bytes,
|
||||
port: resource.port,
|
||||
info: info_bytes,
|
||||
};
|
||||
|
||||
println!("Now it's a bridgeline: {:?}", bridgeline);
|
||||
context.add_unreachable(bridgeline);
|
||||
}
|
||||
}
|
||||
}
|
||||
context.encrypt_table();
|
||||
sleep(Duration::from_millis(1)).await;
|
||||
}
|
||||
Request { req, sender } => {
|
||||
let response = handle(context.clone(), req).await;
|
||||
if let Err(e) = sender.send(response) {
|
||||
eprintln!("Server Response Error: {:?}", e);
|
||||
};
|
||||
sleep(Duration::from_millis(1)).await;
|
||||
}
|
||||
_ = shutdown_rx.recv().await;
|
||||
println!("Receiving shutdown signals. . .");
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
|
||||
let context_manager = spawn(async move { create_context_manager(context_rx, kill_context).await });
|
||||
|
||||
//Sender is resource stream and receiver is bridgedb function (add_openinv_bridges)
|
||||
let (tx, rx) = async_channel::unbounded();
|
||||
// to populate the bridge db
|
||||
let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types)
|
||||
.await
|
||||
.unwrap();
|
||||
let rdsys_stream_handler = spawn(async { rdsys_sender(rstream, tx).await });
|
||||
let rdsys_stream_handler = spawn(async { rdsys_stream(rtype, tx, kill_stream).await });
|
||||
|
||||
let rdsys_resource_receiver = spawn(async { parse_bridges(rdsys_tx, rx).await });
|
||||
let rdsys_resource_receiver = spawn(async { rdsys_bridge_parser(rdsys_tx, rx, kill_parser).await });
|
||||
|
||||
let make_service = make_service_fn(move |_conn: &AddrStream| {
|
||||
let request_tx = request_tx.clone();
|
||||
|
|
@ -561,11 +608,18 @@ async fn main() {
|
|||
|
||||
let addr = SocketAddr::from(([127, 0, 0, 1], 8001));
|
||||
let server = Server::bind(&addr).serve(make_service);
|
||||
let graceful = server.with_graceful_shutdown(shutdown_signal());
|
||||
println!("Listening on {}", addr);
|
||||
if let Err(e) = graceful.await {
|
||||
eprintln!("server error: {}", e);
|
||||
}
|
||||
future::join_all([rdsys_stream_handler, rdsys_resource_receiver, context_manager]).await;
|
||||
let graceful = server.with_graceful_shutdown(shutdown_signal());
|
||||
println!("Listening on {}", addr);
|
||||
if let Err(e) = graceful.await {
|
||||
eprintln!("server error: {}", e);
|
||||
}
|
||||
future::join_all([
|
||||
rdsys_stream_handler,
|
||||
rdsys_resource_receiver,
|
||||
context_manager,
|
||||
shutdown_handler,
|
||||
])
|
||||
.await;
|
||||
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue