Test Single Worker Queue. Single worker queues are the same as running with MAX_WORKERS_PER_QUEUE set to 1.
(self)
| 67 | worker2.join() |
| 68 | |
| 69 | def test_single_worker_queue(self): |
| 70 | """ |
| 71 | Test Single Worker Queue. |
| 72 | |
| 73 | Single worker queues are the same as running with MAX_WORKERS_PER_QUEUE |
| 74 | set to 1. |
| 75 | """ |
| 76 | |
| 77 | # Queue two tasks |
| 78 | task = Task(self.tiger, long_task_ok, queue="swq") |
| 79 | task.delay() |
| 80 | task = Task(self.tiger, long_task_ok, queue="swq") |
| 81 | task.delay() |
| 82 | self._ensure_queues(queued={"swq": 2}) |
| 83 | |
| 84 | # Start a worker and wait until it starts processing. |
| 85 | # It should start processing one task and hold a lock on the queue |
| 86 | worker = Process(target=external_worker) |
| 87 | worker.start() |
| 88 | |
| 89 | # Wait for task to start |
| 90 | wait_for_long_task() |
| 91 | |
| 92 | # This worker should fail to get the queue lock and exit immediately |
| 93 | Worker(self.tiger).run(once=True, force_once=True) |
| 94 | self._ensure_queues(active={"swq": 1}, queued={"swq": 1}) |
| 95 | # Wait for external worker |
| 96 | worker.join() |
| 97 | |
| 98 | # Clear out second task |
| 99 | Worker(self.tiger).run(once=True, force_once=True) |
| 100 | self.conn.delete("long_task_ok") |
| 101 | |
| 102 | # Retest using a non-single worker queue |
| 103 | # Queue two tasks |
| 104 | task = Task(self.tiger, long_task_ok, queue="not_swq") |
| 105 | task.delay() |
| 106 | task = Task(self.tiger, long_task_ok, queue="not_swq") |
| 107 | task.delay() |
| 108 | self._ensure_queues(queued={"not_swq": 2}) |
| 109 | |
| 110 | # Start a worker and wait until it starts processing. |
| 111 | # It should start processing one task |
| 112 | worker = Process(target=external_worker) |
| 113 | worker.start() |
| 114 | |
| 115 | # Wait for task to start processing |
| 116 | wait_for_long_task() |
| 117 | |
| 118 | # This worker should process the second task |
| 119 | Worker(self.tiger).run(once=True, force_once=True) |
| 120 | |
| 121 | # Queues should be empty since the first task will have to |
| 122 | # have finished before the second task finishes. |
| 123 | self._ensure_queues() |
| 124 | |
| 125 | worker.join() |
| 126 |