File: nanobind_runner.py

package info (click to toggle)
ispc 1.28.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 97,620 kB
  • sloc: cpp: 77,067; python: 8,303; yacc: 3,337; lex: 1,126; ansic: 631; sh: 475; makefile: 17
file content (147 lines) | stat: -rw-r--r-- 5,115 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
#!/usr/bin/env python3
"""
Isolated test runner for ISPC nanobind tests.
This module runs test functions in a separate subprocess to provide isolation
and avoid conflicts between tests.
"""

import sys
import os
import numpy
import importlib.util
import importlib.machinery

def call_test_function(module_name, test_sig, func_sig, width, verbose=False):
    """
    Call the test function from the compiled ISPC module.

    Args:
        module_name: Name of the compiled module
        test_sig: Test signature identifier (0-34)
        func_sig: Function signature string (e.g., 'f_v(')
        width: Vector width for the test
        verbose: Enable verbose output for debugging

    Returns:
        Status enum value indicating test result
    """
    try:
        # func_sig is 'f_v(', so substitute ( to _cpu_entry_point
        # because f_v_cpu_entry_point is the entry point
        function = func_sig.replace('(', '_cpu_entry_point')

        # Create a FileFinder for the current directory as we suppose that the
        # script is run from the root project directory.
        finder = importlib.machinery.FileFinder(
            './',
            (importlib.machinery.ExtensionFileLoader, importlib.machinery.EXTENSION_SUFFIXES),
        )

        spec = finder.find_spec(module_name)
        if spec is None:
            raise ImportError(f"Module '{module_name}' not found")

        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        entry_func = getattr(module, function, None)

        # Check if the functions are found in the loaded module
        if test_sig != 7:
            if entry_func is None:
                raise ImportError(f"Function '{function}' not found in module '{module_name}'")

        # Prepare input data
        ARRAY_SIZE = 256
        res = numpy.zeros(ARRAY_SIZE, dtype=numpy.float32)

        dst = numpy.zeros(ARRAY_SIZE, dtype=numpy.float32)
        vfloat = numpy.arange(1, ARRAY_SIZE + 1, dtype=numpy.float32)
        vdouble = numpy.arange(1, ARRAY_SIZE + 1, dtype=numpy.float64)
        vint = numpy.array([2 * (i + 1) for i in range(ARRAY_SIZE)], dtype=numpy.int32)
        vint2 = numpy.array([i + 5 for i in range(ARRAY_SIZE)], dtype=numpy.int32)
        b = 5.0

        # Call corresponding TEST_SIG functions, it should correspond to test_static.cpp
        if test_sig == 0:
            entry_func(dst)
        elif test_sig == 1:
            entry_func(dst, vfloat)
        elif test_sig == 2:
            entry_func(dst, vfloat, b)
        elif test_sig == 3:
            entry_func(dst, vfloat, vint)
        elif test_sig == 4:
            entry_func(dst, vdouble, b)
        elif test_sig == 5:
            entry_func(dst, vdouble, b)
        elif test_sig == 6:
            entry_func(dst, vdouble, vint2)
        elif test_sig == 7:
            struct = getattr(module, f"v{width}_varying_f_sz")
            # TODO: python object has different size than ISPC struct, so just
            # check that we have expected class
            instance = struct()
            return True
        elif test_sig == 32:
            entry_func(b)
        elif test_sig == 33:
            entry_func(vfloat)
        elif test_sig == 34:
            entry_func(vfloat, b)

        if test_sig < 32:
            result_func = getattr(module, 'result_cpu_entry_point', None)
            if result_func is None:
                raise ImportError(f"Function 'result_cpu_entry_point' not found in module '{module_name}'")

            result_func(res)
            if numpy.array_equal(dst[:width], res[:width]):
                return True
            else:
                if verbose:
                    print(f"Test {module_name} failed: expected {res[:width]}, got {dst[:width]}")
                return False
        else:
            print_func = getattr(module, 'print_result_cpu_entry_point', None)

            if print_func is None:
                raise ImportError(f"Function 'print_result_cpu_entry_point' not found in module '{module_name}'")
            print_func()
            return True

    except Exception as e:
        if verbose:
            print(f"Error in test function: {e}")
        return False

def main():
    """
    Main entry point for subprocess execution.
    Expected command line arguments:
    python nanobind_runner.py <module_name> <test_sig> <func_sig> <width> [verbose]
    """
    if len(sys.argv) < 5:
        print("Usage: nanobind_runner.py <module_name> <test_sig> <func_sig> <width> [verbose]")
        sys.exit(1)

    module_name = sys.argv[1]
    test_sig = int(sys.argv[2])
    func_sig = sys.argv[3]
    width = int(sys.argv[4])
    verbose = len(sys.argv) > 5 and sys.argv[5].lower() == 'true'

    try:
        result = call_test_function(module_name, test_sig, func_sig, width, verbose)
        # Return the status value as exit code
        if result:
            sys.exit(0)
        else:
            sys.exit(1)
    except Exception as e:
        if verbose:
            print(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()