(self)
| 85 | self.id = id |
| 86 | |
| 87 | def run(self): |
| 88 | with self.default_sess(): |
| 89 | while not self.stopped(): |
| 90 | batched, futures = self.fetch_batch() |
| 91 | try: |
| 92 | outputs = self.func(*batched) |
| 93 | except tf.errors.CancelledError: |
| 94 | for f in futures: |
| 95 | f.cancel() |
| 96 | logger.warn("In PredictorWorkerThread id={}, call was cancelled.".format(self.id)) |
| 97 | return |
| 98 | # print "Worker {} batched {} Queue {}".format( |
| 99 | # self.id, len(futures), self.queue.qsize()) |
| 100 | # debug, for speed testing |
| 101 | # if not hasattr(self, 'xxx'): |
| 102 | # self.xxx = outputs = self.func(batched) |
| 103 | # else: |
| 104 | # outputs = [[self.xxx[0][0]] * len(batched[0]), [self.xxx[1][0]] * len(batched[0])] |
| 105 | |
| 106 | for idx, f in enumerate(futures): |
| 107 | f.set_result([k[idx] for k in outputs]) |
| 108 | |
| 109 | def fetch_batch(self): |
| 110 | """ Fetch a batch of data without waiting""" |
no test coverage detected