Tests that a single batched tensor executes together and only once.
(self)
| 46 | # Test for only non eager mode as batching in eager context as a functionality |
| 47 | # is TBD. |
| 48 | def testBasicBatch(self): |
| 49 | """Tests that a single batched tensor executes together and only once.""" |
| 50 | if context.executing_eagerly(): |
| 51 | return |
| 52 | with self.cached_session() as sess: |
| 53 | inp = array_ops.placeholder(dtype=dtypes.int32, shape=[1]) |
| 54 | batched, index, _ = batch_ops.batch( |
| 55 | [inp], num_batch_threads=1, max_batch_size=2, |
| 56 | batch_timeout_micros=36000000, grad_timeout_micros=0, |
| 57 | batching_queue="") |
| 58 | thread_results = [] |
| 59 | |
| 60 | def worker(): |
| 61 | thread_results.extend( |
| 62 | sess.run([batched, index], feed_dict={inp: [1]})) |
| 63 | |
| 64 | worker_thread = threading.Thread(target=worker) |
| 65 | worker_thread.start() |
| 66 | main_results = sess.run([batched, index], feed_dict={inp: [2]}) |
| 67 | worker_thread.join() |
| 68 | |
| 69 | # At this point either the thread or the main did the batch and the other |
| 70 | # should have empty results. |
| 71 | if list(thread_results[0][0]): |
| 72 | batch_t = thread_results[0][0] |
| 73 | index_t = thread_results[1] |
| 74 | empty_b = main_results[0][0] |
| 75 | empty_m = main_results[1] |
| 76 | else: |
| 77 | batch_t = main_results[0][0] |
| 78 | index_t = main_results[1] |
| 79 | empty_b = thread_results[0][0] |
| 80 | empty_m = thread_results[1] |
| 81 | |
| 82 | # Check that both the inputs made it out exactly once. |
| 83 | self.assertAllEqual(sorted(batch_t), (1, 2)) |
| 84 | # Check that we get 2 rows in the index tensor. |
| 85 | self.assertEqual(len(index_t), 2) |
| 86 | # Check that the other ones are empty. |
| 87 | self.assertEqual(len(empty_b), 0) |
| 88 | self.assertEqual(len(empty_m), 0) |
| 89 | |
| 90 | def testBatchWithPadding(self): |
| 91 | """Test that batching with padding up to an allowed batch size works.""" |
nothing calls this directly
no test coverage detected