|
|
@ -1,16 +1,18 @@
|
|
|
|
|
|
|
|
|
|
|
|
from __future__ import division, print_function, unicode_literals
|
|
|
|
from __future__ import division, print_function, unicode_literals
|
|
|
|
import threading
|
|
|
|
import threading
|
|
|
|
|
|
|
|
import abc
|
|
|
|
|
|
|
|
import uuid
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
import queue
|
|
|
|
import queue
|
|
|
|
except ImportError:
|
|
|
|
except ImportError:
|
|
|
|
import Queue as queue
|
|
|
|
import Queue as queue
|
|
|
|
from datetime import datetime
|
|
|
|
from datetime import datetime
|
|
|
|
|
|
|
|
from collections import namedtuple
|
|
|
|
|
|
|
|
|
|
|
|
from cps import calibre_db
|
|
|
|
from cps import calibre_db
|
|
|
|
from cps import logger
|
|
|
|
from cps import logger
|
|
|
|
import abc
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
log = logger.create()
|
|
|
|
log = logger.create()
|
|
|
|
|
|
|
|
|
|
|
@ -20,6 +22,11 @@ STAT_FAIL = 1
|
|
|
|
STAT_STARTED = 2
|
|
|
|
STAT_STARTED = 2
|
|
|
|
STAT_FINISH_SUCCESS = 3
|
|
|
|
STAT_FINISH_SUCCESS = 3
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Only retain this many tasks in dequeued list
|
|
|
|
|
|
|
|
TASK_CLEANUP_TRIGGER = 20
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
QueuedTask = namedtuple('QueuedTask', 'user, added, task')
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _get_main_thread():
|
|
|
|
def _get_main_thread():
|
|
|
|
for t in threading.enumerate():
|
|
|
|
for t in threading.enumerate():
|
|
|
@ -51,10 +58,7 @@ class WorkerThread(threading.Thread):
|
|
|
|
def __init__(self):
|
|
|
|
def __init__(self):
|
|
|
|
threading.Thread.__init__(self)
|
|
|
|
threading.Thread.__init__(self)
|
|
|
|
|
|
|
|
|
|
|
|
self.finished = list()
|
|
|
|
self.dequeued = list()
|
|
|
|
|
|
|
|
|
|
|
|
self.db_queue = queue.Queue()
|
|
|
|
|
|
|
|
calibre_db.add_queue(self.db_queue)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self.doLock = threading.Lock()
|
|
|
|
self.doLock = threading.Lock()
|
|
|
|
self.queue = ImprovedQueue()
|
|
|
|
self.queue = ImprovedQueue()
|
|
|
@ -64,43 +68,41 @@ class WorkerThread(threading.Thread):
|
|
|
|
@classmethod
|
|
|
|
@classmethod
|
|
|
|
def add(cls, user, task):
|
|
|
|
def add(cls, user, task):
|
|
|
|
ins = cls.getInstance()
|
|
|
|
ins = cls.getInstance()
|
|
|
|
ins.queue.put((user, task))
|
|
|
|
ins.queue.put(QueuedTask(
|
|
|
|
|
|
|
|
user=user,
|
|
|
|
|
|
|
|
added=datetime.now(),
|
|
|
|
|
|
|
|
task=task,
|
|
|
|
|
|
|
|
))
|
|
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
@property
|
|
|
|
def tasks(self):
|
|
|
|
def tasks(self):
|
|
|
|
with self.doLock:
|
|
|
|
with self.doLock:
|
|
|
|
tasks = list(self.queue.to_list()) + self.finished
|
|
|
|
tasks = list(self.queue.to_list()) + self.dequeued
|
|
|
|
return tasks # todo: order by data added
|
|
|
|
return sorted(tasks, key=lambda x: x.added)
|
|
|
|
|
|
|
|
|
|
|
|
# Main thread loop starting the different tasks
|
|
|
|
# Main thread loop starting the different tasks
|
|
|
|
def run(self):
|
|
|
|
def run(self):
|
|
|
|
main_thread = _get_main_thread()
|
|
|
|
main_thread = _get_main_thread()
|
|
|
|
while main_thread.is_alive():
|
|
|
|
while main_thread.is_alive():
|
|
|
|
user, item = self.queue.get()
|
|
|
|
item = self.queue.get()
|
|
|
|
|
|
|
|
|
|
|
|
# add to list so that in-progress tasks show up
|
|
|
|
# add to list so that in-progress tasks show up
|
|
|
|
with self.doLock:
|
|
|
|
with self.doLock:
|
|
|
|
self.finished.append((user, item))
|
|
|
|
# Remove completed tasks if needed
|
|
|
|
|
|
|
|
if len(self.dequeued) > TASK_CLEANUP_TRIGGER:
|
|
|
|
|
|
|
|
# sort first (just to be certain), then lob off the extra
|
|
|
|
|
|
|
|
self.dequeued = sorted(self.dequeued, key=lambda x: x.added)[-1 * TASK_CLEANUP_TRIGGER:]
|
|
|
|
|
|
|
|
self.dequeued.append(item)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
user, added, task = item
|
|
|
|
|
|
|
|
|
|
|
|
# sometimes tasks (like Upload) don't actually have work to do and are created as already finished
|
|
|
|
# sometimes tasks (like Upload) don't actually have work to do and are created as already finished
|
|
|
|
if item.stat is STAT_WAITING:
|
|
|
|
if task.stat is STAT_WAITING:
|
|
|
|
# CalibreTask.start() should wrap all exceptions in it's own error handling
|
|
|
|
# CalibreTask.start() should wrap all exceptions in it's own error handling
|
|
|
|
item.start(self)
|
|
|
|
task.start(self)
|
|
|
|
|
|
|
|
|
|
|
|
self.queue.task_done()
|
|
|
|
self.queue.task_done()
|
|
|
|
|
|
|
|
|
|
|
|
def _delete_completed_tasks(self):
|
|
|
|
|
|
|
|
raise NotImplementedError()
|
|
|
|
|
|
|
|
# for index, task in reversed(list(enumerate(self.UIqueue))):
|
|
|
|
|
|
|
|
# if task['progress'] == "100 %":
|
|
|
|
|
|
|
|
# # delete tasks
|
|
|
|
|
|
|
|
# self.queue.pop(index)
|
|
|
|
|
|
|
|
# self.UIqueue.pop(index)
|
|
|
|
|
|
|
|
# # if we are deleting entries before the current index, adjust the index
|
|
|
|
|
|
|
|
# if index <= self.current and self.current:
|
|
|
|
|
|
|
|
# self.current -= 1
|
|
|
|
|
|
|
|
# self.last = len(self.queue)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class CalibreTask:
|
|
|
|
class CalibreTask:
|
|
|
|
__metaclass__ = abc.ABCMeta
|
|
|
|
__metaclass__ = abc.ABCMeta
|
|
|
|
|
|
|
|
|
|
|
@ -111,6 +113,7 @@ class CalibreTask:
|
|
|
|
self.start_time = None
|
|
|
|
self.start_time = None
|
|
|
|
self.end_time = None
|
|
|
|
self.end_time = None
|
|
|
|
self.message = message
|
|
|
|
self.message = message
|
|
|
|
|
|
|
|
self.id = uuid.uuid4()
|
|
|
|
|
|
|
|
|
|
|
|
@abc.abstractmethod
|
|
|
|
@abc.abstractmethod
|
|
|
|
def run(self, worker_thread):
|
|
|
|
def run(self, worker_thread):
|
|
|
|