Troplo/PlexKodiConnect
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Add a ton of debug logging for library sync to find deadlock

Browse source
This commit is contained in:
croneter 2021-01-28 10:02:55 +01:00
parent 21788624b9
commit 06f7d88d22
2 changed files with 30 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
resources/lib/backgroundthread.py
View file

@ -143,6 +143,7 @@ class ProcessingQueue(queue.Queue, object):
PKC customization of Queue.put. item needs to be the tuple PKC customization of Queue.put. item needs to be the tuple
(count [int], {'section': [Section], 'xml': [etree xml]}) (count [int], {'section': [Section], 'xml': [etree xml]})
""" """
LOG.debug('Entering ProcessingQueue.put with %s', item)
self.not_full.acquire() self.not_full.acquire()
try: try:
if self.maxsize > 0: if self.maxsize > 0:
@ -151,6 +152,7 @@ class ProcessingQueue(queue.Queue, object):
raise queue.Full raise queue.Full
elif timeout is None: elif timeout is None:
while self._total_qsize() == self.maxsize: while self._total_qsize() == self.maxsize:
LOG.debug('ProcessingQueue._get - waiting in put')
self.not_full.wait() self.not_full.wait()
elif timeout < 0: elif timeout < 0:
raise ValueError("'timeout' must be a non-negative number") raise ValueError("'timeout' must be a non-negative number")
@ -160,14 +162,17 @@ class ProcessingQueue(queue.Queue, object):
remaining = endtime - _time() remaining = endtime - _time()
if remaining <= 0.0: if remaining <= 0.0:
raise queue.Full raise queue.Full
LOG.debug('ProcessingQueue._get - waiting 2 in put')
self.not_full.wait(remaining) self.not_full.wait(remaining)
self._put(item) self._put(item)
self.unfinished_tasks += 1 self.unfinished_tasks += 1
self.not_empty.notify() self.not_empty.notify()
finally: finally:
self.not_full.release() self.not_full.release()
LOG.debug('Exiting ProcessingQueue.put with %s', item)
def _put(self, item): def _put(self, item):
LOG.debug('Entering ProcessingQueue._put with %s', item)
for i, section in enumerate(self._sections): for i, section in enumerate(self._sections):
if item[1]['section'] == section: if item[1]['section'] == section:
self._queues[i]._put(item) self._queues[i]._put(item)
@ -183,6 +188,7 @@ class ProcessingQueue(queue.Queue, object):
Once the get()-method returns None, you've received the sentinel and Once the get()-method returns None, you've received the sentinel and
you've thus exhausted the queue you've thus exhausted the queue
""" """
LOG.debug('Entering ProcessingQueue.add_sentinel with %s', section)
self.not_full.acquire() self.not_full.acquire()
try: try:
section.number_of_items = 1 section.number_of_items = 1
@ -193,6 +199,7 @@ class ProcessingQueue(queue.Queue, object):
self.not_empty.notify() self.not_empty.notify()
finally: finally:
self.not_full.release() self.not_full.release()
LOG.debug('Exiting ProcessingQueue.add_sentinel with %s', section)
def add_section(self, section): def add_section(self, section):
""" """
@ -202,13 +209,16 @@ class ProcessingQueue(queue.Queue, object):
Be sure to set section.number_of_items correctly as it will signal Be sure to set section.number_of_items correctly as it will signal
when processing is completely done for a specific section! when processing is completely done for a specific section!
""" """
LOG.debug('Entering ProcessingQueue.add_section with %s', section)
self.mutex.acquire() self.mutex.acquire()
try: try:
self._add_section(section) self._add_section(section)
finally: finally:
self.mutex.release() self.mutex.release()
LOG.debug('Exiting ProcessingQueue.add_section with %s', section)
def _add_section(self, section): def _add_section(self, section):
LOG.debug('Entering ProcessingQueue._add_section with %s', section)
self._sections.append(section) self._sections.append(section)
self._queues.append( self._queues.append(
OrderedQueue() if section.plex_type == v.PLEX_TYPE_ALBUM OrderedQueue() if section.plex_type == v.PLEX_TYPE_ALBUM
@ -220,18 +230,24 @@ class ProcessingQueue(queue.Queue, object):
""" """
Call only when a section has been completely exhausted Call only when a section has been completely exhausted
""" """
LOG.debug('Entering ProcessingQueue._init_next_section')
self._sections.popleft() self._sections.popleft()
self._queues.popleft() self._queues.popleft()
self._activate_next_section() self._activate_next_section()
def _activate_next_section(self): def _activate_next_section(self):
LOG.debug('Entering ProcessingQueue._activate_next_section')
self._counter = 0 self._counter = 0
self._current_section = self._sections[0] if self._sections else None self._current_section = self._sections[0] if self._sections else None
self._current_queue = self._queues[0] if self._queues else None self._current_queue = self._queues[0] if self._queues else None
def _get(self): def _get(self):
LOG.debug('Entering ProcessingQueue._get')
item = self._current_queue._get() item = self._current_queue._get()
LOG.debug('ProcessingQueue._get - got item %s', item)
self._counter += 1 self._counter += 1
LOG.debug('ProcessingQueue._get - counter %s of %s',
self._counter, self._current_section.number_of_items)
if self._counter == self._current_section.number_of_items: if self._counter == self._current_section.number_of_items:
self._init_next_section() self._init_next_section()
return item[1] return item[1]

14
resources/lib/library_sync/process_metadata.py
View file

@ -24,6 +24,7 @@ class ProcessMetadataThread(common.LibrarySyncMixin,
super(ProcessMetadataThread, self).__init__() super(ProcessMetadataThread, self).__init__()
def start_section(self, section): def start_section(self, section):
LOG.debug('Entering start_section')
if section != self.last_section: if section != self.last_section:
if self.last_section: if self.last_section:
self.finish_last_section() self.finish_last_section()
@ -33,8 +34,10 @@ class ProcessMetadataThread(common.LibrarySyncMixin,
app.SYNC.path_verified = False app.SYNC.path_verified = False
else: else:
LOG.debug('Resume processing section %s', section) LOG.debug('Resume processing section %s', section)
LOG.debug('Exiting start_section')
def finish_last_section(self): def finish_last_section(self):
LOG.debug('Entering finish_last_section')
if (not self.should_cancel() and self.last_section and if (not self.should_cancel() and self.last_section and
self.last_section.sync_successful): self.last_section.sync_successful):
# Check for should_cancel() because we cannot be sure that we # Check for should_cancel() because we cannot be sure that we
@ -48,12 +51,17 @@ class ProcessMetadataThread(common.LibrarySyncMixin,
elif self.last_section and not self.last_section.sync_successful: elif self.last_section and not self.last_section.sync_successful:
LOG.warn('Sync not successful for section %s', self.last_section) LOG.warn('Sync not successful for section %s', self.last_section)
self.successful = False self.successful = False
LOG.debug('Exiting finish_last_section')
def _get(self): def _get(self):
LOG.debug('Entering _get')
item = {'xml': None} item = {'xml': None}
while item and item['xml'] is None: while item and item['xml'] is None:
LOG.debug('_get: getting item')
item = self.processing_queue.get() item = self.processing_queue.get()
LOG.debug('_get: gotten item')
self.processing_queue.task_done() self.processing_queue.task_done()
LOG.debug('Exiting _get')
return item return item
def _run(self): def _run(self):
@ -65,6 +73,7 @@ class ProcessMetadataThread(common.LibrarySyncMixin,
processed = 0 processed = 0
self.start_section(section) self.start_section(section)
while not self.should_cancel(): while not self.should_cancel():
LOG.debug('In loop')
if item is None: if item is None:
break break
elif item['section'] != section: elif item['section'] != section:
@ -72,12 +81,14 @@ class ProcessMetadataThread(common.LibrarySyncMixin,
self.start_section(item['section']) self.start_section(item['section'])
section = item['section'] section = item['section']
with section.context(self.current_time) as context: with section.context(self.current_time) as context:
LOG.debug('Processing %s', section)
while not self.should_cancel(): while not self.should_cancel():
if item is None or item['section'] != section: if item is None or item['section'] != section:
break break
self.update_progressbar(section, self.update_progressbar(section,
item['xml'][0].get('title'), item['xml'][0].get('title'),
section.count) section.count)
LOG.debug('Start add_update')
context.add_update(item['xml'][0], context.add_update(item['xml'][0],
section_name=section.name, section_name=section.name,
section_id=section.section_id, section_id=section.section_id,
@ -86,6 +97,9 @@ class ProcessMetadataThread(common.LibrarySyncMixin,
section.count += 1 section.count += 1
if processed == COMMIT_TO_DB_EVERY_X_ITEMS: if processed == COMMIT_TO_DB_EVERY_X_ITEMS:
processed = 0 processed = 0
LOG.debug('Commiting now')
context.commit() context.commit()
item = self._get() item = self._get()
LOG.debug('Gotten a new item')
LOG.debug('start finish_last_section')
self.finish_last_section() self.finish_last_section()