Commit 2c6c2079 authored by echel0n's avatar echel0n
Browse files

Fixed issue with thread naming for post-processing

parent 5c4b8228
# Changelog
- * dcebc53 - 2017-12-11: refactored remaining GET database calls
- * 8c6be21 - 2017-12-11: Fixed issue with thread naming for post-processing
- * 5c4b822 - 2017-12-11: refactored remaining GET database calls
- * c07404b - 2017-12-11: Fixed issues with SkyTorrents
- * 8923500 - 2017-12-11: Fixed issues with failed and cache database md5 checksums
- * 93c5ad5 - 2017-12-11: Fixed issues with failed and cache database cleanup on new revs
......
......@@ -20,7 +20,7 @@
from __future__ import unicode_literals
try:
from Queue import PriorityQueue, Empty
from Queue import PriorityQueue, Empty, Queue
except ImportError:
from queue import PriorityQueue, Empty
......@@ -42,6 +42,7 @@ class srQueue(threading.Thread):
super(srQueue, self).__init__(name=name)
self.daemon = True
self._queue = PriorityQueue()
self._result_queue = Queue()
self.currentItem = None
self.min_priority = 0
self.amActive = False
......@@ -89,6 +90,7 @@ class srQueue(threading.Thread):
"""
item.added = datetime.datetime.now()
item.name = "{}-{}".format(self.name, item.name)
item.result_queue = self._result_queue
self._queue.put((item.priority, time.time(), item), *args, **kwargs)
return item
......@@ -119,3 +121,5 @@ class srQueueItem(threading.Thread):
self.priority = srQueuePriorities.NORMAL
self.action_id = action_id
self.added = None
self.result = None
self.result_queue = None
......@@ -54,7 +54,8 @@ class PostProcessorQueue(srQueue):
:return: instance of PostProcessorItem or None
"""
for __, __, cur_item in self.queue + [(None, None, self.currentItem)]:
if isinstance(cur_item, PostProcessorItem) and cur_item.dirName == dirName and cur_item.proc_type == proc_type:
if isinstance(cur_item,
PostProcessorItem) and cur_item.dirName == dirName and cur_item.proc_type == proc_type:
return cur_item
return None
......@@ -152,20 +153,18 @@ class PostProcessorQueue(srQueue):
"An item with directory {} is already being processed in the queue, item updated".format(dirName))
return message + "<br\><span class='hidden'>Processing succeeded</span>"
else:
item = PostProcessorItem(dirName, nzbName, process_method, force, is_priority, delete_on, failed, proc_type)
super(PostProcessorQueue, self).put(
PostProcessorItem(dirName, nzbName, process_method, force, is_priority, delete_on, failed, proc_type)
)
if force_next:
with postprocessor_queue_lock:
item.run()
message = item.result
return message
return self._result_queue.get()
else:
super(PostProcessorQueue, self).put(item)
message = logHelper(
"{} post-processing job for {} has been added to the queue".format(proc_type.title(), dirName))
"{} post-processing job for {} has been added to the queue".format(proc_type.title(), dirName)
)
return message + "<br\><span class='hidden'>Processing succeeded</span>"
class PostProcessorItem(srQueueItem):
def __init__(self, dirName, nzbName=None, process_method=None, force=False, is_priority=None, delete_on=False,
failed=False, proc_type="auto"):
......@@ -183,8 +182,6 @@ class PostProcessorItem(srQueueItem):
self.priority = (srQueuePriorities.HIGH, srQueuePriorities.NORMAL)[proc_type == 'auto']
self.result = None
def run(self):
"""
Runs the task
......@@ -213,3 +210,5 @@ class PostProcessorItem(srQueueItem):
sickrage.app.log.debug(traceback.format_exc())
self.result = '{}'.format(traceback.format_exc())
self.result += 'Processing Failed'
self.result_queue.put(self.result)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment