PlexKodiConnect/resources/lib/backgroundthread.py
2019-12-08 16:24:14 +01:00

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()