HaydenandXiaoJiang0208
d12cb25710
A 200 answer to a Range request was appended onto the partial file, so the temp file grew past the real resource and the next resume asked for a range beyond its end - the 416 from #210. Read the total from content-range on 206, and replace rather than append when the response is a full 200. Also recover a temp file that is already bigger than the resource instead of asking for an out-of-bounds range, so users currently stuck on 416 get fixed without deleting the file by hand. Worker: close the event loop, and drop the task id from running_tasks even when the coroutine raised - a failed task used to block its own re-submit forever. Co-authored-by: XiaoJiang0208 <jjm1837080396@gmail.com>
63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
import asyncio
|
|
import threading
|
|
import queue
|
|
|
|
from . import utils
|
|
|
|
|
|
class DownloadThreadPool:
|
|
def __init__(self) -> None:
|
|
self.workers_count = 0
|
|
self.task_queue = queue.Queue()
|
|
self.running_tasks = set()
|
|
self._lock = threading.Lock()
|
|
|
|
default_max_workers = 5
|
|
max_workers: int = default_max_workers
|
|
self.max_worker = max_workers
|
|
|
|
def submit(self, task, task_id):
|
|
with self._lock:
|
|
if task_id in self.running_tasks:
|
|
return "Existing"
|
|
self.running_tasks.add(task_id)
|
|
self.task_queue.put((task, task_id))
|
|
return self._adjust_worker_count()
|
|
|
|
def _adjust_worker_count(self):
|
|
if self.workers_count < self.max_worker:
|
|
self._start_worker()
|
|
return "Running"
|
|
else:
|
|
return "Waiting"
|
|
|
|
def _start_worker(self):
|
|
t = threading.Thread(target=self._worker, daemon=True)
|
|
t.start()
|
|
with self._lock:
|
|
self.workers_count += 1
|
|
|
|
def _worker(self):
|
|
loop = asyncio.new_event_loop()
|
|
|
|
while True:
|
|
try:
|
|
task, task_id = self.task_queue.get(timeout=5)
|
|
except queue.Empty:
|
|
with self._lock:
|
|
if self.task_queue.empty():
|
|
self.workers_count -= 1
|
|
break
|
|
continue
|
|
|
|
try:
|
|
loop.run_until_complete(task(task_id))
|
|
except Exception as e:
|
|
utils.print_error(f"worker run error: {str(e)}")
|
|
finally:
|
|
with self._lock:
|
|
if task_id in self.running_tasks:
|
|
self.running_tasks.remove(task_id)
|
|
|
|
loop.close()
|