diff --git a/asic-rs-core/src/util.rs b/asic-rs-core/src/util.rs index f55de7cb..0108ca22 100644 --- a/asic-rs-core/src/util.rs +++ b/asic-rs-core/src/util.rs @@ -15,6 +15,13 @@ use crate::errors::{ModelSelectionError, RPCError}; /// Default read timeout for RPC stream responses. pub const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(5); +/// Default TCP connection timeout for firmware construction HTTP requests. +pub const DEFAULT_DISCOVERY_HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(2); +/// Default total timeout for each firmware construction HTTP request. +/// +/// This includes connection establishment, the request, and response-body +/// processing. +pub const DEFAULT_DISCOVERY_HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(3); pub fn unix_timestamp_secs() -> u64 { SystemTime::now().duration_since(UNIX_EPOCH).map_or_else( @@ -27,7 +34,11 @@ pub fn unix_timestamp_secs() -> u64 { } pub fn build_discovery_client() -> Result { + // Bound connection establishment separately while enforcing a total + // deadline across connection, request, and response-body processing. reqwest::Client::builder() + .connect_timeout(DEFAULT_DISCOVERY_HTTP_CONNECT_TIMEOUT) + .timeout(DEFAULT_DISCOVERY_HTTP_REQUEST_TIMEOUT) .pool_max_idle_per_host(0) .build() .map_err(|_| ModelSelectionError::NoModelResponse) diff --git a/asic-rs-firmwares/epic/src/firmware.rs b/asic-rs-firmwares/epic/src/firmware.rs index 58fca2b4..3d0ab13f 100644 --- a/asic-rs-firmwares/epic/src/firmware.rs +++ b/asic-rs-firmwares/epic/src/firmware.rs @@ -112,7 +112,6 @@ impl MinerFirmware for EPicFirmware { .send() .await .map_err(|_| ModelSelectionError::NoModelResponse)?; - let json_data = response .json::() .await diff --git a/docs/api.md b/docs/api.md index 614d03ae..191d9363 100644 --- a/docs/api.md +++ b/docs/api.md @@ -8,13 +8,23 @@ differences. `MinerFactory` owns the scan range and discovery tuning. +Network scans first probe common miner ports concurrently and only then run +firmware identification. The default limits are 256 addresses for TCP probing +and 128 addresses for identification, with no TCP retry. `with_concurrent_limit` +controls identification; connectivity can be tuned independently with the +`with_connectivity_*` methods. TCP-negative hosts are excluded by default; set +`with_strict_port_check(false)` to give them a fallback identification pass. +Firmware construction is capped at five seconds after identification. Its HTTP +requests use a two-second connection timeout and a three-second total timeout. + === "Rust" ```rust let factory = MinerFactory::from_subnet("192.168.1.0/24")? - .with_concurrent_limit(2500) - .with_connectivity_timeout_secs(1) - .with_identification_timeout_secs(10); + .with_concurrent_limit(128) + .with_connectivity_timeout_millis(500) + .with_identification_timeout_secs(3) + .with_miner_construction_timeout_secs(5); ``` === "Python" @@ -22,9 +32,10 @@ differences. ```python factory = ( MinerFactory.from_subnet("192.168.1.0/24") - .with_concurrent_limit(2500) - .with_connectivity_timeout_secs(1) - .with_identification_timeout_secs(10) + .with_concurrent_limit(128) + .with_connectivity_timeout_millis(500) + .with_identification_timeout_secs(3) + .with_miner_construction_timeout_secs(5) ) ``` diff --git a/docs/getting-started.md b/docs/getting-started.md index fc0474e3..90e59810 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -57,6 +57,10 @@ construct the matching miner implementation. Use a subnet, octet selectors, or range string when the exact IP address is not known. Large scans use bounded concurrency. +TCP reachability and application identification are separate stages. The +default probe limit is 256 addresses with no retry; responsive addresses are +then identified with a default concurrency of 128. Probe-negative addresses +are excluded unless strict port checking is disabled. === "Rust" @@ -66,7 +70,7 @@ known. Large scans use bounded concurrency. #[tokio::main] async fn main() -> anyhow::Result<()> { let miners = MinerFactory::from_subnet("192.168.1.0/24")? - .with_concurrent_limit(2500) + .with_concurrent_limit(128) .scan() .await?; @@ -86,7 +90,7 @@ known. Large scans use bounded concurrency. async def main() -> None: miners = await ( MinerFactory.from_subnet("192.168.1.0/24") - .with_concurrent_limit(2500) + .with_concurrent_limit(128) .scan() ) diff --git a/python/pyasic_rs/asic_rs.pyi b/python/pyasic_rs/asic_rs.pyi index 54bbcea0..bb42b1e7 100644 --- a/python/pyasic_rs/asic_rs.pyi +++ b/python/pyasic_rs/asic_rs.pyi @@ -607,10 +607,15 @@ class MinerFactory: def scan_stream_with_ip(self, /) -> AsyncIterator[tuple[IPv4Address |IPv6Address, Miner |None]]: ... def with_concurrent_limit(self, /, limit: int) -> MinerFactory: ... def with_connectivity_retries(self, /, retries: int) -> MinerFactory: ... + def with_connectivity_concurrent_limit(self, /, limit: int) -> MinerFactory: ... + def with_connectivity_retry_backoff_millis(self, /, backoff_millis: int) -> MinerFactory: ... + def with_connectivity_retry_concurrent_limit(self, /, limit: int) -> MinerFactory: ... + def with_connectivity_timeout_millis(self, /, timeout_millis: int) -> MinerFactory: ... def with_connectivity_timeout_secs(self, /, timeout_secs: int) -> MinerFactory: ... def with_identification_timeout_secs(self, /, timeout_secs: int) -> MinerFactory: ... def with_octets(self, /, octet1: "str | int", octet2: "str | int", octet3: "str | int", octet4: "str | int") -> "MinerFactory": ... def with_port_check(self, /, enabled: bool) -> MinerFactory: ... + def with_strict_port_check(self, /, enabled: bool) -> MinerFactory: ... def with_range(self, /, range: str) -> MinerFactory: ... def with_subnet(self, /, subnet: str) -> MinerFactory: ... diff --git a/src/factory.rs b/src/factory.rs index 9399e257..86b27be7 100644 --- a/src/factory.rs +++ b/src/factory.rs @@ -29,19 +29,23 @@ use ipnet::IpNet; use rand::seq::SliceRandom; use tokio::{net::TcpStream, time::timeout}; -const IDENTIFICATION_TIMEOUT: Duration = Duration::from_secs(10); -const CONNECTIVITY_TIMEOUT: Duration = Duration::from_secs(1); -const CONNECTIVITY_RETRIES: u32 = 3; +const IDENTIFICATION_TIMEOUT: Duration = Duration::from_secs(3); +const MINER_CONSTRUCTION_TIMEOUT: Duration = Duration::from_secs(5); +const IDENTIFICATION_CONCURRENCY: usize = 128; +const CONNECTIVITY_TIMEOUT: Duration = Duration::from_millis(500); +const CONNECTIVITY_RETRIES: u32 = 0; +const CONNECTIVITY_CONCURRENCY: usize = 256; +const CONNECTIVITY_RETRY_CONCURRENCY: usize = 64; +const CONNECTIVITY_RETRY_BACKOFF: Duration = Duration::from_millis(100); +const SCAN_RETRIES: u32 = 0; +const SCAN_RETRY_BACKOFF: Duration = Duration::from_millis(250); +const IDENTIFICATION_RETRY_CONCURRENCY: usize = 8; const NOFILE_PER_CONCURRENCY: u64 = 8; const MIN_NOFILE_LIMIT: u64 = 2048; +const MINER_PORTS: [u16; 4] = [80, 4028, 4029, 8889]; fn calculate_optimal_concurrency(ip_count: usize) -> usize { - match ip_count { - 0..=1000 => 1000, - 1001..=5000 => 2500, - 5001..=10000 => 5000, - _ => 10000, - } + ip_count.clamp(1, IDENTIFICATION_CONCURRENCY) } fn calculate_desired_nofile_limit(concurrency: usize) -> u64 { @@ -50,6 +54,10 @@ fn calculate_desired_nofile_limit(concurrency: usize) -> u64 { .max(MIN_NOFILE_LIMIT) } +fn calculate_retry_concurrency(concurrency: usize) -> usize { + concurrency.clamp(1, IDENTIFICATION_RETRY_CONCURRENCY) +} + async fn check_port_open(ip: IpAddr, port: u16, connectivity_timeout: Duration) -> bool { let addr: SocketAddr = (ip, port).into(); let stream = match timeout(connectivity_timeout, TcpStream::connect(addr)).await { @@ -60,6 +68,40 @@ async fn check_port_open(ip: IpAddr, port: u16, connectivity_timeout: Duration) true } +async fn check_miner_ports(ip: IpAddr, connectivity_timeout: Duration) -> bool { + let mut probes = stream::iter(MINER_PORTS) + .map(|port| check_port_open(ip, port, connectivity_timeout)) + .buffer_unordered(MINER_PORTS.len()); + + while let Some(open) = probes.next().await { + if open { + return true; + } + } + false +} + +async fn probe_ip_batch( + ips: Vec, + connectivity_timeout: Duration, + concurrency: usize, +) -> (Vec, Vec) { + let mut reachable = Vec::new(); + let mut unreachable = Vec::new(); + let mut probes = stream::iter(ips) + .map(|ip| async move { (ip, check_miner_ports(ip, connectivity_timeout).await) }) + .buffer_unordered(concurrency.max(1)); + + while let Some((ip, is_reachable)) = probes.next().await { + if is_reachable { + reachable.push(ip); + } else { + unreachable.push(ip); + } + } + (reachable, unreachable) +} + async fn get_miner_type_from_command( ip: IpAddr, command: MinerCommand, @@ -238,12 +280,40 @@ pub struct MinerFactory { ips: Vec, discovery_auth_by_firmware: HashMap, identification_timeout: Duration, + miner_construction_timeout: Duration, connectivity_timeout: Duration, connectivity_retries: u32, + connectivity_concurrent: usize, + connectivity_retry_concurrent: usize, + connectivity_retry_backoff: Duration, + scan_retries: u32, + scan_retry_backoff: Duration, + retry_concurrent: Option, concurrent: Option, nofile_limit: Option, nofile_adjustment: bool, check_port: bool, + strict_port_check: bool, +} + +#[allow(clippy::type_complexity)] +fn identify_ip_stream( + factory: Arc, + ips: Vec, + concurrency: usize, +) -> impl Stream>)> + Send { + stream::iter(ips) + .map(move |ip| { + let factory = Arc::clone(&factory); + async move { + let miner = factory.get_miner(ip).await.unwrap_or_else(|error| { + tracing::debug!(%ip, %error, "miner identification failed"); + None + }); + (ip, miner) + } + }) + .buffer_unordered(concurrency.max(1)) } impl std::fmt::Debug for MinerFactory { @@ -259,12 +329,29 @@ impl std::fmt::Debug for MinerFactory { &self.discovery_auth_by_firmware.len(), ) .field("identification_timeout", &self.identification_timeout) + .field( + "miner_construction_timeout", + &self.miner_construction_timeout, + ) .field("connectivity_timeout", &self.connectivity_timeout) .field("connectivity_retries", &self.connectivity_retries) + .field("connectivity_concurrent", &self.connectivity_concurrent) + .field( + "connectivity_retry_concurrent", + &self.connectivity_retry_concurrent, + ) + .field( + "connectivity_retry_backoff", + &self.connectivity_retry_backoff, + ) + .field("scan_retries", &self.scan_retries) + .field("scan_retry_backoff", &self.scan_retry_backoff) + .field("retry_concurrent", &self.retry_concurrent) .field("concurrent", &self.concurrent) .field("nofile_limit", &self.nofile_limit) .field("nofile_adjustment", &self.nofile_adjustment) .field("check_port", &self.check_port) + .field("strict_port_check", &self.strict_port_check) .finish() } } @@ -278,25 +365,26 @@ impl Default for MinerFactory { impl MinerFactory { #[tracing::instrument(level = "debug", skip(self))] pub async fn scan_miner(&self, ip: IpAddr) -> Result>> { - if (1..self.connectivity_retries).next().is_some() { - if !self.check_port { - return self.get_miner(ip).await; - } - if check_port_open(ip, 80, self.connectivity_timeout).await { - return self.get_miner(ip).await; - } - if check_port_open(ip, 4028, self.connectivity_timeout).await { - return self.get_miner(ip).await; - } - if check_port_open(ip, 4029, self.connectivity_timeout).await { - return self.get_miner(ip).await; + if !self.check_port { + return self.get_miner(ip).await; + } + + for attempt in 0..=self.connectivity_retries { + if attempt > 0 { + let multiplier = 1_u32 << (attempt - 1).min(10); + tokio::time::sleep(self.connectivity_retry_backoff.saturating_mul(multiplier)) + .await; } - if check_port_open(ip, 8889, self.connectivity_timeout).await { + if check_miner_ports(ip, self.connectivity_timeout).await { return self.get_miner(ip).await; } } tracing::trace!("no response from any miner-specific ports"); - Ok(None) + if self.strict_port_check { + Ok(None) + } else { + self.get_miner(ip).await + } } /// Discover and construct a miner at the given IP. @@ -391,13 +479,26 @@ impl MinerFactory { match found { Some(fw) => { + let firmware = fw.to_string(); let auth = self.discovery_auth_by_firmware.get(&fw.to_string()); - match fw.build_miner(ip, auth).await { - Ok(miner) => Ok(Some(miner)), - Err(e) => { + let build_result = + timeout(self.miner_construction_timeout, fw.build_miner(ip, auth)).await; + + match build_result { + Ok(Ok(miner)) => Ok(Some(miner)), + Ok(Err(e)) => { tracing::debug!("failed to build miner for {ip}: {e}"); Ok(None) } + Err(_) => { + tracing::warn!( + %ip, + %firmware, + timeout_ms = self.miner_construction_timeout.as_millis(), + "miner construction timed out" + ); + Ok(None) + } } } None => { @@ -417,25 +518,44 @@ impl MinerFactory { ips: Vec::new(), discovery_auth_by_firmware: HashMap::new(), identification_timeout: IDENTIFICATION_TIMEOUT, + miner_construction_timeout: MINER_CONSTRUCTION_TIMEOUT, connectivity_timeout: CONNECTIVITY_TIMEOUT, connectivity_retries: CONNECTIVITY_RETRIES, + connectivity_concurrent: CONNECTIVITY_CONCURRENCY, + connectivity_retry_concurrent: CONNECTIVITY_RETRY_CONCURRENCY, + connectivity_retry_backoff: CONNECTIVITY_RETRY_BACKOFF, + scan_retries: SCAN_RETRIES, + scan_retry_backoff: SCAN_RETRY_BACKOFF, + retry_concurrent: None, concurrent: None, nofile_limit: None, nofile_adjustment: true, check_port: true, + strict_port_check: true, } } - /// Enable or disable the quick TCP port check before miner identification. + /// Enable or disable TCP reachability prioritization before identification. /// - /// Port checking reduces wasted identification attempts during scans by - /// probing common miner ports first. Disable it when a network filters TCP - /// probes but still responds to the firmware-specific discovery requests. + /// Responsive hosts are identified after the reachability stage. By + /// default, hosts that miss the probe are excluded; call + /// `with_strict_port_check(false)` to enable the conservative fallback. pub fn with_port_check(mut self, enabled: bool) -> Self { self.check_port = enabled; self } + /// Exclude hosts that miss all TCP reachability probes. + /// + /// This can reduce scan time on sparse networks, but may produce false + /// negatives when ARP resolution, switches, or embedded TCP stacks drop + /// probe bursts. The default is `true`; set this to `false` to run one + /// application-identification fallback pass for probe-negative hosts. + pub fn with_strict_port_check(mut self, enabled: bool) -> Self { + self.strict_port_check = enabled; + self + } + /// Set credentials for a specific firmware entry to use during /// miner construction after identification. pub fn with_firmware_discovery_auth( @@ -448,11 +568,46 @@ impl MinerFactory { self } - /// Set the maximum number of addresses scanned at the same time. + /// Set the maximum number of responsive addresses identified concurrently. /// - /// If unset, scan concurrency is chosen from the number of queued hosts. + /// TCP reachability probing has an independent concurrency limit. If this + /// is unset, identification uses a default of 128 addresses. pub fn with_concurrent_limit(mut self, limit: usize) -> Self { - self.concurrent = Some(limit); + self.concurrent = Some(limit.max(1)); + self + } + + /// Set the number of additional attempts for addresses that fail scanning. + /// + /// Retries run as a separate lower-concurrency batch after the primary + /// identification pass. The default is zero retries. + pub fn with_scan_retries(mut self, retries: u32) -> Self { + self.scan_retries = retries; + self + } + + /// Set the maximum number of failed addresses retried concurrently. + /// + /// If unset, retry concurrency is capped at 8. This conservative cap + /// prioritizes recovery on saturated networks; callers can explicitly + /// select a higher limit. + pub fn with_retry_concurrent_limit(mut self, limit: usize) -> Self { + self.retry_concurrent = Some(limit.max(1)); + self + } + + /// Set the initial delay before retrying a failed address. + /// + /// Additional retries use exponential backoff. Concurrency remains bounded + /// by [`Self::with_retry_concurrent_limit`]. + pub fn with_scan_retry_backoff(mut self, backoff: Duration) -> Self { + self.scan_retry_backoff = backoff; + self + } + + /// Set the initial scan retry delay in milliseconds. + pub fn with_scan_retry_backoff_millis(mut self, backoff_millis: u64) -> Self { + self.scan_retry_backoff = Duration::from_millis(backoff_millis); self } @@ -502,6 +657,18 @@ impl MinerFactory { self } + /// Set the maximum time allowed to construct a miner after identification. + pub fn with_miner_construction_timeout(mut self, timeout: Duration) -> Self { + self.miner_construction_timeout = timeout; + self + } + + /// Set the miner construction timeout in seconds. + pub fn with_miner_construction_timeout_secs(mut self, timeout_secs: u64) -> Self { + self.miner_construction_timeout = Duration::from_secs(timeout_secs); + self + } + /// Set the timeout for quick connectivity probes during scans. pub fn with_connectivity_timeout(mut self, timeout: Duration) -> Self { self.connectivity_timeout = timeout; @@ -514,12 +681,45 @@ impl MinerFactory { self } - /// Set how many connectivity attempts are made before identification. + /// Set the connectivity probe timeout in milliseconds. + pub fn with_connectivity_timeout_millis(mut self, timeout_millis: u64) -> Self { + self.connectivity_timeout = Duration::from_millis(timeout_millis); + self + } + + /// Set the number of connectivity retries after the initial attempt. + /// + /// A value of zero performs one initial connectivity pass without retries, + /// which is the default. pub fn with_connectivity_retries(mut self, retries: u32) -> Self { self.connectivity_retries = retries; self } + /// Set the maximum number of addresses probed concurrently on the first pass. + pub fn with_connectivity_concurrent_limit(mut self, limit: usize) -> Self { + self.connectivity_concurrent = limit.max(1); + self + } + + /// Set the maximum number of addresses probed concurrently on retry passes. + pub fn with_connectivity_retry_concurrent_limit(mut self, limit: usize) -> Self { + self.connectivity_retry_concurrent = limit.max(1); + self + } + + /// Set the initial delay before retrying connectivity probes. + pub fn with_connectivity_retry_backoff(mut self, backoff: Duration) -> Self { + self.connectivity_retry_backoff = backoff; + self + } + + /// Set the initial connectivity retry delay in milliseconds. + pub fn with_connectivity_retry_backoff_millis(mut self, backoff_millis: u64) -> Self { + self.connectivity_retry_backoff = Duration::from_millis(backoff_millis); + self + } + /// Override the firmware registry with a custom list. pub fn with_firmwares(mut self, firmwares: Vec>) -> Self { self.search_firmwares = Some(firmwares); @@ -688,6 +888,120 @@ impl MinerFactory { self.ips.is_empty() } + #[allow(clippy::type_complexity)] + fn scan_outcomes_stream( + &self, + ) -> Pin>)> + Send + use<>>> { + let identification_concurrency = self + .concurrent + .unwrap_or(calculate_optimal_concurrency(self.ips.len())) + .max(1); + let identification_retry_concurrency = self + .retry_concurrent + .unwrap_or_else(|| calculate_retry_concurrency(identification_concurrency)); + let descriptor_concurrency = self + .connectivity_concurrent + .saturating_mul(MINER_PORTS.len()) + .max(identification_concurrency); + + if let Some(desired_nofile) = self.nofile_limit.or_else(|| { + self.nofile_adjustment + .then(|| calculate_desired_nofile_limit(descriptor_concurrency)) + }) { + maybe_adjust_nofile_limit(desired_nofile); + } + + // The returned stream must own the factory configuration rather than + // borrow `self`, and every in-flight identification task shares it. + let factory = Arc::new(self.clone()); + let queued_ips = self.ips.clone(); + Box::pin(async_stream::stream! { + let (mut candidates, unreachable) = if factory.check_port { + let (mut reachable, mut unreachable) = probe_ip_batch( + queued_ips, + factory.connectivity_timeout, + factory.connectivity_concurrent, + ).await; + + for retry_index in 0..factory.connectivity_retries { + if unreachable.is_empty() { + break; + } + let multiplier = 1_u32 << retry_index.min(10); + tokio::time::sleep( + factory + .connectivity_retry_backoff + .saturating_mul(multiplier), + ) + .await; + let (recovered, still_unreachable) = probe_ip_batch( + unreachable, + factory.connectivity_timeout, + factory.connectivity_retry_concurrent, + ) + .await; + reachable.extend(recovered); + unreachable = still_unreachable; + } + (reachable, unreachable) + } else { + (queued_ips, Vec::new()) + }; + + for attempt in 0..=factory.scan_retries { + if candidates.is_empty() { + break; + } + if attempt > 0 { + let multiplier = 1_u32 << (attempt - 1).min(10); + tokio::time::sleep(factory.scan_retry_backoff.saturating_mul(multiplier)).await; + } + let concurrency = if attempt == 0 { + identification_concurrency + } else { + identification_retry_concurrency + }; + let mut unidentified = Vec::new(); + let mut identification = Box::pin(identify_ip_stream( + Arc::clone(&factory), + candidates, + concurrency, + )); + while let Some((ip, miner)) = identification.next().await { + if miner.is_some() { + yield (ip, miner); + } else { + unidentified.push(ip); + } + } + candidates = unidentified; + } + + for ip in candidates { + yield (ip, None); + } + + if factory.strict_port_check { + for ip in unreachable { + yield (ip, None); + } + } else { + // A TCP probe is an optimization, not proof that a host is + // absent. Preserve correctness with one conservative fallback + // pass after the responsive-host fast path. Fallback misses are + // not retried because most are genuinely unused addresses. + let mut fallback = Box::pin(identify_ip_stream( + Arc::clone(&factory), + unreachable, + identification_concurrency, + )); + while let Some(outcome) = fallback.next().await { + yield outcome; + } + } + }) + } + /// Scan all queued addresses and return every successfully identified miner. /// /// Unsupported hosts and failed identification attempts are skipped. An @@ -698,92 +1012,30 @@ impl MinerFactory { "No IPs to scan. Use with_subnet, with_octets, or with_range to set IPs." )); } - - let concurrency = self - .concurrent - .unwrap_or(calculate_optimal_concurrency(self.ips.len())); - - if let Some(desired_nofile) = self.nofile_limit.or_else(|| { - self.nofile_adjustment - .then(|| calculate_desired_nofile_limit(concurrency)) - }) { - maybe_adjust_nofile_limit(desired_nofile); - } - - let miners: Vec> = stream::iter(self.ips.iter().copied()) - .map(|ip| async move { self.scan_miner(ip).await.ok().flatten() }) - .buffer_unordered(concurrency) - .filter_map(|miner_opt| async move { miner_opt }) - .collect() - .await; - - Ok(miners) + Ok(self.scan_stream().collect().await) } /// Scan queued addresses as a stream of successfully identified miners. /// - /// Use this when callers should process miners as soon as they are found - /// instead of waiting for the full scan to finish. + /// The TCP reachability pass completes before application identification. + /// Responsive addresses are identified with a separate, lower concurrency + /// limit so embedded web servers are not overloaded. pub fn scan_stream(&self) -> Pin> + Send + use<>>> { - let concurrency = self - .concurrent - .unwrap_or(calculate_optimal_concurrency(self.ips.len())); - - if let Some(desired_nofile) = self.nofile_limit.or_else(|| { - self.nofile_adjustment - .then(|| calculate_desired_nofile_limit(concurrency)) - }) { - maybe_adjust_nofile_limit(desired_nofile); - } - - let factory = Arc::new(self.clone()); - let ips: Arc<[IpAddr]> = Arc::from(self.ips.as_slice()); - - let ip_count = ips.len(); - let stream = stream::iter(0..ip_count) - .map(move |i| { - let factory = Arc::clone(&factory); - let ips = Arc::clone(&ips); - async move { factory.scan_miner(ips[i]).await.ok().flatten() } - }) - .buffer_unordered(concurrency) - .filter_map(|miner_opt| async move { miner_opt }); - - Box::pin(stream) + Box::pin( + self.scan_outcomes_stream() + .filter_map(|(_, miner)| async move { miner }), + ) } /// Scan queued addresses as a stream that preserves every attempted IP. /// - /// Stream items are `(ip, miner)` pairs. `miner` is `None` when the host did - /// not identify as a supported miner. + /// Stream items are `(ip, miner)` pairs. `miner` is `None` when the host was + /// unreachable or did not identify as a supported miner. #[allow(clippy::type_complexity)] pub fn scan_stream_with_ip( &self, ) -> Pin>)> + Send + use<>>> { - let concurrency = self - .concurrent - .unwrap_or(calculate_optimal_concurrency(self.ips.len())); - - if let Some(desired_nofile) = self.nofile_limit.or_else(|| { - self.nofile_adjustment - .then(|| calculate_desired_nofile_limit(concurrency)) - }) { - maybe_adjust_nofile_limit(desired_nofile); - } - - let factory = Arc::new(self.clone()); - let ips: Arc<[IpAddr]> = Arc::from(self.ips.as_slice()); - - let ip_count = ips.len(); - let stream = stream::iter(0..ip_count) - .map(move |i| { - let factory = Arc::clone(&factory); - let ips = Arc::clone(&ips); - async move { (ips[i], factory.scan_miner(ips[i]).await.ok().flatten()) } - }) - .buffer_unordered(concurrency); - - Box::pin(stream) + self.scan_outcomes_stream() } /// Append an octet range, scan it, and return identified miners. @@ -941,6 +1193,41 @@ mod tests { assert!(ips.contains(&IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2)))); } + #[test] + fn retry_concurrency_scales_below_primary_concurrency() { + assert_eq!(calculate_retry_concurrency(0), 1); + assert_eq!(calculate_retry_concurrency(1), 1); + assert_eq!(calculate_retry_concurrency(8), 8); + assert_eq!(calculate_retry_concurrency(9), 8); + assert_eq!(calculate_retry_concurrency(64), 8); + assert_eq!(calculate_retry_concurrency(512), 8); + assert_eq!(calculate_retry_concurrency(10_000), 8); + } + + #[test] + fn staged_scan_defaults_are_bounded() { + let factory = MinerFactory::new(); + assert_eq!(calculate_optimal_concurrency(1024), 128); + assert_eq!(factory.identification_timeout, Duration::from_secs(3)); + assert_eq!(factory.miner_construction_timeout, Duration::from_secs(5)); + assert_eq!(factory.connectivity_timeout, Duration::from_millis(500)); + assert_eq!(factory.connectivity_retries, 0); + assert_eq!(factory.scan_retries, 0); + assert_eq!(factory.connectivity_concurrent, 256); + assert_eq!(factory.connectivity_retry_concurrent, 64); + assert!(factory.strict_port_check); + assert_eq!( + factory.connectivity_retry_backoff, + Duration::from_millis(100) + ); + } + + #[test] + fn connectivity_retries_can_be_disabled_without_disabling_initial_attempt() { + let factory = MinerFactory::new().with_connectivity_retries(0); + assert_eq!(factory.connectivity_retries, 0); + } + #[test] #[cfg(feature = "nerdaxe")] fn identify_nerdaxe_web() { diff --git a/src/python/factory.rs b/src/python/factory.rs index cd0f719a..7da86f18 100644 --- a/src/python/factory.rs +++ b/src/python/factory.rs @@ -332,6 +332,36 @@ impl MinerFactory { })) } + /// Set the number of additional attempts for addresses that fail scanning. + pub fn with_scan_retries<'py>( + slf: PyRefMut<'py, Self>, + retries: u32, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_scan_retries(retries) + })) + } + + /// Set the maximum number of failed addresses retried concurrently. + pub fn with_retry_concurrent_limit<'py>( + slf: PyRefMut<'py, Self>, + limit: usize, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_retry_concurrent_limit(limit) + })) + } + + /// Set the initial scan retry backoff in milliseconds. + pub fn with_scan_retry_backoff_millis<'py>( + slf: PyRefMut<'py, Self>, + backoff_millis: u64, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_scan_retry_backoff_millis(backoff_millis) + })) + } + /// Set the maximum seconds spent identifying a miner after it responds. pub fn with_identification_timeout_secs<'py>( slf: PyRefMut<'py, Self>, @@ -342,6 +372,16 @@ impl MinerFactory { })) } + /// Set the maximum seconds allowed to construct a miner after identification. + pub fn with_miner_construction_timeout_secs<'py>( + slf: PyRefMut<'py, Self>, + timeout_secs: u64, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_miner_construction_timeout_secs(timeout_secs) + })) + } + /// Set the timeout, in seconds, for quick connectivity probes. pub fn with_connectivity_timeout_secs<'py>( slf: PyRefMut<'py, Self>, @@ -352,7 +392,17 @@ impl MinerFactory { })) } - /// Set how many connectivity attempts are made before identification. + /// Set the timeout, in milliseconds, for quick connectivity probes. + pub fn with_connectivity_timeout_millis<'py>( + slf: PyRefMut<'py, Self>, + timeout_millis: u64, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_connectivity_timeout_millis(timeout_millis) + })) + } + + /// Set connectivity retries after the initial pass. pub fn with_connectivity_retries<'py>( slf: PyRefMut<'py, Self>, retries: u32, @@ -362,6 +412,36 @@ impl MinerFactory { })) } + /// Set first-pass connectivity probe concurrency. + pub fn with_connectivity_concurrent_limit<'py>( + slf: PyRefMut<'py, Self>, + limit: usize, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_connectivity_concurrent_limit(limit) + })) + } + + /// Set retry-pass connectivity probe concurrency. + pub fn with_connectivity_retry_concurrent_limit<'py>( + slf: PyRefMut<'py, Self>, + limit: usize, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_connectivity_retry_concurrent_limit(limit) + })) + } + + /// Set the initial connectivity retry delay in milliseconds. + pub fn with_connectivity_retry_backoff_millis<'py>( + slf: PyRefMut<'py, Self>, + backoff_millis: u64, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_connectivity_retry_backoff_millis(backoff_millis) + })) + } + /// Enable or disable quick TCP port checks before miner identification. pub fn with_port_check<'py>( slf: PyRefMut<'py, Self>, @@ -372,6 +452,16 @@ impl MinerFactory { })) } + /// Exclude hosts that miss all TCP connectivity probes. + pub fn with_strict_port_check<'py>( + slf: PyRefMut<'py, Self>, + enabled: bool, + ) -> PyResult> { + Ok(Self::update_inner(slf, |inner| { + inner.with_strict_port_check(enabled) + })) + } + /// Await a scan of all queued addresses and return every supported miner. /// /// Unsupported hosts are skipped. Raises `ValueError` when no addresses have