Internal method to process a task batch from the given queue. Args: queue: Queue name to be processed Returns: Task IDs: List of tasks that were processed (even if there was an error so that client code can assume the queue
(self, queue: str)
| 528 | return processed_count |
| 529 | |
| 530 | def _process_from_queue(self, queue: str) -> Tuple[List[str], int]: |
| 531 | """ |
| 532 | Internal method to process a task batch from the given queue. |
| 533 | |
| 534 | Args: |
| 535 | queue: Queue name to be processed |
| 536 | |
| 537 | Returns: |
| 538 | Task IDs: List of tasks that were processed (even if there was an |
| 539 | error so that client code can assume the queue is empty |
| 540 | if nothing was returned) |
| 541 | Count: The number of tasks that were attempted to be executed or |
| 542 | -1 if the queue lock couldn't be acquired. |
| 543 | """ |
| 544 | now = time.time() |
| 545 | |
| 546 | log: BoundLogger = self.log.bind(queue=queue) |
| 547 | assert isinstance(log, BoundLogger) |
| 548 | |
| 549 | batch_size = self._get_queue_batch_size(queue) |
| 550 | |
| 551 | queue_lock, failed_to_acquire = self._get_queue_lock(queue, log) |
| 552 | if failed_to_acquire: |
| 553 | return [], -1 |
| 554 | |
| 555 | # Move an item to the active queue, if available. |
| 556 | # We need to be careful when moving unique tasks: We currently don't |
| 557 | # support concurrent processing of multiple unique tasks. If the task |
| 558 | # is already in the ACTIVE queue, we need to execute the queued task |
| 559 | # later, i.e. move it to the SCHEDULED queue (prefer the earliest |
| 560 | # time if it's already scheduled). We want to make sure that the last |
| 561 | # queued instance of the task always gets executed no earlier than it |
| 562 | # was queued. |
| 563 | later = time.time() + self.config["LOCK_RETRY"] |
| 564 | |
| 565 | task_ids = self.scripts.zpoppush( |
| 566 | self._key(QUEUED, queue), |
| 567 | self._key(ACTIVE, queue), |
| 568 | batch_size, |
| 569 | None, |
| 570 | now, |
| 571 | if_exists=("add", self._key(SCHEDULED, queue), later, "min"), |
| 572 | on_success=( |
| 573 | "update_sets", |
| 574 | queue, |
| 575 | self._key(QUEUED), |
| 576 | self._key(ACTIVE), |
| 577 | self._key(SCHEDULED), |
| 578 | ), |
| 579 | ) |
| 580 | log.debug( |
| 581 | "moved tasks", |
| 582 | src_queue=QUEUED, |
| 583 | dest_queue=ACTIVE, |
| 584 | qty=len(task_ids), |
| 585 | ) |
| 586 | |
| 587 | processed_count = 0 |
no test coverage detected