Repository: scipy/scipy
File: scipy/interpolate/tests/test_bsplines.py (Python)

Code before:
    def test_read_only(self):
        # BSpline must work on read-only knots and coefficients.
        t = np.array([0, 1])
        c = np.array([3.0])
        t.setflags(write=False)
        c.setflags(write=False)

        xx = np.linspace(0, 1, 10)
        xx.setflags(write=False)

        b = BSpline(t=t, c=c, k=0)
        xp_assert_close(b(xx), np.ones_like(xx) * 3.0)

    def test_concurrency(self, xp):
        # Check that no segfaults appear with concurrent access to BSpline
        b = _make_random_spline(xp=xp)

        def worker_fn(_, b):
            t, _, k = b.tck
            xx = xp.linspace(t[k], t[-k-1], 10000)
            b(xx)

        _run_concurrent_barrier(10, worker_fn, b)


    @pytest.mark.xfail(
        sys.platform == "cygwin",
        reason="threading.get_native_id not implemented",
        raises=AttributeError
    )
    def test_memmap(self, tmpdir):
        # Make sure that memmaps can be used as t and c atrributes after the
        # spline has been constructed. This is similar to what happens in a
        # scikit-learn context, where joblib can create read-only memmap to
        # share objects between workers. For more details, see
        # https://github.com/scipy/scipy/issues/22143
        b = _make_random_spline()
        xx = np.linspace(0, 1, 10)

        expected = b(xx)

        tid = threading.get_native_id()
        t_mm = np.memmap(str(tmpdir.join(f't{tid}.dat')), mode='w+',
                         dtype=b.t.dtype, shape=b.t.shape)
        t_mm[:] = b.t
        c_mm = np.memmap(str(tmpdir.join(f'c{tid}.dat')), mode='w+',
                         dtype=b.c.dtype, shape=b.c.shape)
        c_mm[:] = b.c
        b.t = t_mm
        b.c = c_mm


Review comment (suggestion):
If it's needed, could you also separate it into its own PR (can be the same PR as `Python.h` additions, just a separate commit).

Code after:
        t.setflags(write=False)
        c.setflags(write=False)

        xx = np.linspace(0, 1, 10)
        xx.setflags(write=False)

        b = BSpline(t=t, c=c, k=0)
        xp_assert_close(b(xx), np.ones_like(xx) * 3.0)

    def test_concurrency(self, xp):
        # Check that no segfaults appear with concurrent access to BSpline
        b = _make_random_spline(xp=xp)

        def worker_fn(_, b):
            t, _, k = b.tck
            xx = xp.linspace(t[k], t[-k-1], 10000)
            b(xx)

        _run_concurrent_barrier(10, worker_fn, b)


    @pytest.mark.xfail(
        sys.platform == "cygwin",
        reason="threading.get_native_id not implemented",
        raises=AttributeError
    )
    def test_memmap(self, tmpdir):
        # Make sure that memmaps can be used as t and c atrributes after the
        # spline has been constructed. This is similar to what happens in a
        # scikit-learn context, where joblib can create read-only memmap to
        # share objects between workers. For more details, see
        # https://github.com/scipy/scipy/issues/22143
        b = _make_random_spline()
        xx = np.linspace(0, 1, 10)

        expected = b(xx)

        tid = threading.get_native_id()
        t_mm = np.memmap(str(tmpdir.join(f't{tid}.dat')), mode='w+',
                         dtype=b.t.dtype, shape=b.t.shape)
        t_mm[:] = b.t
        c_mm = np.memmap(str(tmpdir.join(f'c{tid}.dat')), mode='w+',
                         dtype=b.c.dtype, shape=b.c.shape)
        c_mm[:] = b.c
        b.t = t_mm
        b.c = c_mm

        xp_assert_close(b(xx), expected)


@make_xp_test_case(BSpline)