(self)
| 615 | |
| 616 | @test_util.run_deprecated_v1 |
| 617 | def testManyThreads(self): |
| 618 | with self.cached_session() as sess: |
| 619 | batch_size = 10 |
| 620 | num_batches = 3 |
| 621 | zero64 = constant_op.constant(0, dtype=dtypes.int64) |
| 622 | |
| 623 | examples = variables.Variable(zero64) |
| 624 | counter = examples.count_up_to(num_batches * batch_size) |
| 625 | sparse_counter = sparse_tensor.SparseTensor( |
| 626 | indices=array_ops.reshape(zero64, [1, 1]), |
| 627 | values=array_ops.stack([math_ops.cast(counter, dtypes.float32)]), |
| 628 | dense_shape=[1]) |
| 629 | batched = inp.batch( |
| 630 | [counter, sparse_counter, "string"], |
| 631 | batch_size=batch_size, |
| 632 | num_threads=4) |
| 633 | self.evaluate(variables.global_variables_initializer()) |
| 634 | variables.local_variables_initializer().run() |
| 635 | threads = queue_runner_impl.start_queue_runners() |
| 636 | |
| 637 | all_counts = [] |
| 638 | for i in range(num_batches): |
| 639 | results = self.evaluate(batched) |
| 640 | tf_logging.info("Batch %d: %s", i, results[0]) |
| 641 | self.assertEqual(len(results[0]), batch_size) |
| 642 | self.assertAllEqual(results[0], results[1].values) |
| 643 | self.assertAllEqual( |
| 644 | results[1].indices, |
| 645 | np.vstack((np.arange(batch_size), np.zeros(batch_size))).T) |
| 646 | self.assertAllEqual(results[1].dense_shape, [batch_size, 1]) |
| 647 | all_counts.extend(results[0]) |
| 648 | self.assertAllEqual(results[2], [b"string"] * batch_size) |
| 649 | self.assertItemsEqual(all_counts, range(num_batches * batch_size)) |
| 650 | |
| 651 | # Reached the limit. |
| 652 | with self.assertRaises(errors_impl.OutOfRangeError): |
| 653 | self.evaluate(batched) |
| 654 | for thread in threads: |
| 655 | thread.join() |
| 656 | |
| 657 | @test_util.run_deprecated_v1 |
| 658 | def testOneThreadSmallerBatch(self): |
nothing calls this directly
no test coverage detected