get next transformed sample
(self)
| 213 | return abnormal_num == 0 |
| 214 | |
| 215 | def next(self): |
| 216 | """ get next transformed sample |
| 217 | """ |
| 218 | if self._epoch < 0: |
| 219 | self.reset() |
| 220 | |
| 221 | if self.drained(): |
| 222 | raise StopIteration() |
| 223 | |
| 224 | while not self._exit: |
| 225 | try: |
| 226 | sample = self._outq.get(timeout=3) |
| 227 | except Empty as e: |
| 228 | if not self._consumer_healthy(): |
| 229 | raise StopIteration() |
| 230 | else: |
| 231 | continue |
| 232 | |
| 233 | if isinstance(sample, EndSignal): |
| 234 | self._consumer_endsig[sample.id] = sample |
| 235 | logger.warn("recv endsignal from outq with errmsg[{}]" \ |
| 236 | .format(sample.errmsg)) |
| 237 | |
| 238 | if len(self._consumer_endsig.keys()) < len(self._consumers): |
| 239 | self._inq.put(sample) |
| 240 | else: |
| 241 | self._exit = True |
| 242 | raise StopIteration("all consumers exited, no more samples") |
| 243 | else: |
| 244 | self._consumed += 1 |
| 245 | return sample |
| 246 | |
| 247 | raise StopIteration() |
| 248 | |
| 249 | def reset(self): |
| 250 | """ reset for a new epoch of samples |