From 2b36e8e8af7f5807a04f95ee1ee2ec980210ff9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mateusz=20Gruszczy=C5=84ski?= Date: Mon, 1 Jun 2026 08:29:14 +0200 Subject: [PATCH] change in smart queue --- pytorrent/services/smart_queue.py | 86 ++++++++++--------------------- 1 file changed, 27 insertions(+), 59 deletions(-) diff --git a/pytorrent/services/smart_queue.py b/pytorrent/services/smart_queue.py index 4c2e6c2..ad7d137 100644 --- a/pytorrent/services/smart_queue.py +++ b/pytorrent/services/smart_queue.py @@ -437,38 +437,6 @@ def _clear_stalled_label(client: Any, torrent_hash: str, current_label: str = '' return False -def _read_int_rpc(client: Any, method: str, torrent_hash: str, default: int = 0) -> int: - """Read an integer rTorrent field without failing the queue pass.""" - # Note: Smart Queue uses this for last-second safety checks before stopping torrents. - try: - return int(client.call(method, torrent_hash) or 0) - except Exception: - return int(default) - - -def _read_live_transfer_activity(client: Any, torrent_hash: str, stalled_seconds: int) -> dict[str, Any]: - """Read live transfer activity directly from rTorrent before a stop decision.""" - # Note: The torrent list can be a stale snapshot; this live guard prevents stopping a torrent that is currently downloading. - down_rate = _read_int_rpc(client, 'd.down.rate', torrent_hash) - up_rate = _read_int_rpc(client, 'd.up.rate', torrent_hash) - last_activity = _read_int_rpc(client, 'd.timestamp.last_active', torrent_hash) - state = _read_int_rpc(client, 'd.state', torrent_hash) - active = _read_int_rpc(client, 'd.is_active', torrent_hash) - if not last_activity and (down_rate > 0 or up_rate > 0): - last_activity = int(time.time()) - recent = bool(last_activity and time.time() - last_activity < max(1, int(stalled_seconds or 0))) - transferring = bool(down_rate > 0 or up_rate > 0) - return { - 'hash': torrent_hash, - 'state': state, - 'active': active, - 'down_rate': down_rate, - 'up_rate': up_rate, - 'last_activity': last_activity, - 'recent_activity': recent, - 'transferring': transferring, - 'protected': bool((state or active) and (transferring or recent)), - } def _remember_auto_label(profile_id: int, torrent_hash: str, previous_label: str) -> None: now = utcnow() @@ -1568,8 +1536,8 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = ignored_seed_peer_count = 0 ignored_speed_count = 0 - live_activity_protected: list[dict[str, Any]] = [] - live_activity_protected_hashes: set[str] = set() + snapshot_activity_protected: list[str] = [] + snapshot_activity_protected_hashes: set[str] = set() with connect() as conn: for t in downloading: @@ -1614,23 +1582,23 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = max_active = max(1, int(settings.get('max_active_downloads') or 5)) stalled_hashes = {str(t.get('hash') or '') for t in stalled} - # Enforce the hard active-download cap across the whole started queue, including manual starts. - # Note: Weak/no-source torrents are stopped first, but the cap is still enforced when the overflow is larger. + # Enforce the active-download cap using only torrents that the current snapshot already proves idle/weak. + # Note: A transferring or recently active torrent is never stopped just because the cap is exceeded. over_limit = max(0, len(downloading) - max_active) stop_eligible_hashes = {str(t.get('hash') or '') for t in stop_eligible} stop_rank = sorted( - downloading, + stop_eligible, key=lambda t: ( 0 if str(t.get('hash') or '') in stalled_hashes else 1, - 0 if str(t.get('hash') or '') in stop_eligible_hashes else 1, int(t.get('down_rate') or 0), int(t.get('seeds') or 0), int(t.get('peers') or 0), ), ) + capped_over_limit = min(over_limit, len(stop_rank)) # Note: The user-defined batch limit caps all automatic stops in one pass. # Hard cap overflow is handled first, then stalled replacement uses only proven spare candidate capacity. - to_stop: list[dict[str, Any]] = stop_rank[:min(over_limit, stop_batch_size)] + to_stop: list[dict[str, Any]] = stop_rank[:min(capped_over_limit, stop_batch_size)] stop_hashes = {str(t.get('hash') or '') for t in to_stop} remaining_stop_budget = max(0, stop_batch_size - len(to_stop)) free_slots_before_stop = max(0, max_active - len(downloading)) @@ -1655,14 +1623,13 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = rtorrent_cap = _ensure_rtorrent_download_cap(c, max_active) for t in downloading: h = str(t.get('hash') or '') - if not h or h in live_activity_protected_hashes or not _has_stalled_label(str(t.get('label') or '')): + if not h or not _has_stalled_label(str(t.get('label') or '')): continue - live_activity = _read_live_transfer_activity(c, h, stalled_seconds) - if live_activity.get('protected'): - # Note: A torrent that resumes real transfer is removed from Stalled without waiting for another manual action. - live_activity_protected.append(live_activity) - live_activity_protected_hashes.add(h) - _clear_stalled_label(c, h, _read_label(c, h, str(t.get('label') or ''))) + if _has_recent_transfer_activity(t, stalled_seconds): + # Note: Snapshot activity is enough to remove Stalled; no per-torrent live RPC guard is needed. + snapshot_activity_protected.append(h) + snapshot_activity_protected_hashes.add(h) + _clear_stalled_label(c, h, str(t.get('label') or '')) with connect() as conn: conn.execute('DELETE FROM smart_queue_stalled WHERE profile_id=? AND torrent_hash=?', (profile_id, h)) stopped_by_queue: list[str] = [] @@ -1678,19 +1645,18 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = for t in to_stop: h = str(t.get('hash') or '') try: - if h in live_activity_protected_hashes: + if not h or h in snapshot_activity_protected_hashes: continue - live_activity = _read_live_transfer_activity(c, h, stalled_seconds) - if live_activity.get('protected'): - # Note: A live rTorrent re-check wins over the stale queue snapshot, so active downloads are never stopped as Stalled. - live_activity_protected.append(live_activity) - live_activity_protected_hashes.add(h) - _clear_stalled_label(c, h, _read_label(c, h, str(t.get('label') or ''))) + if _has_recent_transfer_activity(t, stalled_seconds): + # Note: Snapshot activity wins; active torrents are protected without slow per-item live checks. + snapshot_activity_protected.append(h) + snapshot_activity_protected_hashes.add(h) + _clear_stalled_label(c, h, str(t.get('label') or '')) with connect() as conn: conn.execute('DELETE FROM smart_queue_stalled WHERE profile_id=? AND torrent_hash=?', (profile_id, h)) continue # Note: Smart Queue stops with the same low-level d.stop command used by the manual Stop action. - # This avoids extra pre-check RPCs and keeps large queues from failing after only a few items. + # This avoids extra pre-check RPCs and keeps large queues fast even with many candidates. c.call('d.stop', h) if h in stalled_hashes: if _ensure_stalled_label(c, h, _read_label(c, h, str(t.get('label') or ''))): @@ -1758,6 +1724,7 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = 'active_after_stop': active_after_stop, 'active_after_expected': active_after_stop + len(started_by_queue), 'over_limit': over_limit, + 'stoppable_over_limit': capped_over_limit, 'stopped': stopped_by_queue, 'started': started_by_queue, 'start_requested': start_requested, @@ -1773,8 +1740,8 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = 'stalled_stopped_hashes': _hash_sample(stalled_stopped_hashes), 'stalled_labeled': stalled_labeled, 'protected_stalled': protected_stalled, - 'live_activity_protected': len(live_activity_protected), - 'live_activity_protected_hashes': _hash_sample([item.get('hash') for item in live_activity_protected]), + 'snapshot_activity_protected': len(snapshot_activity_protected), + 'snapshot_activity_protected_hashes': _hash_sample(snapshot_activity_protected), 'stalled_replacement_allowed': stalled_replacement_allowed, 'excluded': len(user_excluded), 'excluded_stalled': len(stalled_label_hashes), @@ -1807,10 +1774,11 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = 'active_after_expected': active_after_stop + len(started_by_queue), 'max_active_downloads': max_active, 'over_limit': over_limit, + 'stoppable_over_limit': capped_over_limit, 'stopped': len(stopped_by_queue), 'stalled': len(stalled), 'protected_stalled': protected_stalled, - 'live_activity_protected': len(live_activity_protected), + 'snapshot_activity_protected': len(snapshot_activity_protected), 'stalled_stopped': len(stalled_stopped_hashes), 'stalled_stopped_hashes': _hash_sample(stalled_stopped_hashes, 20), 'stop_eligible': len(stop_eligible), @@ -1843,7 +1811,7 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = }, 'rtorrent_cap': rtorrent_cap, 'to_stop': _diagnostics_torrents(to_stop), - 'live_activity_protected': _diagnostics_sample(live_activity_protected), + 'snapshot_activity_protected': _diagnostics_sample(snapshot_activity_protected), 'stalled': _diagnostics_torrents(stalled), 'stop_eligible': _diagnostics_torrents(stop_eligible), 'to_start': _diagnostics_torrents(to_start), @@ -1861,4 +1829,4 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool = mark_run(profile_id, user_id) settings = get_settings(profile_id, user_id) remaining = cooldown_remaining(settings) - return {'ok': True, 'enabled': bool(settings.get('enabled')), 'paused': stopped_by_queue, 'resumed': started_by_queue, 'stopped': stopped_by_queue, 'started': started_by_queue, 'start_requested': start_requested, 'start_batch_size': start_summary['start_batch_size'], 'start_verify_attempts': start_summary['start_verify_attempts'], 'start_verify_delay_seconds': start_summary['start_verify_delay_seconds'], 'waiting_labeled': len(to_label_waiting), 'stalled_labeled': stalled_labeled, 'excluded_stalled': len(stalled_label_hashes), 'manual_labeled_running': len(manual_labeled_running), 'labels_restored': restored, 'labels_failed': label_failed, 'stop_failed': stop_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'start_pending_confirmation': start_pending_confirmation, 'active_verified': active_verified, 'active_before': len(downloading), 'active_after_stop': active_after_stop, 'over_limit': over_limit, 'stop_eligible': len(stop_eligible), 'start_source_skipped': len(source_skipped), 'ignore_seed_peer': ignore_seed_peer, 'ignore_speed': ignore_speed, 'ignored_seed_peer_count': ignored_seed_peer_count if ignore_seed_peer else 0, 'ignored_speed_count': ignored_speed_count if ignore_speed else 0, 'stalled_seconds': stalled_seconds, 'stalled_timer_key': timer_key, 'stop_batch_size': stop_batch_size, 'start_grace_seconds': start_grace_seconds, 'protect_active_below_cap': protect_active_below_cap, 'prefer_partial_progress': prefer_partial_progress, 'auto_stop_idle': bool(int(settings.get('auto_stop_idle') or 0)), 'stalled_replacement_allowed': stalled_replacement_allowed, 'start_grace_protected': len(start_grace_hashes), 'replacement_capacity': replacement_capacity, 'protected_stalled': protected_stalled, 'healthy_active_protected': len(live_activity_protected), 'live_activity_protected': live_activity_protected, 'rtorrent_cap': rtorrent_cap, 'checked': len(torrents), 'excluded': len(user_excluded), 'settings': settings, 'cooldown_remaining_seconds': remaining, 'surge_refill_remaining_seconds': surge_refill_remaining(settings)} + return {'ok': True, 'enabled': bool(settings.get('enabled')), 'paused': stopped_by_queue, 'resumed': started_by_queue, 'stopped': stopped_by_queue, 'started': started_by_queue, 'start_requested': start_requested, 'start_batch_size': start_summary['start_batch_size'], 'start_verify_attempts': start_summary['start_verify_attempts'], 'start_verify_delay_seconds': start_summary['start_verify_delay_seconds'], 'waiting_labeled': len(to_label_waiting), 'stalled_labeled': stalled_labeled, 'excluded_stalled': len(stalled_label_hashes), 'manual_labeled_running': len(manual_labeled_running), 'labels_restored': restored, 'labels_failed': label_failed, 'stop_failed': stop_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'start_pending_confirmation': start_pending_confirmation, 'active_verified': active_verified, 'active_before': len(downloading), 'active_after_stop': active_after_stop, 'over_limit': over_limit, 'stoppable_over_limit': capped_over_limit, 'stop_eligible': len(stop_eligible), 'start_source_skipped': len(source_skipped), 'ignore_seed_peer': ignore_seed_peer, 'ignore_speed': ignore_speed, 'ignored_seed_peer_count': ignored_seed_peer_count if ignore_seed_peer else 0, 'ignored_speed_count': ignored_speed_count if ignore_speed else 0, 'stalled_seconds': stalled_seconds, 'stalled_timer_key': timer_key, 'stop_batch_size': stop_batch_size, 'start_grace_seconds': start_grace_seconds, 'protect_active_below_cap': protect_active_below_cap, 'prefer_partial_progress': prefer_partial_progress, 'auto_stop_idle': bool(int(settings.get('auto_stop_idle') or 0)), 'stalled_replacement_allowed': stalled_replacement_allowed, 'start_grace_protected': len(start_grace_hashes), 'replacement_capacity': replacement_capacity, 'protected_stalled': protected_stalled, 'healthy_active_protected': len(snapshot_activity_protected), 'snapshot_activity_protected': snapshot_activity_protected, 'rtorrent_cap': rtorrent_cap, 'checked': len(torrents), 'excluded': len(user_excluded), 'settings': settings, 'cooldown_remaining_seconds': remaining, 'surge_refill_remaining_seconds': surge_refill_remaining(settings)}