PlexKodiConnect/resources/lib/backgroundthread.py

545 lines
16 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
from . import utils, app, variables as v
WORKER_COUNT = 3
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()
def unblock_callers(self):
"""
Ensures that any other thread that requested this thread's suspension
is released
"""
self._suspension_reached.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
add_sentinel()
"""
def _init(self, maxsize):
self.queue = deque()
self._sections = deque()
self._queues = deque()
self._current_section = None
self._current_queue = None
# Item-index for the currently active queue
self._counter = 0
def _qsize(self):
return self._current_queue._qsize() if self._current_queue else 0
def _total_qsize(self):
"""
This method is BROKEN as it can lead to a deadlock when a single item
from the current section takes longer to download then any new items
coming in
"""
return sum(q._qsize() for q in self._queues) if self._queues else 0
def put(self, item, block=True, timeout=None):
"""
PKC customization of Queue.put. item needs to be the tuple
(count [int], {'section': [Section], 'xml': [etree xml]})
"""
self.not_full.acquire()
try:
if self.maxsize > 0:
if not block:
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)
self._put(item)
self.unfinished_tasks += 1
self.not_empty.notify()
finally:
self.not_full.release()
def _put(self, item):
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])
def add_sentinel(self, section):
"""
Adds a new empty section as a sentinel. Call with an empty Section()
object. Call this method immediately after having added all sections
with add_section().
Once the get()-method returns None, you've received the sentinel and
you've thus exhausted the queue
"""
self.not_full.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
self.not_empty.notify()
finally:
self.not_full.release()
def add_section(self, section):
"""
Add a new Section() to this Queue. Each section will be entirely
processed before moving on to the next section.
Be sure to set section.number_of_items correctly as it will signal
when processing is completely done for a specific section!
"""
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._activate_next_section()
def _init_next_section(self):
"""
Call only when a section has been completely exhausted
"""
self._sections.popleft()
self._queues.popleft()
self._activate_next_section()
def _activate_next_section(self):
self._counter = 0
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 app.APP.monitor.abortRequested()
def isValid(self):
return not self.finished and not self._canceled
class ShutdownSentinel(Task):
def run(self):
pass
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(notify=True)
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(notify=True)
def abort(self):
self._abort = True
return self
def aborted(self):
return self._abort or app.APP.monitor.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, block=True):
self.abort()
if self._task:
self._task.cancel()
if block and 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):
LOG.debug('Starting Worker %s', self.name)
while not self.aborted():
self._task = self._queue.get()
if self._task is ShutdownSentinel:
break
self._working = True
self._runTask(self._task)
self._working = False
self._queue.task_done()
self._task = None
LOG.debug('Exiting Worker %s', self.name)
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 app.APP.monitor.abortRequested()
def shutdown(self, block=True):
self.abort()
self.addTasksToFront([ShutdownSentinel() for _ in self.workers])
for w in self.workers:
w.shutdown(block)
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=NonstoppingBackgroundWorker,
worker_count=WORKER_COUNT):
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, block=True):
self.threader.shutdown(block)
for a in self.abandoned:
a.shutdown(block)
BGThreader = ThreaderManager()