Method test_cancellation

python/pyarrow/tests/test_csv.py:1563–1627

(self)

    @pytest.mark.threading
    def test_cancellation(self):
        if (threading.current_thread().ident !=
                threading.main_thread().ident):
            pytest.skip("test only works from main Python thread")

        def signal_from_thread():
            # Give our workload a chance to start up
            time.sleep(0.2)
            signal.raise_signal(signal.SIGINT)

        # We start with a small CSV reading workload and increase its size
        # until it's large enough to get an interruption during it, even in
        # release mode on fast machines.
        last_duration = 0.0
        workload_size = 100_000
        attempts = 0

        while last_duration < 5.0 and attempts < 10:
            print("workload size:", workload_size)
            large_csv = b"a,b,c\n" + b"1,2,3\n" * workload_size
            exc_info = None

            try:
                # We use a signal fd to reliably ensure that the signal
                # has been delivered to Python, regardless of how exactly
                # it was caught.
                with util.signal_wakeup_fd() as sigfd:
                    try:
                        t = threading.Thread(target=signal_from_thread)
                        t.start()
                        t1 = time.time()
                        try:
                            self.read_bytes(large_csv)
                        except KeyboardInterrupt as e:
                            exc_info = e
                            last_duration = time.time() - t1
                    finally:
                        # Wait for signal to arrive if it didn't already,
                        # to avoid getting a KeyboardInterrupt after the
                        # `except` block below.
                        select.select([sigfd], [], [sigfd], 10.0)

            except KeyboardInterrupt:
                # KeyboardInterrupt didn't interrupt `read_bytes` above.
                pass

            if exc_info is not None:
                # We managed to get `self.read_bytes` interrupted, see if it
                # was actually interrupted inside Arrow C++ or in the Python
                # scaffolding.
                if exc_info.__context__ is not None:
                    # Interrupted inside Arrow C++, we're satisfied now
                    break

            # Increase workload size to get a better chance
            workload_size = workload_size * 3

        if exc_info is None: