(size)
| 820 | @cuda_ipc |
| 821 | @pytest.mark.parametrize("size", [0, 1, 1000]) |
| 822 | def test_IPC(size): |
| 823 | import multiprocessing |
| 824 | ctx = multiprocessing.get_context('spawn') |
| 825 | arr, cbuf = make_random_buffer(size=size, target='device') |
| 826 | ipc_handle = cbuf.export_for_ipc() |
| 827 | handle_buffer = ipc_handle.serialize() |
| 828 | p = ctx.Process(target=other_process_for_test_IPC, |
| 829 | args=(handle_buffer, arr)) |
| 830 | p.start() |
| 831 | p.join() |
| 832 | assert p.exitcode == 0 |
| 833 | |
| 834 | |
| 835 | def test_copy_to(): |
nothing calls this directly
no test coverage detected