@@ 163-173 (lines=11) @@ | ||
160 | self.priority_queue.put(task) |
|
161 | self.mutex.release() |
|
162 | ||
163 | def _check_processing(self): |
|
164 | now = time.time() |
|
165 | self.mutex.acquire() |
|
166 | while self.processing.qsize() and self.processing.top and self.processing.top.exetime < now: |
|
167 | task = self.processing.get_nowait() |
|
168 | if task.taskid is None: |
|
169 | continue |
|
170 | task.exetime = 0 |
|
171 | self.priority_queue.put(task) |
|
172 | logger.info("processing: retry %s", task.taskid) |
|
173 | self.mutex.release() |
|
174 | ||
175 | def put(self, taskid, priority=0, exetime=0): |
|
176 | '''Put a task into task queue''' |
|
@@ 154-161 (lines=8) @@ | ||
151 | self._check_time_queue() |
|
152 | self._check_processing() |
|
153 | ||
154 | def _check_time_queue(self): |
|
155 | now = time.time() |
|
156 | self.mutex.acquire() |
|
157 | while self.time_queue.qsize() and self.time_queue.top and self.time_queue.top.exetime < now: |
|
158 | task = self.time_queue.get_nowait() |
|
159 | task.exetime = 0 |
|
160 | self.priority_queue.put(task) |
|
161 | self.mutex.release() |
|
162 | ||
163 | def _check_processing(self): |
|
164 | now = time.time() |