()
| 666 | |
| 667 | |
| 668 | def test_repeat_last_queue(): |
| 669 | @pipeline_def |
| 670 | def pipeline(): |
| 671 | cpu = fn.external_source(name="es_cpu", repeat_last=True) |
| 672 | gpu = fn.external_source(name="es_gpu", repeat_last=True, device="gpu") |
| 673 | return cpu, gpu |
| 674 | |
| 675 | pipe = pipeline(batch_size=4, num_threads=4, device_id=0, prefetch_queue_depth=2) |
| 676 | data1 = [ |
| 677 | np.array([1], dtype=np.int32), |
| 678 | np.array([3], dtype=np.int32), |
| 679 | np.array([42], dtype=np.int32), |
| 680 | np.array([666], dtype=np.int32), |
| 681 | ] |
| 682 | data2 = [ |
| 683 | np.array([11], dtype=np.int32), |
| 684 | np.array([33], dtype=np.int32), |
| 685 | np.array([422], dtype=np.int32), |
| 686 | np.array([6666], dtype=np.int32), |
| 687 | ] |
| 688 | data3 = data1 |
| 689 | |
| 690 | pipe.feed_input("es_cpu", data1) |
| 691 | pipe.feed_input("es_gpu", data1) |
| 692 | a, b = pipe.run() |
| 693 | check_batch(a, data1) |
| 694 | check_batch(b, data1) |
| 695 | a, b = pipe.run() |
| 696 | check_batch(a, data1) |
| 697 | check_batch(b, data1) |
| 698 | |
| 699 | pipe.feed_input("es_cpu", data2) |
| 700 | a, b = pipe.run() |
| 701 | check_batch(a, data1) # <- still the old value |
| 702 | check_batch(b, data1) |
| 703 | |
| 704 | pipe.feed_input("es_gpu", data3) |
| 705 | a, b = pipe.run() |
| 706 | check_batch(a, data2) # <- new value visible |
| 707 | check_batch(b, data1) # <- still old |
| 708 | |
| 709 | pipe.feed_input("es_cpu", data3) |
| 710 | a, b = pipe.run() |
| 711 | check_batch(a, data2) # <- still 2, the most recent change not visible |
| 712 | check_batch(b, data3) # <- new |
| 713 | |
| 714 | |
| 715 | def _check_repeat_last_var_batch(device): |
nothing calls this directly
no test coverage detected