File: threadmappings.py

package info (click to toggle)
pypy3 7.3.19%2Bdfsg-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 212,236 kB
  • sloc: python: 2,098,316; ansic: 540,565; sh: 21,462; asm: 14,419; cpp: 4,451; makefile: 4,209; objc: 761; xml: 530; exp: 499; javascript: 314; pascal: 244; lisp: 45; csh: 12; awk: 4
file content (55 lines) | stat: -rw-r--r-- 2,111 bytes parent folder | download | duplicates (2)
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

def _current_frames(space):
    """_current_frames() -> dictionary

    Return a dictionary mapping each current thread T's thread id to T's
    current stack frame.  Functions in the traceback module can build the
    call stack given such a frame.

    Note that in PyPy with the JIT, calling this function causes a runtime
    penalty in all threads, depending on the internal JIT state.  In each
    thread, the penalty should only be noticeable if this call was done
    while in the middle of a long-running function.

    This function should be used for specialized purposes only."""
    w_result = space.newdict()
    ecs = space.threadlocals.getallvalues()
    for thread_ident, ec in ecs.items():
        w_topframe = ec.gettopframe_nohidden()
        if w_topframe is None:
            continue
        space.setitem(w_result,
                      space.newint(thread_ident),
                      w_topframe)
    return w_result


def _current_exceptions(space):
    """_current_exceptions() -> dictionary

    Return a dict mapping each thread's identifier to its current raised
    exception.

    Note that in PyPy with the JIT, calling this function causes a runtime
    penalty in all threads, depending on the internal JIT state.  In each
    thread, the penalty should only be noticeable if this call was done
    while in the middle of a long-running function.

    This function should be used for specialized purposes only."""
    w_result = space.newdict()
    ecs = space.threadlocals.getallvalues()
    for thread_ident, ec in ecs.items():
        operror = ec.sys_exc_info()
        if not operror:
            space.setitem(w_result,
                          space.newint(thread_ident),
                          space.newtuple([space.w_None] * 3))
        else:
            space.setitem(w_result, 
                          space.newint(thread_ident),
                          space.newtuple([operror.w_type,
                                          operror.get_w_value(space),
                                          operror.get_w_traceback(space)]))
    return w_result