code: def test_duplex_data_transferring():
      r"""
      Test the heterogeneous execution of a simple network where data
      transferring occurs back-and-forth between the target device and host
      processor.
      The host processor is always assumed to be cpu, and the target device
      varies.
      """
      host = "cpu"
      target_host = "llvm"
      host_dev = tvm.device(host)
      if not tvm.runtime.enabled(target_host):
          print("Skip test because llvm is not enabled.")
          return
  
      def check_device(device, target_device):
          if not tvm.runtime.enabled(target_device):
              print(
                  "Skip test because {} is not enabled.".format(
                      target_device
                  )
              )
              return
  
          device_dev = tvm.device(device)
          graph = get_duplex_graph(
              host_dev.device_type, device_dev.device_type
          )
          shape = (4,)
  
          # Insert copy nodes for data transferring between add and sub nodes.
          # Transfers data from gpu to cpu.
          copy_add_sub = te.placeholder(shape, name="__copy0")
          # Transfers data from cpu to gpu.
          copy_sub_add = te.placeholder(shape, name="__copy1")
  
          # Create a module containing adds on the device.
          tensor_a = te.placeholder(shape, name="A")
          tensor_b = te.placeholder(shape, name="B")
          tensor_d = te.placeholder(shape, name="D")
          elemwise_add0 = te.compute(
              shape,
              lambda *i: tensor_a(*i) + tensor_b(*i),
              name="elemwise_add0",
          )
          elemwise_add1 = te.compute(
              shape,
              lambda *i: copy_sub_add(*i) + tensor_d(*i),
              name="elemwise_add1",
          )
          target = topi.cpp.TEST_create_target(device)
          add_schedule0 = topi.cpp.cuda.schedule_injective(
              target, [elemwise_add0]
          )
          lower_add0 = tvm.lower(
              add_schedule0,
              [tensor_a, tensor_b, elemwise_add0],
              name="elemwise_add0",
          )
          add_schedule1 = topi.cpp.cuda.schedule_injective(
              target, [elemwise_add1]
          )
          lower_add1 = tvm.lower(
              add_schedule1,
              [tensor_d, copy_sub_add, elemwise_add1],
              name="elemwise_add1",
          )
          # Create module for sub whose target is the host.
          tensor_c = te.placeholder(shape, name="C")
          elemwise_sub = te.compute(
              shape,
              lambda *i: copy_add_sub(*i) - tensor_c(*i),
              name="elemwise_sub",
          )
          sub_schedule = te.create_schedule(elemwise_sub.op)
          lower_sub = tvm.lower(
              sub_schedule,
              [copy_add_sub, tensor_c, elemwise_sub],
              name="elemwise_sub",
          )
  
          lower_add0.update(lower_add1)
          target_flist = {
              target_device: lower_add0,
              target_host: lower_sub,
          }
          target = tvm.target.Target(target, target_host)
          mhost = tvm.build(target_flist, target=target)
          dev = [host_dev, device_dev]
          params = {}
          params["A"] = tensor_a = np.random.uniform(size=shape).astype(
              tensor_a.dtype
          )
          params["B"] = tensor_b = np.random.uniform(size=shape).astype(
              tensor_b.dtype
          )
          params["C"] = tensor_c = np.random.uniform(size=shape).astype(
              tensor_c.dtype
          )
          params["D"] = tensor_d = np.random.uniform(size=shape).astype(
              tensor_d.dtype
          )
  
          def check_verify():
              mod = graph_executor.create(graph, mhost, dev)
              mod.set_input(**params)
              mod.run()
              out = mod.get_output(0, tvm.nd.empty(shape))
              np.testing.assert_equal(
                  out.numpy(), tensor_a + tensor_b - tensor_c + tensor_d
              )
  
          def check_load_module():
              temp = utils.tempdir()
              path_lib = temp.relpath("deploy.so")
              mhost.export_library(path_lib)
              with open(temp.relpath("deploy.json"), "w") as out_file:
                  out_file.write(graph)
              loaded_lib = tvm.runtime.load_module(path_lib)
              loaded_graph = open(temp.relpath("deploy.json")).read()
              mod = graph_executor.create(loaded_graph, loaded_lib, dev)
              mod.set_input(**params)
              mod.run()
              out = mod.get_output(0, tvm.nd.empty(shape))
              np.testing.assert_equal(
                  out.numpy(), tensor_a + tensor_b - tensor_c + tensor_d
              )
  
          check_verify()
          check_load_module()
  
      dev_tar = {"cuda": "cuda", "opencl": "opencl"}
      for device, target in dev_tar.items():
          with tvm.target.Target(device):
              check_device(device, target)
  
file_path: tvm/tests/python/runtime/test_runtime_heterogeneous.py
imports: import json
  import numpy
  import tvm
  from tvm import te
  from tvm.contrib import graph_executor, utils
  from tvm import topi
label_desc: Use image node_id 4 for calling a global function with example usage: test_duplex_data_transferring() without return types