559 lines
17 KiB
Python
559 lines
17 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
from __future__ import absolute_import, division, unicode_literals
|
|
from logging import getLogger
|
|
from time import time as _time
|
|
import threading
|
|
import Queue
|
|
import heapq
|
|
from collections import deque
|
|
|
|
import xbmc
|
|
|
|
from . import utils, app, variables as v
|
|
|
|
LOG = getLogger('PLEX.threads')
|
|
|
|
|
|
class KillableThread(threading.Thread):
|
|
def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
|
|
self._canceled = False
|
|
self._suspended = False
|
|
self._is_not_suspended = threading.Event()
|
|
self._is_not_suspended.set()
|
|
self._suspension_reached = threading.Event()
|
|
self._is_not_asleep = threading.Event()
|
|
self._is_not_asleep.set()
|
|
self.suspension_timeout = None
|
|
super(KillableThread, self).__init__(group, target, name, args, kwargs)
|
|
|
|
def should_cancel(self):
|
|
"""
|
|
Returns True if the thread should be stopped immediately
|
|
"""
|
|
return self._canceled or app.APP.stop_pkc
|
|
|
|
def cancel(self):
|
|
"""
|
|
Call from another thread to stop this current thread
|
|
"""
|
|
self._canceled = True
|
|
# Make sure thread is running in order to exit quickly
|
|
self._is_not_asleep.set()
|
|
self._is_not_suspended.set()
|
|
|
|
def should_suspend(self):
|
|
"""
|
|
Returns True if the current thread should be suspended immediately
|
|
"""
|
|
return self._suspended
|
|
|
|
def suspend(self, block=False, timeout=None):
|
|
"""
|
|
Call from another thread to suspend the current thread. Provide a
|
|
timeout [float] in seconds optionally. block=True will block the caller
|
|
until the thread-to-be-suspended is indeed suspended
|
|
Will wake a thread that is asleep!
|
|
"""
|
|
self.suspension_timeout = timeout
|
|
self._suspended = True
|
|
self._is_not_suspended.clear()
|
|
# Make sure thread wakes up in order to suspend
|
|
self._is_not_asleep.set()
|
|
if block:
|
|
self._suspension_reached.wait()
|
|
|
|
def resume(self):
|
|
"""
|
|
Call from another thread to revive a suspended or asleep current thread
|
|
back to life
|
|
"""
|
|
self._suspended = False
|
|
self._is_not_asleep.set()
|
|
self._is_not_suspended.set()
|
|
|
|
def wait_while_suspended(self):
|
|
"""
|
|
Blocks until thread is not suspended anymore or the thread should
|
|
exit or for a period of self.suspension_timeout (set by the caller of
|
|
suspend())
|
|
Returns the value of should_cancel()
|
|
"""
|
|
self._suspension_reached.set()
|
|
self._is_not_suspended.wait(self.suspension_timeout)
|
|
self._suspension_reached.clear()
|
|
return self.should_cancel()
|
|
|
|
def is_suspended(self):
|
|
"""
|
|
Check from another thread whether the current thread is suspended
|
|
"""
|
|
return self._suspension_reached.is_set()
|
|
|
|
def sleep(self, timeout):
|
|
"""
|
|
Only call from the current thread in order to sleep for a period of
|
|
timeout [float, seconds]. Will unblock immediately if thread should
|
|
cancel (should_cancel()) or the thread should_suspend
|
|
"""
|
|
self._is_not_asleep.clear()
|
|
self._is_not_asleep.wait(timeout)
|
|
self._is_not_asleep.set()
|
|
|
|
def is_asleep(self):
|
|
"""
|
|
Check from another thread whether the current thread is asleep
|
|
"""
|
|
return not self._is_not_asleep.is_set()
|
|
|
|
|
|
class ProcessingQueue(Queue.Queue, object):
|
|
"""
|
|
Queue of queues that processes a queue completely before moving on to the
|
|
next queue. There's one queue per Section(). You need to initialize each
|
|
section with add_section(section) first.
|
|
Put tuples (count, item) into this queue, with count being the respective
|
|
position of the item in the queue, starting with 0 (zero).
|
|
(None, None) is the sentinel for a single queue being exhausted, added by
|
|
put_sentinel()
|
|
"""
|
|
def _init(self, maxsize):
|
|
self.queue = deque()
|
|
self._sections = deque()
|
|
self._queues = deque()
|
|
self._current_section = None
|
|
self._current_queue = None
|
|
self._counter = 0
|
|
|
|
def _qsize(self):
|
|
return self._current_queue._qsize() if self._current_queue else 0
|
|
|
|
def total_size(self):
|
|
"""
|
|
Return the approximate total size of all queues (not reliable!)
|
|
"""
|
|
self.mutex.acquire()
|
|
n = sum(q._qsize() for q in self._queues) if self._queues else 0
|
|
self.mutex.release()
|
|
return n
|
|
|
|
def put(self, item, block=True, timeout=None):
|
|
"""Put an item into the queue.
|
|
|
|
If optional args 'block' is true and 'timeout' is None (the default),
|
|
block if necessary until a free slot is available. If 'timeout' is
|
|
a non-negative number, it blocks at most 'timeout' seconds and raises
|
|
the Full exception if no free slot was available within that time.
|
|
Otherwise ('block' is false), put an item on the queue if a free slot
|
|
is immediately available, else raise the Full exception ('timeout'
|
|
is ignored in that case).
|
|
"""
|
|
self.not_full.acquire()
|
|
try:
|
|
if self.maxsize > 0:
|
|
if not block:
|
|
# Use >= instead of == due to OrderedQueue!
|
|
if self._qsize() >= self.maxsize:
|
|
raise Queue.Full
|
|
elif timeout is None:
|
|
while self._qsize() >= self.maxsize:
|
|
self.not_full.wait()
|
|
elif timeout < 0:
|
|
raise ValueError("'timeout' must be a non-negative number")
|
|
else:
|
|
endtime = _time() + timeout
|
|
while self._qsize() >= self.maxsize:
|
|
remaining = endtime - _time()
|
|
if remaining <= 0.0:
|
|
raise Queue.Full
|
|
self.not_full.wait(remaining)
|
|
if self._put(item) == 0:
|
|
# Only notify one waiting thread if this item is put into the
|
|
# current queue
|
|
self.not_empty.notify()
|
|
else:
|
|
# Be sure to signal not_empty only once!
|
|
self._unlock_after_section_change()
|
|
self.unfinished_tasks += 1
|
|
finally:
|
|
self.not_full.release()
|
|
|
|
def _put(self, item):
|
|
"""
|
|
Returns the index of the section in whose subqueue we need to put the
|
|
item into
|
|
"""
|
|
for i, section in enumerate(self._sections):
|
|
if item[1]['section'] == section:
|
|
self._queues[i]._put(item)
|
|
break
|
|
else:
|
|
raise RuntimeError('Could not find section for item %s' % item[1])
|
|
return i
|
|
|
|
def _unlock_after_section_change(self):
|
|
"""
|
|
Ugly work-around if we expected more items to be synced, but we had
|
|
to lower our section.number_of_items because PKC decided that nothing
|
|
changed and we don't need to sync the respective item(s).
|
|
get() thus might block indefinitely
|
|
"""
|
|
while (self._current_section and
|
|
self._counter == self._current_section.number_of_items):
|
|
LOG.debug('Signaling completion of current section')
|
|
self._init_next_section()
|
|
if self._current_queue and self._current_queue._qsize():
|
|
LOG.debug('Signaling not_empty')
|
|
self.not_empty.notify()
|
|
|
|
def put_sentinel(self, section):
|
|
"""
|
|
Adds a new empty section as a sentinel. Call with an empty Section()
|
|
object.
|
|
Once the get()-method returns None, you've received the sentinel and
|
|
you've thus exhausted the queue
|
|
"""
|
|
self.not_empty.acquire()
|
|
try:
|
|
section.number_of_items = 1
|
|
self._add_section(section)
|
|
# Add the actual sentinel to the queue we just added
|
|
self._queues[-1]._put((None, None))
|
|
self.unfinished_tasks += 1
|
|
if len(self._queues) == 1:
|
|
# queue was already exhausted!
|
|
self._switch_queues()
|
|
self._counter = 0
|
|
self.not_empty.notify()
|
|
else:
|
|
self._unlock_after_section_change()
|
|
finally:
|
|
self.not_empty.release()
|
|
|
|
def add_section(self, section):
|
|
"""
|
|
Be sure to add all sections first before starting to pop items off this
|
|
queue or adding them to the queue
|
|
"""
|
|
self.mutex.acquire()
|
|
try:
|
|
self._add_section(section)
|
|
finally:
|
|
self.mutex.release()
|
|
|
|
def _add_section(self, section):
|
|
self._sections.append(section)
|
|
self._queues.append(
|
|
OrderedQueue() if section.plex_type == v.PLEX_TYPE_ALBUM
|
|
else Queue.Queue())
|
|
if self._current_section is None:
|
|
self._switch_queues()
|
|
|
|
def _init_next_section(self):
|
|
self._sections.popleft()
|
|
self._queues.popleft()
|
|
self._counter = 0
|
|
self._switch_queues()
|
|
|
|
def _switch_queues(self):
|
|
self._current_section = self._sections[0] if self._sections else None
|
|
self._current_queue = self._queues[0] if self._queues else None
|
|
|
|
def _get(self):
|
|
item = self._current_queue._get()
|
|
self._counter += 1
|
|
if self._counter == self._current_section.number_of_items:
|
|
self._init_next_section()
|
|
return item[1]
|
|
|
|
|
|
class OrderedQueue(Queue.PriorityQueue, object):
|
|
"""
|
|
Queue that enforces an order on the items it returns. An item you push
|
|
onto the queue must be a tuple
|
|
(index, item)
|
|
where index=-1 is the item that will be returned first. The Queue will block
|
|
until index=-1, 0, 1, 2, 3, ... is then made available
|
|
|
|
maxsize will be rather fuzzy, as _qsize returns 0 if we're still waiting
|
|
for the next smalles index. put() thus might not block always when it
|
|
should.
|
|
"""
|
|
def __init__(self, maxsize=0):
|
|
self.next_index = 0
|
|
super(OrderedQueue, self).__init__(maxsize)
|
|
|
|
def _qsize(self, len=len):
|
|
try:
|
|
return len(self.queue) if self.queue[0][0] == self.next_index else 0
|
|
except IndexError:
|
|
return 0
|
|
|
|
def _get(self, heappop=heapq.heappop):
|
|
self.next_index += 1
|
|
return heappop(self.queue)
|
|
|
|
|
|
class Tasks(list):
|
|
def add(self, task):
|
|
for t in self:
|
|
if not t.isValid():
|
|
self.remove(t)
|
|
|
|
if isinstance(task, list):
|
|
self += task
|
|
else:
|
|
self.append(task)
|
|
|
|
def cancel(self):
|
|
while self:
|
|
self.pop().cancel()
|
|
|
|
|
|
class Task(object):
|
|
def __init__(self, priority=None):
|
|
self.priority = priority
|
|
self._canceled = False
|
|
self.finished = False
|
|
|
|
def __cmp__(self, other):
|
|
return self.priority - other.priority
|
|
|
|
def start(self):
|
|
BGThreader.addTask(self)
|
|
|
|
def _run(self):
|
|
self.run()
|
|
self.finished = True
|
|
|
|
def run(self):
|
|
raise NotImplementedError
|
|
|
|
def cancel(self):
|
|
self._canceled = True
|
|
|
|
def should_cancel(self):
|
|
return self._canceled or xbmc.abortRequested
|
|
|
|
def isValid(self):
|
|
return not self.finished and not self._canceled
|
|
|
|
|
|
class FunctionAsTask(Task):
|
|
def __init__(self, function, callback, *args, **kwargs):
|
|
self._function = function
|
|
self._callback = callback
|
|
self._args = args
|
|
self._kwargs = kwargs
|
|
super(FunctionAsTask, self).__init__()
|
|
|
|
def run(self):
|
|
result = self._function(*self._args, **self._kwargs)
|
|
if self._callback:
|
|
self._callback(result)
|
|
|
|
|
|
class MutablePriorityQueue(Queue.PriorityQueue):
|
|
def _get(self, heappop=heapq.heappop):
|
|
self.queue.sort()
|
|
return heappop(self.queue)
|
|
|
|
def lowest(self):
|
|
"""Return the lowest priority item in the queue (not reliable!)."""
|
|
self.mutex.acquire()
|
|
try:
|
|
lowest = self.queue and min(self.queue) or None
|
|
except Exception:
|
|
lowest = None
|
|
utils.ERROR()
|
|
finally:
|
|
self.mutex.release()
|
|
return lowest
|
|
|
|
|
|
class BackgroundWorker(object):
|
|
def __init__(self, queue, name=None):
|
|
self._queue = queue
|
|
self.name = name
|
|
self._thread = None
|
|
self._abort = False
|
|
self._task = None
|
|
|
|
@staticmethod
|
|
def _runTask(task):
|
|
if task._canceled:
|
|
return
|
|
try:
|
|
task._run()
|
|
except Exception:
|
|
utils.ERROR()
|
|
|
|
def abort(self):
|
|
self._abort = True
|
|
return self
|
|
|
|
def aborted(self):
|
|
return self._abort or xbmc.abortRequested
|
|
|
|
def start(self):
|
|
if self._thread and self._thread.isAlive():
|
|
return
|
|
|
|
self._thread = KillableThread(target=self._queueLoop, name='BACKGROUND-WORKER({0})'.format(self.name))
|
|
self._thread.start()
|
|
|
|
def _queueLoop(self):
|
|
if self._queue.empty():
|
|
return
|
|
|
|
LOG.debug('(%s): Active', self.name)
|
|
try:
|
|
while not self.aborted():
|
|
self._task = self._queue.get_nowait()
|
|
self._runTask(self._task)
|
|
self._queue.task_done()
|
|
self._task = None
|
|
except Queue.Empty:
|
|
LOG.debug('(%s): Idle', self.name)
|
|
|
|
def shutdown(self):
|
|
self.abort()
|
|
|
|
if self._task:
|
|
self._task.cancel()
|
|
|
|
if self._thread and self._thread.isAlive():
|
|
LOG.debug('thread (%s): Waiting...', self.name)
|
|
self._thread.join()
|
|
LOG.debug('thread (%s): Done', self.name)
|
|
|
|
def working(self):
|
|
return self._thread and self._thread.isAlive()
|
|
|
|
|
|
class NonstoppingBackgroundWorker(BackgroundWorker):
|
|
def __init__(self, queue, name=None):
|
|
self._working = False
|
|
super(NonstoppingBackgroundWorker, self).__init__(queue, name)
|
|
|
|
def _queueLoop(self):
|
|
while not self.aborted():
|
|
try:
|
|
self._task = self._queue.get_nowait()
|
|
self._working = True
|
|
self._runTask(self._task)
|
|
self._working = False
|
|
self._queue.task_done()
|
|
self._task = None
|
|
except Queue.Empty:
|
|
app.APP.monitor.waitForAbort(0.05)
|
|
|
|
def working(self):
|
|
return self._working
|
|
|
|
|
|
class BackgroundThreader:
|
|
def __init__(self, name=None, worker=BackgroundWorker, worker_count=6):
|
|
self.name = name
|
|
self._queue = MutablePriorityQueue()
|
|
self._abort = False
|
|
self.priority = -1
|
|
self.workers = [worker(self._queue, 'queue.{0}:worker.{1}'.format(self.name, x)) for x in range(worker_count)]
|
|
|
|
def _nextPriority(self):
|
|
self.priority += 1
|
|
return self.priority
|
|
|
|
def abort(self):
|
|
self._abort = True
|
|
for w in self.workers:
|
|
w.abort()
|
|
return self
|
|
|
|
def aborted(self):
|
|
return self._abort or xbmc.abortRequested
|
|
|
|
def shutdown(self):
|
|
self.abort()
|
|
|
|
for w in self.workers:
|
|
w.shutdown()
|
|
|
|
def addTask(self, task):
|
|
task.priority = self._nextPriority()
|
|
self._queue.put(task)
|
|
self.startWorkers()
|
|
|
|
def addTasks(self, tasks):
|
|
for t in tasks:
|
|
t.priority = self._nextPriority()
|
|
self._queue.put(t)
|
|
|
|
self.startWorkers()
|
|
|
|
def addTasksToFront(self, tasks):
|
|
lowest = self.getLowestPrority()
|
|
if lowest is None:
|
|
return self.addTasks(tasks)
|
|
|
|
p = lowest - len(tasks)
|
|
for t in tasks:
|
|
t.priority = p
|
|
self._queue.put(t)
|
|
p += 1
|
|
|
|
self.startWorkers()
|
|
|
|
def startWorkers(self):
|
|
for w in self.workers:
|
|
w.start()
|
|
|
|
def working(self):
|
|
return not self._queue.empty() or self.hasTask()
|
|
|
|
def hasTask(self):
|
|
return any([w.working() for w in self.workers])
|
|
|
|
def getLowestPrority(self):
|
|
lowest = self._queue.lowest()
|
|
if not lowest:
|
|
return None
|
|
|
|
return lowest.priority
|
|
|
|
def moveToFront(self, qitem):
|
|
lowest = self.getLowestPrority()
|
|
if lowest is None:
|
|
return
|
|
|
|
qitem.priority = lowest - 1
|
|
|
|
|
|
class ThreaderManager:
|
|
def __init__(self, worker=BackgroundWorker, worker_count=6):
|
|
self.index = 0
|
|
self.abandoned = []
|
|
self._workerhandler = worker
|
|
self.threader = BackgroundThreader(name=str(self.index),
|
|
worker=worker,
|
|
worker_count=worker_count)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self.threader, name)
|
|
|
|
def reset(self):
|
|
if self.threader._queue.empty() and not self.threader.hasTask():
|
|
return
|
|
|
|
self.index += 1
|
|
self.abandoned.append(self.threader.abort())
|
|
self.threader = BackgroundThreader(name=str(self.index),
|
|
worker=self._workerhandler)
|
|
|
|
def shutdown(self):
|
|
self.threader.shutdown()
|
|
for a in self.abandoned:
|
|
a.shutdown()
|
|
|
|
|
|
BGThreader = ThreaderManager()
|