YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-03-31 09:45:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

update ws dependencies as well as http rustls dependency

Browse source
This commit is contained in:
Joakim Frostegård 2024-01-07 10:40:50 +01:00
parent 071f088d8b
commit 3042539101
16 changed files with 499 additions and 295 deletions
Download patch file Download diff file Expand all files Collapse all files

16
crates/ws/Cargo.toml
View file

@ -29,31 +29,31 @@ aquatic_toml_config.workspace = true
aquatic_ws_protocol.workspace = true
anyhow = "1"
async-tungstenite = "0.23"
async-tungstenite = "0.24"
arc-swap = "1"
cfg-if = "1"
futures = "0.3"
futures-lite = "1"
futures-rustls = "0.24"
futures-rustls = "0.25"
glommio = "0.8"
hashbrown = { version = "0.14", features = ["serde"] }
httparse = "1"
indexmap = "2"
log = "0.4"
metrics = { version = "0.21", optional = true }
metrics-util = { version = "0.15", optional = true }
metrics-exporter-prometheus = { version = "0.12", optional = true, default-features = false, features = ["http-listener"] }
metrics = { version = "0.22", optional = true }
metrics-util = { version = "0.16", optional = true }
metrics-exporter-prometheus = { version = "0.13", optional = true, default-features = false, features = ["http-listener"] }
mimalloc = { version = "0.1", default-features = false }
privdrop = "0.5"
rand = { version = "0.8", features = ["small_rng"] }
rustls = "0.21"
rustls-pemfile = "1"
rustls = "0.22"
rustls-pemfile = "2"
serde = { version = "1", features = ["derive"] }
signal-hook = { version = "0.3" }
slab = "0.4"
slotmap = "1"
socket2 = { version = "0.5", features = ["all"] }
tungstenite = "0.20"
tungstenite = "0.21"
[dev-dependencies]
quickcheck = "1"

4
crates/ws/src/main.rs
View file

@ -1,8 +1,8 @@
use aquatic_common::cli::run_app_with_cli_and_config;
use aquatic_ws::config::Config;
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
// #[global_allocator]
// static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
fn main() {
run_app_with_cli_and_config::<Config>(

142
crates/ws/src/workers/socket/connection.rs
View file

@ -9,7 +9,6 @@ use anyhow::Context;
use aquatic_common::access_list::{create_access_list_cache, AccessListArcSwap, AccessListCache};
use aquatic_common::rustls_config::RustlsConfig;
use aquatic_common::ServerStartInstant;
use aquatic_peer_id::PeerClient;
use aquatic_ws_protocol::*;
use arc_swap::ArcSwap;
use async_tungstenite::WebSocketStream;
@ -26,6 +25,9 @@ use hashbrown::hash_map::Entry;
use hashbrown::HashMap;
use slab::Slab;
#[cfg(feature = "metrics")]
use metrics::{Counter, Gauge};
use crate::common::*;
use crate::config::Config;
use crate::workers::socket::calculate_in_message_consumer_index;
@ -60,6 +62,12 @@ impl ConnectionRunner {
announced_info_hashes: Default::default(),
ip_version: self.ip_version,
opt_peer_client: Default::default(),
#[cfg(feature = "metrics")]
active_connections_gauge: ::metrics::gauge!(
"aquatic_active_connections",
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.get().to_string(),
),
};
clean_up_data.before_open();
@ -74,7 +82,7 @@ impl ConnectionRunner {
clean_up_data
) async move {
if let Err(err) = self.run_inner(clean_up_data, stream).await {
::log::debug!("connection {:?} error: {:#}", connection_id, err);
::log::debug!("connection {:?} closed: {:#}", connection_id, err);
}
}),
tq_regular,
@ -101,7 +109,8 @@ impl ConnectionRunner {
mut stream: TcpStream,
) -> anyhow::Result<()> {
if let Some(tls_config) = self.opt_tls_config.as_ref() {
let tls_acceptor: TlsAcceptor = tls_config.load_full().into();
let tls_config = tls_config.load_full();
let tls_acceptor = TlsAcceptor::from(tls_config);
let stream = tls_acceptor.accept(stream).await?;
@ -177,6 +186,20 @@ impl ConnectionRunner {
ip_version: self.ip_version,
connection_id: self.connection_id,
clean_up_data: clean_up_data.clone(),
#[cfg(feature = "metrics")]
total_announce_requests_counter: ::metrics::counter!(
"aquatic_requests_total",
"type" => "announce",
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
),
#[cfg(feature = "metrics")]
total_scrape_requests_counter: ::metrics::counter!(
"aquatic_requests_total",
"type" => "scrape",
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
)
};
reader.run_in_message_loop().await
@ -219,6 +242,10 @@ struct ConnectionReader<S> {
ip_version: IpVersion,
connection_id: ConnectionId,
clean_up_data: ConnectionCleanupData,
#[cfg(feature = "metrics")]
total_announce_requests_counter: Counter,
#[cfg(feature = "metrics")]
total_scrape_requests_counter: Counter,
}
impl<S: futures::AsyncRead + futures::AsyncWrite + Unpin> ConnectionReader<S> {
@ -270,12 +297,7 @@ impl<S: futures::AsyncRead + futures::AsyncWrite + Unpin> ConnectionReader<S> {
async fn handle_announce_request(&mut self, request: AnnounceRequest) -> anyhow::Result<()> {
#[cfg(feature = "metrics")]
::metrics::increment_counter!(
"aquatic_requests_total",
"type" => "announce",
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
self.total_announce_requests_counter.increment(1);
let info_hash = request.info_hash;
@ -315,24 +337,28 @@ impl<S: futures::AsyncRead + futures::AsyncWrite + Unpin> ConnectionReader<S> {
&& self.clean_up_data.opt_peer_client.borrow().is_none()
{
let peer_id = aquatic_peer_id::PeerId(request.peer_id.0);
let client = peer_id.client();
let prefix = peer_id.first_8_bytes_hex().to_string();
::metrics::increment_gauge!(
let peer_client_gauge = ::metrics::gauge!(
"aquatic_peer_clients",
1.0,
"client" => client.to_string(),
"client" => peer_id.client().to_string(),
);
if self.config.metrics.peer_id_prefixes {
::metrics::increment_gauge!(
"aquatic_peer_id_prefixes",
1.0,
"prefix_hex" => prefix.to_string(),
);
}
peer_client_gauge.increment(1.0);
*self.clean_up_data.opt_peer_client.borrow_mut() = Some((client, prefix));
let opt_peer_id_prefix_gauge =
self.config.metrics.peer_id_prefixes.then(|| {
let g = ::metrics::gauge!(
"aquatic_peer_id_prefixes",
"prefix_hex" => peer_id.first_8_bytes_hex().to_string(),
);
g.increment(1.0);
g
});
*self.clean_up_data.opt_peer_client.borrow_mut() =
Some((peer_client_gauge, opt_peer_id_prefix_gauge));
};
}
}
@ -370,12 +396,7 @@ impl<S: futures::AsyncRead + futures::AsyncWrite + Unpin> ConnectionReader<S> {
async fn handle_scrape_request(&mut self, request: ScrapeRequest) -> anyhow::Result<()> {
#[cfg(feature = "metrics")]
::metrics::increment_counter!(
"aquatic_requests_total",
"type" => "scrape",
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
self.total_scrape_requests_counter.increment(1);
let info_hashes = if let Some(info_hashes) = request.info_hashes {
info_hashes
@ -548,32 +569,24 @@ impl<S: futures::AsyncRead + futures::AsyncWrite + Unpin> ConnectionWriter<S> {
OutMessage::ErrorResponse(_) => "error",
};
::metrics::increment_counter!(
::metrics::counter!(
"aquatic_responses_total",
"type" => out_message_type,
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
)
.increment(1);
if let Some((peer_client, prefix)) =
// As long as connection is still alive, increment peer client
// gauges by zero to prevent them from being removed due to
// idleness
if let Some((peer_client_gauge, opt_peer_id_prefix_gauge)) =
self.clean_up_data.opt_peer_client.borrow().as_ref()
{
// As long as connection is still alive, increment peer client
// gauges by zero to prevent them from being removed due to
// idleness
peer_client_gauge.increment(0.0);
::metrics::increment_gauge!(
"aquatic_peer_clients",
0.0,
"client" => peer_client.to_string(),
);
if self.config.metrics.peer_id_prefixes {
::metrics::increment_gauge!(
"aquatic_peer_id_prefixes",
0.0,
"prefix_hex" => prefix.to_string(),
);
if let Some(g) = opt_peer_id_prefix_gauge {
g.increment(0.0);
}
}
}
@ -587,18 +600,15 @@ impl<S: futures::AsyncRead + futures::AsyncWrite + Unpin> ConnectionWriter<S> {
struct ConnectionCleanupData {
announced_info_hashes: Rc<RefCell<HashMap<InfoHash, PeerId>>>,
ip_version: IpVersion,
opt_peer_client: Rc<RefCell<Option<(PeerClient, String)>>>,
opt_peer_client: Rc<RefCell<Option<(Gauge, Option<Gauge>)>>>,
#[cfg(feature = "metrics")]
active_connections_gauge: Gauge,
}
impl ConnectionCleanupData {
fn before_open(&self) {
#[cfg(feature = "metrics")]
::metrics::increment_gauge!(
"aquatic_active_connections",
1.0,
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.get().to_string(),
);
self.active_connections_gauge.increment(1.0);
}
async fn after_close(
&self,
@ -625,28 +635,14 @@ impl ConnectionCleanupData {
}
#[cfg(feature = "metrics")]
{
::metrics::decrement_gauge!(
"aquatic_active_connections",
1.0,
"ip_version" => ip_version_to_metrics_str(self.ip_version),
"worker_index" => WORKER_INDEX.get().to_string(),
);
self.active_connections_gauge.decrement(1.0);
if let Some((peer_client, prefix)) = self.opt_peer_client.borrow().as_ref() {
::metrics::decrement_gauge!(
"aquatic_peer_clients",
1.0,
"client" => peer_client.to_string(),
);
#[cfg(feature = "metrics")]
if let Some((peer_client_gauge, opt_peer_id_prefix_gauge)) = self.opt_peer_client.take() {
peer_client_gauge.decrement(1.0);
if config.metrics.peer_id_prefixes {
::metrics::decrement_gauge!(
"aquatic_peer_id_prefixes",
1.0,
"prefix_hex" => prefix.to_string(),
);
}
if let Some(g) = opt_peer_id_prefix_gauge {
g.decrement(1.0);
}
}
}

12
crates/ws/src/workers/socket/mod.rs
View file

@ -253,20 +253,20 @@ async fn clean_connections(
let worker_index = WORKER_INDEX.with(|index| index.get()).to_string();
if config.network.address.is_ipv4() || !config.network.only_ipv6 {
::metrics::increment_gauge!(
::metrics::gauge!(
"aquatic_active_connections",
0.0,
"ip_version" => "4",
"worker_index" => worker_index.clone(),
);
)
.increment(0.0);
}
if config.network.address.is_ipv6() {
::metrics::increment_gauge!(
::metrics::gauge!(
"aquatic_active_connections",
0.0,
"ip_version" => "6",
"worker_index" => worker_index,
);
)
.increment(0.0);
}
}

131
crates/ws/src/workers/swarm/storage.rs
View file

@ -2,6 +2,7 @@ use std::sync::Arc;
use aquatic_common::access_list::{create_access_list_cache, AccessListArcSwap, AccessListCache};
use hashbrown::HashMap;
use metrics::Gauge;
use rand::rngs::SmallRng;
use aquatic_common::{
@ -16,10 +17,42 @@ use crate::workers::swarm::WORKER_INDEX;
type TorrentMap = IndexMap<InfoHash, TorrentData>;
type PeerMap = IndexMap<PeerId, Peer>;
#[derive(Default)]
pub struct TorrentMaps {
ipv4: TorrentMap,
ipv6: TorrentMap,
peers_gauge_ipv4: Gauge,
peers_gauge_ipv6: Gauge,
torrents_gauge_ipv4: Gauge,
torrents_gauge_ipv6: Gauge,
}
impl Default for TorrentMaps {
fn default() -> Self {
Self {
ipv4: Default::default(),
ipv6: Default::default(),
peers_gauge_ipv4: ::metrics::gauge!(
"aquatic_peers",
"ip_version" => "4",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
),
peers_gauge_ipv6: ::metrics::gauge!(
"aquatic_peers",
"ip_version" => "6",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
),
torrents_gauge_ipv4: ::metrics::gauge!(
"aquatic_torrents",
"ip_version" => "4",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
),
torrents_gauge_ipv6: ::metrics::gauge!(
"aquatic_torrents",
"ip_version" => "6",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
),
}
}
}
impl TorrentMaps {
@ -33,12 +66,11 @@ impl TorrentMaps {
request_sender_meta: InMessageMeta,
request: AnnounceRequest,
) {
let (torrent_data, ip_version): (&mut TorrentData, &'static str) =
if let IpVersion::V4 = request_sender_meta.ip_version {
(self.ipv4.entry(request.info_hash).or_default(), "4")
} else {
(self.ipv6.entry(request.info_hash).or_default(), "6")
};
let torrent_data = if let IpVersion::V4 = request_sender_meta.ip_version {
self.ipv4.entry(request.info_hash).or_default()
} else {
self.ipv6.entry(request.info_hash).or_default()
};
// If there is already a peer with this peer_id, check that connection id
// is same as that of request sender. Otherwise, ignore request. Since
@ -89,12 +121,10 @@ impl TorrentMaps {
}
#[cfg(feature = "metrics")]
::metrics::decrement_gauge!(
"aquatic_peers",
1.0,
"ip_version" => ip_version,
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
match request_sender_meta.ip_version {
IpVersion::V4 => self.peers_gauge_ipv4.decrement(1.0),
IpVersion::V6 => self.peers_gauge_ipv6.decrement(1.0),
}
return;
}
@ -112,12 +142,10 @@ impl TorrentMaps {
entry.insert(peer);
#[cfg(feature = "metrics")]
::metrics::increment_gauge!(
"aquatic_peers",
1.0,
"ip_version" => ip_version,
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
match request_sender_meta.ip_version {
IpVersion::V4 => self.peers_gauge_ipv4.increment(1.0),
IpVersion::V6 => self.peers_gauge_ipv6.increment(1.0),
}
}
PeerStatus::Seeding => {
torrent_data.num_seeders += 1;
@ -133,12 +161,10 @@ impl TorrentMaps {
entry.insert(peer);
#[cfg(feature = "metrics")]
::metrics::increment_gauge!(
"aquatic_peers",
1.0,
"ip_version" => ip_version,
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
match request_sender_meta.ip_version {
IpVersion::V4 => self.peers_gauge_ipv4.increment(1.0),
IpVersion::V6 => self.peers_gauge_ipv6.increment(1.0),
}
}
PeerStatus::Stopped => return,
},
@ -311,8 +337,20 @@ impl TorrentMaps {
let mut access_list_cache = create_access_list_cache(access_list);
let now = server_start_instant.seconds_elapsed();
Self::clean_torrent_map(config, &mut access_list_cache, &mut self.ipv4, now, "4");
Self::clean_torrent_map(config, &mut access_list_cache, &mut self.ipv6, now, "6");
Self::clean_torrent_map(
config,
&mut access_list_cache,
&mut self.ipv4,
now,
&self.peers_gauge_ipv4,
);
Self::clean_torrent_map(
config,
&mut access_list_cache,
&mut self.ipv6,
now,
&self.peers_gauge_ipv6,
);
}
fn clean_torrent_map(
@ -320,7 +358,7 @@ impl TorrentMaps {
access_list_cache: &mut AccessListCache,
torrent_map: &mut TorrentMap,
now: SecondsSinceServerStart,
ip_version: &'static str,
peers_gauge: &Gauge,
) {
let mut total_num_peers = 0u64;
@ -357,31 +395,14 @@ impl TorrentMaps {
torrent_map.shrink_to_fit();
let total_num_peers = total_num_peers as f64;
#[cfg(feature = "metrics")]
::metrics::gauge!(
"aquatic_peers",
total_num_peers,
"ip_version" => ip_version,
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
peers_gauge.set(total_num_peers as f64)
}
#[cfg(feature = "metrics")]
pub fn update_torrent_count_metrics(&self) {
::metrics::gauge!(
"aquatic_torrents",
self.ipv4.len() as f64,
"ip_version" => "4",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
::metrics::gauge!(
"aquatic_torrents",
self.ipv6.len() as f64,
"ip_version" => "6",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
self.torrents_gauge_ipv4.set(self.ipv4.len() as f64);
self.torrents_gauge_ipv6.set(self.ipv6.len() as f64);
}
pub fn handle_connection_closed(
@ -397,24 +418,14 @@ impl TorrentMaps {
torrent_data.remove_peer(peer_id);
#[cfg(feature = "metrics")]
::metrics::decrement_gauge!(
"aquatic_peers",
1.0,
"ip_version" => "4",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
self.peers_gauge_ipv4.decrement(1.0);
}
} else {
if let Some(torrent_data) = self.ipv6.get_mut(&info_hash) {
torrent_data.remove_peer(peer_id);
#[cfg(feature = "metrics")]
::metrics::decrement_gauge!(
"aquatic_peers",
1.0,
"ip_version" => "6",
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
);
self.peers_gauge_ipv6.decrement(1.0);
}
}
}