Tests that batch and unbatch work together.
(self)
| 177 | 0) |
| 178 | |
| 179 | def testBasicUnbatch(self): |
| 180 | """Tests that batch and unbatch work together.""" |
| 181 | if context.executing_eagerly(): |
| 182 | return |
| 183 | with self.cached_session() as sess: |
| 184 | inp = array_ops.placeholder(dtype=dtypes.int32, shape=[1]) |
| 185 | batched, index, id_t = batch_ops.batch( |
| 186 | [inp], num_batch_threads=1, max_batch_size=10, |
| 187 | batch_timeout_micros=100000, # 100ms |
| 188 | allowed_batch_sizes=[3, 10], |
| 189 | grad_timeout_micros=0, batching_queue="") |
| 190 | computation = batched[0] + 1 |
| 191 | result = batch_ops.unbatch(computation, index, id_t, |
| 192 | timeout_micros=1000000, shared_name="unbatch") |
| 193 | thread_results = [] |
| 194 | |
| 195 | def worker(): |
| 196 | thread_results.extend(sess.run([result], feed_dict={inp: [1]})) |
| 197 | |
| 198 | worker_thread = threading.Thread(target=worker) |
| 199 | worker_thread.start() |
| 200 | main_results = sess.run([result], feed_dict={inp: [2]}) |
| 201 | worker_thread.join() |
| 202 | self.assertEqual(thread_results[0], [2]) |
| 203 | self.assertEqual(main_results[0], [3]) |
| 204 | |
| 205 | def testBasicUnbatchDecorated(self): |
| 206 | """Tests that the batch_function decorator works.""" |
nothing calls this directly
no test coverage detected