()
| 619 | |
| 620 | |
| 621 | def test_repeat_last(): |
| 622 | @pipeline_def |
| 623 | def pipeline(): |
| 624 | cpu = fn.external_source(name="es_cpu", repeat_last=True) |
| 625 | gpu = fn.external_source(name="es_gpu", repeat_last=True, device="gpu", no_copy=True) |
| 626 | return cpu, gpu |
| 627 | |
| 628 | pipe = pipeline(batch_size=4, num_threads=4, device_id=0, prefetch_queue_depth=1) |
| 629 | data1 = [ |
| 630 | np.array([1], dtype=np.int32), |
| 631 | np.array([3], dtype=np.int32), |
| 632 | np.array([42], dtype=np.int32), |
| 633 | np.array([666], dtype=np.int32), |
| 634 | ] |
| 635 | data2 = [ |
| 636 | np.array([11], dtype=np.int32), |
| 637 | np.array([33], dtype=np.int32), |
| 638 | np.array([422], dtype=np.int32), |
| 639 | np.array([6666], dtype=np.int32), |
| 640 | ] |
| 641 | data1_gpu = to_tensor_list_gpu(data1) |
| 642 | data2_gpu = to_tensor_list_gpu(data2) |
| 643 | pipe.feed_input("es_cpu", data1) |
| 644 | pipe.feed_input("es_gpu", data1_gpu) |
| 645 | a, b = pipe.run() |
| 646 | check_batch(a, data1) |
| 647 | check_batch(b, data1) |
| 648 | a, b = pipe.run() |
| 649 | check_batch(a, data1) |
| 650 | check_batch(b, data1) |
| 651 | |
| 652 | pipe.feed_input("es_cpu", data2) |
| 653 | a, b = pipe.run() |
| 654 | check_batch(a, data2) |
| 655 | check_batch(b, data1) |
| 656 | |
| 657 | pipe.feed_input("es_gpu", data2_gpu) |
| 658 | a, b = pipe.run() |
| 659 | check_batch(a, data2) |
| 660 | check_batch(b, data2) |
| 661 | |
| 662 | pipe.feed_input("es_cpu", data1) |
| 663 | a, b = pipe.run() |
| 664 | check_batch(a, data1) |
| 665 | check_batch(b, data2) |
| 666 | |
| 667 | |
| 668 | def test_repeat_last_queue(): |
nothing calls this directly
no test coverage detected