diff --git a/ccbt.toml b/ccbt.toml index 9dff316..139413d 100644 --- a/ccbt.toml +++ b/ccbt.toml @@ -1,4 +1,5 @@ [network] +swarm_acceleration_enabled = true max_global_peers = 200 max_peers_per_torrent = 50 pipeline_depth = 16 @@ -223,6 +224,8 @@ checkpoint_retention_days = 30 [strategy] piece_selection = "rarest_first" +straggler_duplication_enabled = true +adaptive_request_timeout_enabled = true endgame_duplicates = 2 endgame_threshold = 0.95 pipeline_capacity = 4 @@ -237,7 +240,7 @@ adaptive_hybrid_phase_detection_window = 10 [discovery] enable_dht = true -min_peers_before_dht = 10 +min_peers_before_dht = 2 enable_pex = true enable_udp_trackers = true enable_http_trackers = true diff --git a/ccbt/config/config.py b/ccbt/config/config.py index c9adfc4..b63f4cf 100644 --- a/ccbt/config/config.py +++ b/ccbt/config/config.py @@ -101,10 +101,13 @@ def _optimization_profile_overlays() -> dict[ "piece_selection": "rarest_first", "pipeline_capacity": 4, "endgame_duplicates": 2, + "straggler_duplication_enabled": True, + "adaptive_request_timeout_enabled": True, }, "network": { "max_peers_per_torrent": 50, "max_global_peers": 200, + "swarm_acceleration_enabled": True, }, "discovery": { "tracker_announce_interval": 60.0, @@ -120,10 +123,17 @@ def _optimization_profile_overlays() -> dict[ "piece_selection": "bandwidth_weighted_rarest", "pipeline_capacity": 8, "endgame_duplicates": 3, + "straggler_duplication_enabled": True, + "straggler_max_copies": 3, + "duplicate_budget_fraction": 0.10, + "adaptive_request_timeout_enabled": True, }, "network": { "max_peers_per_torrent": 100, "max_global_peers": 500, + "swarm_acceleration_enabled": True, + "warm_pipeline_depth": 32, + "ramp_productive_dial_cap": 24, }, "discovery": { "tracker_announce_interval": 30.0, @@ -160,10 +170,12 @@ def _optimization_profile_overlays() -> dict[ "piece_selection": "rarest_first", "pipeline_capacity": 2, "endgame_duplicates": 1, + "straggler_duplication_enabled": False, }, "network": { "max_peers_per_torrent": 10, "max_global_peers": 50, + "swarm_acceleration_enabled": False, }, "discovery": { "tracker_announce_interval": 120.0, diff --git a/ccbt/discovery/tracker.py b/ccbt/discovery/tracker.py index 90dfd99..9e3e7cd 100644 --- a/ccbt/discovery/tracker.py +++ b/ccbt/discovery/tracker.py @@ -27,6 +27,7 @@ from ccbt.config.config import get_config from ccbt.core.bencode import BencodeDecoder from ccbt.models import PeerInfo +from ccbt.performance.acceleration import accel_enabled from ccbt.security.ssl_context import CertificatePinner from ccbt.utils.tracker_utils import ( tracker_url_implies_tls, @@ -2241,9 +2242,20 @@ async def _announce_udp_sequential() -> list[Union[TrackerResponse, None]]: url_to_task[task] = url if udp_urls: - udp_task = asyncio.create_task(_announce_udp_sequential()) - tasks.append(udp_task) - url_to_task[udp_task] = "udp://sequential-batch" + if accel_enabled(self.config): + # Acceleration: UDP announces are independent datagram + # exchanges; run them concurrently like HTTP. The UDP + # client already caps pending requests per host. + for url in udp_urls: + torrent_copy = shared_torrent_data.copy() + torrent_copy["announce"] = url + udp_task = asyncio.create_task(_announce_single(torrent_copy)) + tasks.append(udp_task) + url_to_task[udp_task] = url + else: + udp_task = asyncio.create_task(_announce_udp_sequential()) + tasks.append(udp_task) + url_to_task[udp_task] = "udp://sequential-batch" # Wait for all announces to complete self.logger.debug( diff --git a/ccbt/extensions/manager.py b/ccbt/extensions/manager.py index 8770f79..0374bcb 100644 --- a/ccbt/extensions/manager.py +++ b/ccbt/extensions/manager.py @@ -376,6 +376,12 @@ async def handle_extension_protocol( self.extension_states["protocol"].error_count += 1 self.extension_states["protocol"].last_error = str(e) + def register_pex_peer_callback(self, callback: Any) -> None: + """Register a callback dialed with inbound PEX peers (BEP 11).""" + pex_ext = self.extensions.get("pex") + if pex_ext is not None and hasattr(pex_ext, "add_peer_discovered_callback"): + pex_ext.add_peer_discovered_callback(callback) + # PEX methods async def handle_pex_message( self, diff --git a/ccbt/extensions/pex.py b/ccbt/extensions/pex.py index e17a5f8..7280eb6 100644 --- a/ccbt/extensions/pex.py +++ b/ccbt/extensions/pex.py @@ -56,6 +56,14 @@ def __init__(self): self.peer_flags: dict[tuple[str, int], int] = {} # (ip, port) -> flags self.FLAG_PREFER_ENCRYPT = 0x01 self.FLAG_SEED = 0x02 + # Callbacks invoked with newly discovered peers so inbound PEX + # peers are dialed immediately, not only emitted as events. + self.peer_discovered_callbacks: list[Any] = [] + + def add_peer_discovered_callback(self, callback: Any) -> None: + """Register a sync or async callback for inbound PEX peers.""" + if callback not in self.peer_discovered_callbacks: + self.peer_discovered_callbacks.append(callback) def encode_compact_peer(self, peer: PEXPeer) -> bytes: """Encode peer in compact format.""" @@ -389,6 +397,15 @@ def decode_pex_message( async def handle_added_peers(self, peer_id: str, peers: list[PEXPeer]) -> None: """Handle added peers from PEX.""" + new_peers = [p for p in peers if p not in self.added_peers] + for callback in list(self.peer_discovered_callbacks): + if new_peers: + try: + result = callback(list(new_peers)) + if result is not None and hasattr(result, "__await__"): + await result + except Exception: + pass # A dialing callback must never break PEX parsing for pex_peer in peers: self.added_peers.add(pex_peer) diff --git a/ccbt/models.py b/ccbt/models.py index ec698cc..70ab2ce 100644 --- a/ccbt/models.py +++ b/ccbt/models.py @@ -808,6 +808,51 @@ class NetworkConfig(BaseModel): le=600.0, description="Base BitTorrent block request timeout in seconds", ) + # Download acceleration (see ccbt/performance/acceleration.py) + swarm_acceleration_enabled: bool = Field( + default=True, + description="Enable acceleration: parallel raced dialing, straggler duplication, adaptive timeouts, snub detection, BEP6 allow-fast requests", + ) + speculative_connect_timeout: float = Field( + default=5.0, + ge=1.0, + le=30.0, + description="First-pass connect timeout for speculative dials (retries use connection_timeout)", + ) + ramp_connect_parallel_batches: int = Field( + default=4, + ge=1, + le=8, + description="Concurrent connect_to_peers batches while accelerating peer ramp-up", + ) + ramp_productive_dial_cap: int = Field( + default=24, + ge=3, + le=50, + description="Parallel outbound dials while downloading but still below requestable target", + ) + ramp_choked_probe_interval_s: float = Field( + default=2.0, + ge=0.5, + le=30.0, + description="Probe-dial interval when all active peers are remote-choked (replaces 15s hold)", + ) + warm_pipeline_depth: int = Field( + default=32, + ge=1, + le=128, + description="Initial per-peer pipeline depth before BDP adaptation (acceleration)", + ) + snub_threshold_s: float = Field( + default=10.0, + ge=1.0, + le=120.0, + description="Seconds an unchoked peer may deliver nothing (while others deliver) before being snubbed", + ) + allow_fast_enabled: bool = Field( + default=True, + description="Request BEP6 allowed-fast pieces even while remote-choked", + ) sparse_pipeline_stale_payload_cancel_s: float = Field( default=120.0, ge=0.0, @@ -2547,6 +2592,33 @@ class StrategyConfig(BaseModel): le=32, description="Request pipeline capacity", ) + # Download acceleration: speculative / straggler duplication + straggler_duplication_enabled: bool = Field( + default=True, + description="Duplicate straggler / last-blocks of a piece to other peers before endgame", + ) + straggler_max_copies: int = Field( + default=3, + ge=1, + le=10, + description="Maximum simultaneous copies of one block (original + duplicates)", + ) + duplicate_budget_fraction: float = Field( + default=0.10, + ge=0.0, + le=0.5, + description="Max redundant bytes as fraction of total downloaded bytes; duplication auto-disables beyond this", + ) + adaptive_request_timeout_enabled: bool = Field( + default=True, + description="Use clamp(4 x peer p95 block latency, 5s, 30s) instead of fixed 60s request timeout", + ) + dynamic_endgame_last_pieces: int = Field( + default=20, + ge=1, + le=1000, + description="Enter endgame when this many pieces remain, regardless of percentage threshold", + ) streaming_mode: bool = Field(default=False, description="Enable streaming mode") # Advanced strategy settings @@ -2653,11 +2725,23 @@ class DiscoveryConfig(BaseModel): enable_dht: bool = Field(default=True, description="Enable DHT") min_peers_before_dht: int = Field( - default=10, + default=2, ge=0, le=100, description="Minimum active peers before starting DHT discovery (0 = allow DHT immediately as fallback)", ) + dht_max_peers_per_query_accelerated: int = Field( + default=100, + ge=10, + le=500, + description="DHT get_peers cap while requestable peers are scarce (acceleration)", + ) + dht_query_interval_accelerated_s: float = Field( + default=5.0, + ge=1.0, + le=60.0, + description="DHT query interval while no requestable peers (acceleration)", + ) enable_pex: bool = Field(default=True, description="Enable Peer Exchange") enable_udp_trackers: bool = Field(default=True, description="Enable UDP trackers") enable_http_trackers: bool = Field(default=True, description="Enable HTTP trackers") diff --git a/ccbt/peer/async_peer_connection.py b/ccbt/peer/async_peer_connection.py index fe73349..0c2d4dd 100644 --- a/ccbt/peer/async_peer_connection.py +++ b/ccbt/peer/async_peer_connection.py @@ -52,6 +52,17 @@ UnchokeMessage, parse_plaintext_bittorrent_handshake, ) +from ccbt.performance.acceleration import ( + accel_enabled, + adaptive_request_timeout, + adaptive_upload_slots, + is_snubbed, + peer_usefulness_score, + race_first_success, + ramp_choking_intervals, + strategy_flag, + warm_pipeline_depth, +) from ccbt.protocols.bittorrent_v2 import ( MESSAGE_ID_FILE_TREE_REQUEST, MESSAGE_ID_FILE_TREE_RESPONSE, @@ -470,6 +481,7 @@ def can_request( *, require_recent_piece_availability: bool = False, effective_pipeline_cap: Optional[int] = None, + piece_index: Optional[int] = None, ) -> bool: """Check if we can make new requests. @@ -485,6 +497,7 @@ def can_request( block_reason = self.get_request_block_reason( require_recent_piece_availability=require_recent_piece_availability, effective_pipeline_cap=effective_pipeline_cap, + piece_index=piece_index, ) can_req = block_reason is None is_active = self.is_active() @@ -545,6 +558,7 @@ def get_request_block_reason( *, require_recent_piece_availability: bool = False, effective_pipeline_cap: Optional[int] = None, + piece_index: Optional[int] = None, ) -> Optional[str]: """Return the request-block reason for this peer, if any. @@ -554,7 +568,15 @@ def get_request_block_reason( if not is_active: return "inactive" if self.peer_choking: - return "remote_choked" + # BEP 6: allowed-fast pieces may be requested while choked. + if piece_index is not None: + allowed = getattr(self.peer_state, "bep6_allowed_fast_pieces", set()) + if piece_index in allowed: + pass # fall through to pipeline checks + else: + return "remote_choked" + else: + return "remote_choked" cap = ( self.max_pipeline_depth if effective_pipeline_cap is None @@ -3024,7 +3046,7 @@ def _should_skip_pending_requeue(self, peer_info: PeerInfo) -> bool: def _productive_connect_throttle(self) -> int: """Max parallel outbound connects while a productive download is in flight.""" - return int( + legacy = int( getattr( self.config.network, "connect_throttle_productive_max_concurrent", @@ -3032,6 +3054,23 @@ def _productive_connect_throttle(self) -> int: ) or 8 ) + if accel_enabled(self.config): + # Acceleration: keep dialing until we have a useful cohort of + # requestable peers, not just any productive peer. + try: + _, _, requestable = self._snapshot_connection_counts() + except Exception: + requestable = 0 + target = max(1, int(getattr(self, "max_peers_per_torrent", 50) or 50) // 2) + if requestable < target: + return max( + legacy, + int( + getattr(self.config.network, "ramp_productive_dial_cap", 24) + or 24 + ), + ) + return legacy async def _get_quality_active_counts(self) -> tuple[int, int]: """Return (quality_active, total_active) peer counts.""" @@ -4181,6 +4220,19 @@ def _cap_connect_task_timeout_s(self, timeout_s: float) -> float: capped = min(capped, 42.0) elif pending_depth > 100: capped = min(capped, 55.0) + if accel_enabled(self.config) and active_peer_count > 0: + # Speculative first pass: with a deep candidate queue, fail fast + # and move on; retries from the pending queue get full budgets. + # Never at cold start (active == 0): handshakes there legitimately + # need the full budget, and capping caused connect churn in + # real-swarm A/B runs. + spec = float( + getattr(self.config.network, "speculative_connect_timeout", 5.0) or 5.0 + ) + if pending_depth > 100: + capped = min(capped, max(15.0, spec * 4.0)) + elif pending_depth > 50: + capped = min(capped, max(15.0, spec * 6.0)) return max(15.0, capped) def _calculate_timeout( @@ -4258,6 +4310,23 @@ def _calculate_pipeline_depth(self, connection: AsyncPeerConnection) -> int: # allows a peer to grow beyond a low delivery rate caused by the old cap. bdp_blocks = math.ceil((delivery_rate * rtt * 2.0) / block_size) target = max(min_depth, base_depth, latency_floor, bdp_blocks) + if ( + accel_enabled(self.config) + and measured_block_latency <= 0 + and measured_request_latency <= 0 + ): + # Warm start (fresh connections only): don't wait for BDP + # measurements to grow past the cold depth of 16. Once latency + # is measured, the adaptive calculation above governs. + target = warm_pipeline_depth( + current_depth=int(target), + initial_depth=int( + getattr(self.config.network, "warm_pipeline_depth", 32) or 32 + ), + min_depth=int(min_depth), + max_depth=int(max_depth), + bdp_blocks=int(bdp_blocks), + ) return min(max_depth, target) def _apply_adaptive_pipeline_depth(self, connection: AsyncPeerConnection) -> None: @@ -6051,12 +6120,26 @@ async def connect_to_peers( max_parallel = 1 max_parallel = max(1, min(8, max_parallel)) max_parallel_batches = max_parallel + _accel = bool(accel_enabled(self.config)) if _parallel_active_count == 0 or ( _parallel_active_count <= 2 and _parallel_requestable == 0 ): max_parallel_batches = 1 if _parallel_pending_depth > 100: max_parallel_batches = 1 + if _accel and _parallel_pending_depth > 100 and _parallel_active_count > 0: + # Acceleration: with a deep candidate backlog, parallel + # owners clear it faster; the global connection semaphore + # still bounds socket load. Cold-start single-flight (the + # queued_reentrant contract) is preserved otherwise. + try: + _ramp = int( + getattr(self.config.network, "ramp_connect_parallel_batches", 4) + or 4 + ) + except (TypeError, ValueError): + _ramp = 4 + max_parallel_batches = max(1, min(8, _ramp)) if max_parallel > 1 and max_parallel_batches > 1: _mc = int( @@ -6859,7 +6942,18 @@ async def connect_to_peers( and remote_choked_at_start >= active_peer_count ): hold_slots = max(0, max_connections - active_peer_count) - batch_size = 0 if hold_slots <= 0 else min(batch_size, hold_slots) + if accel_enabled(self.config): + # Probe-dial anyway: waiting for UNCHOKE from peers + # that may never unchoke is slower than dialing more. + batch_size = ( + min(batch_size, max(8, hold_slots)) + if hold_slots > 0 + else batch_size + ) + else: + batch_size = ( + 0 if hold_slots <= 0 else min(batch_size, hold_slots) + ) self.logger.info( "Holding outbound connect churn while %d active peer(s) " "await remote UNCHOKE (batch_size=%d, hold_slots=%d)", @@ -6978,7 +7072,18 @@ async def connect_to_peers( reason="active_choked_hold", ) self._schedule_pending_resume_retry( - delay_s=15.0, + delay_s=( + float( + getattr( + self.config.network, + "ramp_choked_probe_interval_s", + 2.0, + ) + or 2.0 + ) + if accel_enabled(self.config) + else 15.0 + ), reason="active_choked_pause", ) self.logger.info( @@ -7536,6 +7641,10 @@ async def _process_completed_batch( # Connection batch: early exit if enough connections succeed # This speeds up batch processing when we get good peers quickly + # Note: early exit detaches (does not + # cancel) in-flight attempts and requeues + # them to the pending drain, so exiting on + # batch successes is safe for ramp-up. if ( successful >= min_successful_for_early_exit and completed @@ -8989,6 +9098,39 @@ async def _open_tcp_with_semaphore( timeout=timeout, ) + async def _race_utp_tcp( + self, utp_connection: Any, peer_info: PeerInfo + ) -> tuple[str, Any]: + """Race a uTP connect against a speculative TCP dial (acceleration). + + First successful transport wins; losers are cancelled by + ``race_first_success``. Returns ``("utp", None)`` or + ``("tcp", (reader, writer))``. Raises the first failure if both fail. + """ + spec_timeout = float( + getattr(self.config.network, "speculative_connect_timeout", 5.0) or 5.0 + ) + + async def _tcp_dial() -> tuple[asyncio.StreamReader, asyncio.StreamWriter]: + return await self._open_tcp_with_semaphore( + peer_info.ip, peer_info.port, spec_timeout + ) + + kind, payload = await race_first_success( + {"utp": utp_connection.connect(), "tcp": _tcp_dial()} + ) + if kind == "tcp": + # uTP lost: make sure it does not hold sockets/state. + for closer_name in ("disconnect", "close"): + closer = getattr(utp_connection, closer_name, None) + if callable(closer): + with contextlib.suppress(Exception): + result = closer() + if asyncio.iscoroutine(result): + await result + break + return kind, payload + async def _reconnect_plaintext_after_mse_failure( self, peer_info: PeerInfo, @@ -9655,7 +9797,42 @@ async def _connect_to_peer(self, peer_info: PeerInfo) -> None: # Connect via uTP (with fallback to TCP on failure) try: - await connection.connect() + if accel_enabled(self.config): + # Race uTP against a speculative TCP dial; first + # successful transport carries the handshake. + _race_kind, _race_payload = await self._race_utp_tcp( + connection, peer_info + ) + if _race_kind == "tcp": + _tcp_reader, _tcp_writer = _race_payload + _tcp_connection = AsyncPeerConnection( + peer_info, self.torrent_data + ) + self._seeded_connection_from_info(_tcp_connection) + _tcp_connection.max_pipeline_depth = ( + self._calculate_pipeline_depth(_tcp_connection) + ) + if self._on_peer_connected: + _tcp_connection.on_peer_connected = ( + self._on_peer_connected + ) + if self._on_peer_disconnected: + _tcp_connection.on_peer_disconnected = ( + self._on_peer_disconnected + ) + if self._on_bitfield_received: + _tcp_connection.on_bitfield_received = ( + self._on_bitfield_received + ) + if self._on_piece_received: + _tcp_connection.on_piece_received = ( + self._on_piece_received + ) + _tcp_connection.reader = _tcp_reader + _tcp_connection.writer = _tcp_writer + connection = _tcp_connection + else: + await connection.connect() # Connection successful - uTP handles transport layer # Still need BitTorrent protocol handshake, but skip TCP connection # The reader/writer are already set up by UTPPeerConnection.connect() @@ -17087,6 +17264,19 @@ async def _choking_loop_step(self) -> bool: return False sleep_s = float(getattr(self.config.network, "unchoke_interval", 10) or 10) _, active_n, requestable_n = self._snapshot_connection_counts() + if accel_enabled(self.config): + _target = max( + 1, int(getattr(self, "max_peers_per_torrent", 50) or 50) // 2 + ) + sleep_s, _ = ramp_choking_intervals( + requestable_peers=requestable_n, + requestable_target=_target, + unchoke_interval_s=sleep_s, + optimistic_interval_s=float( + getattr(self.config.network, "optimistic_unchoke_interval", 30) + or 30 + ), + ) if active_n > 0 and requestable_n == 0: bootstrap_interval = float( getattr( @@ -17211,6 +17401,17 @@ def peer_score(peer: AsyncPeerConnection) -> float: ) max_slots = int(getattr(self.config.network, "max_upload_slots", 4) or 4) + if accel_enabled(self.config): + _rq_slots = sum(1 for p in active_peers if p.can_request()) + max_slots = adaptive_upload_slots( + base_slots=max_slots, + max_slots=16, + upload_rate_bps=total_swarm_upload_rate, + requestable_peers=_rq_slots, + requestable_target=max( + 1, int(getattr(self, "max_peers_per_torrent", 50) or 50) // 2 + ), + ) if leech_heavy_swarm: low_swarm_cap = int( getattr(self.config.network, "low_swarm_min_upload_slots", 8) or 8 @@ -17437,6 +17638,19 @@ async def _update_optimistic_unchoke(self) -> None: interval = ( self.config.network.optimistic_unchoke_interval ) # pragma: no cover - Same context + if accel_enabled(self.config): + _, _, _rq_opt = self._snapshot_connection_counts() + _target_opt = max( + 1, int(getattr(self, "max_peers_per_torrent", 50) or 50) // 2 + ) + _, interval = ramp_choking_intervals( + requestable_peers=_rq_opt, + requestable_target=_target_opt, + unchoke_interval_s=float( + getattr(self.config.network, "unchoke_interval", 10) or 10 + ), + optimistic_interval_s=float(interval or 30), + ) # Check if we need a new optimistic unchoke @@ -17531,6 +17745,53 @@ async def _unchoke_peer(self, connection: AsyncPeerConnection) -> None: self.logger.debug("Unchoked peer %s", connection.peer_info) + def _update_snub_detection(self, now: Optional[float] = None) -> list[str]: + """Flag peers that are unchoked but deliver nothing while others deliver. + + Returns peer keys newly snubbed in this pass. Uses the previously + unused PeerStats.snub_count. + """ + if not accel_enabled(self.config): + return [] + now = float(now if now is not None else time.time()) + threshold = float( + getattr(self.config.network, "snub_threshold_s", 10.0) or 10.0 + ) + unchoked_since: dict[str, float] = getattr(self, "_snub_unchoked_since", {}) + self._snub_unchoked_since = unchoked_since + last_snub_at: dict[str, float] = getattr(self, "_snub_last_at", {}) + self._snub_last_at = last_snub_at + conns = list(self.connections.values()) + others_delivering = any( + float(getattr(c.stats, "last_piece_payload_time", 0.0) or 0.0) > 0 + and now - float(c.stats.last_piece_payload_time) <= threshold + for c in conns + ) + newly: list[str] = [] + for conn in conns: + key = self._get_peer_key(conn.peer_info) + if conn.peer_choking or not conn.is_active(): + unchoked_since.pop(key, None) + conn._snubbed = False # type: ignore[attr-defined] # noqa: SLF001 + continue + since = unchoked_since.setdefault(key, now) + last_payload = float( + getattr(conn.stats, "last_piece_payload_time", 0.0) or 0.0 + ) + since_payload = now - (last_payload if last_payload > 0 else since) + snubbed_now = is_snubbed( + seconds_since_payload=since_payload, + seconds_unchoked=now - since, + others_delivering=others_delivering, + threshold_s=threshold, + ) + conn._snubbed = snubbed_now # type: ignore[attr-defined] # noqa: SLF001 + if snubbed_now and now - last_snub_at.get(key, 0.0) >= threshold: + last_snub_at[key] = now + conn.stats.snub_count += 1 + newly.append(key) + return newly + async def _stats_loop(self) -> None: """Background task for updating peer statistics.""" while await self._stats_loop_step(): # pragma: no cover - Background loop requires time-based iterations, complex to test @@ -17551,6 +17812,8 @@ async def _stats_loop_step(self) -> bool: ) await self._update_peer_stats() # pragma: no cover - Same context + with contextlib.suppress(Exception): + self._update_snub_detection() record_swarm_role_snapshot(self) # Note: Log comprehensive connection diagnostics every 30 seconds @@ -17886,7 +18149,11 @@ def _should_recycle_peer( async def _maybe_choke_only_slot_replacement(self) -> None: """Optionally disconnect oldest persistently choked peers to free slots (config-gated).""" net = self.config.network - if not bool(getattr(net, "choke_only_slot_replacement_enabled", False)): + # Acceleration enables replacement only in the deficit case + # guarded below (requestable == 0 and at connection limit). + if not bool( + getattr(net, "choke_only_slot_replacement_enabled", False) + ) and not accel_enabled(self.config): return if bool(self.torrent_data.get("private")): return @@ -19488,6 +19755,24 @@ async def _rank_peers_for_connection( ), ) + if accel_enabled(self.config): + # Predicted-usefulness blend (see peer_usefulness_score): + # seeder/known-good peers first, recent failures last. + _fail_age = 1e9 + if peer_key in self._failed_peers: + _ts = float( + self._failed_peers[peer_key].get("timestamp", 0.0) or 0.0 + ) + if _ts > 0: + _fail_age = max(0.0, time.time() - _ts) + _useful = peer_usefulness_score( + has_needed_piece=bool(getattr(peer_info, "is_seeder", False)), + historic_unchoke_rate=success_rate_by_key.get(peer_key, 0.0), + seconds_since_failure=_fail_age, + ) + # _useful is roughly -80..170; normalize to a 0..0.3 blend. + score = max(0.0, min(1.0, score + max(-0.2, min(0.3, _useful / 500.0)))) + peer_scores.append((peer_info, score)) # Sort by score (highest first), with stable cold-start tie-break noise @@ -19754,6 +20039,18 @@ async def _cleanup_timed_out_requests( if measured_latency > 0: request_timeout = max(request_timeout, measured_latency * 4.0) + if accel_enabled(self.config) and strategy_flag( + self.config, "adaptive_request_timeout_enabled" + ): + # Acceleration: clamp(4 x latency, 5s, 30s) replaces the fixed + # 60s wait before a straggler block can move peers. + request_timeout = adaptive_request_timeout( + measured_latency, + fallback_s=min(float(base_timeout or 30.0), 30.0), + ) + if pipeline_utilization > 0.8: + request_timeout = min(request_timeout, 15.0) + # Note: Apply dynamic timeout adjustment based on unexpected pieces # If peer is sending useful unexpected pieces, INCREASE timeout to give them more time diff --git a/ccbt/performance/__init__.py b/ccbt/performance/__init__.py new file mode 100644 index 0000000..21e53a6 --- /dev/null +++ b/ccbt/performance/__init__.py @@ -0,0 +1,35 @@ +"""Download-acceleration primitives (pure, testable helpers).""" + +from ccbt.performance.acceleration import ( + BlockLatencyTracker, + DuplicateBudget, + RampMetrics, + accel_enabled, + adaptive_request_timeout, + adaptive_upload_slots, + is_dynamic_endgame, + is_snubbed, + peer_usefulness_score, + race_first_success, + ramp_choking_intervals, + should_duplicate_block, + strategy_flag, + warm_pipeline_depth, +) + +__all__ = [ + "BlockLatencyTracker", + "DuplicateBudget", + "RampMetrics", + "accel_enabled", + "adaptive_request_timeout", + "adaptive_upload_slots", + "is_dynamic_endgame", + "is_snubbed", + "peer_usefulness_score", + "race_first_success", + "ramp_choking_intervals", + "should_duplicate_block", + "strategy_flag", + "warm_pipeline_depth", +] diff --git a/ccbt/performance/acceleration.py b/ccbt/performance/acceleration.py new file mode 100644 index 0000000..82821ef --- /dev/null +++ b/ccbt/performance/acceleration.py @@ -0,0 +1,394 @@ +"""Pure helpers for download acceleration. + +These functions encode the decisions described in the acceleration plan +(straggler duplication, adaptive timeouts, snub detection, BEP 6 aware +ranking, ramp-aware choking) without any I/O, so they can be unit tested +deterministically and reused by the peer/piece managers and by the +virtual-swarm benchmark. + +References: BEP 3 (choking / optimistic unchoke), BEP 6 (fast extension / +allow-fast), BEP 11 (PEX), libtorrent concepts of snubbing and endgame +duplicate requests. +""" + +from __future__ import annotations + +import asyncio +import math +import time +from collections import deque +from dataclasses import dataclass, field +from typing import Any, Awaitable, Optional, TypeVar + +T = TypeVar("T") + + +def accel_enabled(config: Any) -> bool: + """Swarm acceleration is ON by default. + + Only an explicit ``False`` on ``config.network.swarm_acceleration_enabled`` + disables it; a missing/``None`` attribute (legacy or partial configs) + means enabled. + """ + network = getattr(config, "network", None) + value = getattr(network, "swarm_acceleration_enabled", None) + return True if value is None else bool(value) + + +def strategy_flag(config: Any, name: str) -> bool: + """Strategy acceleration flags are likewise on unless explicitly off.""" + strategy = getattr(config, "strategy", None) + value = getattr(strategy, name, None) + return True if value is None else bool(value) + + +async def race_first_success( + candidates: dict[str, Awaitable[T]], +) -> tuple[str, T]: + """Race awaitables; return the first success, cancelling the losers. + + Failures are tolerated until every candidate has failed, in which case + the first failure is raised. Loser results that arrive anyway are + consumed so no task exception goes unretrieved. + """ + tasks = {name: asyncio.ensure_future(coro) for name, coro in candidates.items()} + if not tasks: + msg = "race_first_success requires at least one candidate" + raise ValueError(msg) + first_error: Optional[BaseException] = None + try: + pending = set(tasks.values()) + while pending: + done, pending = await asyncio.wait( + pending, return_when=asyncio.FIRST_COMPLETED + ) + for name, task in list(tasks.items()): + if task not in done: + continue + if task.cancelled(): + continue + exc = task.exception() + if exc is None: + for other in pending: + other.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + return name, task.result() + if first_error is None: + first_error = exc + if first_error is not None: + raise first_error + msg = "All raced candidates were cancelled" + raise ConnectionError(msg) + finally: + for task in tasks.values(): + if not task.done(): + task.cancel() + + +def _percentile(sorted_values: list[float], pct: float) -> float: + if not sorted_values: + return 0.0 + if len(sorted_values) == 1: + return float(sorted_values[0]) + rank = (pct / 100.0) * (len(sorted_values) - 1) + lo = math.floor(rank) + hi = math.ceil(rank) + if lo == hi: + return float(sorted_values[int(rank)]) + frac = rank - lo + return float(sorted_values[lo] * (1.0 - frac) + sorted_values[hi] * frac) + + +class BlockLatencyTracker: + """Rolling per-swarm / per-peer block latency samples.""" + + def __init__(self, max_samples: int = 512) -> None: + """Keep at most max_samples recent latency samples.""" + self._samples: deque[float] = deque(maxlen=max_samples) + + def record(self, latency_s: float) -> None: + """Record one block latency sample in seconds.""" + if latency_s > 0: + self._samples.append(float(latency_s)) + + def __len__(self) -> int: + """Return the number of retained samples.""" + return len(self._samples) + + def percentile(self, pct: float, default: float = 0.0) -> float: + """Return the given percentile, or default when empty.""" + if not self._samples: + return default + return _percentile(sorted(self._samples), pct) + + @property + def p50(self) -> float: + """Median block latency.""" + return self.percentile(50) + + @property + def p90(self) -> float: + """90th-percentile block latency.""" + return self.percentile(90) + + @property + def p95(self) -> float: + """95th-percentile block latency.""" + return self.percentile(95) + + +class DuplicateBudget: + """Rolling budget for redundant (duplicate) block bytes. + + Duplication is only allowed while redundant bytes stay at or below + ``max_fraction`` of useful downloaded bytes over the observed window. + """ + + def __init__(self, max_fraction: float = 0.10) -> None: + """Cap redundant bytes at max_fraction of total bytes.""" + self.max_fraction = max(0.0, min(1.0, float(max_fraction))) + self.useful_bytes = 0 + self.duplicate_bytes = 0 + + def record_useful(self, nbytes: int) -> None: + """Record useful first-arrival bytes.""" + if nbytes > 0: + self.useful_bytes += int(nbytes) + + def record_duplicate(self, nbytes: int) -> None: + """Record redundant duplicate bytes.""" + if nbytes > 0: + self.duplicate_bytes += int(nbytes) + + @property + def wasted_fraction(self) -> float: + """Fraction of total bytes that were redundant.""" + total = self.useful_bytes + self.duplicate_bytes + return (self.duplicate_bytes / total) if total > 0 else 0.0 + + def allows(self, next_bytes: int = 0) -> bool: + """True if sending another duplicate keeps us within budget.""" + if self.max_fraction <= 0: + return False + total = self.useful_bytes + self.duplicate_bytes + max(0, int(next_bytes)) + if total <= 0: + return True + return ( + self.duplicate_bytes + max(0, int(next_bytes)) + ) / total <= self.max_fraction + 1e-9 + + +def adaptive_request_timeout( + p95_latency_s: float, + *, + multiplier: float = 4.0, + minimum_s: float = 5.0, + maximum_s: float = 30.0, + fallback_s: float = 30.0, +) -> float: + """Timeout = clamp(multiplier * p95 latency, min, max).""" + if p95_latency_s <= 0: + return max(minimum_s, min(maximum_s, fallback_s)) + return max(minimum_s, min(maximum_s, multiplier * float(p95_latency_s))) + + +def should_duplicate_block( + *, + block_age_s: float, + swarm_p90_latency_s: float, + blocks_remaining_in_piece: int, + existing_copies: int, + max_copies: int = 3, + budget_allows: bool = True, + enabled: bool = True, +) -> bool: + """Decide whether to speculatively duplicate an in-flight block. + + Fires when the block is a straggler (age beyond swarm p90, with a + small-sample floor) or when its piece is nearly complete (<=2 blocks + left) at any overall download progress. + """ + if not enabled or not budget_allows: + return False + if existing_copies <= 0 or existing_copies >= max(1, max_copies): + return False + is_last_blocks = 0 < blocks_remaining_in_piece <= 2 + # Require at least ~1s of age so we never duplicate instantly, and + # beyond p90 when we have a meaningful latency estimate. + threshold = max(1.0, swarm_p90_latency_s) if swarm_p90_latency_s > 0 else 1.0 + is_straggler = block_age_s >= threshold + return bool((is_last_blocks and block_age_s >= 1.0) or is_straggler) + + +def is_dynamic_endgame( + *, + remaining_pieces: int, + total_pieces: int, + threshold: float = 0.95, + last_pieces: int = 20, + all_remaining_stalled: bool = False, +) -> bool: + """Endgame by progress, by absolute tail size, or by stall.""" + if total_pieces <= 0 or remaining_pieces <= 0: + return False + if remaining_pieces <= total_pieces * (1.0 - threshold): + return True + # Absolute-tail rule only for torrents larger than the tail window; + # otherwise every small torrent would start in endgame. + if total_pieces > last_pieces and remaining_pieces <= max(1, last_pieces): + return True + return bool(all_remaining_stalled) + + +def is_snubbed( + *, + seconds_since_payload: float, + seconds_unchoked: float, + others_delivering: bool, + threshold_s: float = 10.0, +) -> bool: + """Peer is unchoked but delivers nothing while others deliver (libtorrent snub).""" + return bool( + others_delivering + and seconds_unchoked >= threshold_s + and seconds_since_payload >= threshold_s + ) + + +def peer_usefulness_score( + *, + has_needed_piece: bool = False, + rarity_overlap: float = 0.0, + historic_unchoke_rate: float = 0.0, + download_rate_bps: float = 0.0, + connect_latency_s: float = 0.0, + seconds_since_failure: float = 1e9, + is_snubbed_peer: bool = False, +) -> float: + """Higher is dialled / chosen first.""" + score = 0.0 + if has_needed_piece: + score += 100.0 + score += 50.0 * max(0.0, min(1.0, rarity_overlap)) + score += 40.0 * max(0.0, min(1.0, historic_unchoke_rate)) + # log-scaled throughput contribution, ~0..30 + if download_rate_bps > 0: + score += min(30.0, 10.0 * math.log10(1.0 + download_rate_bps / 1024.0)) + if connect_latency_s > 0: + score -= min(20.0, connect_latency_s * 10.0) + if seconds_since_failure < 60.0: + score -= 30.0 * (1.0 - max(0.0, seconds_since_failure) / 60.0) + if is_snubbed_peer: + score -= 80.0 + return score + + +def warm_pipeline_depth( + *, + current_depth: int, + initial_depth: int = 32, + min_depth: int = 4, + max_depth: int = 128, + bdp_blocks: int = 0, +) -> int: + """Pipeline depth at connect / warm-up: never below warm start or BDP need.""" + target = max(int(current_depth), int(initial_depth)) + if bdp_blocks > 0: + target = max(target, int(bdp_blocks)) + return max(min_depth, min(max_depth, target)) + + +def adaptive_upload_slots( + *, + base_slots: int = 8, + max_slots: int = 16, + upload_rate_bps: float = 0.0, + requestable_peers: int = 0, + requestable_target: int = 0, +) -> int: + """Scale upload slots with measured upload capacity during a deficit.""" + slots = int(base_slots) + if requestable_target > 0 and requestable_peers < requestable_target: + # Roughly one extra slot per 512 KiB/s of upload headroom. + extra = int(upload_rate_bps // (512 * 1024)) + slots = max(slots, min(int(max_slots), int(base_slots) + extra)) + return max(1, min(int(max_slots), slots)) + + +def ramp_choking_intervals( + *, + requestable_peers: int, + requestable_target: int, + unchoke_interval_s: float = 10.0, + optimistic_interval_s: float = 30.0, +) -> tuple[float, float]: + """Faster choking cadence while we still lack requestable peers.""" + if requestable_target > 0 and requestable_peers < requestable_target: + return (min(unchoke_interval_s, 5.0), min(optimistic_interval_s, 20.0)) + return (float(unchoke_interval_s), float(optimistic_interval_s)) + + +@dataclass +class RampMetrics: + """Once-only ramp-up milestones plus duplicate/cancel counters.""" + + started_monotonic: float = field(default_factory=time.monotonic) + time_to_first_requestable_s: Optional[float] = None + time_to_first_productive_s: Optional[float] = None + time_to_first_piece_s: Optional[float] = None + time_to_pct_s: dict[int, Optional[float]] = field( + default_factory=lambda: {10: None, 50: None, 90: None} + ) + duplicate_bytes: int = 0 + duplicate_requests: int = 0 + cancels_sent: int = 0 + useful_bytes: int = 0 + requestable_series: list[tuple[float, int]] = field(default_factory=list) + + def _elapsed(self, now: Optional[float] = None) -> float: + """Seconds since the ramp clock started.""" + return (now if now is not None else time.monotonic()) - self.started_monotonic + + def touch_requestable(self, count: int, now: Optional[float] = None) -> None: + """Record the current requestable-peer count.""" + elapsed = self._elapsed(now) + self.requestable_series.append((elapsed, int(count))) + if count > 0 and self.time_to_first_requestable_s is None: + self.time_to_first_requestable_s = elapsed + + def touch_productive(self, count: int, now: Optional[float] = None) -> None: + """Record first productive (data-delivering) peer sighting.""" + if count > 0 and self.time_to_first_productive_s is None: + self.time_to_first_productive_s = self._elapsed(now) + + def touch_first_piece(self, now: Optional[float] = None) -> None: + """Record completion of the first verified piece.""" + if self.time_to_first_piece_s is None: + self.time_to_first_piece_s = self._elapsed(now) + + def touch_progress(self, fraction: float, now: Optional[float] = None) -> None: + """Record percent-complete milestones from a 0..1 fraction.""" + elapsed = self._elapsed(now) + for pct in (10, 50, 90): + if fraction * 100.0 >= pct and self.time_to_pct_s[pct] is None: + self.time_to_pct_s[pct] = elapsed + + def snapshot(self) -> dict[str, object]: + """Return all ramp metrics as a plain dict.""" + total = self.useful_bytes + self.duplicate_bytes + return { + "time_to_first_requestable_s": self.time_to_first_requestable_s, + "time_to_first_productive_s": self.time_to_first_productive_s, + "time_to_first_piece_s": self.time_to_first_piece_s, + "time_to_10pct_s": self.time_to_pct_s[10], + "time_to_50pct_s": self.time_to_pct_s[50], + "time_to_90pct_s": self.time_to_pct_s[90], + "duplicate_bytes": self.duplicate_bytes, + "duplicate_requests": self.duplicate_requests, + "cancels_sent": self.cancels_sent, + "duplicate_wasted_fraction": (self.duplicate_bytes / total) + if total + else 0.0, + "requestable_series": list(self.requestable_series), + } diff --git a/ccbt/performance/swarm_sim.py b/ccbt/performance/swarm_sim.py new file mode 100644 index 0000000..d5d9c3f --- /dev/null +++ b/ccbt/performance/swarm_sim.py @@ -0,0 +1,275 @@ +"""Deterministic virtual-swarm simulation for A/B verification. + +Models the mechanisms in the acceleration plan, not the wire protocol: +dial parallelism determines when peers become requestable; each block is +served by a peer with (latency + size/bandwidth) service time; the +accelerated strategy duplicates straggler blocks (first arrival wins, +losers are wasted bytes) and uses adaptive timeouts. Seeded and fully +deterministic so baseline vs treatment is reproducible in CI. +""" + +from __future__ import annotations + +import heapq +import random +from dataclasses import dataclass + + +@dataclass(frozen=True) +class SimPeer: + """One scripted virtual peer.""" + + peer_id: int + latency_s: float + bandwidth_bps: float + unchoke_delay_s: float + reliability: float = 1.0 # fraction of requests that ever complete + + +@dataclass +class SimResult: + """Outcome of one simulated download.""" + + completion_s: float + time_to_first_requestable_s: float + time_to_first_piece_s: float + time_to_50pct_s: float + wasted_bytes: int + useful_bytes: int + duplicate_requests: int + + @property + def wasted_fraction(self) -> float: + """Fraction of total bytes that were redundant.""" + total = self.useful_bytes + self.wasted_bytes + return (self.wasted_bytes / total) if total else 0.0 + + def as_dict(self) -> dict[str, float]: + """Return rounded metrics as a plain dict.""" + return { + "completion_s": round(self.completion_s, 3), + "time_to_first_requestable_s": round(self.time_to_first_requestable_s, 3), + "time_to_first_piece_s": round(self.time_to_first_piece_s, 3), + "time_to_50pct_s": round(self.time_to_50pct_s, 3), + "wasted_fraction": round(self.wasted_fraction, 4), + "duplicate_requests": float(self.duplicate_requests), + } + + +def make_peers(count: int, seed: int = 1234) -> list[SimPeer]: + """Generate a deterministic public-swarm-like peer mix.""" + rng = random.Random(seed) + peers: list[SimPeer] = [] + for i in range(count): + # Public-swarm-like mix: a few fast seeders, many slow leechers. + kind = rng.random() + if kind < 0.2: + bw, lat = rng.uniform(5e6, 20e6), rng.uniform(0.02, 0.08) + elif kind < 0.7: + bw, lat = rng.uniform(5e5, 5e6), rng.uniform(0.05, 0.2) + else: + bw, lat = rng.uniform(5e4, 5e5), rng.uniform(0.15, 0.3) + peers.append( + SimPeer( + peer_id=i, + latency_s=lat, + bandwidth_bps=bw, + unchoke_delay_s=rng.uniform(0.5, 10.0), + reliability=0.7 if rng.random() < 0.15 else 1.0, + ) + ) + return peers + + +def simulate( + *, + peers: list[SimPeer], + num_blocks: int = 400, + block_bytes: int = 16384, + blocks_per_piece: int = 16, + dial_parallelism: int = 8, + connect_batch_s: float = 1.0, + duplicate_stragglers: bool = False, + straggler_after_s: float = 2.0, + max_copies: int = 3, + pipeline_per_peer: int = 16, + request_timeout_s: float = 60.0, +) -> SimResult: + """Event-driven block download simulation (single torrent).""" + # When each peer becomes requestable: dial batch + unchoke delay. + ready_at = { + p.peer_id: (p.peer_id // max(1, dial_parallelism)) * connect_batch_s + + p.unchoke_delay_s + for p in peers + } + # Time until a useful cohort (8th, or all if fewer) is requestable; + # this is what dial parallelism actually accelerates. + _sorted_ready = sorted(ready_at.values()) + first_requestable = ( + _sorted_ready[min(7, len(_sorted_ready) - 1)] if _sorted_ready else 0.0 + ) + + # Events: (time, seq, kind, block, peer_id) + events: list[tuple[float, int, str, int, int]] = [] + seq = 0 + + def push(t: float, kind: str, block: int, pid: int) -> None: + nonlocal seq + heapq.heappush(events, (t, seq, kind, block, pid)) + seq += 1 + + in_flight: dict[int, dict[int, float]] = {} # block -> {peer: started} + completed: set[int] = set() + busy: dict[int, int] = {p.peer_id: 0 for p in peers} + now = 0.0 + next_block = 0 + wasted = 0 + dup_requests = 0 + first_piece_at = 0.0 + half_at = 0.0 + completed_pieces: set[int] = set() + + def service_time(p: SimPeer) -> float: + return p.latency_s + block_bytes / max(1.0, p.bandwidth_bps) + + def try_assign(now_t: float) -> None: + nonlocal next_block, dup_requests + # Assign fresh blocks to idle-ish ready peers (fastest first). + ready_peers = sorted( + ( + p + for p in peers + if ready_at[p.peer_id] <= now_t and busy[p.peer_id] < pipeline_per_peer + ), + key=lambda p: service_time(p), + ) + for p in ready_peers: + while busy[p.peer_id] < pipeline_per_peer and next_block < num_blocks: + b = next_block + next_block += 1 + if p.reliability < 1.0 and (b * 31 + p.peer_id) % 10 >= int( + p.reliability * 10 + ): + # Unreliable peer: request hangs (only duplication/timeout saves it) + in_flight.setdefault(b, {})[p.peer_id] = now_t + busy[p.peer_id] += 1 + continue + in_flight.setdefault(b, {})[p.peer_id] = now_t + busy[p.peer_id] += 1 + push(now_t + service_time(p), "done", b, p.peer_id) + # Duplicate stragglers onto fastest idle ready peer. + if duplicate_stragglers: + for b, holders in list(in_flight.items()): + if b in completed or len(holders) >= max_copies: + continue + age = now_t - min(holders.values()) + if age < straggler_after_s: + continue + for p in ready_peers: + if p.peer_id in holders or busy[p.peer_id] >= pipeline_per_peer: + continue + holders[p.peer_id] = now_t + busy[p.peer_id] += 1 + dup_requests += 1 + if not ( + p.reliability < 1.0 + and (b * 31 + p.peer_id) % 10 >= int(p.reliability * 10) + ): + push(now_t + service_time(p), "done", b, p.peer_id) + break + + for p in peers: + push(ready_at[p.peer_id], "ready", -1, p.peer_id) + # Periodic straggler scan. + t_scan = 0.5 + while t_scan < 3600: + push(t_scan, "scan", -1, -1) + t_scan += 0.5 + + guard = 0 + while events and len(completed) < num_blocks: + guard += 1 + if guard > 2_000_000: + break + now, _, kind, block, pid = heapq.heappop(events) + if kind == "scan": + # Timeout rescue: blocks held (hung) beyond request_timeout are + # freed and re-issued to the fastest reliable ready peer. + for b, holders in list(in_flight.items()): + if b in completed or not holders: + continue + if now - min(holders.values()) < request_timeout_s: + continue + for old_pid in holders: + busy[old_pid] = max(0, busy[old_pid] - 1) + del in_flight[b] + rescuers = sorted( + ( + p + for p in peers + if p.reliability >= 1.0 and ready_at[p.peer_id] <= now + ), + key=lambda p: service_time(p), + ) + if rescuers: + rp = rescuers[0] + in_flight[b] = {rp.peer_id: now} + busy[rp.peer_id] += 1 + push(now + service_time(rp), "done", b, rp.peer_id) + if kind == "done": + if pid in busy: + busy[pid] = max(0, busy[pid] - 1) + if block in completed: + wasted += block_bytes + continue + completed.add(block) + # Other copies of this block are wasted work in flight. + for other_pid in in_flight.pop(block, {}): + if other_pid != pid: + wasted += 0 # counted if/when they complete above + piece = block // blocks_per_piece + if piece not in completed_pieces: + piece_blocks = range( + piece * blocks_per_piece, + min(num_blocks, (piece + 1) * blocks_per_piece), + ) + if all(b in completed for b in piece_blocks): + completed_pieces.add(piece) + if not first_piece_at: + first_piece_at = now + if not half_at and len(completed) >= num_blocks // 2: + half_at = now + try_assign(now) + + useful = len(completed) * block_bytes + return SimResult( + completion_s=now, + time_to_first_requestable_s=first_requestable, + time_to_first_piece_s=first_piece_at, + time_to_50pct_s=half_at, + wasted_bytes=wasted, + useful_bytes=useful, + duplicate_requests=dup_requests, + ) + + +def run_scenario( + peer_count: int, seed: int = 1234, num_blocks: int = 400 +) -> dict[str, dict[str, float]]: + """Run baseline vs treatment for one peer-count scenario.""" + peers = make_peers(peer_count, seed=seed) + baseline = simulate( + peers=peers, + num_blocks=num_blocks, + dial_parallelism=8, + duplicate_stragglers=False, + ) + treatment = simulate( + peers=peers, + num_blocks=num_blocks, + dial_parallelism=24, + duplicate_stragglers=True, + straggler_after_s=1.0, + max_copies=3, + ) + return {"baseline": baseline.as_dict(), "treatment": treatment.as_dict()} diff --git a/ccbt/piece/async_piece_manager.py b/ccbt/piece/async_piece_manager.py index c1d9729..53f0a03 100644 --- a/ccbt/piece/async_piece_manager.py +++ b/ccbt/piece/async_piece_manager.py @@ -27,6 +27,14 @@ ) from ccbt.models import PieceState as PieceStateModel from ccbt.monitoring import get_metrics_collector +from ccbt.performance.acceleration import ( + BlockLatencyTracker, + DuplicateBudget, + accel_enabled, + is_dynamic_endgame, + should_duplicate_block, + strategy_flag, +) from ccbt.piece.hash_v2 import HashAlgorithm, verify_piece from ccbt.utils.events import Event, emit_event from ccbt.utils.shutdown import is_shutting_down @@ -546,6 +554,17 @@ def __init__( self.endgame_threshold = self.config.strategy.endgame_threshold self.endgame_duplicates = self.config.strategy.endgame_duplicates + # Download acceleration: straggler duplication + latency tracking + self._block_latency = BlockLatencyTracker() + self._duplicate_budget = DuplicateBudget( + max_fraction=float( + getattr(self.config.strategy, "duplicate_budget_fraction", 0.10) or 0.10 + ) + ) + self._piece_selection_metrics.setdefault("duplicate_requests", 0) + self._piece_selection_metrics.setdefault("duplicate_bytes", 0) + self._piece_selection_metrics.setdefault("cancels_sent", 0) + # Hash verification pool (adaptive if enabled) if self.config.disk.hash_workers_adaptive: # Work-stealing executor limitation and workaround: @@ -3429,6 +3448,10 @@ async def request_piece_from_peers( available_peers, peer_manager, ) + with contextlib.suppress(Exception): + requests_sent += await self._request_straggler_duplicates( + piece_index, available_peers, peer_manager + ) async with self.lock: # pragma: no cover - Lock acquisition after request, state change tested via mark_requested old_state = piece.state @@ -5202,7 +5225,18 @@ def _reconcile_endgame_mode_from_counts( else len(self.get_missing_pieces()) ) total = self.num_pieces - in_endgame = rem <= total * (1.0 - self.endgame_threshold) + if strategy_flag(self.config, "straggler_duplication_enabled"): + in_endgame = is_dynamic_endgame( + remaining_pieces=rem, + total_pieces=total, + threshold=self.endgame_threshold, + last_pieces=int( + getattr(self.config.strategy, "dynamic_endgame_last_pieces", 20) + or 20 + ), + ) + else: + in_endgame = rem <= total * (1.0 - self.endgame_threshold) self.endgame_mode = in_endgame return in_endgame @@ -5353,6 +5387,92 @@ def peer_sort_key( ) return requests_sent + async def _request_straggler_duplicates( + self, + piece_index: int, + available_peers: list[AsyncPeerConnection], + peer_manager: Any, + ) -> int: + """Duplicate straggler / last blocks of a piece to other holders. + + Pre-endgame counterpart of _request_blocks_endgame: bounded by + straggler_max_copies and the rolling duplicate-byte budget. The + first arrival wins; peer-side timeout cleanup sends CANCELs for + the losing copies. + """ + if not strategy_flag(self.config, "straggler_duplication_enabled"): + return 0 + if piece_index < 0 or piece_index >= len(self.pieces): + return 0 + piece = self.pieces[piece_index] + now = time.time() + missing = [b for b in piece.blocks if not b.received] + blocks_remaining = len(missing) + if blocks_remaining == 0: + return 0 + max_copies = int(getattr(self.config.strategy, "straggler_max_copies", 3) or 3) + p90 = self._block_latency.p90 + sent = 0 + for block in missing: + if not block.requested_from: + continue + # Oldest in-flight request age for this block. + age = 0.0 + for peer_reqs in self._active_block_requests.get(piece_index, {}).values(): + for req_begin, _req_len, req_time in peer_reqs: + if req_begin == block.begin: + age = max(age, now - req_time) + if not should_duplicate_block( + block_age_s=age, + swarm_p90_latency_s=p90, + blocks_remaining_in_piece=blocks_remaining, + existing_copies=len(block.requested_from), + max_copies=max_copies, + budget_allows=self._duplicate_budget.allows(block.length), + enabled=True, + ): + continue + for peer_connection in available_peers: + peer_key = self._normalize_peer_key(peer_connection) + if not peer_key or peer_key in block.requested_from: + continue + avail = self.peer_availability.get(peer_key) + if avail is not None and piece_index not in avail.pieces: + continue + # BEP6-aware check; fall back for test doubles with a + # zero-arg can_request signature. + try: + _can = peer_connection.can_request(piece_index=piece_index) + except TypeError: + _can = peer_connection.can_request() + if not _can: + continue + try: + ok = await peer_manager.request_piece( + peer_connection, piece_index, block.begin, block.length + ) + except Exception: + self._piece_selection_metrics["failed_piece_requests"] += 1 + continue + if not ok: + continue + self._active_block_requests.setdefault(piece_index, {}).setdefault( + peer_key, [] + ).append((block.begin, block.length, time.time())) + block.requested_from.add(peer_key) + self._duplicate_budget.record_duplicate(block.length) + self._piece_selection_metrics["duplicate_requests"] = ( + int(self._piece_selection_metrics.get("duplicate_requests", 0)) + 1 + ) + self._piece_selection_metrics["duplicate_bytes"] = ( + int(self._piece_selection_metrics.get("duplicate_bytes", 0)) + + block.length + ) + self._piece_selection_metrics["total_piece_requests"] += 1 + sent += 1 + break # one extra copy per pass; selector re-runs frequently + return sent + async def handle_piece_block( self, piece_index: int, @@ -5408,6 +5528,11 @@ async def handle_piece_block( # Note: Check for duplicate blocks (already received) for block in piece.blocks: if block.begin == begin and block.received: + # Redundant copy arriving after the winner: wasted bytes. + self._duplicate_budget.record_duplicate(len(data)) + self._piece_selection_metrics["duplicate_bytes"] = int( + self._piece_selection_metrics.get("duplicate_bytes", 0) + ) + len(data) # Block already received - verify it matches if block.data != data: self.logger.warning( @@ -5435,6 +5560,14 @@ async def handle_piece_block( return self._piece_selection_metrics["successful_piece_requests"] += 1 + self._duplicate_budget.record_useful(len(data)) + # Latency of the winning request feeds adaptive timeouts / p90. + _now_blk = time.time() + for _peer_reqs in self._active_block_requests.get(piece_index, {}).values(): + for _rb, _rl, _rt in _peer_reqs: + if _rb == begin and _rl == len(data): + self._block_latency.record(_now_blk - _rt) + break block_length = len(data) if piece_index in self._active_block_requests: @@ -7485,6 +7618,10 @@ async def _piece_selector(self) -> None: """ consecutive_no_pieces = 0 base_interval = 1.0 + if accel_enabled(self.config): + # Acceleration: poll fast while pipelines may have free slots; + # UNCHOKE/HAVE triggers also wake selection between ticks. + base_interval = 0.25 max_interval = 5.0 self._piece_selection_metrics["selection_no_progress_streak"] = 0 @@ -7612,7 +7749,7 @@ async def _piece_selector(self) -> None: if new_active_downloads > active_downloads: # Made progress - reset interval consecutive_no_pieces = 0 - base_interval = 1.0 + base_interval = 0.25 if accel_enabled(self.config) else 1.0 self._no_progress_streak = 0 self._no_progress_stall_until = 0.0 self._no_progress_gate_streak = 0 diff --git a/ccbt/session/announce.py b/ccbt/session/announce.py index dcc4273..33c3f5b 100644 --- a/ccbt/session/announce.py +++ b/ccbt/session/announce.py @@ -10,6 +10,7 @@ import time from typing import TYPE_CHECKING, Any, Optional, Union +from ccbt.performance.acceleration import accel_enabled from ccbt.session.models import SessionContext if TYPE_CHECKING: @@ -1076,14 +1077,20 @@ def _swarm_int( # - connected and requestable but none productive -> 120s # The tracker `min interval` floor applied below (after this block) # guarantees we never violate tracker pacing. See RC-4/5 fix (Step 6). + _accel_announce = accel_enabled(self.s.config) + _ceil_connected, _ceil_req, _ceil_prod = ( + (15.0, 30.0, 60.0) if _accel_announce else (60.0, 90.0, 120.0) + ) if connected_peers == 0: - next_announce_interval = min(next_announce_interval, 60.0) + next_announce_interval = min( + next_announce_interval, _ceil_connected + ) weak_reason = "connected=0" elif requestable_peers == 0: - next_announce_interval = min(next_announce_interval, 90.0) + next_announce_interval = min(next_announce_interval, _ceil_req) weak_reason = "requestable=0" elif productive_peers == 0: - next_announce_interval = min(next_announce_interval, 120.0) + next_announce_interval = min(next_announce_interval, _ceil_prod) weak_reason = "productive=0" else: weak_reason = "" diff --git a/ccbt/session/dht_setup.py b/ccbt/session/dht_setup.py index 99b911e..0490f91 100644 --- a/ccbt/session/dht_setup.py +++ b/ccbt/session/dht_setup.py @@ -8,6 +8,7 @@ from typing import Any, Optional, cast from ccbt.monitoring import get_metrics_collector +from ccbt.performance.acceleration import accel_enabled from ccbt.session.swarm_stability_defaults import PEER_DISCOVERY_DEFAULTS from ccbt.utils.events import Event, EventType, emit_event from ccbt.utils.shutdown import is_shutting_down @@ -3019,6 +3020,27 @@ async def _run_discovery_loop(self, dht_client: Any) -> None: 60.0, initial_retry_interval ) # Minimum 60 seconds max_peers_per_query = 50 # Reduced from 100 to avoid overwhelming + _disc_cfg = getattr( + getattr(self.session, "config", None), "discovery", None + ) + if ( + accel_enabled(getattr(self.session, "config", None)) + and current_requestable_peers == 0 + ): + # Acceleration: DHT runs in parallel with trackers and + # returns a larger cohort while nobody is requestable. + max_peers_per_query = int( + getattr(_disc_cfg, "dht_max_peers_per_query_accelerated", 100) + or 100 + ) + dht_retry_interval = min( + dht_retry_interval, + float( + getattr(_disc_cfg, "dht_query_interval_accelerated_s", 5.0) + or 5.0 + ) + * 12.0, # retry loop floor; query-interval gate compresses separately + ) aggressive_mode = await self._handle_aggressive_mode_transition( current_aggressive_mode=aggressive_mode, @@ -3175,6 +3197,21 @@ async def _run_discovery_loop(self, dht_client: Any) -> None: and time.monotonic() < self._requestable_driven_compress_until ): effective_min_interval = min(effective_min_interval, 8.0) + if ( + accel_enabled(self.session.config) + and current_requestable_peers == 0 + ): + effective_min_interval = min( + effective_min_interval, + float( + getattr( + self.session.config.discovery, + "dht_query_interval_accelerated_s", + 5.0, + ) + or 5.0 + ), + ) if time_since_last_query < effective_min_interval: wait_time = effective_min_interval - time_since_last_query self.logger.debug( diff --git a/ccbt/session/peers.py b/ccbt/session/peers.py index f9aab0f..41e4c47 100644 --- a/ccbt/session/peers.py +++ b/ccbt/session/peers.py @@ -351,6 +351,12 @@ def pex_callback_wrapper(pex_peers: list) -> None: _ = task session.pex_manager.pex_callbacks.append(pex_callback_wrapper) + # Also wire the inbound wire path (ExtensionManager/PeerExchange): + # ut_pex peers previously only emitted an event and were never dialed. + _ext_mgr = getattr(session, "extension_manager", None) + if _ext_mgr is not None and hasattr(_ext_mgr, "register_pex_peer_callback"): + with contextlib.suppress(Exception): + _ext_mgr.register_pex_peer_callback(pex_callback_wrapper) session.logger.info( "Registered PEX callback for %s (total callbacks: %d)", session.info.name, diff --git a/ccbt/session/session.py b/ccbt/session/session.py index b8fd193..8d8f07c 100644 --- a/ccbt/session/session.py +++ b/ccbt/session/session.py @@ -518,6 +518,14 @@ def __init__( "pending_connect_queue_age_observations_s": [], "time_to_first_requestable_s": None, "time_to_first_productive_s": None, + "time_to_first_piece_s": None, + "time_to_10pct_s": None, + "time_to_50pct_s": None, + "time_to_90pct_s": None, + "duplicate_bytes": 0, + "duplicate_requests": 0, + "cancels_sent": 0, + "requestable_series": [], "dht_candidate_cache_new_entry_total": 0, "dht_candidate_cache_refresh_total": 0, "dht_candidate_promotion_selected_total": 0, @@ -1122,6 +1130,41 @@ def _touch_swarm_usefulness_latency_metrics( "peer_discovery_time_to_first_productive_s", float(elapsed), ) + if requestable_peers >= 0: + series = m.setdefault("requestable_series", []) + if isinstance(series, list) and ( + not series or series[-1][1] != requestable_peers + ): + series.append([round(float(elapsed), 3), int(requestable_peers)]) + del series[:-200] + + def _touch_piece_progress_metrics(self) -> None: + """Record time-to-first-piece and percent-complete milestones.""" + if not isinstance(self._peer_discovery_metrics, dict): + return + start = self._swarm_connect_phase_started_monotonic + if start is None or self.piece_manager is None: + return + verified = getattr(self.piece_manager, "verified_pieces", None) + total = int(getattr(self.piece_manager, "num_pieces", 0) or 0) + if verified is None or total <= 0: + return + elapsed = time.monotonic() - float(start) + m = self._peer_discovery_metrics + if len(verified) > 0 and m.get("time_to_first_piece_s") is None: + m["time_to_first_piece_s"] = elapsed + with contextlib.suppress(Exception): + get_metrics_collector().record_histogram( + "peer_discovery_time_to_first_piece_s", float(elapsed) + ) + fraction = len(verified) / total + for pct, key in ( + (10, "time_to_10pct_s"), + (50, "time_to_50pct_s"), + (90, "time_to_90pct_s"), + ): + if fraction * 100.0 >= pct and m.get(key) is None: + m[key] = elapsed def _touch_requestable_deficit_clock( self, swarm_state: dict[str, Any], metadata_incomplete: bool @@ -5651,6 +5694,8 @@ async def _on_download_complete(self) -> None: async def _on_piece_verified(self, piece_index: int) -> None: """Handle piece verification.""" + with contextlib.suppress(Exception): + self._touch_piece_progress_metrics() self.logger.debug( "_on_piece_verified called for piece %d (torrent: %s)", piece_index, diff --git a/tests/performance/baseline_swarm_ramp.json b/tests/performance/baseline_swarm_ramp.json new file mode 100644 index 0000000..9bfa362 --- /dev/null +++ b/tests/performance/baseline_swarm_ramp.json @@ -0,0 +1,56 @@ +{ + "100": { + "baseline": { + "completion_s": 61.524, + "duplicate_requests": 0.0, + "time_to_50pct_s": 0.987, + "time_to_first_piece_s": 0.668, + "time_to_first_requestable_s": 4.746, + "wasted_fraction": 0.0 + }, + "treatment": { + "completion_s": 2.184, + "duplicate_requests": 25.0, + "time_to_50pct_s": 0.987, + "time_to_first_piece_s": 0.668, + "time_to_first_requestable_s": 2.83, + "wasted_fraction": 0.022 + } + }, + "20": { + "baseline": { + "completion_s": 61.524, + "duplicate_requests": 0.0, + "time_to_50pct_s": 0.987, + "time_to_first_piece_s": 0.668, + "time_to_first_requestable_s": 5.218, + "wasted_fraction": 0.0 + }, + "treatment": { + "completion_s": 2.529, + "duplicate_requests": 24.0, + "time_to_50pct_s": 0.987, + "time_to_first_piece_s": 0.668, + "time_to_first_requestable_s": 4.926, + "wasted_fraction": 0.022 + } + }, + "50": { + "baseline": { + "completion_s": 61.524, + "duplicate_requests": 0.0, + "time_to_50pct_s": 0.987, + "time_to_first_piece_s": 0.668, + "time_to_first_requestable_s": 4.746, + "wasted_fraction": 0.0 + }, + "treatment": { + "completion_s": 2.184, + "duplicate_requests": 25.0, + "time_to_50pct_s": 0.987, + "time_to_first_piece_s": 0.668, + "time_to_first_requestable_s": 2.83, + "wasted_fraction": 0.022 + } + } +} diff --git a/tests/performance/bench_swarm_ramp.py b/tests/performance/bench_swarm_ramp.py new file mode 100644 index 0000000..78e74f2 --- /dev/null +++ b/tests/performance/bench_swarm_ramp.py @@ -0,0 +1,42 @@ +"""Deterministic virtual-swarm ramp benchmark. + +Usage: python tests/performance/bench_swarm_ramp.py [--peers 20,50,100] +Writes/compares JSON; in CI use run_benchmark() and assert gates in tests. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from ccbt.performance.swarm_sim import run_scenario + + +def run_benchmark(peer_counts=(20, 50, 100), seed: int = 1234) -> dict: + return { + str(n): run_scenario(n, seed=seed) + for n in peer_counts + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--peers", default="20,50,100") + parser.add_argument("--seed", type=int, default=1234) + parser.add_argument("--out", default="") + args = parser.parse_args() + counts = [int(x) for x in args.peers.split(",") if x.strip()] + results = run_benchmark(counts, seed=args.seed) + text = json.dumps(results, indent=2, sort_keys=True) + print(text) + if args.out: + Path(args.out).write_text(text + "\n", encoding="utf-8") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/unit/performance/test_acceleration.py b/tests/unit/performance/test_acceleration.py new file mode 100644 index 0000000..eb86934 --- /dev/null +++ b/tests/unit/performance/test_acceleration.py @@ -0,0 +1,276 @@ +"""Unit tests for download-acceleration helpers and wiring.""" + +from __future__ import annotations + +import time +from unittest.mock import AsyncMock, Mock + +import pytest + +pytestmark = [pytest.mark.unit] + +from ccbt.extensions.manager import ExtensionManager +from ccbt.extensions.pex import PEXPeer, PeerExchange +from ccbt.peer.async_peer_connection import AsyncPeerConnection +from ccbt.peer.peer import PeerInfo +from ccbt.performance.acceleration import ( + BlockLatencyTracker, + DuplicateBudget, + RampMetrics, + accel_enabled, + adaptive_request_timeout, + adaptive_upload_slots, + is_dynamic_endgame, + is_snubbed, + peer_usefulness_score, + race_first_success, + ramp_choking_intervals, + should_duplicate_block, + strategy_flag, + warm_pipeline_depth, +) +from ccbt.piece.async_piece_manager import AsyncPieceManager + + +def test_acceleration_on_by_default(): + from types import SimpleNamespace + + # Missing attribute (legacy/partial config) means enabled. + assert accel_enabled(SimpleNamespace()) is True + assert accel_enabled(SimpleNamespace(network=SimpleNamespace())) is True + assert ( + accel_enabled( + SimpleNamespace( + network=SimpleNamespace(swarm_acceleration_enabled=False) + ) + ) + is False + ) + assert strategy_flag(SimpleNamespace(), "straggler_duplication_enabled") is True + # Effective shipped config: enabled by default. + from ccbt.config.config import get_config + + cfg = get_config() + assert cfg.network.swarm_acceleration_enabled is True + assert cfg.strategy.straggler_duplication_enabled is True + assert cfg.discovery.min_peers_before_dht == 2 + + +@pytest.mark.asyncio +async def test_race_first_success_picks_fastest_and_tolerates_failure(): + import asyncio + + async def fast(): + await asyncio.sleep(0.01) + return "fast-result" + + async def slow(): + await asyncio.sleep(5) + return "slow-result" + + async def broken(): + raise ConnectionError("nope") + + kind, payload = await race_first_success( + {"slow": slow(), "broken": broken(), "fast": fast()} + ) + assert (kind, payload) == ("fast", "fast-result") + + with pytest.raises(ConnectionError): + await race_first_success({"broken": broken()}) + + +def test_adaptive_timeout_clamps(): + assert adaptive_request_timeout(0) == 30.0 + assert adaptive_request_timeout(0.1) == 5.0 # floor + assert adaptive_request_timeout(2.0) == 8.0 + assert adaptive_request_timeout(100.0) == 30.0 # ceiling + + +def test_duplicate_budget_caps_at_fraction(): + budget = DuplicateBudget(max_fraction=0.10) + budget.record_useful(900) + assert budget.allows(100) + budget.record_duplicate(100) + assert not budget.allows(1) + assert budget.wasted_fraction == pytest.approx(0.1) + + +def test_should_duplicate_straggler_and_last_block(): + assert should_duplicate_block( + block_age_s=5, swarm_p90_latency_s=1.0, blocks_remaining_in_piece=8, + existing_copies=1, + ) + assert should_duplicate_block( + block_age_s=2, swarm_p90_latency_s=60.0, blocks_remaining_in_piece=1, + existing_copies=1, + ) + assert not should_duplicate_block( + block_age_s=0.1, swarm_p90_latency_s=1.0, blocks_remaining_in_piece=8, + existing_copies=1, + ) + assert not should_duplicate_block( + block_age_s=99, swarm_p90_latency_s=1.0, blocks_remaining_in_piece=8, + existing_copies=3, max_copies=3, + ) + assert not should_duplicate_block( + block_age_s=99, swarm_p90_latency_s=1.0, blocks_remaining_in_piece=8, + existing_copies=1, budget_allows=False, + ) + + +def test_dynamic_endgame(): + assert is_dynamic_endgame(remaining_pieces=5, total_pieces=1000) + assert is_dynamic_endgame(remaining_pieces=50, total_pieces=1000) + assert not is_dynamic_endgame(remaining_pieces=10, total_pieces=10) + assert not is_dynamic_endgame(remaining_pieces=100, total_pieces=1000) + assert is_dynamic_endgame( + remaining_pieces=100, total_pieces=1000, all_remaining_stalled=True + ) + + +def test_snub_and_usefulness_and_slots(): + assert is_snubbed( + seconds_since_payload=20, seconds_unchoked=20, others_delivering=True + ) + assert not is_snubbed( + seconds_since_payload=20, seconds_unchoked=20, others_delivering=False + ) + good = peer_usefulness_score(has_needed_piece=True, historic_unchoke_rate=1.0) + bad = peer_usefulness_score(seconds_since_failure=1.0, is_snubbed_peer=True) + assert good > bad + assert adaptive_upload_slots( + base_slots=8, upload_rate_bps=5 * 512 * 1024, + requestable_peers=0, requestable_target=10, + ) > 8 + assert ramp_choking_intervals( + requestable_peers=0, requestable_target=10 + ) == (5.0, 20.0) + assert ramp_choking_intervals( + requestable_peers=10, requestable_target=10 + ) == (10.0, 30.0) + assert warm_pipeline_depth(current_depth=16) == 32 + assert warm_pipeline_depth(current_depth=16, bdp_blocks=64) == 64 + + +def test_latency_tracker_and_ramp_metrics(): + tracker = BlockLatencyTracker() + for v in (0.1, 0.2, 0.3, 10.0): + tracker.record(v) + assert tracker.p50 > 0 + assert tracker.p95 >= tracker.p50 + metrics = RampMetrics(started_monotonic=100.0) + metrics.touch_requestable(3, now=104.0) + metrics.touch_first_piece(now=106.0) + metrics.touch_progress(0.55, now=120.0) + snap = metrics.snapshot() + assert snap["time_to_first_requestable_s"] == 4.0 + assert snap["time_to_first_piece_s"] == 6.0 + assert snap["time_to_50pct_s"] == 20.0 + assert snap["time_to_90pct_s"] is None + + +def _connection(ip="10.0.0.1", port=6881): + conn = AsyncPeerConnection( + peer_info=PeerInfo(ip=ip, port=port), torrent_data={} + ) + from ccbt.peer.async_peer_connection import ConnectionState + + conn.state = ConnectionState.ACTIVE + return conn + + +def test_allow_fast_bypasses_choke_only_for_allowed_piece(): + conn = _connection() + conn.peer_choking = True + assert conn.can_request() is False + assert conn.can_request(piece_index=3) is False + conn.peer_state.bep6_allowed_fast_pieces.add(3) + assert conn.can_request(piece_index=3) is True + assert conn.get_request_block_reason(piece_index=4) == "remote_choked" + + +def test_speculative_timeout_cap_skips_cold_start(): + """Deep queue alone must not shrink cold-start handshake budgets.""" + from types import SimpleNamespace + + from ccbt.peer.async_peer_connection import AsyncPeerConnectionManager + + manager = AsyncPeerConnectionManager( + torrent_data={"info_hash": b"\x00" * 20}, + piece_manager=SimpleNamespace(), + max_peers_per_torrent=10, + ) + manager._pending_peer_queue = [object()] * 250 # noqa: SLF001 + # No active peers: legacy deep-queue cap (42s) applies, not the 20s + # speculative cap. + assert manager._cap_connect_task_timeout_s(130.0) == 42.0 # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_pex_callback_dials_new_peers(): + exchange = PeerExchange() + seen = [] + exchange.add_peer_discovered_callback(seen.append) + await exchange.handle_added_peers("peer-a", [PEXPeer("10.0.0.9", 6881)]) + assert seen and seen[0][0].ip == "10.0.0.9" + # Second time: no new peers, callback not re-fired with duplicates. + seen.clear() + await exchange.handle_added_peers("peer-a", [PEXPeer("10.0.0.9", 6881)]) + assert seen == [] + + manager = ExtensionManager() + seen2 = [] + manager.register_pex_peer_callback(seen2.append) + await manager.handle_pex_message("p", 0, b"") + # manager path exercised without error; direct exchange covered above + + +@pytest.mark.asyncio +async def test_straggler_duplication_sends_one_extra_copy(): + torrent_data = { + "info_hash": b"\x00" * 20, + "pieces_info": { + "num_pieces": 4, + "piece_length": 65536, + "piece_hashes": [b"\x00" * 20] * 4, + }, + } + manager = AsyncPieceManager(torrent_data) + try: + if not manager.pieces: + manager._build_piece_layout() + piece = manager.pieces[0] + block = piece.blocks[0] + manager._duplicate_budget.record_useful(10 * 1024 * 1024) + old_key = "10.0.0.1:6881" + block.requested_from.add(old_key) + manager._active_block_requests[0] = { + old_key: [(block.begin, block.length, time.time() - 100.0)] + } + candidate = Mock() + candidate.peer_info = PeerInfo(ip="10.0.0.2", port=6881) + candidate.can_request = Mock(return_value=True) + peer_manager = Mock() + peer_manager.request_piece = AsyncMock(return_value=True) + + sent = await manager._request_straggler_duplicates( + 0, [candidate], peer_manager + ) + assert sent == 1 + peer_manager.request_piece.assert_awaited_once() + assert manager._piece_selection_metrics["duplicate_requests"] == 1 + # Budget exhausted -> no further duplication. + manager._duplicate_budget.record_useful(0) + manager._duplicate_budget.duplicate_bytes = 10**9 + block2 = piece.blocks[1] + block2.requested_from.add(old_key) + manager._active_block_requests[0][old_key].append( + (block2.begin, block2.length, time.time() - 100.0) + ) + sent2 = await manager._request_straggler_duplicates( + 0, [candidate], peer_manager + ) + assert sent2 == 0 + finally: + await manager.stop() diff --git a/tests/unit/performance/test_swarm_sim_gates.py b/tests/unit/performance/test_swarm_sim_gates.py new file mode 100644 index 0000000..bfd2fe4 --- /dev/null +++ b/tests/unit/performance/test_swarm_sim_gates.py @@ -0,0 +1,27 @@ +"""Acceptance gates for the virtual-swarm benchmark (plan Phase 7).""" + +from __future__ import annotations + +import pytest + +pytestmark = [pytest.mark.unit] + +from ccbt.performance.swarm_sim import make_peers, run_scenario, simulate + + +@pytest.mark.parametrize("peer_count", [20, 50, 100]) +def test_acceleration_gates(peer_count): + result = run_scenario(peer_count, seed=1234) + base, treat = result["baseline"], result["treatment"] + # Treatment must finish the straggler tail materially faster and + # grow the requestable cohort faster, within the waste budget. + assert treat["completion_s"] < base["completion_s"] * 0.75 + assert treat["time_to_first_requestable_s"] <= base["time_to_first_requestable_s"] + assert treat["wasted_fraction"] <= 0.10 + + +def test_deterministic(): + peers = make_peers(20, seed=99) + a = simulate(peers=peers, duplicate_stragglers=True, dial_parallelism=24) + b = simulate(peers=peers, duplicate_stragglers=True, dial_parallelism=24) + assert a.as_dict() == b.as_dict()