3. Using ModelRunner

The ModelRunner class is a lightweight wrapper around all the functionality provided by Model Runtime (Section 2, Model Runtime overview) to make it easy for you to deploy models from PopEF files.

There are two steps for running a PopEF model using ModelRunner:

1. Create a ModelRunner object by providing either a list of PopEF files or an instance of the popef::Model class.

   In this step, an IPU is acquired and the model is loaded onto it. All necessary threads and classes are created and stored in the ModelRunner internal state.

   You can set several configuration options, for example replication factor, when you create the ModelRunner object using ModelRunnerConfig.

2. Use one of the execution modes (Section 3.1, Execution modes) to send an inference request to the IPU.

Note

The ModelRunner instance must be preserved until the last inference request returns a result. Destruction of the instance causes the model to be stopped and unloaded from the IPU. The state of the requests that were being processed when the object was destroyed is undefined.

3.1. Execution modes

ModelRunner provides two execution modes: synchronous (execute()) and asynchronous (executeAsync()).

• In the synchronous mode, the request blocks until the result is available.

• In the asynchronous mode, the request is queued and a std::future object is returned. The result can be accessed as soon as the IPU finishes computation.

For both modes, you are responsible for the memory allocation of the input tensors. All execute() and executeAsync() functions take an InputMemoryView parameter that contains pointers to all input data. You must ensure that the input data exists and the pointers are valid until the result is returned.

Each of these execution functions come in two flavours, which differ in how the memory for the output tensors is allocated:

1. ModelRunner allocates memory for the output and returns a TensorMemory instance for each output tensor.

   The corresponding functions are:

   • execute(const InputMemoryView &, unsigned)

   • executeAsync(const InputMemoryView &, unsigned)

2. You allocate the output tensor memory and pass an OutputMemoryView object to the execute function. ModelRunner will place the result in the memory you have provided.

   The corresponding functions are:

   • execute(const InputMemoryView &, const OutputMemoryView &output_data, unsigned)

   • executeAsync(const InputMemoryView &, const OutputMemoryView &output_data, unsigned)

You can find out about the tensors that the model accepts as inputs and returns as outputs by calling one of the following ModelRunner methods:

• getExecuteInputs()

• getExecuteOutputs()

• getModelInputs()

• getModelOutputs().

These methods return a collection of DataDesc objects which contain basic information about the tensor:

• name

• shape

• data type

• size in bytes

The Python and C++ examples below send inference requests to the IPU using all available execution modes.

Note

Files used by the examples in this chapter are listed in the examples appendix. They contain helper functions, for example to process command line arguments.

Python exampleC++ example

Download model_runner_execution_modes.py

Listing 3.1 model_runner_execution_modes.py

#!/usr/bin/env python3
# Copyright (c) 2022 Graphcore Ltd. All rights reserved.

import argparse
from datetime import timedelta
from re import L
import numpy as np
import model_runtime
import popef
"""
The example shows loading a model from PopEF files and sending
inference requests using all available ModelRunner execution modes.
"""


def main():
    parser = argparse.ArgumentParser("Model runner simple example.")
    parser.add_argument(
        "-p",
        "--popef",
        type=str,
        metavar='popef_file_path',
        help="A collection of PopEF files containing the model.",
        nargs='+',
        required=True)
    args = parser.parse_args()

    # Create model runner
    config = model_runtime.ModelRunnerConfig()
    config.device_wait_config = model_runtime.DeviceWaitConfig(
        model_runtime.DeviceWaitStrategy.WAIT_WITH_TIMEOUT,
        timeout=timedelta(seconds=600),
        sleepTime=timedelta(seconds=1))

    print("Creating ModelRunner with", config)
    model_runner = model_runtime.ModelRunner(model_runtime.PopefPaths(
        args.popef),
                                             config=config)

    print("Preparing input tensors:")
    input_descriptions = model_runner.getExecuteInputs()
    input_tensors = [
        np.random.randn(*input_desc.shape).astype(input_desc.numpy_data_type())
        for input_desc in input_descriptions
    ]
    input_view = model_runtime.InputMemoryView()

    for input_desc, input_tensor in zip(input_descriptions, input_tensors):
        print("\tname:", input_desc.name, "shape:", input_tensor.shape,
              "dtype:", input_tensor.dtype)
        input_view[input_desc.name] = input_tensor

    print("Running synchronous execution mode. The memory of the output "
          "tensors is allocated by the ModelRunner object.")
    synchronousExecutionModeLibraryAllocatedOutput(model_runner, input_view)

    print("Running synchronous execution mode. The memory of the output "
          "tensors is allocated by the user.")
    synchronousExecutionModeUserAllocatedOutput(model_runner, input_view)

    print("Running asynchronous execution mode. The memory of the output "
          "tensors is allocated by the ModelRunner object.")
    asynchronousExecutionModeLibraryAllocatedOutput(model_runner, input_view)

    print("Running asynchronous execution mode. The memory of the output "
          "tensors is allocated by the user.")
    asynchronousExecutionModeUserAllocatedOutput(model_runner, input_view)

    input_numpy = dict()
    for input_desc, input_tensor in zip(input_descriptions, input_tensors):
        input_numpy[input_desc.name] = input_tensor

    print("Running synchronous execution mode. The input is a numpy array. "
          "The memory of the output tensors is allocated by the ModelRunner "
          "object.")
    synchronousExecutionModeLibraryAllocatedNumpyInputOutput(
        model_runner, input_numpy)

    print("Running synchronous execution mode. The input and the output are "
          "numpy arrays. The memory of the output tensors is allocated by the "
          "user. ")
    synchronousExecutionModeUserAllocatedNumpyInputOutput(
        model_runner, input_numpy)

    print(
        "Running asynchronous execution mode. The input and the output are "
        "numpy arrays . The memory of the output tensors is allocated by the "
        "ModelRunner object.")
    asynchronousExecutionModeLibraryAllocatedNumpyOutput(
        model_runner, input_numpy)

    print(
        "Running asynchronous execution mode. The input and the output are "
        "numpy arrays . The memory of the output tensors is allocated by the "
        "user.")
    asynchronousExecutionModeUserAllocatedNumpyOutput(model_runner,
                                                      input_numpy)

    print("Success: exiting")
    return 0


def synchronousExecutionModeLibraryAllocatedOutput(model_runner, input_view):
    print("Sending single synchronous request with random data. Output "
          "allocated by ModelRunner.")
    result = model_runner.execute(input_view)

    output_descriptions = model_runner.getExecuteOutputs()
    print("Processing output tensors:")
    for output_desc in output_descriptions:
        output_tensor = np.frombuffer(
            result[output_desc.name],
            dtype=output_desc.numpy_data_type()).reshape(output_desc.shape)
        print("\tname:", output_desc.name, "shape:", output_tensor.shape,
              "dtype:", output_tensor.dtype, "\n", output_tensor)


def synchronousExecutionModeUserAllocatedOutput(model_runner, input_view):

    output_descriptions = model_runner.getExecuteOutputs()
    print("Preparing memory for output tensors")
    output_tensors = [
        np.zeros(output_desc.shape, dtype=output_desc.numpy_data_type())
        for output_desc in output_descriptions
    ]

    print("Creating model_runtime.OutputMemoryView()")
    output_view = model_runtime.OutputMemoryView()
    for desc, tensor in zip(output_descriptions, output_tensors):
        print("\tname:", desc.name, "shape:", tensor.shape, "dtype:",
              tensor.dtype)
        output_view[desc.name] = tensor

    print("Sending single synchronous request with random data")
    model_runner.execute(input_view, output_view)
    print("Processing output tensors:")
    for desc, tensor in zip(output_descriptions, output_tensors):
        print("\tname:", desc.name, "shape", tensor.shape, "dtype",
              tensor.dtype, "\n", tensor)


def synchronousExecutionModeLibraryAllocatedNumpyInputOutput(
        model_runner, numpy_input):

    output_descriptions = model_runner.getExecuteOutputs()

    print("Sending single synchronous request random data (numpy array)")
    output_tensors = model_runner.execute(numpy_input)
    print("Processing output tensors (numpy dict):")
    for desc in output_descriptions:
        tensor = output_tensors[desc.name]
        print("\tname:", desc.name, "shape", tensor.shape, "dtype",
              tensor.dtype, "\n", tensor)


def synchronousExecutionModeUserAllocatedNumpyInputOutput(
        model_runner, numpy_input):

    output_descriptions = model_runner.getExecuteOutputs()
    print("Preparing memory for output tensors")
    numpy_output = {}
    for output_desc in output_descriptions:
        numpy_output[output_desc.name] = np.zeros(
            output_desc.shape, dtype=output_desc.numpy_data_type())

    print("Sending single synchronous request with random data")
    model_runner.execute(numpy_input, numpy_output)
    print("Processing output tensors (numpy dict):")
    for desc in output_descriptions:
        tensor = numpy_output[desc.name]
        print("\tname:", desc.name, "shape", tensor.shape, "dtype",
              tensor.dtype, "\n", tensor)


def asynchronousExecutionModeLibraryAllocatedOutput(model_runner, input_view):

    print("Sending single asynchronous request with random data. Output "
          "allocated by ModelRunner.")
    result = model_runner.executeAsync(input_view)

    print("Waiting for output allocated by ModelRunner:")
    result.wait()
    print("Results available")

    output_descriptions = model_runner.getExecuteOutputs()
    print("Processing output tensors:")
    for output_desc in output_descriptions:
        output_tensor = np.frombuffer(
            result[output_desc.name],
            dtype=output_desc.numpy_data_type()).reshape(output_desc.shape)
        print("\tname:", output_desc.name, "shape:", output_tensor.shape,
              "dtype:", output_tensor.dtype, "\n", output_tensor)


def asynchronousExecutionModeUserAllocatedOutput(model_runner, input_view):
    output_descriptions = model_runner.getExecuteOutputs()
    print("Preparing memory for output tensors")
    output_tensors = [
        np.zeros(output_desc.shape, dtype=output_desc.numpy_data_type())
        for output_desc in output_descriptions
    ]

    print("Creating model_runtime.OutputMemoryView()")
    output_view = model_runtime.OutputMemoryView()
    for desc, tensor in zip(output_descriptions, output_tensors):
        print("\tname:", desc.name, "shape:", tensor.shape, "dtype:",
              tensor.dtype)
        output_view[desc.name] = tensor

    print("Sending single asynchronous request with random data")
    future = model_runner.executeAsync(input_view, output_view)

    print("Waiting for the output.")
    future.wait()
    print("Results available.")
    print("Processing output tensors:")
    for desc, tensor in zip(output_descriptions, output_tensors):
        print("\tname:", desc.name, "shape", tensor.shape, "dtype",
              tensor.dtype, "\n", tensor)


def asynchronousExecutionModeLibraryAllocatedNumpyOutput(
        model_runner, numpy_input):
    print("Sending single asynchronous request with random data")
    future = model_runner.executeAsync(numpy_input)

    print("Waiting for the output.")
    future.wait()
    for desc in model_runner.getExecuteOutputs():
        future_py_array = future[desc.name]

        # Create a np.array copy from the future_py_array buffer
        # using numpy() method.
        tensor = future_py_array.numpy()
        print("\tname:", desc.name, "shape", tensor.shape, "dtype",
              tensor.dtype, "tensor id", id(tensor), "\n", tensor)

        # Create a np.array copy from the future_py_array buffer
        # (allocated by ModelRunner instance).
        tensor_copy = np.array(future_py_array, copy=True)
        print("Tensor copy", tensor_copy, "tensor id", id(tensor_copy))

        # Avoid copying. Create a np.array view from the future_py_array buffer
        # (allocated by ModelRunner instance).
        tensor_view = np.array(future_py_array, copy=False)
        print("Tensor view", tensor_view, "tensor id", id(tensor_view))

        assert not np.shares_memory(tensor_view, tensor_copy)
        assert not np.shares_memory(tensor, tensor_copy)
        assert not np.shares_memory(tensor, tensor_view)


def asynchronousExecutionModeUserAllocatedNumpyOutput(model_runner,
                                                      numpy_input):

    output_descriptions = model_runner.getExecuteOutputs()
    print("Preparing memory for output tensors")
    numpy_output = {}
    for output_desc in output_descriptions:
        numpy_output[output_desc.name] = np.zeros(
            output_desc.shape, dtype=output_desc.numpy_data_type())

    print("Sending single asynchronous request with random data")
    future = model_runner.executeAsync(numpy_input, numpy_output)

    print("Waiting for the output.")
    future.wait()
    print("Results available.")
    print("Processing output tensors:")
    for desc in output_descriptions:
        output_tensor = numpy_output[desc.name]
        future_py_array_view = future[desc.name]

        # Create a np.array view from the future_py_array_view using numpy()
        # method, view points to np.array present in numpy_output dict
        tensor_from_future_object = future_py_array_view.numpy()
        print("\tname:", desc.name, "shape", tensor_from_future_object.shape,
              "dtype", tensor_from_future_object.dtype, "\n",
              tensor_from_future_object)
        assert np.shares_memory(output_tensor, tensor_from_future_object)

        # Create a np.array view from the future_py_array_view buffer, view
        # points to np.array present in numpy_output dict
        tensor_view = np.array(future_py_array_view, copy=False)
        assert np.shares_memory(output_tensor, tensor_view)
        assert np.shares_memory(tensor_from_future_object, tensor_view)

        # Create a np.array copy from the future_py_array_view buffer
        tensor_copy = np.array(future_py_array_view, copy=True)
        assert not np.shares_memory(tensor_from_future_object, tensor_copy)
        assert not np.shares_memory(output_tensor, tensor_copy)


if __name__ == "__main__":
    main()

3.2. Replication

You can specify the replication factor inside the ModelRunnerConfig object passed to the ModelRunner constructor. When the replication factor is set, the ModelRunner object will create the number of IPU model replicas specified, up to the number of available IPUs. The last parameter to each execution function is the ID of the replica to which the inference request will be sent.

The Python and C++ examples below create two replicas and send inference requests to each of them.

Python exampleC++ example

Download model_runner_replication.py

Listing 3.3 model_runner_replication.py

#!/usr/bin/env python3
# Copyright (c) 2022 Graphcore Ltd. All rights reserved.

import argparse
from datetime import timedelta
import numpy as np
import model_runtime
import popef
"""
The example shows loading a model from PopEF files, creating 2 model replicas
and sending inference requests to each of them.
"""


def main():
    parser = argparse.ArgumentParser("Model runner simple example.")
    parser.add_argument(
        "-p",
        "--popef",
        type=str,
        metavar='popef_file_path',
        help="A collection of PopEF files containing the model.",
        nargs='+',
        required=True)
    args = parser.parse_args()

    num_replicas = 2
    # Create model runner
    config = model_runtime.ModelRunnerConfig()
    config.replication_factor = num_replicas
    config.device_wait_config = model_runtime.DeviceWaitConfig(
        model_runtime.DeviceWaitStrategy.WAIT_WITH_TIMEOUT,
        timeout=timedelta(seconds=600),
        sleepTime=timedelta(seconds=1))

    print("Creating ModelRunner with", config)
    runner = model_runtime.ModelRunner(model_runtime.PopefPaths(args.popef),
                                       config=config)

    input_descriptions = runner.getExecuteInputs()

    input = model_runtime.InputMemoryView()

    print("Preparing input tensors:")
    input_descriptions = runner.getExecuteInputs()
    input_tensors = [
        np.random.randn(*input_desc.shape).astype(input_desc.numpy_data_type())
        for input_desc in input_descriptions
    ]
    input_view = model_runtime.InputMemoryView()

    for input_desc, input_tensor in zip(input_descriptions, input_tensors):
        print("\tname:", input_desc.name, "shape:", input_tensor.shape,
              "dtype:", input_tensor.dtype)
        input_view[input_desc.name] = input_tensor

    for replica_id in range(num_replicas):
        print("Sending single synchronous request with empty data - replica",
              replica_id, ".")
        result = runner.execute(input_view, replica_id=replica_id)
        output_descriptions = runner.getExecuteOutputs()

        print("Processing output tensors - replica", replica_id, ":")
        for output_desc in output_descriptions:
            output_tensor = np.frombuffer(
                result[output_desc.name],
                dtype=output_desc.numpy_data_type()).reshape(output_desc.shape)
            print("\tname:", output_desc.name, "shape:", output_tensor.shape,
                  "dtype:", output_tensor.dtype, "\n", output_tensor)

    print("Success: exiting")
    return 0


if __name__ == "__main__":
    main()

3.3. Multithreading

By default, ModelRunner is not thread-safe.

When many threads call execute() or executeAsync(), it can lead to race conditions and undefined behaviour. To avoid this when using ModelRunner in a multithreaded environment, you must ensure that appropriate synchronization mechanisms are used between threads.

The alternative is to set thread_safe in ModelRunnerConfig to true. Every subsequent call of execute() or executeAsync() will cause the internal std::mutex instance to lock.

The examples below first create several threads and then each thread sends inference requests to the IPU.

Python exampleC++ example

Download model_runner_multithreading.py

Listing 3.5 model_runner_multithreading.py

#!/usr/bin/env python3
# Copyright (c) 2022 Graphcore Ltd. All rights reserved.

import argparse
import threading
from datetime import timedelta
import numpy as np
import model_runtime
import popef
"""
The example shows loading a model from PopEF files and sending inference
requests to the same model by multiple threads.
"""


def main():
    parser = argparse.ArgumentParser("Model runner simple example.")
    parser.add_argument(
        "-p",
        "--popef",
        type=str,
        metavar='popef_file_path',
        help="A collection of PopEF files containing the model.",
        nargs='+',
        required=True)
    args = parser.parse_args()

    config = model_runtime.ModelRunnerConfig()
    config.thread_safe = True
    config.device_wait_config = model_runtime.DeviceWaitConfig(
        model_runtime.DeviceWaitStrategy.WAIT_WITH_TIMEOUT,
        timeout=timedelta(seconds=600),
        sleepTime=timedelta(seconds=1))

    print("Creating ModelRunner with", config)
    model_runner = model_runtime.ModelRunner(model_runtime.PopefPaths(
        args.popef),
                                             config=config)
    num_workers = 4
    print("Starting", num_workers, "worker threads.")
    threads = [
        threading.Thread(target=workerMain, args=(model_runner, worker_id))
        for worker_id in range(num_workers)
    ]

    for thread in threads:
        thread.start()

    for thread in threads:
        thread.join()

    print("Success: exiting")
    return 0


def workerMain(model_runner, worker_id):
    print("Worker", worker_id, "Starting workerMain()")
    num_requests = 5

    input_descriptions = model_runner.getExecuteInputs()
    input_requests = []

    print("Worker", worker_id, "Allocating input tensors for", num_requests,
          "requests", input_descriptions)
    for _ in range(num_requests):
        input_requests.append([
            np.random.randn(*input_desc.shape).astype(
                input_desc.numpy_data_type())
            for input_desc in input_descriptions
        ])

    futures = []

    for req_id in range(num_requests):
        print("Worker", worker_id, "Sending asynchronous request. Request id",
              req_id)
        input_view = model_runtime.InputMemoryView()
        for input_desc, input_tensor in zip(input_descriptions,
                                            input_requests[req_id]):
            input_view[input_desc.name] = input_tensor
        futures.append(model_runner.executeAsync(input_view))

    print("Worker", worker_id, "Processing outputs.")
    for req_id, future in enumerate(futures):
        print("Worker", worker_id, "Waiting for the result - request", req_id)
        future.wait()
        print("Worker", worker_id, "Result available - request", req_id)


if __name__ == "__main__":
    main()

3.4. Frozen inputs

The ModelRunner class allows you to bind constant tensor data to input tensors by setting frozen_inputs in ModelRunnerConfig. frozen_inputs is an instance of InputMemoryView and contains a mapping from the input tensor names to the constant tensor data you have allocated. You allocate and pass the pointer to the constant data for the input tensors you want to freeze.

If the tensor to be frozen was required as an input during the execution call, this tensor will no longer be required and the constant tensor from frozen_inputs will instead be added to the request. If the tensor to be frozen was saved as PopEF tensor data or feed data, it will be overridden by the constant tensor from frozen_inputs.

The examples below bind a constant value to one of the inputs and sends inference requests to the IPU.

Note

These examples use a PopEF file generated by the code in Section A.3, Generating an example PopEF file.

Python exampleC++ example

Download model_runner_frozen_inputs.py

Listing 3.7 model_runner_frozen_inputs.py

#!/usr/bin/env python3
# Copyright (c) 2022 Graphcore Ltd. All rights reserved.

import os
import argparse
from datetime import timedelta
import numpy as np
import model_runtime
import popef
"""
The example shows loading a model from PopEF file and binding constant tensor
value to one of the inputs. The example is based on the PopEF file generated
by `model_runtime_example_generate_simple_popef` example. Generated PopEF
file consists simple model:

output = (A * weights) + B

where A and B are stream inputs, weights is a tensor saved as popef::TensorData
and  output is result stream output tensor.
"""


def main():
    parser = argparse.ArgumentParser("Model runner simple example.")
    parser.add_argument(
        "-p",
        "--popef",
        type=str,
        metavar='popef_file_path',
        help="A collection of PopEF files containing the model.",
        nargs='+',
        required=True)
    args = parser.parse_args()
    model = load_model(args.popef)

    frozen_input_name = "tensor_B"
    print("Looking for tensor", frozen_input_name, "inside PopEF model.")
    tensor_b_anchor = popef.Anchor()

    for anchor in model.metadata.anchors():
        if anchor.name() == frozen_input_name:
            tensor_b_anchor = anchor
            break
    else:
        raise Exception(f'Anchor {frozen_input_name} not found inside givem '
                        'model. Please make sure that PopEF was generated by '
                        '`model_runtime_example_generate_simple_popef`')

    print("Generating", frozen_input_name, "random values")
    tensor_b_info = tensor_b_anchor.tensorInfo()
    tensor_b = np.random.randn(*tensor_b_info.shape()).astype(
        tensor_b_info.numpyDType())

    config = model_runtime.ModelRunnerConfig()

    frozen_inputs = model_runtime.InputMemoryView()
    frozen_inputs[frozen_input_name] = tensor_b
    config.frozen_inputs = frozen_inputs

    print(
        "Tensor", frozen_input_name, "is frozen - will be treated as "
        "constant in each execution request.")
    config.device_wait_config = model_runtime.DeviceWaitConfig(
        model_runtime.DeviceWaitStrategy.WAIT_WITH_TIMEOUT,
        timeout=timedelta(seconds=600),
        sleepTime=timedelta(seconds=1))

    model_runner = model_runtime.ModelRunner(model, config=config)

    print("Preparing input tensors:")
    input_descriptions = model_runner.getExecuteInputs()
    input_tensors = [
        np.random.randn(*input_desc.shape).astype(input_desc.numpy_data_type())
        for input_desc in input_descriptions
    ]
    input_view = model_runtime.InputMemoryView()

    for input_desc, input_tensor in zip(input_descriptions, input_tensors):
        print("\tname:", input_desc.name, "shape:", input_tensor.shape,
              "dtype:", input_tensor.dtype)
        input_view[input_desc.name] = input_tensor

    print("Sending single synchronous request with empty data.")
    result = model_runner.execute(input_view)
    output_descriptions = model_runner.getExecuteOutputs()

    print("Processing output tensors:")
    for output_desc in output_descriptions:
        output_tensor = np.frombuffer(
            result[output_desc.name],
            dtype=output_desc.numpy_data_type()).reshape(output_desc.shape)
        print("\tname:", output_desc.name, "shape:", output_tensor.shape,
              "dtype:", output_tensor.dtype, "\n", output_tensor)

    print("Success: exiting")

    return 0


def load_model(popef_paths):
    for model_file in popef_paths:
        assert os.path.isfile(model_file) is True
        reader = popef.Reader()
        reader.parseFile(model_file)

        meta = reader.metadata()
        exec = reader.executables()
        return popef.ModelBuilder(reader).createModel()


if __name__ == "__main__":
    main()

3.5. Conditional execution

When a compiled graph contains one model, you must provide data for all input and output anchors for correct execution of the graph on the IPU. This is the default execution for a ModelRunner object. Sometimes, a graph compiled by Poplar can contain multiple models (Fig. 3.1). In this case, the IPU will conditionally execute one of these models based on an input parameter (specified in an input anchor) that selects the appropriate execution path. In order to execute different models through one ModelRunner instance, you must provide data for only the anchors that are required by the IPU to perform the request.

There are two steps you must perform:

• You must disable the validate_io_params option. This option checks if you have provided enough input parameters to complete the request. In other words, it checks that each parameter corresponds to an existing internal queue. By disabling it, you will be able to provide inputs for only the model you want to execute and not all models.

• You must know which anchors are associated with each model in the graph. When you send a request, you need to provide the data only required by the model you want to execute. The PopEF file contains information about the anchors that the graph contains, but it does not contain information about which anchors are associated with a model. You can use the popef_dump tool to display the anchors in the graph.

Note

If you provide inputs for a different model than the one you want to execute, then a timeout will occur. This will be the only information you will receive in the case of an incorrect execution of the request.

Fig. 3.1 An example graph construction with multiple models

3.6. Dynamic batch sizing

By default, ModelRunner only accepts model inputs and outputs with batch sizes that are an integer multiple of the batch size from the PopEF model. By setting the batching_dim configuration option, you can enable dynamic batch sizing which allows you to specify any batch size.

By default, dynamic batch sizing is disabled (batching_dim == 0xFFFFFFFF). In this case, the batch size must be an integer multiple of the batch size from the PopEF model. For example, without dynamic batch sizing enabled and for a PopEF model with an input shape of [3, 4, 2] where dimension 0 specifies the batch size, ModelRunner can only accept inputs with shapes defined by [N * 3, 4, 2], (where N is any positive integer).

To enable dynamic batch sizing, we set batching_dim to the dimension that contains the value for the batch size. The value of batching_dim must be an integer between 0 and max_dimension_model-1.

For example to enable dynamic batch sizing on dimension 1, you would set batching_dim == 1. Now dimension 1 can contain any batch size. Using the example from earlier, ModelRunner can service inference requests with input and output shapes [4, batch_size, 2] where batch_size is any positive integer.

3.7. Improved model fusion and I/O overlap performance

Model fusion is a concept in machine learning where the inputs and outputs of multiple small models are combined to improve the overall predictive performance for a specific problem. While Model Runtime can use the individual inputs and outputs of the smaller models as inputs and outputs into the single large model and can conditionally execute a branch related to a single (small) model, problems arise when input data is unavailable. In this case, the default Model Runtime behaviour is to flush all inputs and outputs. This degrades performance for fused models as this will execute the branches for all the different small models.

A similar situation occurs if I/O overlap is being used to improve model throughput. If there is not enough data to fill a batch, then there is a timeout. This also causes all data to be flushed.

To handle flushing of specific data in fused models or to flush some data during I/O overlap, ModelRunnerConfig contains flush_callback, a configuration option that specifies a pointer to a callback function, which Model Runtime calls once, after a timeout occurs when expected data is unavailable.

This callback function provides a mechanism to specify which input and output data is changed and how the data is changed. Input and output data can either be flushed (with null data) or the values can be updated.

The callback function must have the following parameters:

• tensor_id: The ID of the tensor that ModelRunner was expecting when the timeout occurred.

• inputs: The pointer to the updated input data structure. ModelRunner expects to find the updated input data here after the callback returns and will then update its input queue.

• outputs: The pointer to the updated output data structure. ModelRunner expects to find the updated output data here after the callback returns and will then update its output queue.

You must assign a pointer to the callback function to the flush_callback configuration option.

For example, in your application, you would define:

ModelRunnerConfig config;
if (batch_size == 3)
  config.flush_callback = [&inputs_to_be_flushed, &outputs_to_be_flushed](
                              const std::string &tensor_id,
                              const InputMemoryView *&inputs,
                              const OutputMemoryView *&outputs) {
    BOOST_TEST_MESSAGE(
        "flush_callback called without any change for tensor: "
        << tensor_id);
    inputs = &inputs_to_be_flushed;
    outputs = &outputs_to_be_flushed;
    return;
  };
else if (batch_size == 7)
  config.flush_callback = [](const std::string &tensor_id,
                             const InputMemoryView *&inputs,
                             const OutputMemoryView *&outputs) {
    BOOST_TEST_MESSAGE("Suppress the compile warning inputs: "
                       << inputs << "outputs:" << outputs);
    BOOST_TEST_MESSAGE(
        "Return immediately in flush_callback for tensor: " << tensor_id);
    return;
  };

Note

The application defining the callback function must ensure that the unavailable tensor and the input and output data to be changed or flushed belong to the small model that is currently being executed. If not, inference results may become corrupted.

3.8. Monitoring statistics

ModelRunner can monitor your running inference applications and this can be easily integrated into your server framework and monitoring system. Statistics are available for the phases shown in Fig. 3.2.

Fig. 3.2 The phases that monitoring statistics are available for

You can collect the following statistics on the online server while running inference:

• percentile of the latency in each phase of the request with the getMonitoringStatisticsPercentile() funciton.

• time taken for different phases of the last request with the getTimeTrace() function.

• mean latency in each phase with the getMonitoringStatisticsMean() function.

• total number of requests in each phase with the getMonitoringStatisticsTotalCount() function.

The collection of these metrics does not have any impact on the inference performance.

To use this in ModelRunner, you need to first set the following in ModelRunnerConfig:

• flush_on_waiting_outputs to true.

• request_tracepoints_buffer_size to the size of the request tracepoint buffer. The default value is 1,000. If set to 0, the tracepoint buffer size is infinite.

Then you can call any of the monitoring functions. For example to get the P99.9 latency, you can call getMonitoringStatisticsPercentile(ms_info, 0.999) and to get the mean of the latencies, you can call getMonitoringStatisticsMean(ms_info) where ms_info is a dictionary that contains the monitoring information. You need to pass in an empty dictionary in Python or std::map<std::string, float> in C++. The results returned, for example for Python, are as follows:

• getMonitoringStatisticsPercentile() returns:

  {
  "computation_monitoring_statistics_percentile_us":  <value>,
  "read_queue_monitoring_statistics_percentile_us" :  <value>,
  "request_monitoring_statistics_percentile_us" :  <value>,
  "request_monitoring_statistics_percentile_us" :  <value>,
  }
  

• getTimeTrace() returns:

  {
  "request_duration_us":  <value>,
  "read_preparation_duration_us" :  <value>,
  "read_queue_duration_us" :  <value>,
  "computation_duration_us" :  <value>,
  }
  

• getMonitoringStatisticsMean() returns:

  {
  "computation_monitoring_statistics_mean_us":  <value>,
  "read_preparation_monitoring_statistics_mean_us" :  <value>,
  "read_queue_monitoring_statistics_mean_us" :  <value>,
  "request_monitoring_statistics_mean_us" :  <value>,
  }
  

• getMonitoringStatisticsTotalCount() returns:

  {
  "read_preparation_monitoring_statistics_count":  <value>,
  "read_queue_monitoring_statistics_count" :  <value>,
  "computation_monitoring_statistics_count" :  <value>,
  "request_monitoring_statistics_count" :  <value>,
  }