| 230 | return True |
| 231 | |
| 232 | def put(self, chunk, data_input_coordinator): |
| 233 | if not self._validate_chunk(chunk): |
| 234 | return |
| 235 | |
| 236 | while data_input_coordinator.is_active(): |
| 237 | try: |
| 238 | qsize = self._internal_queue.qsize() |
| 239 | if qsize < 2 and (time.time() - self._last_warning) > LOG_INT_SECS: |
| 240 | log.warning("Warning, data loading lagging behind: " + |
| 241 | "queue size={}, name={}".format(qsize, self._input_source_name)) |
| 242 | self._last_warning = time.time() |
| 243 | self._counter += 1 |
| 244 | self._internal_queue.put(chunk, block=True, timeout=0.5) |
| 245 | self._log_inputs_per_interval(chunk[0].shape[0]) |
| 246 | return |
| 247 | except Queue.Full: |
| 248 | log.debug("Queue full: stalling fetchers...") |
| 249 | continue |
| 250 | |
| 251 | def _enqueue_batch_direct(self, data_input_coordinator): |
| 252 | data = self._get(data_input_coordinator) |