Test big endian rpc when there is a PowerPC RPC server available
()
| 68 | |
| 69 | @pytest.mark.skipif(not env.has_rpc(), reason="need rpc") |
| 70 | def test_bigendian_rpc(): |
| 71 | """Test big endian rpc when there is a PowerPC RPC server available""" |
| 72 | host = os.environ.get("TVM_POWERPC_TEST_HOST", None) |
| 73 | port = os.environ.get("TVM_POWERPC_TEST_PORT", 9090) |
| 74 | if host is None: |
| 75 | return |
| 76 | |
| 77 | def verify_rpc(remote, target, shape, dtype): |
| 78 | A = te.placeholder(shape, dtype=dtype) |
| 79 | B = te.compute(A.shape, lambda i: A[i] + tvm.tirx.const(1, A.dtype)) |
| 80 | f = tvm.compile(te.create_prim_func([A, B]), target=target) |
| 81 | |
| 82 | dev = remote.cpu(0) |
| 83 | a = tvm.runtime.tensor(np.random.randint(0, 256, size=shape).astype(A.dtype), device=dev) |
| 84 | b = tvm.runtime.tensor(np.zeros(shape).astype(A.dtype), device=dev) |
| 85 | temp = utils.tempdir() |
| 86 | path_dso = temp.relpath("dev_lib.o") |
| 87 | f.write_to_file(path_dso) |
| 88 | remote.upload(path_dso) |
| 89 | f = remote.load_module("dev_lib.o") |
| 90 | f(a, b) |
| 91 | tvm.testing.assert_allclose(a.numpy() + 1, b.numpy()) |
| 92 | |
| 93 | print("Test RPC connection to PowerPC...") |
| 94 | remote = rpc.connect(host, port) |
| 95 | target = {"kind": "llvm", "mtriple": "powerpc-linux-gnu"} |
| 96 | for dtype in ["float32", "float64", "int32", "int8"]: |
| 97 | verify_rpc(remote, target, (10,), dtype) |
| 98 | |
| 99 | |
| 100 | @pytest.mark.skipif(not env.has_rpc(), reason="need rpc") |
nothing calls this directly
no test coverage detected
searching dependent graphs…