Drains a queue in a background thread until the wrapped code unblocks. This can be used to unblock joining a child process that might still write to the queue. The join should be wrapped by this context manager.
(queue)
| 109 | |
| 110 | @contextmanager |
| 111 | def drain_queue_async(queue): |
| 112 | """Drains a queue in a background thread until the wrapped code unblocks. |
| 113 | |
| 114 | This can be used to unblock joining a child process that might still write |
| 115 | to the queue. The join should be wrapped by this context manager. |
| 116 | """ |
| 117 | keep_running = True |
| 118 | |
| 119 | def empty_queue(): |
| 120 | elem_count = 0 |
| 121 | while keep_running: |
| 122 | try: |
| 123 | while True: |
| 124 | queue.get(True, 0.1) |
| 125 | elem_count += 1 |
| 126 | if elem_count < 200: |
| 127 | logging.info('Drained an element from queue.') |
| 128 | except Empty: |
| 129 | pass |
| 130 | except: |
| 131 | logging.exception('Error draining queue.') |
| 132 | |
| 133 | emptier = threading.Thread(target=empty_queue) |
| 134 | emptier.start() |
| 135 | yield |
| 136 | keep_running = False |
| 137 | emptier.join() |
| 138 | |
| 139 | |
| 140 | class ContextPool(): |