Function rpc_proxy_check

tests/python/contrib/test_rpc_proxy.py:28–64

This is a simple test function for RPC Proxy It is not included as pytests, because: - It depends on tornado - It relies on the fact that Proxy starts before client and server connects, which is often the case but not always User can directly run this script to verify correct

()

def rpc_proxy_check():
    """This is a simple test function for RPC Proxy

    It is not included as pytests, because:
    - It depends on tornado
    - It relies on the fact that Proxy starts before client and server connects,
      which is often the case but not always

    User can directly run this script to verify correctness.
    """

    try:
        # pylint: disable=import-outside-toplevel
        from tvm.rpc import proxy

        web_port = 8888
        prox = proxy.Proxy("127.0.0.1", web_port=web_port)

        def check():
            if not tvm.runtime.enabled("rpc"):
                return

            server = multiprocessing.Process(
                target=proxy.websocket_proxy_server, args=(f"ws://localhost:{web_port}/ws", "x1")
            )
            # Need to make sure that the connection start after proxy comes up
            time.sleep(0.1)
            server.deamon = True
            server.start()
            client = rpc.connect(prox.host, prox.port, key="x1")
            f1 = client.get_function("testing.echo")
            assert f1(10) == 10
            assert f1("xyz") == "xyz"

        check()
    except ImportError:
        print("Skipping because tornado is not avaliable...")


if __name__ == "__main__":