(self)
| 148 | # self._size = queue.size() |
| 149 | |
| 150 | def run(self): |
| 151 | with self.default_sess(): |
| 152 | try: |
| 153 | self.reinitialize_dataflow() |
| 154 | while True: |
| 155 | # pausable loop |
| 156 | if not self._running.is_set(): |
| 157 | self._running.wait() |
| 158 | |
| 159 | dp = next(self._itr) |
| 160 | feed = _make_feeds(self.placehdrs, dp) |
| 161 | # _, sz = sess.run([self.op, self._sz], feed_dict=feed) |
| 162 | self.op.run(feed_dict=feed) |
| 163 | except (tf.errors.CancelledError, tf.errors.OutOfRangeError): |
| 164 | pass |
| 165 | except DataFlowTerminated: |
| 166 | logger.info("[EnqueueThread] DataFlow has terminated.") |
| 167 | except Exception as e: |
| 168 | if isinstance(e, RuntimeError) and 'closed Session' in str(e): |
| 169 | pass |
| 170 | else: |
| 171 | logger.exception("[EnqueueThread] Exception in thread {}:".format(self.name)) |
| 172 | finally: |
| 173 | try: |
| 174 | self.close_op.run() |
| 175 | except Exception: |
| 176 | pass |
| 177 | logger.info("[EnqueueThread] Thread {} Exited.".format(self.name)) |
| 178 | |
| 179 | def reinitialize_dataflow(self): |
| 180 | self._itr = self.dataflow.__iter__() |
no test coverage detected