mirror of
https://github.com/YGGverse/aquatic.git
synced 2026-04-02 10:45:30 +00:00
aquatic_udp: glommio: read access list on start
This commit is contained in:
parent
119cd0fe46
commit
d0be89388c
4 changed files with 29 additions and 5 deletions
|
|
@ -20,7 +20,11 @@ pub async fn update_access_list(config: Config, access_list: Rc<RefCell<AccessLi
|
||||||
match reader.read_line(&mut buf).await {
|
match reader.read_line(&mut buf).await {
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
if let Err(err) = access_list.borrow_mut().insert_from_line(&buf) {
|
if let Err(err) = access_list.borrow_mut().insert_from_line(&buf) {
|
||||||
::log::error!("Couln't parse access list line '{}': {:?}", buf, err);
|
::log::error!(
|
||||||
|
"Couln't parse access list line '{}': {:?}",
|
||||||
|
buf,
|
||||||
|
err
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
|
|
@ -32,7 +36,7 @@ pub async fn update_access_list(config: Config, access_list: Rc<RefCell<AccessLi
|
||||||
|
|
||||||
yield_if_needed().await;
|
yield_if_needed().await;
|
||||||
}
|
}
|
||||||
},
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
::log::error!("Couldn't open access list file: {:?}", err)
|
::log::error!("Couldn't open access list file: {:?}", err)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,7 @@ pub async fn run_request_worker(
|
||||||
config: Config,
|
config: Config,
|
||||||
request_mesh_builder: MeshBuilder<(usize, AnnounceRequest, SocketAddr), Partial>,
|
request_mesh_builder: MeshBuilder<(usize, AnnounceRequest, SocketAddr), Partial>,
|
||||||
response_mesh_builder: MeshBuilder<(AnnounceResponse, SocketAddr), Partial>,
|
response_mesh_builder: MeshBuilder<(AnnounceResponse, SocketAddr), Partial>,
|
||||||
|
access_list: AccessList,
|
||||||
) {
|
) {
|
||||||
let (_, mut request_receivers) = request_mesh_builder.join(Role::Consumer).await.unwrap();
|
let (_, mut request_receivers) = request_mesh_builder.join(Role::Consumer).await.unwrap();
|
||||||
let (response_senders, _) = response_mesh_builder.join(Role::Producer).await.unwrap();
|
let (response_senders, _) = response_mesh_builder.join(Role::Producer).await.unwrap();
|
||||||
|
|
@ -26,7 +27,7 @@ pub async fn run_request_worker(
|
||||||
let response_senders = Rc::new(response_senders);
|
let response_senders = Rc::new(response_senders);
|
||||||
|
|
||||||
let torrents = Rc::new(RefCell::new(TorrentMaps::default()));
|
let torrents = Rc::new(RefCell::new(TorrentMaps::default()));
|
||||||
let access_list = Rc::new(RefCell::new(AccessList::default()));
|
let access_list = Rc::new(RefCell::new(access_list));
|
||||||
|
|
||||||
// Periodically clean torrents and update access list
|
// Periodically clean torrents and update access list
|
||||||
TimerActionRepeat::repeat(enclose!((config, torrents, access_list) move || {
|
TimerActionRepeat::repeat(enclose!((config, torrents, access_list) move || {
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,7 @@
|
||||||
|
|
||||||
use std::sync::{atomic::AtomicUsize, Arc};
|
use std::sync::{atomic::AtomicUsize, Arc};
|
||||||
|
|
||||||
|
use aquatic_common::access_list::AccessList;
|
||||||
use glommio::channels::channel_mesh::MeshBuilder;
|
use glommio::channels::channel_mesh::MeshBuilder;
|
||||||
use glommio::prelude::*;
|
use glommio::prelude::*;
|
||||||
|
|
||||||
|
|
@ -23,6 +24,12 @@ pub fn run(config: Config) -> anyhow::Result<()> {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let access_list = if config.access_list.mode.is_on() {
|
||||||
|
AccessList::create_from_path(&config.access_list.path).expect("Load access list")
|
||||||
|
} else {
|
||||||
|
AccessList::default()
|
||||||
|
};
|
||||||
|
|
||||||
let num_peers = config.socket_workers + config.request_workers;
|
let num_peers = config.socket_workers + config.request_workers;
|
||||||
|
|
||||||
let request_mesh_builder = MeshBuilder::partial(num_peers, SHARED_CHANNEL_SIZE);
|
let request_mesh_builder = MeshBuilder::partial(num_peers, SHARED_CHANNEL_SIZE);
|
||||||
|
|
@ -37,6 +44,7 @@ pub fn run(config: Config) -> anyhow::Result<()> {
|
||||||
let request_mesh_builder = request_mesh_builder.clone();
|
let request_mesh_builder = request_mesh_builder.clone();
|
||||||
let response_mesh_builder = response_mesh_builder.clone();
|
let response_mesh_builder = response_mesh_builder.clone();
|
||||||
let num_bound_sockets = num_bound_sockets.clone();
|
let num_bound_sockets = num_bound_sockets.clone();
|
||||||
|
let access_list = access_list.clone();
|
||||||
|
|
||||||
let mut builder = LocalExecutorBuilder::default();
|
let mut builder = LocalExecutorBuilder::default();
|
||||||
|
|
||||||
|
|
@ -50,6 +58,7 @@ pub fn run(config: Config) -> anyhow::Result<()> {
|
||||||
request_mesh_builder,
|
request_mesh_builder,
|
||||||
response_mesh_builder,
|
response_mesh_builder,
|
||||||
num_bound_sockets,
|
num_bound_sockets,
|
||||||
|
access_list,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
});
|
});
|
||||||
|
|
@ -61,6 +70,7 @@ pub fn run(config: Config) -> anyhow::Result<()> {
|
||||||
let config = config.clone();
|
let config = config.clone();
|
||||||
let request_mesh_builder = request_mesh_builder.clone();
|
let request_mesh_builder = request_mesh_builder.clone();
|
||||||
let response_mesh_builder = response_mesh_builder.clone();
|
let response_mesh_builder = response_mesh_builder.clone();
|
||||||
|
let access_list = access_list.clone();
|
||||||
|
|
||||||
let mut builder = LocalExecutorBuilder::default();
|
let mut builder = LocalExecutorBuilder::default();
|
||||||
|
|
||||||
|
|
@ -70,7 +80,13 @@ pub fn run(config: Config) -> anyhow::Result<()> {
|
||||||
}
|
}
|
||||||
|
|
||||||
let executor = builder.spawn(|| async move {
|
let executor = builder.spawn(|| async move {
|
||||||
handlers::run_request_worker(config, request_mesh_builder, response_mesh_builder).await
|
handlers::run_request_worker(
|
||||||
|
config,
|
||||||
|
request_mesh_builder,
|
||||||
|
response_mesh_builder,
|
||||||
|
access_list,
|
||||||
|
)
|
||||||
|
.await
|
||||||
});
|
});
|
||||||
|
|
||||||
executors.push(executor);
|
executors.push(executor);
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,7 @@ pub async fn run_socket_worker(
|
||||||
request_mesh_builder: MeshBuilder<(usize, AnnounceRequest, SocketAddr), Partial>,
|
request_mesh_builder: MeshBuilder<(usize, AnnounceRequest, SocketAddr), Partial>,
|
||||||
response_mesh_builder: MeshBuilder<(AnnounceResponse, SocketAddr), Partial>,
|
response_mesh_builder: MeshBuilder<(AnnounceResponse, SocketAddr), Partial>,
|
||||||
num_bound_sockets: Arc<AtomicUsize>,
|
num_bound_sockets: Arc<AtomicUsize>,
|
||||||
|
access_list: AccessList,
|
||||||
) {
|
) {
|
||||||
let (local_sender, local_receiver) = new_unbounded();
|
let (local_sender, local_receiver) = new_unbounded();
|
||||||
|
|
||||||
|
|
@ -57,6 +58,7 @@ pub async fn run_socket_worker(
|
||||||
response_consumer_index,
|
response_consumer_index,
|
||||||
local_sender,
|
local_sender,
|
||||||
socket.clone(),
|
socket.clone(),
|
||||||
|
access_list,
|
||||||
))
|
))
|
||||||
.detach();
|
.detach();
|
||||||
|
|
||||||
|
|
@ -77,6 +79,7 @@ async fn read_requests(
|
||||||
response_consumer_index: usize,
|
response_consumer_index: usize,
|
||||||
local_sender: LocalSender<(Response, SocketAddr)>,
|
local_sender: LocalSender<(Response, SocketAddr)>,
|
||||||
socket: Rc<UdpSocket>,
|
socket: Rc<UdpSocket>,
|
||||||
|
access_list: AccessList,
|
||||||
) {
|
) {
|
||||||
let mut rng = StdRng::from_entropy();
|
let mut rng = StdRng::from_entropy();
|
||||||
|
|
||||||
|
|
@ -84,7 +87,7 @@ async fn read_requests(
|
||||||
|
|
||||||
let max_connection_age = config.cleaning.max_connection_age;
|
let max_connection_age = config.cleaning.max_connection_age;
|
||||||
let connection_valid_until = Rc::new(RefCell::new(ValidUntil::new(max_connection_age)));
|
let connection_valid_until = Rc::new(RefCell::new(ValidUntil::new(max_connection_age)));
|
||||||
let access_list = Rc::new(RefCell::new(AccessList::default()));
|
let access_list = Rc::new(RefCell::new(access_list));
|
||||||
let connections = Rc::new(RefCell::new(ConnectionMap::default()));
|
let connections = Rc::new(RefCell::new(ConnectionMap::default()));
|
||||||
|
|
||||||
// Periodically update connection_valid_until
|
// Periodically update connection_valid_until
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue