from __future__ import print_function, absolute_import, division

import time
import ctypes

import numpy as np

from numba import unittest_support as unittest
from numba.tests.support import captured_stdout
from numba import vectorize, guvectorize


class TestParUfuncIssues(unittest.TestCase):
    def test_thread_response(self):
        """
        Related to #89.
        This does not test #89 but tests the fix for it.
        We want to make sure the worker threads can be used multiple times
        and with different time gap between each execution.
        """

        @vectorize('float64(float64, float64)', target='parallel')
        def fnv(a, b):
            return a + b

        sleep_time = 1   # 1 second
        while sleep_time > 0.00001:    # 10us
            time.sleep(sleep_time)
            a = b = np.arange(10**5)
            np.testing.assert_equal(a + b, fnv(a, b))
            # Reduce sleep time
            sleep_time /= 2

    def test_gil_reacquire_deadlock(self):
        """
        Testing issue #1998 due to GIL reacquiring
        """
        # make a ctypes callback that requires the GIL
        proto = ctypes.CFUNCTYPE(None, ctypes.c_int32)
        characters = 'abcdefghij'

        def bar(x):
            print(characters[x])

        cbar = proto(bar)

        # our unit under test
        @vectorize(['int32(int32)'], target='parallel', nopython=True)
        def foo(x):
            print(x % 10)  # this reacquires the GIL
            cbar(x % 10)   # this reacquires the GIL
            return x * 2

        # Numpy ufunc has a heuristic to determine whether to release the GIL
        # during execution.  Small input size (10) seems to not release the GIL.
        # Large input size (1000) seems to release the GIL.
        for nelem in [1, 10, 100, 1000]:
            # inputs
            a = np.arange(nelem, dtype=np.int32)
            acopy = a.copy()
            # run and capture stdout
            with captured_stdout() as buf:
                got = foo(a)
            stdout = buf.getvalue()
            buf.close()
            # process outputs from print
            got_output = sorted(map(lambda x: x.strip(), stdout.splitlines()))
            # build expected output
            expected_output = [str(x % 10) for x in range(nelem)]
            expected_output += [characters[x % 10] for x in range(nelem)]
            expected_output = sorted(expected_output)
            # verify
            self.assertEqual(got_output, expected_output)
            np.testing.assert_equal(got, 2 * acopy)



class TestParGUfuncIssues(unittest.TestCase):
    def test_gil_reacquire_deadlock(self):
        """
        Testing similar issue to #1998 due to GIL reacquiring for Gufunc
        """
        # make a ctypes callback that requires the GIL
        proto = ctypes.CFUNCTYPE(None, ctypes.c_int32)
        characters = 'abcdefghij'

        def bar(x):
            print(characters[x])

        cbar = proto(bar)

        # our unit under test
        @guvectorize(['(int32, int32[:])'], "()->()",
                     target='parallel', nopython=True)
        def foo(x, out):
            print(x % 10)  # this reacquires the GIL
            cbar(x % 10)   # this reacquires the GIL
            out[0] = x * 2

        # Numpy ufunc has a heuristic to determine whether to release the GIL
        # during execution.  Small input size (10) seems to not release the GIL.
        # Large input size (1000) seems to release the GIL.
        for nelem in [1, 10, 100, 1000]:
            # inputs
            a = np.arange(nelem, dtype=np.int32)
            acopy = a.copy()
            # run and capture stdout
            with captured_stdout() as buf:
                got = foo(a)
            stdout = buf.getvalue()
            buf.close()
            # process outputs from print
            got_output = sorted(map(lambda x: x.strip(), stdout.splitlines()))
            # build expected output
            expected_output = [str(x % 10) for x in range(nelem)]
            expected_output += [characters[x % 10] for x in range(nelem)]
            expected_output = sorted(expected_output)
            # verify
            self.assertEqual(got_output, expected_output)
            np.testing.assert_equal(got, 2 * acopy)


if __name__ == '__main__':
    unittest.main()