mirror of
https://github.com/saymrwulf/onnxruntime.git
synced 2026-07-21 19:18:55 +00:00
c444b9d76a
* Add CUDA option to run copy in default stream This change fixes #4829. Thanks @maherzog for providing the repro! The bug is caused by memory reuse in BFC arena, where copy and compute stream in CUDA has a racing condition. BFC arena is an arena allocator on top of cudaMalloc/Free to reduce the cost in syncing CPU and GPU when alloc/free. It means when CPU alloc/free the memory, GPU might not finished previous work on the memory, so that CPU and GPU could run asynchronously. This is OK if there's only one stream, where the execution order in CPU and GPU are consistent. For example, if we have two kernels A and B, CPU runs allocA->computeA->freeA->allocB->computeB->freeB, A and B could shares the same memory since computeA and computeB will not have racing as long as they run in the same GPU compute stream. However, if CPU runs allocA->CopyA->freeA->allocB->computeB->freeB, the order of execution in GPU could have copyA happen after computeB, if copy and compute happens in different GPU streams. This change makes copy to run in default compute stream, while adding an option to fall back to previous behavior if there's perf hit. This is a short term fix before BFC arena could support multiple streams. User may use following options to revert to previous behavior: C API: struct OrtCUDAProviderOptions cudaProviderOpt; cudaProviderOpt.do_copy_in_default_stream = false; C++ API: CUDAExecutionProviderInfo cudaEPInfo; cudaEPInfo.do_copy_in_default_stream = false; C# API: pending... Python: import onnxruntime onnxruntime.capi._pybind_state.set_do_copy_in_default_stream(False) * Confirmed the test failes in CI when doing copy in separate stream Revert the test to get CI pass now * Fix Windows test * Address CR
788 lines
37 KiB
Python
788 lines
37 KiB
Python
# Copyright (c) Microsoft Corporation. All rights reserved.
|
|
# Licensed under the MIT License.
|
|
|
|
# -*- coding: UTF-8 -*-
|
|
import unittest
|
|
import os
|
|
import numpy as np
|
|
import onnxruntime as onnxrt
|
|
import threading
|
|
import sys
|
|
from helper import get_name
|
|
|
|
class TestInferenceSession(unittest.TestCase):
|
|
|
|
def run_model(self, session_object, run_options):
|
|
x = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
input_name = session_object.get_inputs()[0].name
|
|
res = session_object.run([], {input_name: x}, run_options=run_options)
|
|
output_expected = np.array([[1.0, 4.0], [9.0, 16.0], [25.0, 36.0]], dtype=np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
|
|
def testModelSerialization(self):
|
|
so = onnxrt.SessionOptions()
|
|
so.log_verbosity_level = 1
|
|
so.logid = "TestModelSerialization"
|
|
so.optimized_model_filepath = "./PythonApiTestOptimizedModel.onnx"
|
|
onnxrt.InferenceSession(get_name("mul_1.onnx"), sess_options=so)
|
|
self.assertTrue(os.path.isfile(so.optimized_model_filepath))
|
|
|
|
def testGetProviders(self):
|
|
self.assertTrue('CPUExecutionProvider' in onnxrt.get_available_providers())
|
|
# get_all_providers() returns the default EP order from highest to lowest.
|
|
# CPUExecutionProvider should always be last.
|
|
self.assertTrue('CPUExecutionProvider' == onnxrt.get_all_providers()[-1])
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
self.assertTrue('CPUExecutionProvider' in sess.get_providers())
|
|
|
|
def testSetProviders(self):
|
|
if 'CUDAExecutionProvider' in onnxrt.get_available_providers():
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
# confirm that CUDA Provider is in list of registered providers.
|
|
self.assertTrue('CUDAExecutionProvider' in sess.get_providers())
|
|
# reset the session and register only CPU Provider.
|
|
sess.set_providers(['CPUExecutionProvider'])
|
|
# confirm only CPU Provider is registered now.
|
|
self.assertEqual(['CPUExecutionProvider'], sess.get_providers())
|
|
|
|
def testSetProvidersWithOptions(self):
|
|
if 'CUDAExecutionProvider' in onnxrt.get_available_providers():
|
|
import sys
|
|
import ctypes
|
|
CUDA_SUCCESS = 0
|
|
|
|
def runBaseTest1():
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
self.assertTrue('CUDAExecutionProvider' in sess.get_providers())
|
|
|
|
option1 = {'device_id': 0}
|
|
sess.set_providers(['CUDAExecutionProvider'], [option1])
|
|
self.assertEqual(['CUDAExecutionProvider', 'CPUExecutionProvider'], sess.get_providers())
|
|
option2 = {'device_id': -1}
|
|
with self.assertRaises(RuntimeError):
|
|
sess.set_providers(['CUDAExecutionProvider'], [option2])
|
|
sess.set_providers(['CUDAExecutionProvider', 'CPUExecutionProvider'], [option1, {}])
|
|
self.assertEqual(['CUDAExecutionProvider', 'CPUExecutionProvider'], sess.get_providers())
|
|
|
|
def runBaseTest2():
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
self.assertTrue('CUDAExecutionProvider' in sess.get_providers())
|
|
|
|
# test get/set of "cuda_mem_limit" configuration.
|
|
options = sess.get_provider_options()
|
|
self.assertTrue('CUDAExecutionProvider' in options)
|
|
option = options['CUDAExecutionProvider']
|
|
self.assertTrue('cuda_mem_limit' in option)
|
|
ori_mem_limit = option['cuda_mem_limit']
|
|
new_mem_limit = int(ori_mem_limit) // 2
|
|
option['cuda_mem_limit'] = new_mem_limit
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
options = sess.get_provider_options()
|
|
self.assertEqual(options['CUDAExecutionProvider']['cuda_mem_limit'], str(new_mem_limit))
|
|
|
|
option['cuda_mem_limit'] = ori_mem_limit
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
options = sess.get_provider_options()
|
|
self.assertEqual(options['CUDAExecutionProvider']['cuda_mem_limit'], ori_mem_limit)
|
|
|
|
# test get/set of "arena_extend_strategy" configuration.
|
|
options = sess.get_provider_options()
|
|
self.assertTrue('CUDAExecutionProvider' in options)
|
|
option = options['CUDAExecutionProvider']
|
|
self.assertTrue('arena_extend_strategy' in option)
|
|
for strategy in ['kNextPowerOfTwo', 'kSameAsRequested']:
|
|
option['arena_extend_strategy'] = strategy
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
options = sess.get_provider_options()
|
|
self.assertEqual(options['CUDAExecutionProvider']['arena_extend_strategy'], strategy)
|
|
|
|
#
|
|
# Note: Tests that throw an exception leave an empty session due to how set_providers currently works,
|
|
# so run them last. Each set_providers call will attempt to re-create a session, so it's
|
|
# fine for a test that fails to run immediately after another one that fails.
|
|
# Alternatively a valid call to set_providers could be used to recreate the underlying session
|
|
# after a failed call.
|
|
#
|
|
option['arena_extend_strategy'] = 'wrong_value'
|
|
with self.assertRaises(RuntimeError):
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
|
|
option['cuda_mem_limit'] = -1024
|
|
with self.assertRaises(RuntimeError):
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
|
|
option['cuda_mem_limit'] = 1024.1024
|
|
with self.assertRaises(RuntimeError):
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
|
|
option['cuda_mem_limit'] = 'wrong_value'
|
|
with self.assertRaises(RuntimeError):
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
|
|
def getCudaDeviceCount():
|
|
import ctypes
|
|
|
|
num_device = ctypes.c_int()
|
|
result = ctypes.c_int()
|
|
error_str = ctypes.c_char_p()
|
|
|
|
result = cuda.cuInit(0)
|
|
result = cuda.cuDeviceGetCount(ctypes.byref(num_device))
|
|
if result != CUDA_SUCCESS:
|
|
cuda.cuGetErrorString(result, ctypes.byref(error_str))
|
|
print("cuDeviceGetCount failed with error code %d: %s" % (result, error_str.value.decode()))
|
|
return -1
|
|
|
|
return num_device.value
|
|
|
|
def setDeviceIdTest(i):
|
|
import ctypes
|
|
import onnxruntime as onnxrt
|
|
|
|
device = ctypes.c_int()
|
|
result = ctypes.c_int()
|
|
error_str = ctypes.c_char_p()
|
|
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
option = {'device_id': i}
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
self.assertEqual(['CUDAExecutionProvider', 'CPUExecutionProvider'], sess.get_providers())
|
|
result = cuda.cuCtxGetDevice(ctypes.byref(device))
|
|
if result != CUDA_SUCCESS:
|
|
cuda.cuGetErrorString(result, ctypes.byref(error_str))
|
|
print("cuCtxGetDevice failed with error code %d: %s" % (result, error_str.value.decode()))
|
|
|
|
self.assertEqual(result, CUDA_SUCCESS)
|
|
self.assertEqual(i, device.value)
|
|
|
|
def runAdvancedTest():
|
|
num_device = getCudaDeviceCount()
|
|
if num_device < 0:
|
|
return
|
|
|
|
# Configure session to be ready to run on all available cuda devices
|
|
for i in range(num_device):
|
|
setDeviceIdTest(i)
|
|
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
|
|
# configure session with not legit option values and that shloud fail
|
|
with self.assertRaises(RuntimeError):
|
|
option = {'device_id': num_device}
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
option = {'device_id': 'non_legit_value'}
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
|
|
# configure session with not legit option should cause no effect
|
|
option = {'device_id': 0}
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
option = {'non_legit_option': num_device}
|
|
sess.set_providers(['CUDAExecutionProvider'], [option])
|
|
self.assertEqual(['CUDAExecutionProvider', 'CPUExecutionProvider'], sess.get_providers())
|
|
|
|
|
|
|
|
libnames = ('libcuda.so', 'libcuda.dylib', 'cuda.dll')
|
|
for libname in libnames:
|
|
try:
|
|
cuda = ctypes.CDLL(libname)
|
|
runBaseTest1()
|
|
runBaseTest2()
|
|
runAdvancedTest()
|
|
|
|
except OSError:
|
|
continue
|
|
else:
|
|
break
|
|
else:
|
|
runBaseTest1()
|
|
runBaseTest2()
|
|
# raise OSError("could not load any of: " + ' '.join(libnames))
|
|
|
|
def testInvalidSetProviders(self):
|
|
with self.assertRaises(ValueError) as context:
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
sess.set_providers(['InvalidProvider'])
|
|
self.assertTrue(
|
|
'[\'InvalidProvider\'] does not contain a subset of available providers' in str(context.exception))
|
|
|
|
def testSessionProviders(self):
|
|
if 'CUDAExecutionProvider' in onnxrt.get_available_providers():
|
|
# create session from scratch, but constrain it to only use the CPU.
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"), providers=['CPUExecutionProvider'])
|
|
self.assertEqual(['CPUExecutionProvider'], sess.get_providers())
|
|
|
|
def testRunModel(self):
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
x = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
input_name = sess.get_inputs()[0].name
|
|
self.assertEqual(input_name, "X")
|
|
input_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(input_shape, [3, 2])
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "Y")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [3, 2])
|
|
res = sess.run([output_name], {input_name: x})
|
|
output_expected = np.array([[1.0, 4.0], [9.0, 16.0], [25.0, 36.0]], dtype=np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
|
|
def testRunModelFromBytes(self):
|
|
with open(get_name("mul_1.onnx"), "rb") as f:
|
|
content = f.read()
|
|
sess = onnxrt.InferenceSession(content)
|
|
x = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
input_name = sess.get_inputs()[0].name
|
|
self.assertEqual(input_name, "X")
|
|
input_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(input_shape, [3, 2])
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "Y")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [3, 2])
|
|
res = sess.run([output_name], {input_name: x})
|
|
output_expected = np.array([[1.0, 4.0], [9.0, 16.0], [25.0, 36.0]], dtype=np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
|
|
def testRunModel2(self):
|
|
sess = onnxrt.InferenceSession(get_name("matmul_1.onnx"))
|
|
x = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
input_name = sess.get_inputs()[0].name
|
|
self.assertEqual(input_name, "X")
|
|
input_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(input_shape, [3, 2])
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "Y")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [3, 1])
|
|
res = sess.run([output_name], {input_name: x})
|
|
output_expected = np.array([[5.0], [11.0], [17.0]], dtype=np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
|
|
def testRunModel2Contiguous(self):
|
|
sess = onnxrt.InferenceSession(get_name("matmul_1.onnx"))
|
|
x = np.array([[2.0, 1.0], [4.0, 3.0], [6.0, 5.0]], dtype=np.float32)[:, [1, 0]]
|
|
input_name = sess.get_inputs()[0].name
|
|
self.assertEqual(input_name, "X")
|
|
input_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(input_shape, [3, 2])
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "Y")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [3, 1])
|
|
res = sess.run([output_name], {input_name: x})
|
|
output_expected = np.array([[5.0], [11.0], [17.0]], dtype=np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
xcontiguous = np.ascontiguousarray(x)
|
|
rescontiguous = sess.run([output_name], {input_name: xcontiguous})
|
|
np.testing.assert_allclose(output_expected, rescontiguous[0], rtol=1e-05, atol=1e-08)
|
|
|
|
def testRunModelMultipleThreads(self):
|
|
available_providers = onnxrt.get_available_providers()
|
|
|
|
# Skip this test for a "pure" DML onnxruntime python wheel. We keep this test enabled for instances where both DML and CUDA
|
|
# EPs are available (Windows GPU CI pipeline has this config) - this test will pass because CUDA has higher precendence than DML
|
|
# and the nodes are assigned to only the CUDA EP (which supports this test)
|
|
if ('DmlExecutionProvider' in available_providers and not 'CUDAExecutionProvider' in available_providers):
|
|
print("Skipping testRunModelMultipleThreads as the DML EP does not support calling Run() on different threads using the same session object ")
|
|
else:
|
|
so = onnxrt.SessionOptions()
|
|
so.log_verbosity_level = 1
|
|
so.logid = "MultiThreadsTest"
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"), sess_options=so)
|
|
ro1 = onnxrt.RunOptions()
|
|
ro1.logid = "thread1"
|
|
t1 = threading.Thread(target=self.run_model, args=(sess, ro1))
|
|
ro2 = onnxrt.RunOptions()
|
|
ro2.logid = "thread2"
|
|
t2 = threading.Thread(target=self.run_model, args=(sess, ro2))
|
|
t1.start()
|
|
t2.start()
|
|
t1.join()
|
|
t2.join()
|
|
|
|
def testListAsInput(self):
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
x = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
input_name = sess.get_inputs()[0].name
|
|
res = sess.run([], {input_name: x.tolist()})
|
|
output_expected = np.array([[1.0, 4.0], [9.0, 16.0], [25.0, 36.0]], dtype=np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
|
|
def testStringListAsInput(self):
|
|
sess = onnxrt.InferenceSession(get_name("identity_string.onnx"))
|
|
x = np.array(['this', 'is', 'identity', 'test'], dtype=np.str).reshape((2, 2))
|
|
x_name = sess.get_inputs()[0].name
|
|
res = sess.run([], {x_name: x.tolist()})
|
|
np.testing.assert_equal(x, res[0])
|
|
|
|
def testRunDevice(self):
|
|
device = onnxrt.get_device()
|
|
self.assertTrue('CPU' in device or 'GPU' in device)
|
|
|
|
def testRunModelSymbolicInput(self):
|
|
sess = onnxrt.InferenceSession(get_name("matmul_2.onnx"))
|
|
x = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
input_name = sess.get_inputs()[0].name
|
|
self.assertEqual(input_name, "X")
|
|
input_shape = sess.get_inputs()[0].shape
|
|
# Input X has an unknown dimension.
|
|
self.assertEqual(input_shape, ['None', 2])
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "Y")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
# Output X has an unknown dimension.
|
|
self.assertEqual(output_shape, ['None', 1])
|
|
res = sess.run([output_name], {input_name: x})
|
|
output_expected = np.array([[5.0], [11.0], [17.0]], dtype=np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
|
|
def testBooleanInputs(self):
|
|
sess = onnxrt.InferenceSession(get_name("logicaland.onnx"))
|
|
a = np.array([[True, True], [False, False]], dtype=np.bool)
|
|
b = np.array([[True, False], [True, False]], dtype=np.bool)
|
|
|
|
# input1:0 is first in the protobuf, and input:0 is second
|
|
# and we maintain the original order.
|
|
a_name = sess.get_inputs()[0].name
|
|
self.assertEqual(a_name, "input1:0")
|
|
a_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(a_shape, [2, 2])
|
|
a_type = sess.get_inputs()[0].type
|
|
self.assertEqual(a_type, 'tensor(bool)')
|
|
|
|
b_name = sess.get_inputs()[1].name
|
|
self.assertEqual(b_name, "input:0")
|
|
b_shape = sess.get_inputs()[1].shape
|
|
self.assertEqual(b_shape, [2, 2])
|
|
b_type = sess.get_inputs()[0].type
|
|
self.assertEqual(b_type, 'tensor(bool)')
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output:0")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [2, 2])
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'tensor(bool)')
|
|
|
|
output_expected = np.array([[True, False], [False, False]], dtype=np.bool)
|
|
res = sess.run([output_name], {a_name: a, b_name: b})
|
|
np.testing.assert_equal(output_expected, res[0])
|
|
|
|
def testStringInput1(self):
|
|
sess = onnxrt.InferenceSession(get_name("identity_string.onnx"))
|
|
x = np.array(['this', 'is', 'identity', 'test'], dtype=np.str).reshape((2, 2))
|
|
|
|
x_name = sess.get_inputs()[0].name
|
|
self.assertEqual(x_name, "input:0")
|
|
x_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(x_shape, [2, 2])
|
|
x_type = sess.get_inputs()[0].type
|
|
self.assertEqual(x_type, 'tensor(string)')
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output:0")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [2, 2])
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'tensor(string)')
|
|
|
|
res = sess.run([output_name], {x_name: x})
|
|
np.testing.assert_equal(x, res[0])
|
|
|
|
def testStringInput2(self):
|
|
sess = onnxrt.InferenceSession(get_name("identity_string.onnx"))
|
|
x = np.array(['Olá', '你好', '여보세요', 'hello'], dtype=np.unicode).reshape((2, 2))
|
|
|
|
x_name = sess.get_inputs()[0].name
|
|
self.assertEqual(x_name, "input:0")
|
|
x_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(x_shape, [2, 2])
|
|
x_type = sess.get_inputs()[0].type
|
|
self.assertEqual(x_type, 'tensor(string)')
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output:0")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [2, 2])
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'tensor(string)')
|
|
|
|
res = sess.run([output_name], {x_name: x})
|
|
np.testing.assert_equal(x, res[0])
|
|
|
|
def testInputBytes(self):
|
|
sess = onnxrt.InferenceSession(get_name("identity_string.onnx"))
|
|
x = np.array([b'this', b'is', b'identity', b'test']).reshape((2, 2))
|
|
|
|
x_name = sess.get_inputs()[0].name
|
|
self.assertEqual(x_name, "input:0")
|
|
x_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(x_shape, [2, 2])
|
|
x_type = sess.get_inputs()[0].type
|
|
self.assertEqual(x_type, 'tensor(string)')
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output:0")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [2, 2])
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'tensor(string)')
|
|
|
|
res = sess.run([output_name], {x_name: x})
|
|
np.testing.assert_equal(x, res[0].astype('|S8'))
|
|
|
|
def testInputObject(self):
|
|
sess = onnxrt.InferenceSession(get_name("identity_string.onnx"))
|
|
x = np.array(['this', 'is', 'identity', 'test'], object).reshape((2, 2))
|
|
|
|
x_name = sess.get_inputs()[0].name
|
|
self.assertEqual(x_name, "input:0")
|
|
x_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(x_shape, [2, 2])
|
|
x_type = sess.get_inputs()[0].type
|
|
self.assertEqual(x_type, 'tensor(string)')
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output:0")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [2, 2])
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'tensor(string)')
|
|
|
|
res = sess.run([output_name], {x_name: x})
|
|
np.testing.assert_equal(x, res[0])
|
|
|
|
def testInputVoid(self):
|
|
sess = onnxrt.InferenceSession(get_name("identity_string.onnx"))
|
|
x = np.array([b'this', b'is', b'identity', b'test'], np.void).reshape((2, 2))
|
|
|
|
x_name = sess.get_inputs()[0].name
|
|
self.assertEqual(x_name, "input:0")
|
|
x_shape = sess.get_inputs()[0].shape
|
|
self.assertEqual(x_shape, [2, 2])
|
|
x_type = sess.get_inputs()[0].type
|
|
self.assertEqual(x_type, 'tensor(string)')
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output:0")
|
|
output_shape = sess.get_outputs()[0].shape
|
|
self.assertEqual(output_shape, [2, 2])
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'tensor(string)')
|
|
|
|
res = sess.run([output_name], {x_name: x})
|
|
|
|
expr = np.array([['this\x00\x00\x00\x00', 'is\x00\x00\x00\x00\x00\x00'], ['identity', 'test\x00\x00\x00\x00']],
|
|
dtype=object)
|
|
np.testing.assert_equal(expr, res[0])
|
|
|
|
def testRaiseWrongNumInputs(self):
|
|
with self.assertRaises(ValueError) as context:
|
|
sess = onnxrt.InferenceSession(get_name("logicaland.onnx"))
|
|
a = np.array([[True, True], [False, False]], dtype=np.bool)
|
|
res = sess.run([], {'input:0': a})
|
|
|
|
self.assertTrue('Model requires 2 inputs' in str(context.exception))
|
|
|
|
def testModelMeta(self):
|
|
model_path = "../models/opset8/test_squeezenet/model.onnx"
|
|
if not os.path.exists(model_path):
|
|
return
|
|
sess = onnxrt.InferenceSession(model_path)
|
|
modelmeta = sess.get_modelmeta()
|
|
self.assertEqual('onnx-caffe2', modelmeta.producer_name)
|
|
self.assertEqual('squeezenet_old', modelmeta.graph_name)
|
|
self.assertEqual('', modelmeta.domain)
|
|
self.assertEqual('', modelmeta.description)
|
|
|
|
def testProfilerWithSessionOptions(self):
|
|
so = onnxrt.SessionOptions()
|
|
so.enable_profiling = True
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"), sess_options=so)
|
|
x = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
sess.run([], {'X': x})
|
|
profile_file = sess.end_profiling()
|
|
|
|
tags = ['pid', 'dur', 'ts', 'ph', 'X', 'name', 'args']
|
|
with open(profile_file) as f:
|
|
lines = f.readlines()
|
|
self.assertTrue('[' in lines[0])
|
|
for i in range(1, 8):
|
|
for tag in tags:
|
|
self.assertTrue(tag in lines[i])
|
|
self.assertTrue(']' in lines[8])
|
|
|
|
def testGraphOptimizationLevel(self):
|
|
opt = onnxrt.SessionOptions()
|
|
# default should be all optimizations optimization
|
|
self.assertEqual(opt.graph_optimization_level, onnxrt.GraphOptimizationLevel.ORT_ENABLE_ALL)
|
|
opt.graph_optimization_level = onnxrt.GraphOptimizationLevel.ORT_ENABLE_EXTENDED
|
|
self.assertEqual(opt.graph_optimization_level, onnxrt.GraphOptimizationLevel.ORT_ENABLE_EXTENDED)
|
|
sess = onnxrt.InferenceSession(get_name("logicaland.onnx"), sess_options=opt)
|
|
a = np.array([[True, True], [False, False]], dtype=np.bool)
|
|
b = np.array([[True, False], [True, False]], dtype=np.bool)
|
|
|
|
res = sess.run([], {'input1:0': a, 'input:0': b})
|
|
|
|
def testSequenceLength(self):
|
|
sess = onnxrt.InferenceSession(get_name("sequence_length.onnx"))
|
|
x = [
|
|
np.array([1.0, 0.0, 3.0, 44.0, 23.0, 11.0], dtype=np.float32).reshape((2, 3)),
|
|
np.array([1.0, 0.0, 3.0, 44.0, 23.0, 11.0], dtype=np.float32).reshape((2, 3))
|
|
]
|
|
|
|
x_name = sess.get_inputs()[0].name
|
|
self.assertEqual(x_name, "X")
|
|
x_type = sess.get_inputs()[0].type
|
|
self.assertEqual(x_type, 'seq(tensor(float))')
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "Y")
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'tensor(int64)')
|
|
|
|
output_expected = np.array(2, dtype=np.int64)
|
|
res = sess.run([output_name], {x_name: x})
|
|
self.assertEqual(output_expected, res[0])
|
|
|
|
def testSequenceConstruct(self):
|
|
sess = onnxrt.InferenceSession(get_name("sequence_construct.onnx"))
|
|
|
|
self.assertEqual(sess.get_inputs()[0].type, 'tensor(int64)')
|
|
self.assertEqual(sess.get_inputs()[1].type, 'tensor(int64)')
|
|
|
|
self.assertEqual(sess.get_inputs()[0].name, "tensor1")
|
|
self.assertEqual(sess.get_inputs()[1].name, "tensor2")
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output_sequence")
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'seq(tensor(int64))')
|
|
|
|
output_expected = [
|
|
np.array([1, 0, 3, 44, 23, 11], dtype=np.int64).reshape((2, 3)),
|
|
np.array([1, 2, 3, 4, 5, 6], dtype=np.int64).reshape((2, 3))
|
|
]
|
|
|
|
res = sess.run(
|
|
[output_name], {
|
|
"tensor1": np.array([1, 0, 3, 44, 23, 11], dtype=np.int64).reshape((2, 3)),
|
|
"tensor2": np.array([1, 2, 3, 4, 5, 6], dtype=np.int64).reshape((2, 3))
|
|
})
|
|
|
|
np.testing.assert_array_equal(output_expected, res[0])
|
|
|
|
def testSequenceInsert(self):
|
|
opt = onnxrt.SessionOptions()
|
|
opt.execution_mode = onnxrt.ExecutionMode.ORT_SEQUENTIAL
|
|
sess = onnxrt.InferenceSession(get_name("sequence_insert.onnx"), sess_options=opt)
|
|
|
|
self.assertEqual(sess.get_inputs()[0].type, 'seq(tensor(int64))')
|
|
self.assertEqual(sess.get_inputs()[1].type, 'tensor(int64)')
|
|
|
|
self.assertEqual(sess.get_inputs()[0].name, "input_seq")
|
|
self.assertEqual(sess.get_inputs()[1].name, "tensor")
|
|
|
|
output_name = sess.get_outputs()[0].name
|
|
self.assertEqual(output_name, "output_sequence")
|
|
output_type = sess.get_outputs()[0].type
|
|
self.assertEqual(output_type, 'seq(tensor(int64))')
|
|
|
|
output_expected = [np.array([1, 0, 3, 44, 23, 11], dtype=np.int64).reshape((2, 3))]
|
|
res = sess.run([output_name], {
|
|
"tensor": np.array([1, 0, 3, 44, 23, 11], dtype=np.int64).reshape((2, 3)),
|
|
"input_seq": []
|
|
})
|
|
np.testing.assert_array_equal(output_expected, res[0])
|
|
|
|
def testOrtExecutionMode(self):
|
|
opt = onnxrt.SessionOptions()
|
|
self.assertEqual(opt.execution_mode, onnxrt.ExecutionMode.ORT_SEQUENTIAL)
|
|
opt.execution_mode = onnxrt.ExecutionMode.ORT_PARALLEL
|
|
self.assertEqual(opt.execution_mode, onnxrt.ExecutionMode.ORT_PARALLEL)
|
|
|
|
def testLoadingSessionOptionsFromModel(self):
|
|
try:
|
|
os.environ['ORT_LOAD_CONFIG_FROM_MODEL'] = str(1)
|
|
sess = onnxrt.InferenceSession(get_name("model_with_valid_ort_config_json.onnx"))
|
|
session_options = sess.get_session_options()
|
|
|
|
self.assertEqual(session_options.inter_op_num_threads, 5) # from the ORT config
|
|
|
|
self.assertEqual(session_options.intra_op_num_threads, 2) # from the ORT config
|
|
|
|
self.assertEqual(session_options.execution_mode,
|
|
onnxrt.ExecutionMode.ORT_SEQUENTIAL) # default option (not from the ORT config)
|
|
|
|
self.assertEqual(session_options.graph_optimization_level,
|
|
onnxrt.GraphOptimizationLevel.ORT_ENABLE_ALL) # from the ORT config
|
|
|
|
self.assertEqual(session_options.enable_profiling, True) # from the ORT config
|
|
|
|
except Exception:
|
|
raise
|
|
|
|
finally:
|
|
# Make sure the usage of the feature is disabled after this test
|
|
os.environ['ORT_LOAD_CONFIG_FROM_MODEL'] = str(0)
|
|
|
|
def testSessionOptionsAddFreeDimensionOverrideByDenotation(self):
|
|
so = onnxrt.SessionOptions()
|
|
so.add_free_dimension_override_by_denotation("DATA_BATCH", 3)
|
|
so.add_free_dimension_override_by_denotation("DATA_CHANNEL", 5)
|
|
sess = onnxrt.InferenceSession(get_name("abs_free_dimensions.onnx"), so)
|
|
input_name = sess.get_inputs()[0].name
|
|
self.assertEqual(input_name, "x")
|
|
input_shape = sess.get_inputs()[0].shape
|
|
# Free dims with denotations - "DATA_BATCH" and "DATA_CHANNEL" have values assigned to them.
|
|
self.assertEqual(input_shape, [3, 5, 5])
|
|
|
|
def testSessionOptionsAddFreeDimensionOverrideByName(self):
|
|
so = onnxrt.SessionOptions()
|
|
so.add_free_dimension_override_by_name("Dim1", 4)
|
|
so.add_free_dimension_override_by_name("Dim2", 6)
|
|
sess = onnxrt.InferenceSession(get_name("abs_free_dimensions.onnx"), so)
|
|
input_name = sess.get_inputs()[0].name
|
|
self.assertEqual(input_name, "x")
|
|
input_shape = sess.get_inputs()[0].shape
|
|
# "Dim1" and "Dim2" have values assigned to them.
|
|
self.assertEqual(input_shape, [4, 6, 5])
|
|
|
|
def testSessionOptionsAddConfigEntry(self):
|
|
so = onnxrt.SessionOptions()
|
|
key = "CONFIG_KEY"
|
|
val = "CONFIG_VAL"
|
|
so.add_session_config_entry(key, val)
|
|
self.assertEqual(so.get_session_config_entry(key), val)
|
|
|
|
def testInvalidSessionOptionsConfigEntry(self):
|
|
so = onnxrt.SessionOptions()
|
|
invalide_key = "INVALID_KEY"
|
|
with self.assertRaises(RuntimeError) as context:
|
|
so.get_session_config_entry(invalide_key)
|
|
self.assertTrue(
|
|
'SessionOptions does not have configuration with key: ' + invalide_key in str(context.exception))
|
|
|
|
def testSessionOptionsAddInitializer(self):
|
|
# Create an initializer and add it to a SessionOptions instance
|
|
so = onnxrt.SessionOptions()
|
|
# This initializer is different from the actual initializer in the model for "W"
|
|
ortvalue_initializer = onnxrt.OrtValue.ortvalue_from_numpy(np.array([[2.0, 1.0], [4.0, 3.0], [6.0, 5.0]], dtype=np.float32))
|
|
# The user should manage the life cycle of this OrtValue and should keep it in scope
|
|
# as long as any session that is going to be reliant on it is in scope
|
|
so.add_initializer("W", ortvalue_initializer)
|
|
|
|
# Create an InferenceSession that only uses the CPU EP and validate that it uses the
|
|
# initializer provided via the SessionOptions instance (overriding the model initializer)
|
|
# We only use the CPU EP because the initializer we created is on CPU and we want the model to use that
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"), so, ['CPUExecutionProvider'])
|
|
res = sess.run(["Y"], {"X": np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)})
|
|
self.assertTrue(np.array_equal(res[0], np.array([[2.0, 2.0], [12.0, 12.0], [30.0, 30.0]], dtype=np.float32)))
|
|
|
|
def testRegisterCustomOpsLibrary(self):
|
|
if sys.platform.startswith("win"):
|
|
shared_library = 'custom_op_library.dll'
|
|
if not os.path.exists(shared_library):
|
|
raise FileNotFoundError("Unable to find '{0}'".format(shared_library))
|
|
|
|
elif sys.platform.startswith("darwin"):
|
|
shared_library = 'libcustom_op_library.dylib'
|
|
if not os.path.exists(shared_library):
|
|
raise FileNotFoundError("Unable to find '{0}'".format(shared_library))
|
|
|
|
else:
|
|
shared_library = './libcustom_op_library.so'
|
|
if not os.path.exists(shared_library):
|
|
raise FileNotFoundError("Unable to find '{0}'".format(shared_library))
|
|
|
|
this = os.path.dirname(__file__)
|
|
custom_op_model = os.path.join(this, "testdata", "custom_op_library", "custom_op_test.onnx")
|
|
if not os.path.exists(custom_op_model):
|
|
raise FileNotFoundError("Unable to find '{0}'".format(custom_op_model))
|
|
|
|
so1 = onnxrt.SessionOptions()
|
|
so1.register_custom_ops_library(shared_library)
|
|
|
|
# Model loading successfully indicates that the custom op node could be resolved successfully
|
|
sess1 = onnxrt.InferenceSession(custom_op_model, so1)
|
|
#Run with input data
|
|
input_name_0 = sess1.get_inputs()[0].name
|
|
input_name_1 = sess1.get_inputs()[1].name
|
|
output_name = sess1.get_outputs()[0].name
|
|
input_0 = np.ones((3,5)).astype(np.float32)
|
|
input_1 = np.zeros((3,5)).astype(np.float32)
|
|
res = sess1.run([output_name], {input_name_0: input_0, input_name_1: input_1})
|
|
output_expected = np.ones((3,5)).astype(np.float32)
|
|
np.testing.assert_allclose(output_expected, res[0], rtol=1e-05, atol=1e-08)
|
|
|
|
# Create an alias of SessionOptions instance
|
|
# We will use this alias to construct another InferenceSession
|
|
so2 = so1
|
|
|
|
# Model loading successfully indicates that the custom op node could be resolved successfully
|
|
sess2 = onnxrt.InferenceSession(custom_op_model, so2)
|
|
|
|
# Create another SessionOptions instance with the same shared library referenced
|
|
so3 = onnxrt.SessionOptions()
|
|
so3.register_custom_ops_library(shared_library)
|
|
sess3 = onnxrt.InferenceSession(custom_op_model, so3)
|
|
|
|
def testOrtValue(self):
|
|
|
|
numpy_arr_input = np.array([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], dtype=np.float32)
|
|
numpy_arr_output = np.array([[1.0, 4.0], [9.0, 16.0], [25.0, 36.0]], dtype=np.float32)
|
|
|
|
def test_session_with_ortvalue_input(ortvalue):
|
|
sess = onnxrt.InferenceSession(get_name("mul_1.onnx"))
|
|
res = sess.run(["Y"], {"X": ortvalue})
|
|
self.assertTrue(np.array_equal(res[0], numpy_arr_output))
|
|
|
|
ortvalue1 = onnxrt.OrtValue.ortvalue_from_numpy(numpy_arr_input)
|
|
self.assertEqual(ortvalue1.device_name(), "cpu")
|
|
self.assertEqual(ortvalue1.shape(), [3, 2])
|
|
self.assertEqual(ortvalue1.data_type(), "tensor(float)")
|
|
self.assertEqual(ortvalue1.is_tensor(), True)
|
|
self.assertTrue(np.array_equal(ortvalue1.numpy(), numpy_arr_input))
|
|
|
|
# Pass in the constructed OrtValue to a session via Run() and check results
|
|
test_session_with_ortvalue_input(ortvalue1)
|
|
|
|
# The constructed OrtValue should still be valid after being used in a session
|
|
self.assertTrue(np.array_equal(ortvalue1.numpy(), numpy_arr_input))
|
|
|
|
if 'CUDAExecutionProvider' in onnxrt.get_available_providers():
|
|
ortvalue2 = onnxrt.OrtValue.ortvalue_from_numpy(numpy_arr_input, 'cuda', 0)
|
|
self.assertEqual(ortvalue2.device_name(), "cuda")
|
|
self.assertEqual(ortvalue2.shape(), [3, 2])
|
|
self.assertEqual(ortvalue2.data_type(), "tensor(float)")
|
|
self.assertEqual(ortvalue2.is_tensor(), True)
|
|
self.assertTrue(np.array_equal(ortvalue2.numpy(), numpy_arr_input))
|
|
|
|
# Pass in the constructed OrtValue to a session via Run() and check results
|
|
test_session_with_ortvalue_input(ortvalue2)
|
|
|
|
# The constructed OrtValue should still be valid after being used in a session
|
|
self.assertTrue(np.array_equal(ortvalue2.numpy(), numpy_arr_input))
|
|
|
|
def testRunModelWithCudaCopyStream(self):
|
|
available_providers = onnxrt.get_available_providers()
|
|
|
|
if (not 'CUDAExecutionProvider' in available_providers):
|
|
print("Skipping testRunModelWithCudaCopyStream when CUDA is not available")
|
|
else:
|
|
# adapted from issue #4829 for a race condition when copy is not on default stream
|
|
# note:
|
|
# 1. if there are intermittent failure in this test, something is wrong
|
|
# 2. it's easier to repro on slower GPU (like M60, Geforce 1070)
|
|
|
|
# to repro #4829, uncomment the line below to run copy in a separate stream
|
|
#onnxrt.capi._pybind_state.set_do_copy_in_default_stream(False)
|
|
|
|
session = onnxrt.InferenceSession(get_name("issue4829.onnx"))
|
|
shape = np.array([2,2], dtype=np.int64)
|
|
for iteration in range(100000):
|
|
result = session.run(output_names=['output'], input_feed={'shape': shape})
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|