IPv4 and portable atomic

Cargo.toml

@@ -108,7 +108,7 @@ itertools = "0.14"
 leaky-bucket = "1.1"
 libc = "0.2"
 librqbit-dualstack-sockets = "0.6"
-librqbit-utp = "0.6"
+librqbit-utp = "0.6.4"
 lru = "0.16"
 memchr = "2"
 memmap2 = "0.9"
@@ -152,4 +152,4 @@ url = { version = "2", default-features = false }
 urlencoding = "2"
 uuid = "1"
 walkdir = "2"
-windows = "0.62"
+windows = "0.62"

crates/dht/Cargo.toml

@@ -28,6 +28,7 @@ dashmap = { workspace = true, features = ["serde"] }
 clone_to_owned.workspace = true
 librqbit-core.workspace = true
 chrono = { workspace = true, features = ["serde"] }
+portable-atomic.workspace = true
 tokio-util.workspace = true
 bytes.workspace = true
 librqbit-dualstack-sockets.workspace = true

crates/dht/src/dht.rs

@@ -1,6 +1,6 @@
 use std::{
     cmp::Reverse,
-    net::{Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
+    net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
     str::FromStr,
     sync::{
         Arc,
@@ -1283,6 +1283,7 @@ pub struct DhtConfig<'a> {
     pub peer_store: Option<PeerStore>,
     pub cancellation_token: Option<CancellationToken>,
     pub bind_device: Option<&'a BindDevice>,
+    pub ipv4_only: bool,
 }
 
 impl DhtState {
@@ -1296,13 +1297,17 @@ impl DhtState {
     #[inline(never)]
     pub fn with_config<'a>(mut config: DhtConfig<'a>) -> BoxFuture<'a, crate::Result<Arc<Self>>> {
         async move {
-            let addr = config
-                .listen_addr
-                .unwrap_or((Ipv6Addr::UNSPECIFIED, 0).into());
+            let addr = config.listen_addr.unwrap_or_else(|| {
+                if config.ipv4_only {
+                    (Ipv4Addr::UNSPECIFIED, 0).into()
+                } else {
+                    (Ipv6Addr::UNSPECIFIED, 0).into()
+                }
+            });
             let socket = UdpSocket::bind_udp(
                 addr,
                 librqbit_dualstack_sockets::BindOpts {
-                    request_dualstack: true,
+                    request_dualstack: !config.ipv4_only,
                     reuseport: false,
                     device: config.bind_device,
                 },

crates/dht/src/peer_store.rs

@@ -1,4 +1,6 @@
-use std::{collections::VecDeque, net::SocketAddr, str::FromStr, sync::atomic::AtomicU32};
+use std::{collections::VecDeque, net::SocketAddr, str::FromStr};
+
+use portable_atomic::{AtomicU32, Ordering};
 
 use bencode::ByteBufOwned;
 use chrono::{DateTime, Utc};
@@ -68,7 +70,7 @@ impl Serialize for PeerStore {
         s.serialize_field("peers", &SerializePeers { peers: &self.peers })?;
         s.serialize_field(
             "peers_len",
-            &self.peers_len.load(std::sync::atomic::Ordering::SeqCst),
+            &self.peers_len.load(Ordering::SeqCst),
         )?;
         s.end()
     }
@@ -151,7 +153,7 @@ impl PeerStore {
 
         use dashmap::mapref::entry::Entry;
         let peers_entry = self.peers.entry(announce.info_hash);
-        let peers_len = self.peers_len.load(std::sync::atomic::Ordering::SeqCst);
+        let peers_len = self.peers_len.load(Ordering::SeqCst);
         match peers_entry {
             Entry::Occupied(mut occ) => {
                 if let Some(s) = occ.get_mut().iter_mut().find(|s| s.addr == addr) {
@@ -180,7 +182,7 @@ impl PeerStore {
         }
 
         self.peers_len
-            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            .fetch_add(1, Ordering::SeqCst);
         true
     }
 

crates/dht/src/persistence.rs

@@ -154,7 +154,11 @@ impl PersistentDht {
                 if let Some(ref mut addr) = listen_addr {
                     addr.set_port(port);
                 } else {
-                    listen_addr = Some(SocketAddr::from((Ipv6Addr::UNSPECIFIED, port)));
+                    listen_addr = Some(if config.ipv4_only {
+                        SocketAddr::from((Ipv4Addr::UNSPECIFIED, port))
+                    } else {
+                        SocketAddr::from((Ipv6Addr::UNSPECIFIED, port))
+                    });
                 }
             }
             let peer_id = routing_table.as_ref().map(|r| r.id());
@@ -166,6 +170,7 @@ impl PersistentDht {
                 peer_store,
                 cancellation_token,
                 bind_device,
+                ipv4_only: config.ipv4_only,
                 ..Default::default()
             };
             let dht = DhtState::with_config(dht_config).await?;

crates/librqbit/Cargo.toml

@@ -99,6 +99,7 @@ lru = { workspace = true, optional = true }
 mime_guess.workspace = true
 tokio-socks.workspace = true
 async-trait.workspace = true
+portable-atomic.workspace = true
 async-backtrace = { workspace = true, optional = true }
 notify = { workspace = true, optional = true }
 walkdir.workspace = true

crates/librqbit/src/file_ops.rs

@@ -1,8 +1,9 @@
 use std::{
     marker::PhantomData,
-    sync::atomic::{AtomicU64, Ordering},
+    sync::atomic::Ordering,
 };
 
+use portable_atomic::AtomicU64;
 use anyhow::Context;
 use buffers::{ByteBuf, ByteBufOwned};
 use librqbit_core::{

crates/librqbit/src/limits.rs

@@ -14,7 +14,7 @@ pub struct LimitsConfig {
 
 struct Limit {
     limiter: ArcSwapOption<RateLimiter>,
-    current_bps: std::sync::atomic::AtomicU32,
+    current_bps: portable_atomic::AtomicU32,
 }
 
 impl Limit {
@@ -24,7 +24,7 @@ impl Limit {
     }
 
     fn new(bps: Option<NonZeroU32>) -> Self {
-        use std::sync::atomic::AtomicU32;
+        use portable_atomic::AtomicU32;
         Self {
             limiter: ArcSwapOption::new(Self::new_inner(bps)),
             current_bps: AtomicU32::new(bps.map(|v| v.get()).unwrap_or(0)),
@@ -40,15 +40,15 @@ impl Limit {
     }
 
     fn set(&self, limit: Option<NonZeroU32>) {
-        use std::sync::atomic::Ordering;
+        use portable_atomic::Ordering;
         let new = Self::new_inner(limit);
         self.limiter.swap(new);
         self.current_bps
             .store(limit.map(|v| v.get()).unwrap_or(0), Ordering::Relaxed);
     }
 
     fn get(&self) -> Option<NonZeroU32> {
-        use std::sync::atomic::Ordering;
+        use portable_atomic::Ordering;
         NonZeroU32::new(self.current_bps.load(Ordering::Relaxed))
     }
 }

crates/librqbit/src/session.rs

@@ -6,11 +6,16 @@ use std::{
     path::{Component, Path, PathBuf},
     sync::{
         Arc,
-        atomic::{AtomicUsize, Ordering},
+        atomic::Ordering,
     },
     time::Duration,
 };
 
+#[cfg(target_pointer_width = "64")]
+use std::sync::atomic::AtomicUsize;
+#[cfg(target_pointer_width = "32")]
+use portable_atomic::AtomicUsize;
+
 use crate::{
     ApiError, CreateTorrentOptions, FileInfos, ManagedTorrent, ManagedTorrentShared,
     api::TorrentIdOrHash,
@@ -555,12 +560,14 @@ impl Session {
                         bootstrap_addrs: opts.dht_bootstrap_addrs.clone(),
                         cancellation_token: Some(token.child_token()),
                         bind_device: bind_device.as_ref(),
+                        ipv4_only: opts.ipv4_only,
                         ..Default::default()
                     })
                     .await
                     .context("error initializing DHT")?
                 } else {
-                    let pdht_config = opts.dht_config.take().unwrap_or_default();
+                    let mut pdht_config = opts.dht_config.take().unwrap_or_default();
+                    pdht_config.ipv4_only = opts.ipv4_only;
                     PersistentDht::create(
                         Some(pdht_config),
                         Some(token.clone()),
@@ -691,7 +698,11 @@ impl Session {
                 None
             };
 
-            let udp_tracker_client = UdpTrackerClient::new(token.clone(), bind_device.as_ref())
+            let udp_tracker_client = UdpTrackerClient::new(
+                token.clone(), 
+                bind_device.as_ref(),
+                opts.ipv4_only
+            )
                 .await
                 .context("error creating UDP tracker client")?;
 
@@ -702,6 +713,7 @@ impl Session {
                     LocalServiceDiscovery::new(LocalServiceDiscoveryOptions {
                         cancel_token: token.clone(),
                         bind_device: bind_device.as_ref(),
+                        ipv4_only: opts.ipv4_only,
                         ..Default::default()
                     })
                     .await
@@ -1236,7 +1248,7 @@ impl Session {
             p.next_id().await?
         } else {
             self.next_id
-                .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+                .fetch_add(1, Ordering::Relaxed)
         };
 
         let _permit = self.spawner.semaphore().acquire_owned().await?;

crates/librqbit/src/stat_gen.rs

@@ -1,9 +1,9 @@
 macro_rules! stype {
     (atomic u32) => {
-        std::sync::atomic::AtomicU32
+        portable_atomic::AtomicU32
     };
     (atomic u64) => {
-        std::sync::atomic::AtomicU64
+        portable_atomic::AtomicU64
     };
     (u32) => {
         u32
@@ -30,7 +30,7 @@ macro_rules! gen_stats {
             pub fn snapshot(&self) -> $snapshot_name {
                 $snapshot_name {
                     $(
-                        $stat_name: self.$stat_name.load(std::sync::atomic::Ordering::Relaxed),
+                        $stat_name: self.$stat_name.load(portable_atomic::Ordering::Relaxed),
                     )*
 
                     $(
@@ -42,7 +42,7 @@ macro_rules! gen_stats {
             $(
                 #[allow(unused)]
                 pub fn $stat_name(&self, value: $stat_ty) {
-                    self.$stat_name.fetch_add(value, std::sync::atomic::Ordering::Relaxed);
+                    self.$stat_name.fetch_add(value, portable_atomic::Ordering::Relaxed);
                 }
             )*
         }

crates/librqbit/src/torrent_state/initializing.rs

@@ -1,11 +1,16 @@
 use std::{
     sync::{
         Arc,
-        atomic::{AtomicU64, Ordering},
+        atomic::Ordering,
     },
     time::Instant,
 };
 
+#[cfg(target_pointer_width = "64")]
+use std::sync::atomic::AtomicU64;
+#[cfg(target_pointer_width = "32")]
+use portable_atomic::AtomicU64;
+
 use anyhow::Context;
 
 use itertools::Itertools;
@@ -53,7 +58,7 @@ impl TorrentStateInitializing {
 
     pub fn get_checked_bytes(&self) -> u64 {
         self.checked_bytes
-            .load(std::sync::atomic::Ordering::Relaxed)
+            .load(Ordering::Relaxed)
     }
 
     async fn validate_fastresume(

crates/librqbit/src/torrent_state/live/mod.rs

@@ -50,11 +50,16 @@ use std::{
     num::NonZeroU32,
     sync::{
         Arc,
-        atomic::{AtomicBool, AtomicU64, Ordering},
+        atomic::{AtomicBool, Ordering},
     },
     time::{Duration, Instant},
 };
 
+#[cfg(target_pointer_width = "64")]
+use std::sync::atomic::AtomicU64;
+#[cfg(target_pointer_width = "32")]
+use portable_atomic::AtomicU64;
+
 use anyhow::{Context, bail};
 use buffers::{ByteBuf, ByteBufOwned};
 use clone_to_owned::CloneToOwned;

crates/librqbit/src/torrent_state/live/peer/stats/atomic.rs

@@ -1,11 +1,12 @@
 use std::{
     sync::{
         Arc,
-        atomic::{AtomicU32, AtomicU64, Ordering},
+        atomic::Ordering,
     },
     time::Duration,
 };
 
+use portable_atomic::{AtomicU32, AtomicU64};
 use backon::{BackoffBuilder, ExponentialBackoff, ExponentialBuilder};
 
 #[derive(Default, Debug)]

crates/librqbit/src/torrent_state/live/peers/stats/mod.rs

@@ -1,4 +1,4 @@
-use std::sync::atomic::AtomicU32;
+use portable_atomic::AtomicU32;
 
 use crate::{
     stream_connect::ConnectionKind,

crates/librqbit/src/torrent_state/live/stats/atomic.rs

@@ -1,4 +1,4 @@
-use std::sync::atomic::AtomicU64;
+use portable_atomic::AtomicU64;
 
 #[derive(Default, Debug)]
 pub struct AtomicStats {

crates/librqbit/src/torrent_state/streaming.rs

@@ -3,7 +3,7 @@ use std::{
     io::SeekFrom,
     sync::{
         Arc,
-        atomic::{AtomicUsize, Ordering},
+        atomic::Ordering,
     },
     task::{Poll, Waker},
     time::Instant,
@@ -19,6 +19,7 @@ use tokio::{
 };
 use tracing::{debug, trace};
 
+use portable_atomic::AtomicUsize;
 use crate::{ManagedTorrent, file_info::FileInfo, storage::TorrentStorage};
 
 use super::{ManagedTorrentHandle, TorrentMetadata};

crates/librqbit/src/torrent_state/utils.rs

@@ -1,4 +1,5 @@
-use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::atomic::Ordering;
+use portable_atomic::AtomicU32;
 
 pub fn atomic_inc(c: &AtomicU32) -> u32 {
     c.fetch_add(1, Ordering::Relaxed)

crates/librqbit/webui/src/components/compact/PeersTab.tsx

@@ -65,7 +65,7 @@ export const PeersTab: React.FC<PeersTabProps> = ({ torrent }) => {
 
   // Fetch peer stats periodically
   useEffect(() => {
-    if (torrentId == null || !statsResponse?.live) return;
+    if (!torrentId || !statsResponse?.live) return;
 
     return customSetInterval(() => {
       return API.getPeerStats(torrentId).then(

crates/librqbit_core/Cargo.toml

@@ -28,6 +28,7 @@ buffers.workspace = true
 bencode.workspace = true
 clone_to_owned.workspace = true
 sha1w = { workspace = true, optional = true }
+portable-atomic.workspace = true
 itertools.workspace = true
 directories.workspace = true
 tokio-util.workspace = true

crates/librqbit_core/src/speed_estimator.rs

@@ -1,9 +1,11 @@
 use std::{
     collections::VecDeque,
-    sync::atomic::{AtomicU64, Ordering},
+    sync::atomic::Ordering,
     time::{Duration, Instant},
 };
 
+use portable_atomic::AtomicU64;
+
 use parking_lot::Mutex;
 
 #[derive(Clone, Copy)]

crates/librqbit_lsd/Cargo.toml

@@ -17,6 +17,7 @@ parking_lot.workspace = true
 httparse.workspace = true
 atoi.workspace = true
 tracing.workspace = true
+portable-atomic.workspace = true
 
 [dev-dependencies]
 tracing-subscriber.workspace = true