File: zrpy_releasegil_test.py

package info (click to toggle)
pypy 5.6.0%2Bdfsg-4
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 97,040 kB
  • ctags: 185,069
  • sloc: python: 1,147,862; ansic: 49,642; cpp: 5,245; asm: 5,169; makefile: 529; sh: 481; xml: 232; lisp: 45
file content (124 lines) | stat: -rw-r--r-- 4,325 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
from rpython.rtyper.lltypesystem import lltype, llmemory, rffi
from rpython.rlib.jit import dont_look_inside
from rpython.jit.metainterp.optimizeopt import ALL_OPTS_NAMES
from rpython.translator.tool.cbuild import ExternalCompilationInfo
from rpython.rlib import rposix

from rpython.rtyper.annlowlevel import llhelper

from rpython.jit.backend.llsupport.test.zrpy_gc_test import BaseFrameworkTests
from rpython.jit.backend.llsupport.test.zrpy_gc_test import check
from rpython.tool.udir import udir


class ReleaseGILTests(BaseFrameworkTests):
    compile_kwds = dict(enable_opts=ALL_OPTS_NAMES, thread=True)

    def define_simple(self):
        c_strchr = rffi.llexternal('strchr', [rffi.CCHARP, lltype.Signed],
                                   rffi.CCHARP)

        def before(n, x):
            return (n, None, None, None, None, None,
                    None, None, None, None, None, None)
        #
        def f(n, x, *args):
            a = rffi.str2charp(str(n))
            c_strchr(a, ord('0'))
            lltype.free(a, flavor='raw')
            n -= 1
            return (n, x) + args
        return before, f, None

    def test_simple(self):
        self.run('simple')
        assert 'call_release_gil' in udir.join('TestCompileFramework.log').read()

    def define_close_stack(self):
        #
        class Glob(object):
            pass
        glob = Glob()
        class X(object):
            pass
        #
        def callback(p1, p2):
            for i in range(100):
                glob.lst.append(X())
            return rffi.cast(rffi.INT, 1)
        CALLBACK = lltype.Ptr(lltype.FuncType([lltype.Signed,
                                               lltype.Signed], rffi.INT))
        #
        @dont_look_inside
        def alloc1():
            return llmemory.raw_malloc(16)
        @dont_look_inside
        def free1(p):
            llmemory.raw_free(p)

        c_qsort = rffi.llexternal('qsort', [rffi.VOIDP, rffi.SIZE_T,
                                            rffi.SIZE_T, CALLBACK], lltype.Void)
        #
        def f42(n):
            length = len(glob.lst)
            raw = alloc1()
            wrapper = rffi._make_wrapper_for(CALLBACK, callback, None, True)
            fn = llhelper(CALLBACK, wrapper)
            if n & 1:    # to create a loop and a bridge, and also
                pass     # to run the qsort() call in the blackhole interp
            c_qsort(rffi.cast(rffi.VOIDP, raw), rffi.cast(rffi.SIZE_T, 2),
                    rffi.cast(rffi.SIZE_T, 8), fn)
            free1(raw)
            check(len(glob.lst) > length)
            del glob.lst[:]
        #
        def before(n, x):
            glob.lst = []
            
            return (n, None, None, None, None, None,
                    None, None, None, None, None, None)
        #
        def f(n, x, *args):
            f42(n)
            n -= 1
            return (n, x) + args
        return before, f, None

    def test_close_stack(self):
        self.run('close_stack')
        assert 'call_release_gil' in udir.join('TestCompileFramework.log').read()

    # XXX this should also test get/set_alterrno ?
    def define_get_set_errno(self):
        eci = ExternalCompilationInfo(
            post_include_bits=[r'''
                #include <errno.h>
                static int test_get_set_errno(void) {
                    int r = errno;
                    //fprintf(stderr, "read saved errno: %d\n", r);
                    errno = 42;
                    return r;
                }
            '''])

        c_test = rffi.llexternal('test_get_set_errno', [], rffi.INT,
                                 compilation_info=eci,
                                 save_err=rffi.RFFI_FULL_ERRNO)

        def before(n, x):
            return (n, None, None, None, None, None,
                    None, None, None, None, None, None)
        #
        def f(n, x, *args):
            rposix.set_saved_errno(24)
            result1 = c_test()
            result2 = rposix.get_saved_errno()
            assert result1 == 24
            assert result2 == 42
            n -= 1
            return (n, x) + args
        return before, f, None

    def test_get_set_errno(self):
        self.run('get_set_errno')
        assert 'call_release_gil' in udir.join('TestCompileFramework.log').read()