gru/pyTorrent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

change in smart queue

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-06-01 08:29:14 +02:00
parent 6451afa5bc
commit 2b36e8e8af
1 changed files with 27 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pytorrent/services/smart_queue.py
+27 -59
View File
@@ -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)}