File: watchdog.py

package info (click to toggle)
pypy3 7.3.19%2Bdfsg-2
  • links: PTS, VCS
  • area: main
  • in suites: 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 (51 lines) | stat: -rw-r--r-- 1,422 bytes parent folder | download | duplicates (9)
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
import sys, os, signal
import threading

def getsignalname(n):
    for name, value in signal.__dict__.items():
        if value == n and name.startswith('SIG'):
            return name
    return 'signal %d' % (n,)

def childkill():
    global timedout
    timedout = True
    sys.stderr.write("==== test running for %d seconds ====\n" % timeout)
    sys.stderr.write("="*26 + "timedout" + "="*26 + "\n")
    try:
        os.kill(pid, signal.SIGTERM)
    except OSError:
        pass

if __name__ == '__main__':
    timeout = float(sys.argv[1])
    timedout = False

    pid = os.fork()
    if pid == 0:
        os.execvp(sys.argv[2], sys.argv[2:])
    else: # parent
        t = threading.Timer(timeout, childkill)
        t.start()
        while True:
            try:
                pid, status = os.waitpid(pid, 0)
            except KeyboardInterrupt:
                continue
            else:
                t.cancel()
                break
        if os.WIFEXITED(status):
            sys.exit(os.WEXITSTATUS(status))
        else:
            assert os.WIFSIGNALED(status)
            sign = os.WTERMSIG(status)
            if timedout and sign == signal.SIGTERM:
                sys.exit(1)
            signame = getsignalname(sign)
            sys.stderr.write("="*26 + "timedout" + "="*26 + "\n")        
            sys.stderr.write("="*25 + " %-08s " %  signame + "="*25 + "\n")
            sys.exit(1)