(self)
| 169 | self._evt = None |
| 170 | |
| 171 | def reset_state(self): |
| 172 | super(MultiThreadMapData, self).reset_state() |
| 173 | if self._threads: |
| 174 | self._threads[0].stop() |
| 175 | for t in self._threads: |
| 176 | t.join() |
| 177 | |
| 178 | self._in_queue = queue.Queue() |
| 179 | self._out_queue = queue.Queue() |
| 180 | self._evt = threading.Event() |
| 181 | self._threads = [MultiThreadMapData._Worker( |
| 182 | self._in_queue, self._out_queue, self._evt, self.map_func) |
| 183 | for _ in range(self.num_thread)] |
| 184 | for t in self._threads: |
| 185 | t.start() |
| 186 | |
| 187 | self._guard = DataFlowReentrantGuard() |
| 188 | |
| 189 | # Call once at the beginning, to ensure inq+outq has a total of buffer_size elements |
| 190 | self._fill_buffer() |
| 191 | |
| 192 | def _recv(self): |
| 193 | return self._out_queue.get() |
nothing calls this directly
no test coverage detected