image-match/python/py/Lib/site-packages/scipy/interpolate/tests/test_gil.py

import itertools
import threading
import time

import numpy as np
import pytest
import scipy.interpolate


class TestGIL:
    """Check if the GIL is properly released by scipy.interpolate functions."""

    def setup_method(self):
        self.messages = []

    def log(self, message):
        self.messages.append(message)

    def make_worker_thread(self, target, args):
        log = self.log

        class WorkerThread(threading.Thread):
            def run(self):
                log('interpolation started')
                target(*args)
                log('interpolation complete')

        return WorkerThread()

    @pytest.mark.xslow
    @pytest.mark.xfail(reason='race conditions, may depend on system load')
    def test_rectbivariatespline(self):
        def generate_params(n_points):
            x = y = np.linspace(0, 1000, n_points)
            x_grid, y_grid = np.meshgrid(x, y)
            z = x_grid * y_grid
            return x, y, z

        def calibrate_delay(requested_time):
            for n_points in itertools.count(5000, 1000):
                args = generate_params(n_points)
                time_started = time.time()
                interpolate(*args)
                if time.time() - time_started > requested_time:
                    return args

        def interpolate(x, y, z):
            scipy.interpolate.RectBivariateSpline(x, y, z)

        args = calibrate_delay(requested_time=3)
        worker_thread = self.make_worker_thread(interpolate, args)
        worker_thread.start()
        for i in range(3):
            time.sleep(0.5)
            self.log('working')
        worker_thread.join()
        assert self.messages == [
            'interpolation started',
            'working',
            'working',
            'working',
            'interpolation complete',
        ]