diff --git a/.github/workflows/build-wheel.yml b/.github/workflows/build-wheel.yml index b205692bd..5fa5f6f87 100644 --- a/.github/workflows/build-wheel.yml +++ b/.github/workflows/build-wheel.yml @@ -33,7 +33,7 @@ jobs: - "3.12" - "3.13" - "3.14" - # - "3.14t" + - "3.14t" name: py${{ matrix.python-version }} runs-on: ${{ (inputs.host-platform == 'linux-64' && 'linux-amd64-cpu8') || (inputs.host-platform == 'linux-aarch64' && 'linux-arm64-cpu8') || diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index b51dcaed8..b6f8e5e4c 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -232,7 +232,17 @@ jobs: build_type: pull-request script: "ci/test_wheel_deps_wheels.sh" matrix: ${{ needs.compute-matrix.outputs.TEST_MATRIX }} - matrix_filter: map(select(.ARCH == "amd64" and (.CUDA_VER | split(".") | .[0] | tonumber >= 12))) | group_by(.CUDA_VER|split(".")|map(tonumber)|.[0]) | map(max_by([(.PY_VER|split(".")|map(tonumber)), (.CUDA_VER|split(".")|map(tonumber))])) + matrix_filter: >- + map(select( + .ARCH == "amd64" and + (.CUDA_VER | split(".") | .[0] | tonumber >= 12) + )) + | group_by(.CUDA_VER | split(".") | map(tonumber) | .[0]) + | map(max_by([ + (.PY_VER | sub("t$"; "") | split(".") | map(tonumber)), + (.PY_VER | endswith("t") | not), + (.CUDA_VER | split(".") | map(tonumber)) + ])) test-thirdparty-cudf: needs: @@ -244,7 +254,17 @@ jobs: script: "ci/test_thirdparty_cudf.sh" matrix: ${{ needs.compute-matrix.outputs.TEST_MATRIX }} # TODO: Enable for CUDA 13 when a supporting version of cuDF is available - matrix_filter: map(select(.ARCH == "amd64" and (.CUDA_VER | split(".") | .[0] | tonumber == 12))) | group_by(.CUDA_VER|split(".")|map(tonumber)|.[0]) | map(max_by([(.PY_VER|split(".")|map(tonumber)), (.CUDA_VER|split(".")|map(tonumber))])) + matrix_filter: >- + map(select( + .ARCH == "amd64" and + (.CUDA_VER | split(".") | .[0] | tonumber == 12) + )) + | group_by(.CUDA_VER | split(".") | map(tonumber) | .[0]) + | map(max_by([ + (.PY_VER | sub("t$"; "") | split(".") | map(tonumber)), + (.PY_VER | endswith("t") | not), + (.CUDA_VER | split(".") | map(tonumber)) + ])) test-thirdparty-nvmath: needs: @@ -256,7 +276,17 @@ jobs: script: "ci/test_thirdparty_nvmath.sh" matrix: ${{ needs.compute-matrix.outputs.TEST_MATRIX }} # TODO: Enable for CUDA 13 when a supporting version of nvmath-python is available - matrix_filter: map(select(.ARCH == "amd64" and (.CUDA_VER | split(".") | .[0] | tonumber == 12))) | group_by(.CUDA_VER|split(".")|map(tonumber)|.[0]) | map(max_by([(.PY_VER|split(".")|map(tonumber)), (.CUDA_VER|split(".")|map(tonumber))])) + matrix_filter: >- + map(select( + .ARCH == "amd64" and + (.CUDA_VER | split(".") | .[0] | tonumber == 12) + )) + | group_by(.CUDA_VER | split(".") | map(tonumber) | .[0]) + | map(max_by([ + (.PY_VER | sub("t$"; "") | split(".") | map(tonumber)), + (.PY_VER | endswith("t") | not), + (.CUDA_VER | split(".") | map(tonumber)) + ])) test-thirdparty-awkward: needs: @@ -268,7 +298,17 @@ jobs: script: "ci/test_thirdparty_awkward.sh" matrix: ${{ needs.compute-matrix.outputs.TEST_MATRIX }} # TODO: Enable for CUDA 13 in future - matrix_filter: map(select(.ARCH == "amd64" and (.CUDA_VER | split(".") | .[0] | tonumber == 12))) | group_by(.CUDA_VER|split(".")|map(tonumber)|.[0]) | map(max_by([(.PY_VER|split(".")|map(tonumber)), (.CUDA_VER|split(".")|map(tonumber))])) + matrix_filter: >- + map(select( + .ARCH == "amd64" and + (.CUDA_VER | split(".") | .[0] | tonumber == 12) + )) + | group_by(.CUDA_VER | split(".") | map(tonumber) | .[0]) + | map(max_by([ + (.PY_VER | sub("t$"; "") | split(".") | map(tonumber)), + (.PY_VER | endswith("t") | not), + (.CUDA_VER | split(".") | map(tonumber)) + ])) build-docs: needs: @@ -285,7 +325,13 @@ jobs: build_type: pull-request script: "ci/coverage_report.sh" matrix: ${{ needs.compute-matrix.outputs.TEST_MATRIX }} - matrix_filter: 'map(select(.ARCH == "amd64" and .CUDA_VER == "12.9.1" and .PY_VER == "3.12")) | .[0:1]' + matrix_filter: >- + map(select( + .ARCH == "amd64" and + .CUDA_VER == "12.9.1" and + .PY_VER == "3.12" + )) + | .[0:1] # ============================================================= diff --git a/.gitignore b/.gitignore index baf7bf180..479165f06 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ __pycache__ build .*.swp *.so +*.pyd numba_cuda/numba/cuda/tests/cudadrv/test_device_functions.* numba_cuda/numba/cuda/tests/cudadrv/undefined_extern.* testing/*.a diff --git a/ci/test-matrix.yml b/ci/test-matrix.yml index 9d9e3e1be..c6199689d 100644 --- a/ci/test-matrix.yml +++ b/ci/test-matrix.yml @@ -34,9 +34,7 @@ linux: - { ARCH: 'amd64', PY_VER: '3.14', CUDA_VER: '12.9.1', LOCAL_CTK: '0', GPU: 't4', GPU_COUNT: '1', DRIVER: 'latest' } - { ARCH: 'amd64', PY_VER: '3.14', CUDA_VER: '13.0.2', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } - { ARCH: 'amd64', PY_VER: '3.14', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } - # - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '12.9.1', LOCAL_CTK: '1', GPU: 't4', GPU_COUNT: '1', DRIVER: 'latest' } - # - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '13.0.2', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } - # - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } + - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } # linux-aarch64 - { ARCH: 'arm64', PY_VER: '3.10', CUDA_VER: '12.9.1', LOCAL_CTK: '1', GPU: 'a100', GPU_COUNT: '1', DRIVER: 'latest' } - { ARCH: 'arm64', PY_VER: '3.10', CUDA_VER: '13.0.2', LOCAL_CTK: '0', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } @@ -53,9 +51,7 @@ linux: - { ARCH: 'arm64', PY_VER: '3.14', CUDA_VER: '12.9.1', LOCAL_CTK: '0', GPU: 'a100', GPU_COUNT: '1', DRIVER: 'latest' } - { ARCH: 'arm64', PY_VER: '3.14', CUDA_VER: '13.0.2', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } - { ARCH: 'arm64', PY_VER: '3.14', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'a100', GPU_COUNT: '1', DRIVER: 'latest' } - # - { ARCH: 'arm64', PY_VER: '3.14t', CUDA_VER: '12.9.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } - # - { ARCH: 'arm64', PY_VER: '3.14t', CUDA_VER: '13.0.2', LOCAL_CTK: '0', GPU: 'a100', GPU_COUNT: '1', DRIVER: 'latest' } - # - { ARCH: 'arm64', PY_VER: '3.14t', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } + - { ARCH: 'arm64', PY_VER: '3.14t', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest' } # special runners - { ARCH: 'amd64', PY_VER: '3.13', CUDA_VER: '13.0.2', LOCAL_CTK: '1', GPU: 'h100', GPU_COUNT: '1', DRIVER: 'latest' } - { ARCH: 'amd64', PY_VER: '3.13', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'h100', GPU_COUNT: '1', DRIVER: 'latest' } @@ -83,7 +79,5 @@ windows: - { ARCH: 'amd64', PY_VER: '3.14', CUDA_VER: '12.9.1', LOCAL_CTK: '0', GPU: 'v100', GPU_COUNT: '1', DRIVER: 'latest', DRIVER_MODE: 'TCC' } - { ARCH: 'amd64', PY_VER: '3.14', CUDA_VER: '13.0.2', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest', DRIVER_MODE: 'MCDM' } - { ARCH: 'amd64', PY_VER: '3.14', CUDA_VER: '13.2.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest', DRIVER_MODE: 'MCDM' } - # - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '12.9.1', LOCAL_CTK: '1', GPU: 'l4', GPU_COUNT: '1', DRIVER: 'latest', DRIVER_MODE: 'TCC' } - # - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '13.0.2', LOCAL_CTK: '0', GPU: 'a100', GPU_COUNT: '1', DRIVER: 'latest', DRIVER_MODE: 'MCDM' } - # - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '13.2.1', LOCAL_CTK: '0', GPU: 'a100', GPU_COUNT: '1', DRIVER: 'latest', DRIVER_MODE: 'MCDM' } + - { ARCH: 'amd64', PY_VER: '3.14t', CUDA_VER: '13.2.1', LOCAL_CTK: '0', GPU: 'a100', GPU_COUNT: '1', DRIVER: 'latest', DRIVER_MODE: 'MCDM' } nightly: [] diff --git a/docs/source/developer/free_threading.rst b/docs/source/developer/free_threading.rst new file mode 100644 index 000000000..e54e2c0fb --- /dev/null +++ b/docs/source/developer/free_threading.rst @@ -0,0 +1,77 @@ +.. + SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + SPDX-License-Identifier: BSD-2-Clause + +Free-threading +============== + +Free-threaded CPython test coverage requires a free-threaded Python build, +for example Python 3.14t with the ``cp314t`` ABI tag. The regular test suite +contains free-threaded smoke tests that run automatically in such an +environment. The heavier stress tests are opt-in because they create many +threads and subprocesses and are intended for local or dedicated CI runs. + +Run the free-threading stress tests with:: + + $ PYTHON_GIL=0 NUMBA_CUDA_FT_STRESS=1 \ + python -m pytest -q --pyargs numba.cuda.tests.stress + +``PYTHON_GIL=0`` keeps the GIL disabled for free-threaded CPython builds. The +stress tests also check the build metadata and skip when Python is not a +free-threaded build. + +Stress test environment variables +--------------------------------- + +.. envvar:: NUMBA_CUDA_FT_STRESS + + Enables the opt-in free-threading stress tests when set to ``1``, + ``true``, ``yes``, or ``on``. Without this variable, tests in + ``numba.cuda.tests.stress.test_free_threading`` are skipped. + +.. envvar:: NUMBA_CUDA_FT_STRESS_SECONDS + + Controls the duration, in seconds, of timed CUDA stress tests. The default + is ``30`` seconds. + +.. envvar:: NUMBA_CUDA_FT_STRESS_WORKERS + + Controls the thread count used by thread-pool stress tests. Defaults vary + by test and are capped at the detected CPU count unless this variable is + set explicitly. + +.. envvar:: NUMBA_CUDA_FT_STRESS_PROCESSES + + Controls the subprocess count used by cache-concurrency stress tests. + Defaults vary by test and are capped at the detected CPU count unless this + variable is set explicitly. + +.. envvar:: NUMBA_CUDA_FT_STRESS_ITERS + + Overrides iteration counts for loop-based stress tests. Defaults vary by + test, for example fingerprinting, memoryview buffer helpers, and type + conversion stress cases. + +Suggested stress profiles +------------------------- + +Use the defaults for a quick local run. On many-core systems, explicitly set +the worker and process counts to exercise concurrent dispatcher, cache, +driver, and helper-extension paths more aggressively, for example:: + + $ PYTHON_GIL=0 NUMBA_CUDA_FT_STRESS=1 \ + NUMBA_CUDA_FT_STRESS_SECONDS=120 \ + NUMBA_CUDA_FT_STRESS_WORKERS=96 \ + NUMBA_CUDA_FT_STRESS_PROCESSES=24 \ + NUMBA_CUDA_FT_STRESS_ITERS=10000 \ + python -m pytest -q --pyargs numba.cuda.tests.stress + +If a stress failure is hard to reproduce, increase +``NUMBA_CUDA_FT_STRESS_SECONDS`` for CUDA tests or +``NUMBA_CUDA_FT_STRESS_ITERS`` for loop-based no-CUDA tests. + +The CUDA dispatch stress tests deliberately avoid a concurrent tight loop of +full ``gc.collect()`` calls. On many-core CPython 3.14t free-threaded builds, +that pattern timed out in CPython's ``Python/gc_free_threading.c`` and turned +the dispatch stress into a CPython full-GC progress test. The dispatch stress +still performs a full collection after worker shutdown. diff --git a/docs/source/developer/index.rst b/docs/source/developer/index.rst new file mode 100644 index 000000000..3ed2490e2 --- /dev/null +++ b/docs/source/developer/index.rst @@ -0,0 +1,10 @@ +.. + SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + SPDX-License-Identifier: BSD-2-Clause + +Developer documentation +======================= + +.. toctree:: + + free_threading.rst diff --git a/docs/source/index.rst b/docs/source/index.rst index 8bdd398de..d0dc95108 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -28,3 +28,4 @@ Contents user/index.rst reference/index.rst + developer/index.rst diff --git a/docs/source/user/installation.rst b/docs/source/user/installation.rst index 82b3cd21d..1911661ae 100644 --- a/docs/source/user/installation.rst +++ b/docs/source/user/installation.rst @@ -49,6 +49,21 @@ CUDA 13 dependencies can be installed via ``pip`` with:: $ pip install numba-cuda[cu13] +Free-threaded Python +-------------------- + +Free-threaded CPython wheels use the ``cp314t`` ABI tag and are supported for +Python 3.14t. A conda-forge environment for local smoke testing can be created +with:: + + $ mamba create -n numba-cuda-py314t -c conda-forge \ + "python=3.14.*=*_cp314t" numba llvmlite numpy \ + cuda-bindings cuda-core cuda-pathfinder packaging pip + +The ``python=3.14.*=*_cp314t`` constraint selects a free-threaded Python build. +In a supported environment, importing ``numba.cuda`` with ``PYTHON_GIL=0`` must +leave ``sys._is_gil_enabled()`` false. + If you are not using Conda/pip or if you want to use a different version of CUDA toolkit, :ref:`cudatoolkit-lookup` describes how Numba searches for a CUDA toolkit. diff --git a/numba_cuda/numba/cuda/cext/_dispatcher.cpp b/numba_cuda/numba/cuda/cext/_dispatcher.cpp index 8a0c74730..d6e8a7aab 100644 --- a/numba_cuda/numba/cuda/cext/_dispatcher.cpp +++ b/numba_cuda/numba/cuda/cext/_dispatcher.cpp @@ -546,17 +546,40 @@ class Dispatcher { /* A flattened array of argument types to all overloads * (invariant: sizeof(overloads) == argct * sizeof(functions)) */ TypeTable overloads; + PyThread_type_lock dispatcher_lock; + + class LockGuard { + public: + explicit LockGuard(PyThread_type_lock lock) : lock(lock) { + if (lock != NULL) { + PyThread_acquire_lock(lock, 1); + } + } + + ~LockGuard() { + if (lock != NULL) { + PyThread_release_lock(lock); + } + } + + private: + PyThread_type_lock lock; + }; /* Add a new overload. Parameters: - args: An array of Type objects, one for each parameter - callable: The callable implementing this overload. */ - void addDefinition(Type args[], PyObject *callable) { + void addDefinition(Type args[], PyObject *callable, int objectmode) { + LockGuard guard(dispatcher_lock); overloads.reserve(argct + overloads.size()); for (int i=0; iargnames); Py_XDECREF(self->defargs); self->clear(); + if (self->dispatcher_lock != NULL) { + PyThread_free_lock(self->dispatcher_lock); + self->dispatcher_lock = NULL; + } Py_TYPE(self)->tp_free((PyObject*)self); } +static PyObject * +Dispatcher_new(PyTypeObject *type, PyObject *args, PyObject *kwds) +{ + Dispatcher *self = (Dispatcher *) PyType_GenericNew(type, args, kwds); + if (self != NULL) { + self->dispatcher_lock = NULL; + self->fallbackdef = NULL; + } + return (PyObject *) self; +} + + static int Dispatcher_init(Dispatcher *self, PyObject *args, PyObject *kwds) { @@ -643,6 +693,14 @@ Dispatcher_init(Dispatcher *self, PyObject *args, PyObject *kwds) )) { return -1; } + if (self->dispatcher_lock == NULL) { + self->dispatcher_lock = PyThread_allocate_lock(); + if (self->dispatcher_lock == NULL) { + PyErr_SetString(PyExc_RuntimeError, + "failed to allocate dispatcher lock"); + return -1; + } + } Py_INCREF(self->argnames); Py_INCREF(self->defargs); tmaddr = PyLong_AsVoidPtr(tmaddrobj); @@ -703,12 +761,7 @@ Dispatcher_Insert(Dispatcher *self, PyObject *args, PyObject *kwds) /* The reference to cfunc is borrowed; this only works because the derived Python class also stores an (owned) reference to cfunc. */ - self->addDefinition(sig, cfunc); - - /* Add pure python fallback */ - if (!self->fallbackdef && objectmode){ - self->fallbackdef = cfunc; - } + self->addDefinition(sig, cfunc, objectmode); delete[] sig; @@ -947,7 +1000,7 @@ Dispatcher_cuda_call(Dispatcher *self, PyObject *args, PyObject *kws) int i; int prealloc[24]; int matches; - PyObject *cfunc; + PyObject *cfunc = NULL; PyThreadState *ts = PyThreadState_Get(); PyObject *locals = NULL; PyObject *launch_config = NULL; @@ -1037,7 +1090,7 @@ Dispatcher_cuda_call(Dispatcher *self, PyObject *args, PyObject *kws) if (matches == 1) { /* Definition is found */ retval = cfunc; - Py_INCREF(retval); + cfunc = NULL; } else if (matches == 0) { /* No matching definition */ if (self->can_compile) { @@ -1047,13 +1100,17 @@ Dispatcher_cuda_call(Dispatcher *self, PyObject *args, PyObject *kws) goto CLEANUP; } retval = cuda_compile_only(self, args, kws, locals); - } else if (self->fallbackdef) { - /* Have object fallback */ - retval = call_cfunc(self, self->fallbackdef, args, kws, locals); } else { - /* Raise TypeError */ - explain_matching_error((PyObject *) self, args, kws); - retval = NULL; + PyObject *fallbackdef = self->getFallbackDefinition(); + if (fallbackdef) { + /* Have object fallback */ + retval = call_cfunc(self, fallbackdef, args, kws, locals); + Py_DECREF(fallbackdef); + } else { + /* Raise TypeError */ + explain_matching_error((PyObject *) self, args, kws); + retval = NULL; + } } } else if (self->can_compile) { /* Ambiguous, but are allowed to compile */ @@ -1074,6 +1131,7 @@ Dispatcher_cuda_call(Dispatcher *self, PyObject *args, PyObject *kws) delete[] tys; Py_DECREF(args); Py_XDECREF(launch_config); + Py_XDECREF(cfunc); return retval; } @@ -1234,7 +1292,7 @@ static PyMethodDef ext_methods[] = { MOD_INIT(_dispatcher) { PyObject *m; - MOD_DEF(m, "_dispatcher", "No docs", ext_methods) + MOD_DEF_NOGIL(m, "_dispatcher", "No docs", ext_methods) if (m == NULL) return MOD_ERROR_VAL; @@ -1245,7 +1303,7 @@ MOD_INIT(_dispatcher) { return MOD_ERROR_VAL; } - DispatcherType.tp_new = PyType_GenericNew; + DispatcherType.tp_new = Dispatcher_new; if (PyType_Ready(&DispatcherType) < 0) { return MOD_ERROR_VAL; } diff --git a/numba_cuda/numba/cuda/cext/_helpermod.c b/numba_cuda/numba/cuda/cext/_helpermod.c index 2424bda25..7ce54668a 100644 --- a/numba_cuda/numba/cuda/cext/_helpermod.c +++ b/numba_cuda/numba/cuda/cext/_helpermod.c @@ -60,7 +60,7 @@ static PyMethodDef ext_methods[] = { MOD_INIT(_helperlib) { PyObject *m; - MOD_DEF(m, "_helperlib", "No docs", ext_methods) + MOD_DEF_NOGIL(m, "_helperlib", "No docs", ext_methods) if (m == NULL) return MOD_ERROR_VAL; diff --git a/numba_cuda/numba/cuda/cext/_pymodule.h b/numba_cuda/numba/cuda/cext/_pymodule.h index 151ee5cbe..bf46da753 100644 --- a/numba_cuda/numba/cuda/cext/_pymodule.h +++ b/numba_cuda/numba/cuda/cext/_pymodule.h @@ -13,10 +13,27 @@ #define MOD_ERROR_VAL NULL #define MOD_SUCCESS_VAL(val) val #define MOD_INIT(name) PyMODINIT_FUNC PyInit_##name(void) +#if defined(Py_GIL_DISABLED) && Py_GIL_DISABLED +#define MOD_NOGIL(ob) do { \ + if (PyUnstable_Module_SetGIL(ob, Py_MOD_GIL_NOT_USED) < 0) { \ + Py_DECREF(ob); \ + ob = NULL; \ + } \ + } while (0) +#else +#define MOD_NOGIL(ob) do {} while (0) +#endif #define MOD_DEF(ob, name, doc, methods) { \ static struct PyModuleDef moduledef = { \ PyModuleDef_HEAD_INIT, name, doc, -1, methods, NULL, NULL, NULL, NULL }; \ - ob = PyModule_Create(&moduledef); } + ob = PyModule_Create(&moduledef); \ + } +#define MOD_DEF_NOGIL(ob, name, doc, methods) { \ + MOD_DEF(ob, name, doc, methods) \ + if (ob != NULL) { \ + MOD_NOGIL(ob); \ + } \ + } #define MOD_INIT_EXEC(name) PyInit_##name(); #define PyString_AsString PyUnicode_AsUTF8 diff --git a/numba_cuda/numba/cuda/cext/_typeconv.cpp b/numba_cuda/numba/cuda/cext/_typeconv.cpp index ca414e08d..c48aa8e30 100644 --- a/numba_cuda/numba/cuda/cext/_typeconv.cpp +++ b/numba_cuda/numba/cuda/cext/_typeconv.cpp @@ -41,7 +41,7 @@ static PyMethodDef ext_methods[] = { MOD_INIT(_typeconv) { PyObject *m; - MOD_DEF(m, "_typeconv", "No docs", ext_methods) + MOD_DEF_NOGIL(m, "_typeconv", "No docs", ext_methods) if (m == NULL) return MOD_ERROR_VAL; diff --git a/numba_cuda/numba/cuda/cext/_typeof.cpp b/numba_cuda/numba/cuda/cext/_typeof.cpp index 7d5905fb7..31016a61c 100644 --- a/numba_cuda/numba/cuda/cext/_typeof.cpp +++ b/numba_cuda/numba/cuda/cext/_typeof.cpp @@ -67,6 +67,25 @@ static PyObject *structured_dtypes; static PyObject *str_typeof_pyval = NULL; static PyObject *str_value = NULL; static PyObject *str_numba_type = NULL; +static PyThread_type_lock typeof_lock = NULL; + +class TypeofLockGuard { +public: + explicit TypeofLockGuard(PyThread_type_lock lock) : lock(lock) { + if (lock != NULL) { + PyThread_acquire_lock(lock, 1); + } + } + + ~TypeofLockGuard() { + if (lock != NULL) { + PyThread_release_lock(lock); + } + } + +private: + PyThread_type_lock lock; +}; /* * Type fingerprint computation. @@ -242,15 +261,24 @@ compute_dtype_fingerprint(string_writer_t *w, PyArray_Descr *descr) * (e.g. np.recarray(dtype=some_dtype) creates a new dtype * equal to some_dtype) */ - PyObject *interned = PyDict_GetItem(structured_dtypes, - (PyObject *) descr); - if (interned == NULL) { - interned = (PyObject *) descr; - if (PyDict_SetItem(structured_dtypes, interned, interned)) - return -1; + PyObject *interned; + { + TypeofLockGuard guard(typeof_lock); + interned = PyDict_GetItem(structured_dtypes, (PyObject *) descr); + if (interned == NULL) { + interned = (PyObject *) descr; + if (PyDict_SetItem(structured_dtypes, interned, interned)) + return -1; + } + Py_INCREF(interned); } - TRY(string_writer_put_char, w, (char) typenum); - return string_writer_put_intp(w, (npy_intp) interned); + if (string_writer_put_char(w, (char) typenum)) { + Py_DECREF(interned); + return -1; + } + int ret = string_writer_put_intp(w, (npy_intp) interned); + Py_DECREF(interned); + return ret; } #if NPY_API_VERSION >= 0x00000007 if (PyTypeNum_ISDATETIME(typenum)) { @@ -407,16 +435,37 @@ compute_fingerprint(string_writer_t *w, PyObject *val) return compute_dtype_fingerprint(w, PyArray_DESCR(ary)); } if (PyList_Check(val)) { - Py_ssize_t n = PyList_GET_SIZE(val); + Py_ssize_t n; + PyObject *item = NULL; +#if defined(Py_GIL_DISABLED) && Py_GIL_DISABLED + Py_BEGIN_CRITICAL_SECTION(val); +#endif + n = PyList_GET_SIZE(val); + if (n > 0) { + item = PyList_GET_ITEM(val, 0); + Py_XINCREF(item); + } +#if defined(Py_GIL_DISABLED) && Py_GIL_DISABLED + Py_END_CRITICAL_SECTION(); +#endif if (n == 0) { PyErr_SetString(PyExc_ValueError, "cannot compute fingerprint of empty list"); return -1; } + if (item == NULL) { + PyErr_SetString(PyExc_RuntimeError, + "failed to read list item for fingerprint"); + return -1; + } /* Only the first item is considered, as in typeof.py */ - TRY(string_writer_put_char, w, OP_LIST); - TRY(compute_fingerprint, w, PyList_GET_ITEM(val, 0)); - return 0; + if (string_writer_put_char(w, OP_LIST)) { + Py_DECREF(item); + return -1; + } + int ret = compute_fingerprint(w, item); + Py_DECREF(item); + return ret; } /* Note we only accept sets, not frozensets */ if (Py_TYPE(val) == &PySet_Type) { @@ -676,7 +725,13 @@ typecode_using_fingerprint(PyObject *dispatcher, PyObject *val) } return -1; } - if (_Numba_HASHTABLE_GET(fingerprint_hashtable, &w, typecode) > 0) { + int fingerprint_cache_hit; + { + TypeofLockGuard guard(typeof_lock); + fingerprint_cache_hit = + _Numba_HASHTABLE_GET(fingerprint_hashtable, &w, typecode) > 0; + } + if (fingerprint_cache_hit) { /* Cache hit */ string_writer_clear(&w); return typecode; @@ -698,10 +753,18 @@ typecode_using_fingerprint(PyObject *dispatcher, PyObject *val) * to the hash table. */ string_writer_move(key, &w); - if (_Numba_HASHTABLE_SET(fingerprint_hashtable, key, typecode)) { - string_writer_clear(&w); - PyErr_NoMemory(); - return -1; + { + TypeofLockGuard guard(typeof_lock); + if (_Numba_HASHTABLE_GET(fingerprint_hashtable, key, typecode) > 0) { + string_writer_clear(key); + free(key); + } + else if (_Numba_HASHTABLE_SET(fingerprint_hashtable, key, typecode)) { + string_writer_clear(key); + free(key); + PyErr_NoMemory(); + return -1; + } } } return typecode; @@ -767,6 +830,7 @@ static int dtype_num_to_typecode(int type_num) { static int get_cached_typecode(PyArray_Descr* descr) { + TypeofLockGuard guard(typeof_lock); PyObject* tmpobject = PyDict_GetItem(typecache, (PyObject*)descr); if (tmpobject == NULL) return -1; @@ -775,10 +839,14 @@ int get_cached_typecode(PyArray_Descr* descr) { } static -void cache_typecode(PyArray_Descr* descr, int typecode) { +int cache_typecode(PyArray_Descr* descr, int typecode) { PyObject* value = PyLong_FromLong(typecode); - PyDict_SetItem(typecache, (PyObject*)descr, value); + if (value == NULL) + return -1; + TypeofLockGuard guard(typeof_lock); + int ret = PyDict_SetItem(typecache, (PyObject*)descr, value); Py_DECREF(value); + return ret; } static @@ -796,22 +864,35 @@ PyObject* ndarray_key(int ndim, int layout, int readonly, PyArray_Descr* descr) static int get_cached_ndarray_typecode(int ndim, int layout, int readonly, PyArray_Descr* descr) { PyObject* key = ndarray_key(ndim, layout, readonly, descr); + if (key == NULL) + return -1; + TypeofLockGuard guard(typeof_lock); PyObject *tmpobject = PyDict_GetItem(ndarray_typecache, key); - if (tmpobject == NULL) + if (tmpobject == NULL) { + Py_DECREF(key); return -1; + } + int typecode = PyLong_AsLong(tmpobject); Py_DECREF(key); - return PyLong_AsLong(tmpobject); + return typecode; } static -void cache_ndarray_typecode(int ndim, int layout, int readonly, PyArray_Descr* descr, +int cache_ndarray_typecode(int ndim, int layout, int readonly, PyArray_Descr* descr, int typecode) { PyObject* key = ndarray_key(ndim, layout, readonly, descr); PyObject* value = PyLong_FromLong(typecode); - PyDict_SetItem(ndarray_typecache, key, value); + if (key == NULL || value == NULL) { + Py_XDECREF(key); + Py_XDECREF(value); + return -1; + } + TypeofLockGuard guard(typeof_lock); + int ret = PyDict_SetItem(ndarray_typecache, key, value); Py_DECREF(key); Py_DECREF(value); + return ret; } static @@ -847,11 +928,22 @@ int typecode_ndarray(PyObject *dispatcher, PyArrayObject *ary) { assert(ndim <= N_NDIM); assert(dtype < N_DTYPES); - typecode = cached_arycode[ndim - 1][layout][dtype]; + { + TypeofLockGuard guard(typeof_lock); + typecode = cached_arycode[ndim - 1][layout][dtype]; + } if (typecode == -1) { /* First use of this table entry, so it requires populating */ typecode = typecode_fallback_keep_ref(dispatcher, (PyObject*)ary); - cached_arycode[ndim - 1][layout][dtype] = typecode; + if (typecode >= 0) { + TypeofLockGuard guard(typeof_lock); + if (cached_arycode[ndim - 1][layout][dtype] == -1) { + cached_arycode[ndim - 1][layout][dtype] = typecode; + } + else { + typecode = cached_arycode[ndim - 1][layout][dtype]; + } + } } return typecode; @@ -866,9 +958,15 @@ int typecode_ndarray(PyObject *dispatcher, PyArrayObject *ary) { readonly = !PyArray_ISWRITEABLE(ary); typecode = get_cached_ndarray_typecode(ndim, layout, readonly, PyArray_DESCR(ary)); if (typecode == -1) { + if (PyErr_Occurred()) + return -1; /* First use of this type, use fallback and populate the cache */ typecode = typecode_fallback_keep_ref(dispatcher, (PyObject*)ary); - cache_ndarray_typecode(ndim, layout, readonly, PyArray_DESCR(ary), typecode); + if (typecode == -1) + return -1; + if (cache_ndarray_typecode(ndim, layout, readonly, + PyArray_DESCR(ary), typecode)) + return -1; } return typecode; } @@ -885,9 +983,20 @@ int typecode_arrayscalar(PyObject *dispatcher, PyObject* aryscalar) { if (descr->type_num == NPY_VOID) { typecode = get_cached_typecode(descr); if (typecode == -1) { + if (PyErr_Occurred()) { + Py_DECREF(descr); + return -1; + } /* Resolve through fallback then populate cache */ typecode = typecode_fallback_keep_ref(dispatcher, aryscalar); - cache_typecode(descr, typecode); + if (typecode == -1) { + Py_DECREF(descr); + return -1; + } + if (cache_typecode(descr, typecode)) { + Py_DECREF(descr); + return -1; + } } Py_DECREF(descr); return typecode; @@ -1031,6 +1140,17 @@ typeof_init(PyObject *self, PyObject *args) return NULL; } + if (typeof_lock == NULL) { + typeof_lock = PyThread_allocate_lock(); + if (typeof_lock == NULL) { + PyErr_SetString(PyExc_RuntimeError, + "failed to allocate typeof lock"); + return NULL; + } + } + + /* Import-time setup is serialized by Python's import machinery. The + runtime cache operations are protected by typeof_lock after setup. */ #define UNWRAP_TYPE(S) \ if(!(tmpobj = PyDict_GetItemString(dict, #S))) return NULL; \ else { tc_##S = PyLong_AsLong(tmpobj); \ diff --git a/numba_cuda/numba/cuda/cext/mviewbuf.c b/numba_cuda/numba/cuda/cext/mviewbuf.c index 8cc6a5814..957ae377f 100644 --- a/numba_cuda/numba/cuda/cext/mviewbuf.c +++ b/numba_cuda/numba/cuda/cext/mviewbuf.c @@ -38,6 +38,21 @@ static void free_buffer(Py_buffer * buf) PyBuffer_Release(buf); } +static PyObject* +sequence_fast_get_item_ref(PyObject *seq, Py_ssize_t index) +{ + PyObject *item; + +#if PY_VERSION_HEX >= 0x030d0000 + if (PyList_Check(seq)) { + return PyList_GetItemRef(seq, index); + } +#endif + item = PySequence_Fast_GET_ITEM(seq, index); + Py_XINCREF(item); + return item; +} + /** * Return a pointer to the data of a writable buffer from obj. If only a * read-only buffer is available and force is True, a read-write buffer based on @@ -161,6 +176,7 @@ memoryview_get_extents_info(PyObject *self, PyObject *args) PyObject *shape_tuple = NULL; PyObject *strides_tuple = NULL; PyObject *shape = NULL, *strides = NULL; + PyObject *item = NULL; Py_ssize_t itemsize = 0; int ndim = 0; PyObject* res = NULL; @@ -174,33 +190,54 @@ memoryview_get_extents_info(PyObject *self, PyObject *args) } if (itemsize <= 0) { - PyErr_SetString(PyExc_ValueError, "ndim <= 0"); + PyErr_SetString(PyExc_ValueError, "itemsize <= 0"); goto cleanup; } - shape_ary = malloc(sizeof(Py_ssize_t) * ndim + 1); - strides_ary = malloc(sizeof(Py_ssize_t) * ndim + 1); + if (ndim > 0) { + shape_ary = malloc(sizeof(Py_ssize_t) * ndim); + strides_ary = malloc(sizeof(Py_ssize_t) * ndim); + if (shape_ary == NULL || strides_ary == NULL) { + PyErr_NoMemory(); + goto cleanup; + } + } shape_tuple = PySequence_Fast(shape, "shape is not a sequence"); if (!shape_tuple) goto cleanup; + if (PySequence_Fast_GET_SIZE(shape_tuple) < ndim) { + PyErr_SetString(PyExc_ValueError, "shape is shorter than ndim"); + goto cleanup; + } for (i = 0; i < ndim; ++i) { - shape_ary[i] = PyNumber_AsSsize_t( - PySequence_Fast_GET_ITEM(shape_tuple, i), - PyExc_OverflowError); + item = sequence_fast_get_item_ref(shape_tuple, i); + if (item == NULL) goto cleanup; + shape_ary[i] = PyNumber_AsSsize_t(item, PyExc_OverflowError); + Py_DECREF(item); + item = NULL; + if (shape_ary[i] == -1 && PyErr_Occurred()) goto cleanup; } strides_tuple = PySequence_Fast(strides, "strides is not a sequence"); if (!strides_tuple) goto cleanup; + if (PySequence_Fast_GET_SIZE(strides_tuple) < ndim) { + PyErr_SetString(PyExc_ValueError, "strides is shorter than ndim"); + goto cleanup; + } for (i = 0; i < ndim; ++i) { - strides_ary[i] = PyNumber_AsSsize_t( - PySequence_Fast_GET_ITEM(strides_tuple, i), - PyExc_OverflowError); + item = sequence_fast_get_item_ref(strides_tuple, i); + if (item == NULL) goto cleanup; + strides_ary[i] = PyNumber_AsSsize_t(item, PyExc_OverflowError); + Py_DECREF(item); + item = NULL; + if (strides_ary[i] == -1 && PyErr_Occurred()) goto cleanup; } res = get_extents(shape_ary, strides_ary, ndim, itemsize, 0); cleanup: + Py_XDECREF(item); free(shape_ary); free(strides_ary); Py_XDECREF(shape_tuple); @@ -375,7 +412,7 @@ static PyMethodDef core_methods[] = { MOD_INIT(mviewbuf) { PyObject *module; - MOD_DEF(module, "mviewbuf", "No docs", core_methods) + MOD_DEF_NOGIL(module, "mviewbuf", "No docs", core_methods) if (module == NULL) return MOD_ERROR_VAL; diff --git a/numba_cuda/numba/cuda/cext/typeconv.cpp b/numba_cuda/numba/cuda/cext/typeconv.cpp index 5af7d16b3..1f8dcaad6 100644 --- a/numba_cuda/numba/cuda/cext/typeconv.cpp +++ b/numba_cuda/numba/cuda/cext/typeconv.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include "typeconv.hpp" @@ -68,15 +69,18 @@ inline bool Rating::operator == (const Rating &other) const { // ------ TypeManager ------ bool TypeManager::canPromote(Type from, Type to) const { - return isCompatible(from, to) == TCC_PROMOTE; + std::lock_guard guard(mutex); + return _isCompatible(from, to) == TCC_PROMOTE; } bool TypeManager::canSafeConvert(Type from, Type to) const { - return isCompatible(from, to) == TCC_CONVERT_SAFE; + std::lock_guard guard(mutex); + return _isCompatible(from, to) == TCC_CONVERT_SAFE; } bool TypeManager::canUnsafeConvert(Type from, Type to) const { - return isCompatible(from, to) == TCC_CONVERT_UNSAFE; + std::lock_guard guard(mutex); + return _isCompatible(from, to) == TCC_CONVERT_UNSAFE; } void TypeManager::addPromotion(Type from, Type to) { @@ -92,11 +96,17 @@ void TypeManager::addSafeConversion(Type from, Type to) { } void TypeManager::addCompatibility(Type from, Type to, TypeCompatibleCode tcc) { + std::lock_guard guard(mutex); TypePair pair(from, to); tccmap.insert(pair, tcc); } TypeCompatibleCode TypeManager::isCompatible(Type from, Type to) const { + std::lock_guard guard(mutex); + return _isCompatible(from, to); +} + +TypeCompatibleCode TypeManager::_isCompatible(Type from, Type to) const { if (from == to) return TCC_EXACT; TypePair pair(from, to); @@ -109,6 +119,7 @@ int TypeManager::selectOverload(const Type sig[], const Type ovsigs[], int sigsz, int ovct, bool allow_unsafe, bool exact_match_required ) const { + std::lock_guard guard(mutex); int count; if (ovct <= 16) { Rating ratings[16]; @@ -142,7 +153,7 @@ int TypeManager::_selectOverload(const Type sig[], const Type ovsigs[], Rating rate; for (int j = 0; j < sigsz; ++j) { - TypeCompatibleCode tcc = isCompatible(sig[j], entry[j]); + TypeCompatibleCode tcc = _isCompatible(sig[j], entry[j]); if (tcc == TCC_FALSE || (tcc == TCC_CONVERT_UNSAFE && !allow_unsafe) || (tcc != TCC_EXACT && exact_match_required)) { diff --git a/numba_cuda/numba/cuda/cext/typeconv.hpp b/numba_cuda/numba/cuda/cext/typeconv.hpp index da5d87a77..fd7b48a8a 100644 --- a/numba_cuda/numba/cuda/cext/typeconv.hpp +++ b/numba_cuda/numba/cuda/cext/typeconv.hpp @@ -3,6 +3,7 @@ #ifndef NUMBA_TYPECONV_HPP_ #define NUMBA_TYPECONV_HPP_ +#include #include #include @@ -86,11 +87,14 @@ class TypeManager{ ) const; private: + TypeCompatibleCode _isCompatible(Type from, Type to) const; + int _selectOverload(const Type sig[], const Type ovsigs[], int &selected, int sigsz, int ovct, bool allow_unsafe, bool exact_match_required, Rating ratings[], int candidates[]) const; + mutable std::mutex mutex; TCCMap tccmap; }; diff --git a/numba_cuda/numba/cuda/cudadrv/driver.py b/numba_cuda/numba/cuda/cudadrv/driver.py index a981d614d..73120fabd 100644 --- a/numba_cuda/numba/cuda/cudadrv/driver.py +++ b/numba_cuda/numba/cuda/cudadrv/driver.py @@ -263,6 +263,18 @@ def __init__(self): self.is_initialized = False self.initialization_error = None self.pid = None + # Serializes lazy driver initialization so concurrent first-touch from + # multiple threads (notably under free-threaded CPython) cannot observe + # a half-initialized driver. Reentrant because resolving ``cuInit`` + # through ``__getattr__`` calls ``ensure_initialized`` again on the same + # thread. Guard the lock and the in-progress flag separately: + # ``Driver`` is a singleton whose ``__init__`` can be re-entered, and + # each attribute must exist by the time any re-entered ``__init__`` + # returns without clobbering state from an active initialization. + if not hasattr(self, "_initializing"): + self._initializing = False + if not hasattr(self, "_initialization_lock"): + self._initialization_lock = threading.RLock() try: if config.DISABLE_CUDA: msg = ( @@ -280,20 +292,37 @@ def ensure_initialized(self): if self.is_initialized: return - # lazily initialize logger - global _logger - _logger = make_logger() + with self._initialization_lock: + # Another thread may have completed initialization while we waited. + if self.is_initialized: + return + # Resolving ``self.cuInit`` below goes through ``__getattr__``, + # which calls back into ``ensure_initialized`` on this thread. + # ``_initializing`` breaks that recursion without publishing a + # half-initialized driver to threads blocked on the lock. + if self._initializing: + return + self._initializing = True - self.is_initialized = True - try: - _logger.info("init") - self.cuInit(0) - except CudaAPIError as e: - description = f"{e.msg} ({e.code})" - self.initialization_error = description - raise CudaSupportError(f"Error at driver init: {description}") - else: - self.pid = _getpid() + # lazily initialize logger + global _logger + _logger = make_logger() + + try: + _logger.info("init") + self.cuInit(0) + except CudaAPIError as e: + description = f"{e.msg} ({e.code})" + self.initialization_error = description + raise CudaSupportError(f"Error at driver init: {description}") + else: + self.pid = _getpid() + finally: + # Publish completion (success *or* a cached failure) only after + # cuInit has actually run, so the fast-path guard above never + # lets another thread proceed before the driver is ready. + self.is_initialized = True + self._initializing = False @property def is_available(self): @@ -301,6 +330,13 @@ def is_available(self): return self.initialization_error is None def __getattr__(self, fname): + # Only CUDA driver API entry points (e.g. ``cuInit``) are bound lazily + # here. Internal attributes never start with an underscore, so refuse + # to "initialize and bind" those -- doing so would recurse infinitely + # when ``ensure_initialized`` touches its own lock before it is set. + if fname.startswith("_"): + raise AttributeError(fname) + # First request of a driver API function self.ensure_initialized() @@ -985,6 +1021,11 @@ def __init__(self, capacity=_SizeNotSet): self._disable_count = 0 self._size = 0 self.memory_capacity = capacity + # Deallocations are driven by weakref finalizers, which can run on any + # thread and concurrently under free-threaded CPython. Serialize the + # pending list, size accounting and disable counter so a concurrent + # add_item/clear cannot race (e.g. popleft from an emptied deque). + self._lock = threading.RLock() @property def _max_pending_bytes(self): @@ -1000,12 +1041,14 @@ def add_item(self, dtor, handle, size=_SizeNotSet): resources (e.g. CUModule) has an unknown memory footprint on the device. """ _logger.info("add pending dealloc: %s %s bytes", dtor.__name__, size) - self._cons.append((dtor, handle, size)) - self._size += int(size) - if ( - len(self._cons) > config.CUDA_DEALLOCS_COUNT - or self._size > self._max_pending_bytes - ): + with self._lock: + self._cons.append((dtor, handle, size)) + self._size += int(size) + should_clear = ( + len(self._cons) > config.CUDA_DEALLOCS_COUNT + or self._size > self._max_pending_bytes + ) + if should_clear: self.clear() def clear(self): @@ -1013,36 +1056,46 @@ def clear(self): Flush any pending deallocations unless it is disabled. Do nothing if disabled. """ - if not self.is_disabled: - while self._cons: - [dtor, handle, size] = self._cons.popleft() - _logger.info("dealloc: %s %s bytes", dtor.__name__, size) - dtor(handle) - + # Atomically take ownership of the pending list, then run the + # destructors without holding the lock: they call into the CUDA driver + # and must not block other threads queueing deallocations. + with self._lock: + if self.is_disabled: + return + pending = list(self._cons) + self._cons.clear() self._size = 0 + for dtor, handle, size in pending: + _logger.info("dealloc: %s %s bytes", dtor.__name__, size) + dtor(handle) + @contextlib.contextmanager def disable(self): """ Context manager to temporarily disable flushing pending deallocation. This can be nested. """ - self._disable_count += 1 + with self._lock: + self._disable_count += 1 try: yield finally: - self._disable_count -= 1 - assert self._disable_count >= 0 + with self._lock: + self._disable_count -= 1 + assert self._disable_count >= 0 @property def is_disabled(self): - return self._disable_count > 0 + with self._lock: + return self._disable_count > 0 def __len__(self): """ Returns number of pending deallocations. """ - return len(self._cons) + with self._lock: + return len(self._cons) MemoryInfo = namedtuple("MemoryInfo", "free,total") diff --git a/numba_cuda/numba/cuda/dispatcher.py b/numba_cuda/numba/cuda/dispatcher.py index 3b3b30a0b..b940d1335 100644 --- a/numba_cuda/numba/cuda/dispatcher.py +++ b/numba_cuda/numba/cuda/dispatcher.py @@ -857,6 +857,42 @@ def load_overload(self, sig, target_context): with utils.numba_target_override(): return super().load_overload(sig, target_context) + def load_launch_config_sensitive_overload(self, sig, target_context): + with utils.numba_target_override(): + target_context.refresh() + with self._guard_against_spurious_io_errors(): + return self._load_launch_config_sensitive_overload( + sig, target_context + ) + + return None + + def _load_launch_config_sensitive_overload(self, sig, target_context): + if not self._enabled or not self.is_launch_config_sensitive(): + return None + + base_key = super()._index_key(sig, target_context.codegen()) + for key in self._cache_file._load_index(): + if len(key) != len(base_key) + 1: + continue + if key[: len(base_key)] != base_key: + continue + + launch_config_entry = key[-1] + if not ( + isinstance(launch_config_entry, tuple) + and len(launch_config_entry) == 2 + and launch_config_entry[0] == "launch_config" + and launch_config_entry[1] != _NO_LAUNCH_CONFIG_CACHE_KEY + ): + continue + + data = self._cache_file.load(key) + if data is not None: + return self._impl.rebuild(target_context, data) + + return None + class OmittedArg: """ @@ -1722,6 +1758,7 @@ def _configure_cache_for_launch_config(self, launch_config): return self._cache.set_launch_config_key(None) + @global_compiler_lock def _get_launch_config_specialization(self, key): dispatcher = self._launch_config_specializations.get(key) if dispatcher is None: @@ -1773,11 +1810,27 @@ def _update_launch_config_sensitivity(self, kernel, launch_config): return if not self._launch_config_sensitive: self._launch_config_sensitive = True + if launch_config is None: + launch_config = launchconfig.current_launch_config() + if launch_config is None: + return if self._launch_config_default_key is None: self._launch_config_default_key = self._launch_config_key( launch_config ) + def _requires_launch_config_specialization(self, launch_config): + if self._launch_config_is_specialized: + return False + if not self._launch_config_sensitive: + return False + if self._launch_config_default_key is None: + return False + return ( + self._launch_config_key(launch_config) + != self._launch_config_default_key + ) + def forall(self, ntasks, tpb=0, stream=0, sharedmem=0): """Returns a 1D-configured dispatcher for a given number of tasks. @@ -1849,6 +1902,12 @@ def call(self, args, launch_config): self._update_launch_config_sensitivity(kernel, launch_config) + if self._requires_launch_config_specialization(launch_config): + dispatcher = self._get_launch_config_specialization( + self._launch_config_key(launch_config) + ) + return dispatcher.call(args, launch_config) + for callback in launch_config.pre_launch_callbacks: callback(kernel, launch_config) @@ -1867,6 +1926,7 @@ def typeof_pyval(self, val): # the CUDA Array Interface. return typeof(val, Purpose.argument) + @global_compiler_lock def specialize(self, *args): """ Create a new instance of this dispatcher specialized for the given @@ -2135,6 +2195,23 @@ def compile(self, sig): self._cache_launch_config_key(launch_config) ) + if ( + kernel is None + and launch_config is None + and isinstance(self._cache, CUDACache) + and self._cache.is_launch_config_sensitive() + ): + kernel = self._cache.load_launch_config_sensitive_overload( + sig, self.targetctx + ) + if kernel is not None: + self._cache_hits[sig] += 1 + self._launch_config_sensitive = True + # A no-launch direct compile cannot safely bind a concrete + # launch-config-specific kernel as the default. Defer binding + # until a real launch supplies the launch configuration. + return kernel + if kernel is not None: self._cache_hits[sig] += 1 else: @@ -2158,6 +2235,7 @@ def compile(self, sig): self._cache = NullCache() self._cache.save_overload(sig, kernel) + self._update_launch_config_sensitivity(kernel, launch_config) self.add_overload(kernel, argtypes) return kernel diff --git a/numba_cuda/numba/cuda/tests/cudadrv/test_deallocations.py b/numba_cuda/numba/cuda/tests/cudadrv/test_deallocations.py index 63af86a3f..131651190 100644 --- a/numba_cuda/numba/cuda/tests/cudadrv/test_deallocations.py +++ b/numba_cuda/numba/cuda/tests/cudadrv/test_deallocations.py @@ -115,6 +115,65 @@ def test_nested_defer_cleanup(self): deallocs.clear() self.assertEqual(len(deallocs), 0) + @skip_if_external_memmgr("Deallocation specific to Numba memory management") + def test_concurrent_add_and_clear(self): + # Regression: _PendingDeallocs must tolerate concurrent add_item/clear. + # Deallocations run from weakref finalizers on arbitrary threads, which + # execute concurrently under free-threaded CPython; an unsynchronized + # "while self._cons: popleft()" raced and raised IndexError. + import threading + + from numba.cuda.cudadrv.driver import _PendingDeallocs + + # Ensure the driver is initialized (sets up the module logger that the + # dealloc path logs through; deallocations only ever run post-init). + cuda.current_context() + + deallocs = _PendingDeallocs(capacity=2**30) + + freed = [] + freed_lock = threading.Lock() + + def dtor(handle): + with freed_lock: + freed.append(handle) + + n_threads = 16 + per_thread = 500 + barrier = threading.Barrier(n_threads) + errors = [] + errors_lock = threading.Lock() + + def worker(base): + try: + barrier.wait(timeout=30) + for i in range(per_thread): + # Unique handle per queued item, size=1 so the count and + # byte thresholds both drive frequent concurrent clear()s. + deallocs.add_item(dtor, base + i, size=1) + except BaseException as e: # noqa: BLE001 + with errors_lock: + errors.append(e) + + threads = [ + threading.Thread(target=worker, args=(t * per_thread,)) + for t in range(n_threads) + ] + for t in threads: + t.start() + for t in threads: + t.join(timeout=60) + + # Flush whatever remains pending. + deallocs.clear() + + self.assertFalse(any(t.is_alive() for t in threads)) + self.assertEqual(errors, []) + self.assertEqual(len(deallocs), 0) + # Every queued item is freed exactly once. + self.assertEqual(len(freed), n_threads * per_thread) + self.assertEqual(len(set(freed)), n_threads * per_thread) + @skip_if_external_memmgr("Deallocation specific to Numba memory management") def test_exception(self): harr = np.arange(5) diff --git a/numba_cuda/numba/cuda/tests/cudadrv/test_init.py b/numba_cuda/numba/cuda/tests/cudadrv/test_init.py index d018de5ea..3d002487d 100644 --- a/numba_cuda/numba/cuda/tests/cudadrv/test_init.py +++ b/numba_cuda/numba/cuda/tests/cudadrv/test_init.py @@ -68,6 +68,54 @@ def cuda_disabled_test(): return success, msg +# Run in a fresh (spawned) child so the CUDA driver is uninitialized: many +# threads touch the driver for the very first time simultaneously. If +# Driver.ensure_initialized() is not properly serialized, a thread can observe +# is_initialized=True before cuInit() has completed and call a driver API on an +# uninitialized driver (CUDA_ERROR_NOT_INITIALIZED). Returns the list of errors +# seen by the worker threads (empty == success). +def concurrent_first_touch_test(num_threads=32): + import threading + import time + + import numpy as np + + barrier = threading.Barrier(num_threads) + errors = [] + errors_lock = threading.Lock() + + def touch(): + try: + barrier.wait(timeout=30) + # First driver touch for this thread; all threads arrive together. + cuda.to_device(np.zeros(1, dtype=np.float32)) + except BaseException as e: # noqa: BLE001 + with errors_lock: + errors.append(repr(e)) + + threads = [ + threading.Thread( + target=touch, name=f"cuda-first-touch-{i}", daemon=True + ) + for i in range(num_threads) + ] + for t in threads: + t.start() + join_timeout = 30 + deadline = time.monotonic() + join_timeout + for t in threads: + t.join(timeout=max(deadline - time.monotonic(), 0)) + timed_out = [t.name for t in threads if t.is_alive()] + if timed_out: + with errors_lock: + errors.append( + f"{len(timed_out)} worker thread(s) did not finish within " + f"{join_timeout}s: {', '.join(timed_out)}" + ) + + return errors + + # Similar to cuda_disabled_test, but checks cuda.cuda_error() instead of the # exception raised on initialization def cuda_disabled_error_test(): @@ -133,6 +181,18 @@ def test_init_success(self): # there is no error recorded. self.assertIsNone(cuda.cuda_error()) + def test_concurrent_initialization_no_race(self): + # Concurrent first-touch of the driver from many threads must not + # observe a half-initialized driver. Regression test for the + # ensure_initialized() race that fails reliably under free-threaded + # CPython. Runs in a spawned subprocess so the driver starts cold. + with concurrent.futures.ProcessPoolExecutor( + mp_context=mp.get_context("spawn") + ) as exe: + errors = exe.submit(concurrent_first_touch_test).result(timeout=60) + + self.assertEqual(errors, []) + if __name__ == "__main__": unittest.main() diff --git a/numba_cuda/numba/cuda/tests/cudapy/test_caching.py b/numba_cuda/numba/cuda/tests/cudapy/test_caching.py index 4db395ccd..786236d62 100644 --- a/numba_cuda/numba/cuda/tests/cudapy/test_caching.py +++ b/numba_cuda/numba/cuda/tests/cudapy/test_caching.py @@ -1,6 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: BSD-2-Clause +from concurrent.futures import ThreadPoolExecutor import multiprocessing import json import os @@ -10,6 +11,7 @@ import sys import stat import subprocess +import threading from numba import cuda from numba.cuda.core.errors import NumbaWarning @@ -23,14 +25,17 @@ ) from numba.cuda.tests.support import ( TestCase, - temp_directory, import_dynamic, + launch_subprocess_code, + subprocess_marker_results, + temp_directory, ) import numpy as np from pickle import PicklingError # Module-level global for testing that caching rejects global device arrays GLOBAL_DEVICE_ARRAY = None +_LCS_CACHE_RESULT = "__LCS_CACHE_RESULT__" class BaseCacheTest(TestCase): @@ -495,12 +500,20 @@ def test_launch_config_sensitive_cache_keys(self): report = self.run_in_separate_process( report_code="\n".join( [ - "main_hits = sum(mod.lcs_cache_kernel.stats.cache_hits.values())", - "main_misses = sum(mod.lcs_cache_kernel.stats.cache_misses.values())", - "spec = next(iter(mod.lcs_cache_kernel._launch_config_specializations.values()))", + "main_stats = mod.lcs_cache_kernel.stats", + "main_hits = sum(main_stats.cache_hits.values())", + "main_misses = sum(main_stats.cache_misses.values())", + "specs = mod.lcs_cache_kernel", + "specs = specs._launch_config_specializations", + "spec = next(iter(specs.values()))", "spec_hits = sum(spec.stats.cache_hits.values())", "spec_misses = sum(spec.stats.cache_misses.values())", - "report = {'main_hits': main_hits, 'main_misses': main_misses, 'spec_hits': spec_hits, 'spec_misses': spec_misses}", + "report = {", + "'main_hits': main_hits,", + "'main_misses': main_misses,", + "'spec_hits': spec_hits,", + "'spec_misses': spec_misses,", + "}", ] ) ) @@ -510,17 +523,77 @@ def test_launch_config_sensitive_cache_keys(self): self.assertEqual(report["spec_misses"], 0) self.assertEqual(self.get_cache_mtimes(), mtimes) - def test_launch_config_sensitive_compile_requires_active_launch_config( + def test_launch_config_sensitive_compile_defers_default_launch_config( self, ): mod = self.import_module() mod.launch(32) mod.launch(64) sig = mod.lcs_cache_kernel.signatures[0] + mtimes = self.get_cache_mtimes() mod2 = self.import_module() - with self.assertRaisesRegex(RuntimeError, "No launch config set"): - mod2.lcs_cache_kernel.compile(sig) + mod2.lcs_cache_kernel.compile(sig) + self.assertTrue(mod2.lcs_cache_kernel._launch_config_sensitive) + self.assertIsNone(mod2.lcs_cache_kernel._launch_config_default_key) + self.assertEqual(mod2.lcs_cache_kernel.overloads, {}) + self.assertEqual(self.get_cache_mtimes(), mtimes) + + arr = mod2.launch(64) + self.assertEqual(arr[0], 1) + self.assertIsNotNone(mod2.lcs_cache_kernel._launch_config_default_key) + self.assertEqual(self.get_cache_mtimes(), mtimes) + stats = mod2.lcs_cache_kernel.stats + self.assertGreaterEqual(sum(stats.cache_hits.values()), 1) + self.assertEqual(sum(stats.cache_misses.values()), 0) + + def test_concurrent_launch_config_sensitive_cold_cache_threads(self): + self.check_pycache(0) + mod = self.import_module() + blockdims = (32, 64, 32, 64) + barrier = threading.Barrier(len(blockdims)) + + def launch(blockdim): + barrier.wait(timeout=10) + return int(mod.launch(blockdim)[0]) + + with ThreadPoolExecutor(max_workers=len(blockdims)) as executor: + futures = [ + executor.submit(launch, blockdim) for blockdim in blockdims + ] + results = [future.result(timeout=60) for future in futures] + + self.assertEqual(results, [1] * len(blockdims)) + self.assertEqual(self.count_cache_markers(), 1) + cache_contents = self.cache_contents() + self.assertTrue(any(fn.endswith(".nbi") for fn in cache_contents)) + + def test_concurrent_launch_config_sensitive_cold_cache_processes(self): + self.check_pycache(0) + blockdims = (32, 64, 32, 64) + processes = [ + subprocess.Popen( + [ + sys.executable, + "-c", + launch_subprocess_code( + self.tempdir, + self.modname, + blockdim, + _LCS_CACHE_RESULT, + ), + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + for blockdim in blockdims + ] + results = subprocess_marker_results(processes, _LCS_CACHE_RESULT) + + self.assertEqual(results, [1] * len(blockdims)) + self.assertEqual(self.count_cache_markers(), 1) + cache_contents = self.cache_contents() + self.assertTrue(any(fn.endswith(".nbi") for fn in cache_contents)) @skip_on_cudasim("Simulator does not implement caching") diff --git a/numba_cuda/numba/cuda/tests/cudapy/test_dispatcher.py b/numba_cuda/numba/cuda/tests/cudapy/test_dispatcher.py index 06efe7f4d..aa8e55461 100644 --- a/numba_cuda/numba/cuda/tests/cudapy/test_dispatcher.py +++ b/numba_cuda/numba/cuda/tests/cudapy/test_dispatcher.py @@ -1,9 +1,11 @@ # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: BSD-2-Clause +from concurrent.futures import ThreadPoolExecutor from cuda.core._utils.cuda_utils import CUDAError import numpy as np import threading +import time from numba.cuda.types import ( boolean, @@ -15,6 +17,7 @@ void, ) from numba import cuda +import numba.cuda.dispatcher as dispatcher_module from numba.cuda import config, types from numba.cuda import launchconfig from numba.cuda.core.errors import TypingError @@ -100,6 +103,59 @@ def f(x): self.assertEqual(len(f.specializations), 2) self.assertIsNot(f_int32, f_float32) + def test_concurrent_specialize_cache_same(self): + @cuda.jit + def f(x): + x[0] += 1 + + original_kernel = dispatcher_module._Kernel + constructed_kernels = [] + constructed_lock = threading.Lock() + + class CountingKernel(original_kernel): + def __init__(self, *args, **kwargs): + with constructed_lock: + constructed_kernels.append(threading.get_ident()) + # Widen the race window so concurrent specialize() calls all + # reach the shared specialization cache lookup before the first + # kernel can be inserted. + time.sleep(0.05) + super().__init__(*args, **kwargs) + + barrier = threading.Barrier(16) + specializations = [] + specializations_lock = threading.Lock() + errors = [] + errors_lock = threading.Lock() + + def specialize(): + try: + arr = np.zeros(1, dtype=np.int32) + barrier.wait(timeout=10) + specialization = f.specialize(arr) + with specializations_lock: + specializations.append(specialization) + except BaseException as e: + with errors_lock: + errors.append(e) + + threads = [threading.Thread(target=specialize) for _ in range(16)] + dispatcher_module._Kernel = CountingKernel + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=20) + finally: + dispatcher_module._Kernel = original_kernel + + self.assertFalse(any(thread.is_alive() for thread in threads)) + self.assertEqual(errors, []) + self.assertEqual(len(specializations), len(threads)) + self.assertEqual(len({id(s) for s in specializations}), 1) + self.assertEqual(len(f.specializations), 1) + self.assertEqual(len(constructed_kernels), 1) + def test_specialize_cache_same_with_ordering(self): # Ensure that the same dispatcher is returned for the same argument # types, and that different dispatchers are returned for different @@ -354,6 +410,213 @@ def wrapper(): t.join() self.assertFalse(errors) + @skip_on_cudasim("Simulator doesn't compile CUDA kernels") + def test_concurrent_launch_same_signature_compiles_once(self): + constructed_kernels = [] + constructed_lock = threading.Lock() + original_kernel = dispatcher_module._Kernel + + class CountingKernel(original_kernel): + def __init__(self, *args, **kwargs): + with constructed_lock: + constructed_kernels.append(threading.get_ident()) + # Widen the first compile window. Without synchronization + # around the dispatcher overload cache, racing launchers would + # all build their own kernel for the same signature. + time.sleep(0.05) + super().__init__(*args, **kwargs) + + @cuda.jit + def foo(r, x): + r[0] = x + 1 + + barrier = threading.Barrier(16) + errors = [] + errors_lock = threading.Lock() + + def launch(): + try: + r = np.zeros(1, dtype=np.int64) + barrier.wait(timeout=10) + foo[1, 1](r, 1) + self.assertEqual(r[0], 2) + except BaseException as e: + with errors_lock: + errors.append(e) + + threads = [threading.Thread(target=launch) for _ in range(16)] + dispatcher_module._Kernel = CountingKernel + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=20) + finally: + dispatcher_module._Kernel = original_kernel + + self.assertFalse(any(thread.is_alive() for thread in threads)) + self.assertEqual(errors, []) + self.assertEqual(len(foo.overloads), 1) + self.assertEqual(len(constructed_kernels), 1) + + @skip_on_cudasim("Simulator doesn't compile CUDA kernels") + def test_concurrent_launch_many_signatures_and_fresh_kernels(self): + dtypes = (np.int16, np.int32, np.int64, np.float32, np.float64) + + def make_kernel(): + @cuda.jit + def kernel(x, y): + i = cuda.grid(1) + if i < x.size: + y[i] = x[i] + 1 + + return kernel + + shared_kernel = make_kernel() + layout_kernel = make_kernel() + + def launch_shared(seed, barrier): + barrier.wait(timeout=10) + for i in range(4): + dtype = dtypes[(seed + i) % len(dtypes)] + n = i + 4 + host = np.arange(n, dtype=dtype) + inp = cuda.to_device(host) + out = cuda.device_array(n, dtype=dtype) + shared_kernel[1, 32](inp, out) + np.testing.assert_array_equal(out.copy_to_host(), host + 1) + + def launch_fresh(seed, barrier): + barrier.wait(timeout=10) + kernel = make_kernel() + dtype = dtypes[seed % len(dtypes)] + host = np.arange(32, dtype=dtype) + inp = cuda.to_device(host) + out = cuda.device_array(host.size, dtype=dtype) + kernel[1, 32](inp, out) + np.testing.assert_array_equal(out.copy_to_host(), host + 1) + + def launch_layouts(seed, barrier): + barrier.wait(timeout=10) + for i in range(4): + dtype = dtypes[(seed + i) % len(dtypes)] + base = np.arange(64, dtype=dtype).reshape(8, 8) + variants = ( + base.ravel(), + np.asfortranarray(base).ravel(order="F"), + base[::2].ravel(), + ) + host = np.ascontiguousarray(variants[i % len(variants)]) + inp = cuda.to_device(host) + out = cuda.device_array(host.size, dtype=dtype) + layout_kernel[1, 64](inp, out) + np.testing.assert_array_equal(out.copy_to_host(), host + 1) + + workers = ( + launch_shared, + launch_shared, + launch_shared, + launch_fresh, + launch_fresh, + launch_layouts, + ) + barrier = threading.Barrier(len(workers)) + with ThreadPoolExecutor(max_workers=len(workers)) as executor: + futures = [ + executor.submit(worker, i, barrier) + for i, worker in enumerate(workers) + ] + for future in futures: + future.result() + + @skip_on_cudasim("Simulator doesn't compile CUDA kernels") + def test_concurrent_compile_failures_do_not_corrupt_dispatcher(self): + @cuda.jit + def bad_kernel(x): + x[0] = undefined_global_name # noqa: F821 + + @cuda.jit + def good_kernel(x): + i = cuda.grid(1) + if i < x.size: + x[i] += 1 + + def compile_bad_then_launch_good(): + for _ in range(4): + with self.assertRaises(TypingError): + bad_kernel[1, 1](np.zeros(1, dtype=np.int32)) + + arr = cuda.to_device(np.zeros(8, dtype=np.int32)) + good_kernel[1, 8](arr) + np.testing.assert_array_equal( + arr.copy_to_host(), np.ones(8, dtype=np.int32) + ) + + with ThreadPoolExecutor(max_workers=4) as executor: + futures = [ + executor.submit(compile_bad_then_launch_good) for _ in range(4) + ] + for future in futures: + future.result() + + @skip_on_cudasim("Simulator doesn't compile CUDA kernels") + def test_concurrent_configured_launch_callback_mutation(self): + @cuda.jit + def kernel(x): + i = cuda.grid(1) + if i < x.size: + x[i] += 1 + + hits = [] + hits_lock = threading.Lock() + + def callback(kernel, launch_config): + with hits_lock: + hits.append((kernel, launch_config)) + + launch_config = kernel.configure(1, 64) + launch_config.pre_launch_callbacks[:] = [callback] + + def launch(seed, barrier): + arr = cuda.to_device(np.zeros(64, dtype=np.int32)) + barrier.wait(timeout=10) + for _ in range(50): + launch_config(arr) + self.assertGreaterEqual(arr.copy_to_host()[0], 1) + + def mutate_callbacks(seed, barrier): + barrier.wait(timeout=10) + for _ in range(50): + launch_config.pre_launch_callbacks.append(callback) + if len(launch_config.pre_launch_callbacks) > 10: + del launch_config.pre_launch_callbacks[:] + time.sleep(0) + + workers = ( + launch, + launch, + launch, + launch, + mutate_callbacks, + mutate_callbacks, + ) + barrier = threading.Barrier(len(workers)) + try: + with ThreadPoolExecutor(max_workers=len(workers)) as executor: + futures = [ + executor.submit(worker, i, barrier) + for i, worker in enumerate(workers) + ] + for future in futures: + future.result() + if not hits: + launch_config.pre_launch_callbacks[:] = [callback] + launch_config(cuda.to_device(np.zeros(64, dtype=np.int32))) + finally: + del launch_config.pre_launch_callbacks[:] + + self.assertGreater(len(hits), 0) + def _test_explicit_signatures(self, sigs): f = cuda.jit(sigs)(add_kernel) diff --git a/numba_cuda/numba/cuda/tests/cudapy/test_launch_config_sensitive.py b/numba_cuda/numba/cuda/tests/cudapy/test_launch_config_sensitive.py index f01187d8e..8125758d5 100644 --- a/numba_cuda/numba/cuda/tests/cudapy/test_launch_config_sensitive.py +++ b/numba_cuda/numba/cuda/tests/cudapy/test_launch_config_sensitive.py @@ -2,8 +2,11 @@ # SPDX-License-Identifier: BSD-2-Clause import numpy as np +import threading +import time from numba import cuda +import numba.cuda.dispatcher as dispatcher_module from numba.cuda import launchconfig from numba.cuda.core.rewrites import register_rewrite, Rewrite from numba.cuda.testing import skip_on_cudasim, unittest, CUDATestCase @@ -81,6 +84,189 @@ def launch_config_sensitive_kernel(x): self.assertEqual(LAUNCH_CONFIG_LOG[1]["blockdim"], (64, 1, 1)) self.assertEqual(LAUNCH_CONFIG_LOG[1]["griddim"], (1, 1, 1)) + def test_concurrent_launch_config_specialization_compiles_once(self): + @cuda.jit + def launch_config_sensitive_kernel(x): + x[0] = 1 + + arr = np.zeros(1, dtype=np.int32) + launch_config_sensitive_kernel[1, 32](arr) + self.assertTrue(launch_config_sensitive_kernel._launch_config_sensitive) + self.assertEqual(len(LAUNCH_CONFIG_LOG), 1) + + original_init = dispatcher_module.CUDADispatcher.__init__ + constructed_dispatchers = [] + constructed_lock = threading.Lock() + + def counting_init(self, *args, **kwargs): + with constructed_lock: + constructed_dispatchers.append(threading.get_ident()) + # Widen the race window so concurrent launches all see the missing + # launch-config specialization before the first one can finish + # construction. + time.sleep(0.05) + return original_init(self, *args, **kwargs) + + barrier = threading.Barrier(16) + errors = [] + errors_lock = threading.Lock() + + def launch(): + try: + local = np.zeros(1, dtype=np.int32) + barrier.wait(timeout=10) + launch_config_sensitive_kernel[1, 64](local) + self.assertEqual(local[0], 1) + except BaseException as e: + with errors_lock: + errors.append(e) + + threads = [threading.Thread(target=launch) for _ in range(16)] + dispatcher_module.CUDADispatcher.__init__ = counting_init + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=20) + finally: + dispatcher_module.CUDADispatcher.__init__ = original_init + + self.assertFalse(any(thread.is_alive() for thread in threads)) + self.assertEqual(errors, []) + self.assertEqual( + len(launch_config_sensitive_kernel._launch_config_specializations), + 1, + ) + self.assertEqual(len(constructed_dispatchers), 1) + self.assertEqual(len(LAUNCH_CONFIG_LOG), 2) + self.assertEqual(LAUNCH_CONFIG_LOG[1]["blockdim"], (64, 1, 1)) + + def test_concurrent_first_launches_record_sensitivity_before_overload( + self, + ): + @cuda.jit + def launch_config_sensitive_kernel(x): + x[0] = 1 + + original_update = ( + dispatcher_module.CUDADispatcher._update_launch_config_sensitivity + ) + first_update_entered = threading.Event() + release_update = threading.Event() + blocked_once = threading.Event() + + def blocking_update(dispatcher, kernel, launch_config): + if ( + dispatcher is launch_config_sensitive_kernel + and getattr(kernel, "launch_config_sensitive", False) + and dispatcher._launch_config_default_key is None + and not blocked_once.is_set() + ): + blocked_once.set() + first_update_entered.set() + if not release_update.wait(timeout=10): + raise AssertionError( + "timed out waiting to release LCS update" + ) + return original_update(dispatcher, kernel, launch_config) + + barrier = threading.Barrier(2) + errors = [] + errors_lock = threading.Lock() + + def launch(blockdim): + try: + local = np.zeros(1, dtype=np.int32) + barrier.wait(timeout=10) + launch_config_sensitive_kernel[1, blockdim](local) + self.assertEqual(local[0], 1) + except BaseException as e: + with errors_lock: + errors.append(e) + + threads = [ + threading.Thread(target=launch, args=(blockdim,)) + for blockdim in (32, 64) + ] + + dispatcher_module.CUDADispatcher._update_launch_config_sensitivity = ( + blocking_update + ) + try: + for thread in threads: + thread.start() + update_entered = first_update_entered.wait(timeout=20) + # Give the second thread time to reach the same dispatcher while + # the first launch is in the LCS publication window. + time.sleep(0.05) + release_update.set() + for thread in threads: + thread.join(timeout=20) + finally: + release_update.set() + dispatcher_class = dispatcher_module.CUDADispatcher + dispatcher_class._update_launch_config_sensitivity = original_update + + self.assertTrue(update_entered) + self.assertFalse(any(thread.is_alive() for thread in threads)) + self.assertEqual(errors, []) + self.assertEqual(len(LAUNCH_CONFIG_LOG), 2) + self.assertEqual( + {entry["blockdim"] for entry in LAUNCH_CONFIG_LOG}, + {(32, 1, 1), (64, 1, 1)}, + ) + + def test_concurrent_distinct_launch_config_specializations(self): + @cuda.jit + def launch_config_sensitive_kernel(x, mult): + i = cuda.grid(1) + if i < x.size: + x[i] *= mult[0] + + configs = ((1, 32), (1, 64), (2, 32), (2, 64)) + barrier = threading.Barrier(len(configs)) + errors = [] + errors_lock = threading.Lock() + + def launch(config): + try: + blocks, threads = config + n = blocks * threads + arr = cuda.to_device(np.ones(n, dtype=np.int32)) + mult = cuda.to_device(np.array([2], dtype=np.int32)) + barrier.wait(timeout=10) + launch_config_sensitive_kernel[blocks, threads](arr, mult) + np.testing.assert_array_equal( + arr.copy_to_host(), np.full(n, 2, dtype=np.int32) + ) + except BaseException as e: + with errors_lock: + errors.append(e) + + threads = [threading.Thread(target=launch, args=(c,)) for c in configs] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + self.assertFalse(any(thread.is_alive() for thread in threads)) + self.assertEqual(errors, []) + self.assertTrue(launch_config_sensitive_kernel._launch_config_sensitive) + self.assertIsNotNone( + launch_config_sensitive_kernel._launch_config_default_key + ) + self.assertLessEqual( + len(launch_config_sensitive_kernel._launch_config_specializations), + len(configs), + ) + log_configs = { + (entry["griddim"], entry["blockdim"]) for entry in LAUNCH_CONFIG_LOG + } + self.assertEqual( + log_configs, + {((b, 1, 1), (t, 1, 1)) for b, t in configs}, + ) + if __name__ == "__main__": unittest.main() diff --git a/numba_cuda/numba/cuda/tests/nocuda/test_import.py b/numba_cuda/numba/cuda/tests/nocuda/test_import.py index dc9c7ec6c..e49884b4b 100644 --- a/numba_cuda/numba/cuda/tests/nocuda/test_import.py +++ b/numba_cuda/numba/cuda/tests/nocuda/test_import.py @@ -1,9 +1,17 @@ # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: BSD-2-Clause -from numba.cuda.tests.support import run_in_subprocess +import os +import sysconfig import unittest +from numba.cuda.tests.support import run_in_subprocess + + +def sysconfig_var_is_true(name): + value = sysconfig.get_config_var(name) + return value is True or str(value).strip() == "1" + class TestImport(unittest.TestCase): def test_no_impl_import(self): @@ -64,6 +72,34 @@ def test_no_impl_import(self): unexpected = set(banlist) & set(modlist) assert not unexpected + @unittest.skipUnless( + sysconfig_var_is_true("Py_GIL_DISABLED"), + "requires a free-threaded Python build", + ) + def test_free_threaded_import_keeps_gil_disabled(self): + code = """\ +import importlib +import sys + +assert hasattr(sys, "_is_gil_enabled") +assert not sys._is_gil_enabled() + +import numba.cuda +assert not sys._is_gil_enabled() + +for name in ( + "numba.cuda.cext._typeconv", + "numba.cuda.cext.mviewbuf", + "numba.cuda.cext._helperlib", + "numba.cuda.cext._dispatcher", +): + importlib.import_module(name) + assert not sys._is_gil_enabled(), name +""" + env = os.environ.copy() + env["PYTHON_GIL"] = "0" + run_in_subprocess(code, flags=("-W", "error::RuntimeWarning"), env=env) + if __name__ == "__main__": unittest.main() diff --git a/numba_cuda/numba/cuda/tests/nocuda/test_mviewbuf_threading.py b/numba_cuda/numba/cuda/tests/nocuda/test_mviewbuf_threading.py new file mode 100644 index 000000000..e651f1d61 --- /dev/null +++ b/numba_cuda/numba/cuda/tests/nocuda/test_mviewbuf_threading.py @@ -0,0 +1,65 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: BSD-2-Clause + +from concurrent.futures import ThreadPoolExecutor +import unittest + +import numpy as np + +from numba.cuda.cext import mviewbuf + + +class TestMViewBufThreading(unittest.TestCase): + def test_memoryview_get_extents_info_bad_inputs_raise_cleanly(self): + for args in ( + ((3,), (4,), 2, 4), + ((3, 4), (16,), 2, 4), + ((3,), (4,), 1, 0), + ((3,), (4,), 1, -1), + ): + with self.assertRaises((ValueError, OverflowError, TypeError)): + mviewbuf.memoryview_get_extents_info(*args) + + def test_concurrent_memoryview_extent_helpers(self): + def check_extents_info(n): + for k in range(n): + ndim = k % 4 + shape = tuple(range(1, ndim + 1)) if ndim else () + strides = tuple(4 for _ in range(ndim)) + start, end = mviewbuf.memoryview_get_extents_info( + shape, strides, ndim, 4 + ) + self.assertGreaterEqual(end, start) + + start, end = mviewbuf.memoryview_get_extents_info( + (3, 4), (16, 4), 2, 4 + ) + self.assertEqual(end - start, 48) + + def check_extents(n): + for k in range(n): + ary = np.arange((k % 8) + 1, dtype=np.float64) + start, end = mviewbuf.memoryview_get_extents(ary) + self.assertGreaterEqual(end, start) + self.assertEqual(end - start, ary.nbytes) + + def check_bad_inputs(n): + for _ in range(n): + self.test_memoryview_get_extents_info_bad_inputs_raise_cleanly() + + workers = ( + check_extents_info, + check_extents, + check_bad_inputs, + check_extents_info, + check_extents, + check_bad_inputs, + ) + with ThreadPoolExecutor(max_workers=len(workers)) as executor: + futures = [executor.submit(worker, 500) for worker in workers] + for future in futures: + future.result() + + +if __name__ == "__main__": + unittest.main() diff --git a/numba_cuda/numba/cuda/tests/nocuda/test_typeconv_threading.py b/numba_cuda/numba/cuda/tests/nocuda/test_typeconv_threading.py new file mode 100644 index 000000000..1c2e37fb5 --- /dev/null +++ b/numba_cuda/numba/cuda/tests/nocuda/test_typeconv_threading.py @@ -0,0 +1,66 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: BSD-2-Clause + +from concurrent.futures import ThreadPoolExecutor +import unittest + +from numba.cuda import types +from numba.cuda.typeconv import Conversion +from numba.cuda.typeconv.typeconv import TypeManager + + +class TestTypeConvThreading(unittest.TestCase): + def test_concurrent_reads_and_writes(self): + tm = TypeManager() + i16 = types.int16 + i32 = types.int32 + i64 = types.int64 + f32 = types.float32 + f64 = types.float64 + + tm.set_promote(i32, i64) + tm.set_unsafe_convert(i32, f32) + tm.set_promote(i16, i32) + tm.set_safe_convert(f32, f64) + + sig = (i32, f32) + overloads = ( + (i32, i32), + (f32, f32), + (i64, i64), + (i16, i16), + ) + + def write_conversions(): + for _ in range(300): + tm.set_promote(i32, i64) + tm.set_unsafe_convert(i32, f32) + tm.set_promote(i16, i32) + tm.set_safe_convert(f32, f64) + + def read_conversions(): + for _ in range(300): + self.assertEqual( + tm.check_compatible(i32, i64), Conversion.promote + ) + self.assertEqual( + tm.check_compatible(i32, f32), Conversion.unsafe + ) + self.assertEqual( + tm.check_compatible(i16, i32), Conversion.promote + ) + self.assertEqual(tm.check_compatible(f32, f64), Conversion.safe) + self.assertEqual( + tm.select_overload(sig, overloads, True, False), 1 + ) + + with ThreadPoolExecutor(max_workers=8) as executor: + futures = [executor.submit(write_conversions) for _ in range(4)] + [ + executor.submit(read_conversions) for _ in range(4) + ] + for future in futures: + future.result() + + +if __name__ == "__main__": + unittest.main() diff --git a/numba_cuda/numba/cuda/tests/nocuda/test_typeof_threading.py b/numba_cuda/numba/cuda/tests/nocuda/test_typeof_threading.py new file mode 100644 index 000000000..f7de99507 --- /dev/null +++ b/numba_cuda/numba/cuda/tests/nocuda/test_typeof_threading.py @@ -0,0 +1,114 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: BSD-2-Clause + +from concurrent.futures import ThreadPoolExecutor +import threading +import unittest + +import numpy as np + +from numba.cuda.cext import _dispatcher +from numba.cuda.tests.support import ( + fresh_struct_array, + is_free_threaded_python, +) + + +class TestTypeofThreading(unittest.TestCase): + def test_compute_fingerprint_concurrent_dtype_variants(self): + def fingerprint_values(seed): + rng = np.random.default_rng(seed) + for _ in range(250): + i = int(rng.integers(0, 1000)) + values = ( + np.arange(4, dtype=np.int32), + np.arange(6, dtype=np.float64).reshape(2, 3), + fresh_struct_array(i), + fresh_struct_array(i)[0], + (1, 2.0, np.float32(3), np.int64(4)), + [np.arange((i % 5) + 1, dtype=np.int16)], + np.float32(3.5), + np.complex128(1 + 2j), + {i}, + ) + for value in values: + fingerprint = _dispatcher.compute_fingerprint(value) + self.assertIsInstance(fingerprint, bytes) + self.assertGreater(len(fingerprint), 0) + + with ThreadPoolExecutor(max_workers=8) as executor: + futures = [executor.submit(fingerprint_values, i) for i in range(8)] + for future in futures: + future.result() + + @unittest.skipUnless( + is_free_threaded_python(), + "requires a free-threaded Python build", + ) + def test_compute_fingerprint_with_mutating_containers(self): + stop = threading.Event() + list_value = [np.arange(4, dtype=np.int32)] + set_value = {1} + structured = fresh_struct_array(0) + + def mutate_list(): + while not stop.is_set(): + list_value[:] = [np.arange(4, dtype=np.int32)] + list_value.append(np.arange(2, dtype=np.float64)) + list_value.pop() + list_value.clear() + list_value.append(np.arange(1, dtype=np.int16)) + + def mutate_set(): + i = 0 + while not stop.is_set(): + set_value.clear() + set_value.add(i) + i += 1 + + def fingerprint_values(): + benign_empty_container_races = 0 + values = ( + list_value, + set_value, + structured, + structured[0], + (1, 2.0, np.float32(3)), + ) + for _ in range(1000): + for value in values: + try: + fingerprint = _dispatcher.compute_fingerprint(value) + self.assertIsInstance(fingerprint, bytes) + except ValueError as e: + message = str(e) + if "empty list" in message or "empty set" in message: + benign_empty_container_races += 1 + else: + raise + return benign_empty_container_races + + mutators = [ + threading.Thread(target=mutate_list), + threading.Thread(target=mutate_set), + ] + for mutator in mutators: + mutator.start() + + try: + with ThreadPoolExecutor(max_workers=8) as executor: + futures = [ + executor.submit(fingerprint_values) for _ in range(8) + ] + for future in futures: + future.result() + finally: + stop.set() + for mutator in mutators: + mutator.join(timeout=5) + + self.assertFalse(any(mutator.is_alive() for mutator in mutators)) + + +if __name__ == "__main__": + unittest.main() diff --git a/numba_cuda/numba/cuda/tests/stress/__init__.py b/numba_cuda/numba/cuda/tests/stress/__init__.py new file mode 100644 index 000000000..68c87dcd4 --- /dev/null +++ b/numba_cuda/numba/cuda/tests/stress/__init__.py @@ -0,0 +1,2 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: BSD-2-Clause diff --git a/numba_cuda/numba/cuda/tests/stress/test_free_threading.py b/numba_cuda/numba/cuda/tests/stress/test_free_threading.py new file mode 100644 index 000000000..3dcbec4fa --- /dev/null +++ b/numba_cuda/numba/cuda/tests/stress/test_free_threading.py @@ -0,0 +1,744 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: BSD-2-Clause + +from concurrent.futures import ThreadPoolExecutor +import gc +import os +import signal +import shutil +import stat +import subprocess +import sys +import threading +import time +import traceback +import unittest + +import numpy as np + +from numba import cuda +from numba.cuda import launchconfig, types +from numba.cuda.cext import _dispatcher, mviewbuf +from numba.cuda.core.errors import TypingError +from numba.cuda.core.rewrites import Rewrite, register_rewrite, rewrite_registry +from numba.cuda.cudadrv.driver import CudaAPIError +from numba.cuda.testing import CUDATestCase, skip_on_cudasim +from numba.cuda.tests.support import ( + fresh_struct_array, + free_threading_stress_enabled, + is_free_threaded_python, + launch_subprocess_code, + subprocess_marker_results, + temp_directory, +) +from numba.cuda.typeconv import Conversion +from numba.cuda.typeconv.typeconv import TypeManager + + +skip_unless_free_threaded = unittest.skipUnless( + is_free_threaded_python(), "requires a free-threaded Python build" +) +skip_unless_ft_stress = unittest.skipUnless( + free_threading_stress_enabled(), "requires NUMBA_CUDA_FT_STRESS=1" +) + + +def _stress_float(name, default): + try: + return float(os.environ.get(name, default)) + except (TypeError, ValueError): + return default + + +def _stress_int_config(name, default): + raw = os.environ.get(name) + if raw is None: + return default, False + try: + return int(raw), True + except (TypeError, ValueError): + return default, False + + +def _stress_int(name, default): + return _stress_int_config(name, default)[0] + + +def _stress_seconds(default=30.0): + return _stress_float("NUMBA_CUDA_FT_STRESS_SECONDS", default) + + +def _stress_workers(default=32): + configured, explicit = _stress_int_config( + "NUMBA_CUDA_FT_STRESS_WORKERS", default + ) + if explicit: + return max(1, configured) + return max(1, min(configured, os.cpu_count() or 1)) + + +def _stress_processes(default=8): + configured, explicit = _stress_int_config( + "NUMBA_CUDA_FT_STRESS_PROCESSES", default + ) + if explicit: + return max(1, configured) + return max(1, min(configured, os.cpu_count() or 1)) + + +def _stress_iters(default): + return _stress_int("NUMBA_CUDA_FT_STRESS_ITERS", default) + + +def _stress_child_timeout(seconds): + # CUDA stress runs can spend substantial time compiling before the timed + # workload reaches steady state. Bound hard hangs, but leave enough slack + # that normal compilation overhead does not look like a deadlock. + return max(300.0, seconds * 6.0) + + +def _dispatch_stress_child_command(): + test_name = "test_dispatch_lcs_callbacks_and_surfaces_stress" + pyargs = "numba.cuda.tests.stress.test_free_threading" + addopts = "addopts=--benchmark-disable --import-mode=importlib" + command = [ + sys.executable, + "-m", + "pytest", + "-q", + "--override-ini", + addopts, + "--pyargs", + pyargs, + "-k", + test_name, + ] + if os.path.isdir("testing"): + # Source-checkout CUDA tests need testing/conftest.py fixtures. + command.insert(4, "testing") + return command + + +def _stress_child_kwargs(): + if os.name == "nt": + return { + "creationflags": subprocess.CREATE_NEW_PROCESS_GROUP, + } + return { + "start_new_session": True, + } + + +def _kill_stress_child(process): + if process.poll() is not None: + return + if os.name == "nt": + try: + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(process.pid)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + except OSError: + process.kill() + if process.poll() is None: + process.kill() + return + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + +_REWRITE_FLAG = "_numba_cuda_free_threading_stress_rewrite_registered" +_FT_STRESS_CACHE_RESULT = "__FT_STRESS_CACHE_RESULT__" +_FT_STRESS_DISPATCH_CHILD = "_NUMBA_CUDA_FT_STRESS_DISPATCH_CHILD" + +if free_threading_stress_enabled() and not getattr( + rewrite_registry, _REWRITE_FLAG, False +): + + @register_rewrite("after-inference") + class FreeThreadingStressRewrite(Rewrite): + _TARGET_NAMES = { + "ft_stress_lcs_kernel", + "ft_stress_cached_lcs_kernel", + } + + def __init__(self, state): + super().__init__(state) + self._block = None + self._applied = False + + def match(self, func_ir, block, typemap, calltypes): + if self._applied: + return False + if func_ir.func_id.func_name not in self._TARGET_NAMES: + return False + self._block = block + return True + + def apply(self): + cfg = launchconfig.ensure_current_launch_config() + cfg.dispatcher.mark_launch_config_sensitive() + self._applied = True + return self._block + + setattr(rewrite_registry, _REWRITE_FLAG, True) + + +@cuda.jit +def ft_stress_plain_kernel(x, y): + i = cuda.grid(1) + if i < x.size: + y[i] = x[i] + 1 + + +@cuda.jit +def ft_stress_lcs_kernel(x): + i = cuda.grid(1) + if i < x.size: + x[i] += 1 + + +@cuda.jit +def ft_stress_bad_kernel(x): + x[0] = undefined_global_name # noqa: F821 + + +@cuda.jit +def ft_stress_good_kernel(x): + i = cuda.grid(1) + if i < x.size: + x[i] += 1 + + +@skip_unless_ft_stress +@skip_unless_free_threaded +class TestFreeThreadingNoCudaStress(unittest.TestCase): + def test_compute_fingerprint_stress(self): + workers = _stress_workers(16) + iters = _stress_iters(2000) + stop = threading.Event() + shared_list = [np.arange(4, dtype=np.int32)] + shared_set = {1, 2, 3} + + def mutate_list(): + i = 0 + while not stop.is_set(): + shared_list[:] = [np.arange(4, dtype=np.int32)] + shared_list.append(np.arange(2, dtype=np.float64)) + shared_list.pop() + shared_list.clear() + shared_list.append(np.arange((i % 3) + 1, dtype=np.int16)) + i += 1 + + def mutate_set(): + i = 0 + while not stop.is_set(): + shared_set.clear() + shared_set.add(i) + shared_set.add(i + 1) + shared_set.discard(i) + i += 1 + + def fingerprint(seed): + rng = np.random.default_rng(seed) + for _ in range(iters): + i = int(rng.integers(0, 1000)) + values = ( + np.arange(4, dtype=np.int32), + np.arange(6, dtype=np.float64).reshape(2, 3), + fresh_struct_array(i), + fresh_struct_array(i)[0], + (1, 2.0, np.float32(3), np.int64(4)), + [np.arange((i % 5) + 1, dtype=np.int16)], + np.float32(3.5), + np.complex128(1 + 2j), + {i}, + shared_list, + shared_set, + ) + for value in values: + try: + fp = _dispatcher.compute_fingerprint(value) + except ValueError as e: + if "empty" in str(e): + continue + raise + self.assertIsInstance(fp, bytes) + + mutators = [ + threading.Thread(target=mutate_list), + threading.Thread(target=mutate_set), + ] + for mutator in mutators: + mutator.start() + try: + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [ + executor.submit(fingerprint, i) for i in range(workers) + ] + for future in futures: + future.result() + finally: + stop.set() + for mutator in mutators: + mutator.join(timeout=5) + + self.assertFalse(any(mutator.is_alive() for mutator in mutators)) + + def test_mviewbuf_stress(self): + workers = _stress_workers(16) + iters = _stress_iters(20000) + + def check_info(n): + for k in range(n): + ndim = k % 4 + shape = tuple(range(1, ndim + 1)) if ndim else () + strides = tuple(4 for _ in range(ndim)) + start, end = mviewbuf.memoryview_get_extents_info( + shape, strides, ndim, 4 + ) + self.assertGreaterEqual(end, start) + start, end = mviewbuf.memoryview_get_extents_info( + (3, 4), (16, 4), 2, 4 + ) + self.assertEqual(end - start, 48) + + def check_extents(n): + for k in range(n): + ary = np.arange((k % 8) + 1, dtype=np.float64) + start, end = mviewbuf.memoryview_get_extents(ary) + self.assertGreaterEqual(end, start) + + def check_bad_inputs(n): + for _ in range(n): + for args in ( + ((3,), (4,), 2, 4), + ((3, 4), (16,), 2, 4), + ((3,), (4,), 1, 0), + ((3,), (4,), 1, -1), + ): + with self.assertRaises( + (ValueError, OverflowError, TypeError) + ): + mviewbuf.memoryview_get_extents_info(*args) + + jobs = (check_info, check_extents, check_bad_inputs) + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [ + executor.submit(jobs[i % len(jobs)], iters) + for i in range(workers) + ] + for future in futures: + future.result() + + def test_typeconv_stress(self): + workers = _stress_workers(12) + iters = _stress_iters(3000) + manager = TypeManager() + i16, i32, i64 = types.int16, types.int32, types.int64 + f32, f64 = types.float32, types.float64 + + manager.set_promote(i32, i64) + manager.set_unsafe_convert(i32, f32) + manager.set_promote(i16, i32) + manager.set_safe_convert(f32, f64) + + sig = (i32, f32) + overloads = ((i32, i32), (f32, f32), (i64, i64), (i16, i16)) + + def writer(): + for _ in range(iters): + manager.set_promote(i32, i64) + manager.set_unsafe_convert(i32, f32) + manager.set_promote(i16, i32) + manager.set_safe_convert(f32, f64) + + def reader(): + for _ in range(iters): + self.assertEqual( + manager.check_compatible(i32, i64), Conversion.promote + ) + self.assertEqual( + manager.check_compatible(i32, f32), Conversion.unsafe + ) + self.assertEqual( + manager.check_compatible(i16, i32), Conversion.promote + ) + self.assertEqual( + manager.check_compatible(f32, f64), Conversion.safe + ) + manager.select_overload(sig, overloads, True, False) + + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [ + executor.submit(writer if i % 2 else reader) + for i in range(workers) + ] + for future in futures: + future.result() + + +@skip_on_cudasim("stress tests require the CUDA runtime") +@skip_unless_ft_stress +@skip_unless_free_threaded +class TestFreeThreadingCudaStress(CUDATestCase): + def test_dispatch_lcs_callbacks_and_surfaces_stress(self): + seconds = _stress_seconds(30.0) + if os.environ.get(_FT_STRESS_DISPATCH_CHILD) != "1": + env = os.environ.copy() + # The child executes the original workload. The parent only + # contains hangs so the outer pytest process can report them. + env[_FT_STRESS_DISPATCH_CHILD] = "1" + process = subprocess.Popen( + _dispatch_stress_child_command(), + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + **_stress_child_kwargs(), + ) + timeout = _stress_child_timeout(seconds) + try: + out, err = process.communicate(timeout=timeout) + except subprocess.TimeoutExpired as e: + _kill_stress_child(process) + out, err = process.communicate() + stdout = out.decode(errors="replace") + stderr = err.decode(errors="replace") + raise AssertionError( + "dispatch stress child timed out after %.1fs:\n" + "stdout follows\n%s\n" + "stderr follows\n%s\n" % (timeout, stdout, stderr) + ) from e + + stdout = out.decode(errors="replace") + stderr = err.decode(errors="replace") + if process.returncode != 0: + raise AssertionError( + "dispatch stress child failed with code %s:\n" + "stdout follows\n%s\n" + "stderr follows\n%s\n" + % (process.returncode, stdout, stderr) + ) + return + + workers = _stress_workers(32) + stop = threading.Event() + errors = [] + errors_lock = threading.Lock() + callback_hits = [] + callback_lock = threading.Lock() + original_gc_threshold = gc.get_threshold() + dtypes = ( + np.int16, + np.int32, + np.int64, + np.uint16, + np.float32, + np.float64, + ) + configs = [(b, t) for b in (1, 2, 4) for t in (32, 64, 128)] + + def record_error(where): + with errors_lock: + errors.append((where, traceback.format_exc())) + stop.set() + + def callback(kernel, launch_config): + with callback_lock: + callback_hits.append((kernel, launch_config)) + + callback_config = ft_stress_plain_kernel.configure(1, 64) + callback_config.pre_launch_callbacks[:] = [callback] + + def launch_plain(seed): + rng = np.random.default_rng(seed) + try: + while not stop.is_set(): + dtype = dtypes[int(rng.integers(0, len(dtypes)))] + n = int(rng.integers(1, 200)) + host = np.arange(n, dtype=dtype) + inp = cuda.to_device(host) + out = cuda.device_array(n, dtype=dtype) + blocks = (n + 63) // 64 + ft_stress_plain_kernel[blocks, 64](inp, out) + np.testing.assert_array_equal(out.copy_to_host(), host + 1) + except Exception: + record_error("launch_plain") + + def launch_lcs(seed): + rng = np.random.default_rng(seed + 1000) + try: + while not stop.is_set(): + blocks, threads = configs[ + int(rng.integers(0, len(configs))) + ] + n = blocks * threads + arr = cuda.to_device(np.zeros(n, dtype=np.int32)) + ft_stress_lcs_kernel[blocks, threads](arr) + np.testing.assert_array_equal( + arr.copy_to_host(), np.ones(n, dtype=np.int32) + ) + except Exception: + record_error("launch_lcs") + + def launch_fresh(seed): + try: + while not stop.is_set(): + + @cuda.jit + def kernel(x, y): + i = cuda.grid(1) + if i < x.size: + y[i] = x[i] * 2 + + host = np.arange(50, dtype=np.int32) + inp = cuda.to_device(host) + out = cuda.device_array(host.size, dtype=np.int32) + kernel[1, 64](inp, out) + np.testing.assert_array_equal(out.copy_to_host(), host * 2) + except Exception: + record_error("launch_fresh") + + def specialize_fresh(seed): + try: + while not stop.is_set(): + + @cuda.jit + def kernel(x): + i = cuda.grid(1) + if i < x.size: + x[i] += 1 + + host = np.zeros(32, dtype=np.int32) + specialized = kernel.specialize(host) + arr = cuda.to_device(host) + specialized[1, 32](arr) + except Exception: + record_error("specialize_fresh") + + def fingerprint(seed): + shared = [np.arange(4, dtype=np.int32)] + + def mutate(): + i = 0 + while not stop.is_set(): + shared[:] = [np.arange((i % 4) + 1, dtype=np.int16)] + i += 1 + + mutator = threading.Thread(target=mutate) + mutator.start() + try: + while not stop.is_set(): + values = ( + shared, + (1, 2.0, np.float32(3)), + np.zeros(3, np.float64), + ) + for value in values: + try: + _dispatcher.compute_fingerprint(value) + except ValueError as e: + if "empty" not in str(e): + raise + except Exception: + record_error("fingerprint") + finally: + mutator.join(timeout=5) + + def streams_events(seed): + try: + while not stop.is_set(): + stream = cuda.stream() + event = cuda.event() + host = np.arange(64, dtype=np.float32) + arr = cuda.to_device(host, stream=stream) + event.record(stream=stream) + stream.synchronize() + del arr, event, stream + except Exception: + record_error("streams_events") + + def pinned(seed): + try: + while not stop.is_set(): + host = np.full(64, 7, dtype=np.int32) + with cuda.pinned(host): + arr = cuda.to_device(host) + np.testing.assert_array_equal(arr.copy_to_host(), host) + pinned_arr = cuda.pinned_array(32, dtype=np.float32) + pinned_arr[:] = 1.5 + except Exception: + record_error("pinned") + + def managed(seed): + try: + while not stop.is_set(): + arr = cuda.managed_array(32, dtype=np.float32) + arr[:] = 0 + ft_stress_good_kernel[1, 32](arr) + cuda.synchronize() + except Exception: + record_error("managed") + + def compile_failures(seed): + try: + while not stop.is_set(): + with self.assertRaises(TypingError): + ft_stress_bad_kernel[1, 1](np.zeros(1, dtype=np.int32)) + arr = cuda.to_device(np.zeros(8, dtype=np.int32)) + ft_stress_good_kernel[1, 8](arr) + np.testing.assert_array_equal( + arr.copy_to_host(), np.ones(8, dtype=np.int32) + ) + except Exception: + record_error("compile_failures") + + def mutate_callbacks(seed): + try: + while not stop.is_set(): + callback_config.pre_launch_callbacks.append(callback) + if len(callback_config.pre_launch_callbacks) > 50: + del callback_config.pre_launch_callbacks[:] + time.sleep(0) + except Exception: + record_error("mutate_callbacks") + + cuda.to_device(np.zeros(1, dtype=np.float32)).copy_to_host() + managed_supported = True + try: + managed_probe = cuda.managed_array(1, dtype=np.float32) + del managed_probe + except CudaAPIError as e: + message = str(e).lower() + if "not supported" in message or "not_supported" in message: + managed_supported = False + else: + raise + + jobs = ( + launch_plain, + launch_lcs, + launch_fresh, + specialize_fresh, + fingerprint, + streams_events, + pinned, + compile_failures, + mutate_callbacks, + ) + if managed_supported: + jobs = jobs[:7] + (managed,) + jobs[7:] + selected_jobs = [jobs[i % len(jobs)] for i in range(workers)] + deadline = time.monotonic() + seconds + gc.set_threshold(50, 5, 5) + try: + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [ + executor.submit(job, i) + for i, job in enumerate(selected_jobs) + ] + while time.monotonic() < deadline and not errors: + time.sleep(0.2) + stop.set() + for future in futures: + future.result() + finally: + del callback_config.pre_launch_callbacks[:] + gc.set_threshold(*original_gc_threshold) + + # A tight full-GC loop under many dispatch workers timed out on + # CPython 3.14t in Python/gc_free_threading.c. Keep this test aimed at + # numba-cuda dispatch surfaces and collect after worker shutdown. + gc.collect() + + self.assertEqual(errors, []) + if not callback_hits: + callback_config.pre_launch_callbacks[:] = [callback] + try: + arr = cuda.to_device(np.zeros(64, dtype=np.int32)) + callback_config(arr, arr) + finally: + del callback_config.pre_launch_callbacks[:] + self.assertGreater(len(callback_hits), 0) + + def test_launch_config_sensitive_disk_cache_stress(self): + workers = _stress_workers(12) + processes = _stress_processes(8) + tempdir = temp_directory("ft_stress_cache") + modname = "ft_stress_lcs_cache_fodder" + here = os.path.dirname(__file__) + usecase = os.path.join( + here, "..", "cudapy", "cache_launch_config_sensitive_usecases.py" + ) + modfile = os.path.join(tempdir, modname + ".py") + cache_dir = os.path.join(tempdir, "__pycache__") + shutil.copy(usecase, modfile) + os.chmod(modfile, stat.S_IREAD | stat.S_IWRITE) + + def cache_files(): + files = [] + for root, _dirs, filenames in os.walk(cache_dir): + files.extend(os.path.join(root, f) for f in filenames) + return files + + def assert_cache_artifacts(): + files = cache_files() + self.assertTrue(any(f.endswith(".lcs") for f in files), files) + self.assertTrue(any(f.endswith(".nbi") for f in files), files) + + sys.path.insert(0, tempdir) + try: + mod = __import__(modname) + blockdims = [32, 64] * max(1, workers // 2) + barrier = threading.Barrier(len(blockdims)) + + def thread_worker(blockdim): + barrier.wait(timeout=30) + return int(mod.launch(blockdim)[0]) + + with ThreadPoolExecutor(max_workers=len(blockdims)) as executor: + futures = [ + executor.submit(thread_worker, blockdim) + for blockdim in blockdims + ] + results = [future.result(timeout=120) for future in futures] + self.assertEqual(results, [1] * len(blockdims)) + assert_cache_artifacts() + + shutil.rmtree(cache_dir, ignore_errors=True) + blockdims = [32, 64] * max(1, processes // 2) + process_list = [ + subprocess.Popen( + [ + sys.executable, + "-c", + launch_subprocess_code( + tempdir, + modname, + blockdim, + _FT_STRESS_CACHE_RESULT, + ), + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + for blockdim in blockdims + ] + results = subprocess_marker_results( + process_list, + _FT_STRESS_CACHE_RESULT, + timeout=180, + ) + self.assertEqual(results, [1] * len(blockdims)) + assert_cache_artifacts() + finally: + sys.path.remove(tempdir) + sys.modules.pop(modname, None) + + +if __name__ == "__main__": + unittest.main() diff --git a/numba_cuda/numba/cuda/tests/support.py b/numba_cuda/numba/cuda/tests/support.py index a895c97ed..b40ebc27a 100644 --- a/numba_cuda/numba/cuda/tests/support.py +++ b/numba_cuda/numba/cuda/tests/support.py @@ -12,6 +12,7 @@ import io import subprocess import sys +import sysconfig import shutil import warnings import tempfile @@ -60,6 +61,90 @@ def tearDown(self): skip_if_py314 = unittest.skipIf(PYVERSION == (3, 14), "Test unstable on 3.14") +def _sysconfig_var_is_true(name): + value = sysconfig.get_config_var(name) + return value is True or str(value).strip() == "1" + + +def is_free_threaded_python(): + return _sysconfig_var_is_true("Py_GIL_DISABLED") + + +def free_threading_stress_enabled(): + value = os.environ.get("NUMBA_CUDA_FT_STRESS", "") + return value.lower() in {"1", "true", "yes", "on"} + + +def fresh_struct_array(i): + dtype = np.dtype( + [("a", np.int32), ("b", np.float64), (f"c{i % 4}", np.int16)] + ) + return np.zeros(2, dtype=dtype) + + +def launch_subprocess_code(tempdir, modname, blockdim, marker): + return "\n".join( + [ + "import sys", + f"sys.path.insert(0, {tempdir!r})", + f"mod = __import__({modname!r})", + f"out = mod.launch({blockdim!r})", + f"print({marker!r} + str(int(out[0])))", + ] + ) + + +def _decode_subprocess_output(data): + return data.decode(errors="replace") + + +def subprocess_marker_result(process, marker, timeout=120): + try: + out, err = process.communicate(timeout=timeout) + except subprocess.TimeoutExpired as e: + process.kill() + out, err = process.communicate() + stdout = _decode_subprocess_output(out) + stderr = _decode_subprocess_output(err) + raise AssertionError( + "process timed out: \n" + "stdout follows\n%s\n" + "stderr follows\n%s\n" % (stdout, stderr) + ) from e + stdout = _decode_subprocess_output(out) + stderr = _decode_subprocess_output(err) + if process.returncode != 0: + raise AssertionError( + "process failed with code %s: \n" + "stdout follows\n%s\n" + "stderr follows\n%s\n" % (process.returncode, stdout, stderr) + ) + for line in reversed(stdout.splitlines()): + if line.startswith(marker): + return int(line[len(marker) :]) + raise AssertionError( + "subprocess marker missing:\nstdout follows\n%s\n" + "stderr follows\n%s\n" % (stdout, stderr) + ) + + +def subprocess_marker_results(processes, marker, timeout=120): + try: + return [ + subprocess_marker_result( + process, + marker, + timeout=timeout, + ) + for process in processes + ] + finally: + for process in processes: + if process.poll() is None: + process.kill() + process.communicate() + + def expected_failure_py314(fn): if PYVERSION == (3, 14): return unittest.expectedFailure(fn) diff --git a/setup.py b/setup.py index 5ac36d3d6..d58bd282f 100644 --- a/setup.py +++ b/setup.py @@ -3,6 +3,7 @@ import pathlib import sys +import sysconfig from setuptools import setup, Extension from setuptools.command.build_py import build_py @@ -14,6 +15,11 @@ SITE_PACKAGES = pathlib.Path("site-packages") +def sysconfig_var_is_true(name): + value = sysconfig.get_config_var(name) + return value is True or str(value).strip() == "1" + + def get_version(): """Read version from VERSION file.""" version_file = pathlib.Path(__file__).parent / "numba_cuda" / "VERSION" @@ -43,9 +49,14 @@ def get_ext_modules(): if sys.platform == "darwin": install_name_tool_fixer = ["-headerpad_max_install_names"] + free_threaded_macros = [] + if sysconfig_var_is_true("Py_GIL_DISABLED"): + free_threaded_macros.append(("Py_GIL_DISABLED", "1")) + ext_mviewbuf = Extension( name="numba_cuda.numba.cuda.cext.mviewbuf", extra_link_args=install_name_tool_fixer, + define_macros=list(free_threaded_macros), sources=["numba_cuda/numba/cuda/cext/mviewbuf.c"], ) @@ -64,6 +75,7 @@ def get_ext_modules(): "numba_cuda/numba/cuda/cext/_hashtable.h", ], extra_compile_args=["-std=c++11"], + define_macros=list(free_threaded_macros), **np_compile_args, ) @@ -75,6 +87,7 @@ def get_ext_modules(): ], depends=["numba_cuda/numba/cuda/cext/_pymodule.h"], extra_compile_args=["-std=c++11"], + define_macros=list(free_threaded_macros), ) # Append our cext dir to include_dirs @@ -88,6 +101,7 @@ def get_ext_modules(): "numba_cuda/numba/cuda/cext/_helperlib.c", ], include_dirs=["numba_cuda/numba/cuda/cext"], + define_macros=list(free_threaded_macros), ) return [ diff --git a/testing/pytest.ini b/testing/pytest.ini index 3b96ddb4a..376811b8b 100644 --- a/testing/pytest.ini +++ b/testing/pytest.ini @@ -4,8 +4,10 @@ [pytest] minversion = 8.0 consider_namespace_packages = true -# loadscope ensures the grouping required by CUDATestCase -addopts = --benchmark-disable --pyargs numba.cuda.tests +# CI test commands use loadscope for the grouping required by CUDATestCase. +# importlib mode preserves the redirected numba.cuda module name instead of +# deriving cuda.tests from the physical numba_cuda/numba/cuda test path. +addopts = --benchmark-disable --import-mode=importlib --pyargs numba.cuda.tests filterwarnings = error