An offline predictor worker that takes input and produces output by queue. Each process will exit when they see :class:`DIE`.
| 45 | |
| 46 | |
| 47 | class MultiProcessQueuePredictWorker(MultiProcessPredictWorker): |
| 48 | """ |
| 49 | An offline predictor worker that takes input and produces output by queue. |
| 50 | Each process will exit when they see :class:`DIE`. |
| 51 | """ |
| 52 | |
| 53 | def __init__(self, idx, inqueue, outqueue, config): |
| 54 | """ |
| 55 | Args: |
| 56 | idx, config: same as in :class:`MultiProcessPredictWorker`. |
| 57 | inqueue (multiprocessing.Queue): input queue to get data point. elements are (task_id, dp) |
| 58 | outqueue (multiprocessing.Queue): output queue to put result. elements are (task_id, output) |
| 59 | """ |
| 60 | super(MultiProcessQueuePredictWorker, self).__init__(idx, config) |
| 61 | self.inqueue = inqueue |
| 62 | self.outqueue = outqueue |
| 63 | assert isinstance(self.inqueue, multiprocessing.queues.Queue) |
| 64 | assert isinstance(self.outqueue, multiprocessing.queues.Queue) |
| 65 | |
| 66 | def run(self): |
| 67 | self._init_runtime() |
| 68 | while True: |
| 69 | tid, dp = self.inqueue.get() |
| 70 | if tid == DIE: |
| 71 | self.outqueue.put((DIE, None)) |
| 72 | return |
| 73 | else: |
| 74 | self.outqueue.put((tid, self.predictor(*dp))) |
| 75 | |
| 76 | |
| 77 | class PredictorWorkerThread(StoppableThread, ShareSessionThread): |