Tests that multiple batched tensors execute together.
(self)
| 120 | self.assertEqual(len(batch_t), 5) |
| 121 | |
| 122 | def testMultipleBatch(self): |
| 123 | """Tests that multiple batched tensors execute together.""" |
| 124 | if context.executing_eagerly(): |
| 125 | return |
| 126 | with self.cached_session() as sess: |
| 127 | inp0 = array_ops.placeholder(dtype=dtypes.int32, shape=[1]) |
| 128 | inp1 = array_ops.placeholder(dtype=dtypes.int32, shape=[1]) |
| 129 | batched, _, _ = batch_ops.batch( |
| 130 | [inp0, inp1], |
| 131 | num_batch_threads=1, |
| 132 | max_batch_size=2, |
| 133 | batch_timeout_micros=36000000, |
| 134 | grad_timeout_micros=0, |
| 135 | batching_queue="") |
| 136 | thread_results = [] |
| 137 | |
| 138 | def worker(): |
| 139 | thread_results.extend( |
| 140 | sess.run([batched], feed_dict={inp0: [1], |
| 141 | inp1: [2]})) |
| 142 | |
| 143 | worker_thread = threading.Thread(target=worker) |
| 144 | worker_thread.start() |
| 145 | main_results = sess.run([batched], feed_dict={inp0: [2], inp1: [3]}) |
| 146 | worker_thread.join() |
| 147 | |
| 148 | # At this point either the thread or the main did the batch and the other |
| 149 | # should have empty results. |
| 150 | if list(thread_results[0][0]): |
| 151 | batch_t = thread_results[0] |
| 152 | empty_t = main_results[0] |
| 153 | else: |
| 154 | batch_t = main_results[0] |
| 155 | empty_t = thread_results[0] |
| 156 | |
| 157 | # Assert that the tensors were batched together. |
| 158 | self.assertAllEqual(sorted(batch_t[0]), [1, 2]) |
| 159 | self.assertAllEqual(sorted(batch_t[1]), [2, 3]) |
| 160 | self.assertAllEqual(empty_t[0], []) |
| 161 | self.assertAllEqual(empty_t[1], []) |
| 162 | |
| 163 | def testIllegalBatchDifferentDim0Sizes(self): |
| 164 | """Tests illegally feeding tensors with different dim0 sizes.""" |
nothing calls this directly
no test coverage detected