Add new metadata queue. #715
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,7 +2,7 @@ | |
| import logging | ||
|
|
||
| from PyQt5.QtCore import QObject, QThread, pyqtSlot, pyqtSignal | ||
| from queue import PriorityQueue, Full | ||
| from sdclientapi import API, RequestTimeoutError | ||
| from sqlalchemy.orm import scoped_session | ||
| from typing import Optional, Tuple # noqa: F401 | ||
|
|
@@ -61,11 +61,19 @@ class RunnableQueue(QObject): | |
| ''' | ||
| resume = pyqtSignal() | ||
|
|
||
| """ | ||
| Signal emitted when the queue successfully. | ||
| """ | ||
| pinged = pyqtSignal() | ||
|
|
||
| def __init__(self, api_client: API, session_maker: scoped_session, size: int = 0) -> None: | ||
| """ | ||
| A size of zero means there's no upper bound to the queue size. | ||
| """ | ||
| super().__init__() | ||
| self.api_client = api_client | ||
| self.session_maker = session_maker | ||
| self.queue = PriorityQueue(maxsize=size) # type: PriorityQueue[Tuple[int, ApiJob]] | ||
| # `order_number` ensures jobs with equal priority are retrived in FIFO order. This is needed | ||
| # because PriorityQueue is implemented using heapq which does not have sort stability. For | ||
| # more info, see : https://bugs.python.org/issue17794 | ||
|
|
@@ -81,7 +89,12 @@ def add_job(self, job: ApiJob) -> None: | |
| current_order_number = next(self.order_number) | ||
| job.order_number = current_order_number | ||
| priority = self.JOB_PRIORITIES[type(job)] | ||
| try: | ||
| self.queue.put_nowait((priority, job)) | ||
| except Full: | ||
| # Pass silently if the queue is full. For use with MetadataSyncJob. | ||
| # See #652. | ||
| pass | ||
|
|
||
| def re_add_job(self, job: ApiJob) -> None: | ||
| ''' | ||
|
|
@@ -117,6 +130,7 @@ def process(self) -> None: | |
| try: | ||
| session = self.session_maker() | ||
| job._do_call_api(self.api_client, session) | ||
| self.pinged.emit() | ||
| except (RequestTimeoutError, ApiInaccessibleError) as e: | ||
| logger.debug('Job {} raised an exception: {}: {}'.format(self, type(e).__name__, e)) | ||
| self.add_job(PauseQueueJob()) | ||
|
|
@@ -139,27 +153,36 @@ def __init__(self, api_client: API, session_maker: scoped_session) -> None: | |
|
|
||
| self.main_thread = QThread() | ||
| self.download_file_thread = QThread() | ||
| self.metadata_thread = QThread() | ||
|
|
||
| self.main_queue = RunnableQueue(api_client, session_maker) | ||
| self.download_file_queue = RunnableQueue(api_client, session_maker) | ||
| self.metadata_queue = RunnableQueue(api_client, session_maker, size=1) | ||
|
|
||
| self.main_queue.moveToThread(self.main_thread) | ||
| self.download_file_queue.moveToThread(self.download_file_thread) | ||
| self.metadata_queue.moveToThread(self.metadata_thread) | ||
|
|
||
| self.main_thread.started.connect(self.main_queue.process) | ||
| self.download_file_thread.started.connect(self.download_file_queue.process) | ||
| self.metadata_thread.started.connect(self.metadata_queue.process) | ||
|
|
||
| self.main_queue.paused.connect(self.on_queue_paused) | ||
| self.download_file_queue.paused.connect(self.on_queue_paused) | ||
| self.metadata_queue.paused.connect(self.on_queue_paused) | ||
|
|
||
| self.metadata_queue.pinged.connect(self.resume_queues) | ||
|
There was a problem hiding this comment. this queue doesn't need to pause or resume so you can remove this as well as the pinged signal. instead you can just call resume_queues from the Controller on_sync_success if the queues are paused. There was a problem hiding this comment. I'm not sure I follow. The queue decides itself (see the exception handling in the In which case, I'd argue these signals still need to be there. IYSWIM. There was a problem hiding this comment. So currently, we never stop metadata syncs when they fail because of a request timeout. We want them to run in the background until they succeed, which is why it doesn't make sense to pause the metadata sync queue. It looks like your code will try to enqueue another MetadataSyncJob after 60 seconds but it will be dropped because the queue will be full and paused. The right behavior is to not pause the metadata queue and it makes the code simpler because you can remove the |
||
|
|
||
| def logout(self) -> None: | ||
| self.main_queue.api_client = None | ||
| self.download_file_queue.api_client = None | ||
| self.metadata_queue.api_client = None | ||
|
|
||
| def login(self, api_client: API) -> None: | ||
| logger.debug('Passing API token to queues') | ||
| self.main_queue.api_client = api_client | ||
| self.download_file_queue.api_client = api_client | ||
| self.metadata_queue.api_client = api_client | ||
| self.start_queues() | ||
|
|
||
| def start_queues(self) -> None: | ||
|
|
@@ -171,14 +194,25 @@ def start_queues(self) -> None: | |
| logger.debug('Starting download thread') | ||
| self.download_file_thread.start() | ||
|
|
||
| if not self.metadata_thread.isRunning(): | ||
| logger.debug("Starting metadata thread") | ||
| self.metadata_thread.start() | ||
|
|
||
| def on_queue_paused(self) -> None: | ||
| self.paused.emit() | ||
|
|
||
| def resume_queues(self) -> None: | ||
| logger.info("Resuming queues") | ||
| main_paused = not self.main_thread.isRunning() | ||
| download_paused = not self.download_file_thread.isRunning() | ||
| metadata_paused = not self.metadata_thread.isRunning() | ||
| self.start_queues() | ||
| if main_paused: | ||
| self.main_queue.resume.emit() | ||
| if download_paused: | ||
| self.download_file_queue.resume.emit() | ||
| if metadata_paused: | ||
| self.metadata_queue.resume.emit() | ||
|
|
||
| def enqueue(self, job: ApiJob) -> None: | ||
| # Prevent api jobs being added to the queue when not logged in. | ||
|
|
@@ -192,6 +226,9 @@ def enqueue(self, job: ApiJob) -> None: | |
| if isinstance(job, FileDownloadJob): | ||
| logger.debug('Adding job to download queue') | ||
| self.download_file_queue.add_job(job) | ||
| elif isinstance(job, MetadataSyncJob): | ||
| logger.debug("Adding job to metadata queue") | ||
| self.metadata_queue.add_job(job) | ||
| else: | ||
| logger.debug('Adding job to main queue') | ||
| self.main_queue.add_job(job) | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this time seems good for now, we can increase sync frequency once all sync_api calls have been removed
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
also: you need to remove the line of code that resumes the queues in on_sync_failure now that sync is outside of the main queue and we don't need to unpause the queue in order to run the MetadataSyncJob. We should never pause the metadata sync queue.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
one more note: while offline you can stop the thread and restart it on login, which will take care of this issue: #671 since you can remove the sync_api call on login and instead just start the metadata sync thread
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've removed the queue resumption code as requested. The QTimer will kick in (after a minute) and try another MetadataSyncJob.
See my comment in #671.