Mercurial > public > mercurial-scm > hg

comparison tests/testlib/sigpipe-remote.py @ 47557:ed81f2be5527

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
test: use a python script in `test-transaction-rollback-on-sigpipe.t` This still does not work on Windows, but at least this is a python script now. Differential Revision: https://phab.mercurial-scm.org/D10947
author Pierre-Yves David <pierre-yves.david@octobus.net>
date Fri, 02 Jul 2021 20:20:37 +0200
parents
children df6148ca7120
comparison
equal deleted inserted replaced
47556:640fdb7fd67b 47557:ed81f2be5527
1 #!/usr/bin/env python3
2 from __future__ import print_function
3
4 import os
5 import subprocess
6 import sys
7 import threading
8 import time
9
10 # we cannot use mercurial.testing as long as python2 is not dropped as the test will only install the mercurial module for python2 in python2 run
11
12
13 def _timeout_factor():
14 """return the current modification to timeout"""
15 default = int(os.environ.get('HGTEST_TIMEOUT_DEFAULT', 360))
16 current = int(os.environ.get('HGTEST_TIMEOUT', default))
17 if current == 0:
18 return 1
19 return current / float(default)
20
21
22 def wait_file(path, timeout=10):
23 timeout *= _timeout_factor()
24 start = time.time()
25 while not os.path.exists(path):
26 if (time.time() - start) > timeout:
27 raise RuntimeError(b"timed out waiting for file: %s" % path)
28 time.sleep(0.01)
29
30
31 def write_file(path, content=b''):
32 with open(path, 'wb') as f:
33 f.write(content)
34
35
36 # end of mercurial.testing content
37
38 if sys.version_info[0] < 3:
39 print('SIGPIPE-HELPER: script should run with Python 3', file=sys.stderr)
40 sys.exit(255)
41
42
43 def sysbytes(s):
44 return s.encode('utf-8')
45
46
47 def sysstr(s):
48 return s.decode('latin-1')
49
50
51 piped_stdout = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC)
52 piped_stderr = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC)
53
54 stdout_writer = os.fdopen(piped_stdout[1], "rb")
55 stdout_reader = os.fdopen(piped_stdout[0], "rb")
56 stderr_writer = os.fdopen(piped_stderr[1], "rb")
57 stderr_reader = os.fdopen(piped_stderr[0], "rb")
58
59 DEBUG_FILE = os.environ.get('SIGPIPE_REMOTE_DEBUG_FILE')
60 if DEBUG_FILE is None:
61 debug_stream = sys.stderr.buffer
62 else:
63 debug_stream = open(DEBUG_FILE, 'bw', buffering=0)
64
65 SYNCFILE1 = os.environ.get('SYNCFILE1')
66 SYNCFILE2 = os.environ.get('SYNCFILE2')
67 if SYNCFILE1 is None:
68 print('SIGPIPE-HELPER: missing variable $SYNCFILE1', file=sys.stderr)
69 sys.exit(255)
70 if SYNCFILE2 is None:
71 print('SIGPIPE-HELPER: missing variable $SYNCFILE2', file=sys.stderr)
72 sys.exit(255)
73
74 debug_stream.write(b'SIGPIPE-HELPER: Starting\n')
75
76 TESTLIB_DIR = os.path.dirname(sys.argv[0])
77 WAIT_SCRIPT = os.path.join(TESTLIB_DIR, 'wait-on-file')
78
79 hooks_cmd = '%s 10 %s %s'
80 hooks_cmd %= (
81 WAIT_SCRIPT,
82 SYNCFILE2,
83 SYNCFILE1,
84 )
85
86 cmd = ['hg']
87 cmd += sys.argv[1:]
88 sub = subprocess.Popen(
89 cmd,
90 bufsize=0,
91 close_fds=True,
92 stdin=sys.stdin,
93 stdout=stdout_writer,
94 stderr=stderr_writer,
95 )
96
97 debug_stream.write(b'SIGPIPE-HELPER: Mercurial started\n')
98
99
100 shut_down = threading.Event()
101
102 close_lock = threading.Lock()
103
104
105 def _read(stream):
106 try:
107 return stream.read()
108 except ValueError:
109 # read on closed file
110 return None
111
112
113 def forward_stdout():
114 while not shut_down.is_set():
115 c = _read(stdout_reader)
116 while c is not None:
117 sys.stdout.buffer.write(c)
118 c = _read(stdout_reader)
119 time.sleep(0.001)
120 with close_lock:
121 if not stdout_reader.closed:
122 stdout_reader.close()
123 debug_stream.write(b'SIGPIPE-HELPER: stdout closed\n')
124
125
126 def forward_stderr():
127 while not shut_down.is_set():
128 c = _read(stderr_reader)
129 if c is not None:
130 sys.stderr.buffer.write(c)
131 c = _read(stderr_reader)
132 time.sleep(0.001)
133 with close_lock:
134 if not stderr_reader.closed:
135 stderr_reader.close()
136 debug_stream.write(b'SIGPIPE-HELPER: stderr closed\n')
137
138
139 stdout_thread = threading.Thread(target=forward_stdout, daemon=True)
140 stderr_thread = threading.Thread(target=forward_stderr, daemon=True)
141
142 try:
143 stdout_thread.start()
144 stderr_thread.start()
145
146 debug_stream.write(b'SIGPIPE-HELPER: Redirection in place\n')
147
148 try:
149 wait_file(sysbytes(SYNCFILE1))
150 except RuntimeError as exc:
151 msg = sysbytes(str(exc))
152 debug_stream.write(b'SIGPIPE-HELPER: wait failed: %s\n' % msg)
153 else:
154 debug_stream.write(b'SIGPIPE-HELPER: SYNCFILE1 detected\n')
155 with close_lock:
156 if not stdout_reader.closed:
157 stdout_reader.close()
158 if not stderr_reader.closed:
159 stderr_reader.close()
160 sys.stdin.close()
161 debug_stream.write(b'SIGPIPE-HELPER: pipes closed\n')
162 debug_stream.write(b'SIGPIPE-HELPER: creating SYNCFILE2\n')
163 write_file(sysbytes(SYNCFILE2))
164 finally:
165 debug_stream.write(b'SIGPIPE-HELPER: Shutting down\n')
166 shut_down.set()
167 if not sys.stdin.closed:
168 sys.stdin.close()
169 try:
170 sub.wait(timeout=30)
171 except subprocess.TimeoutExpired:
172 msg = b'SIGPIPE-HELPER: Server process failed to terminate\n'
173 debug_stream.write(msg)
174 else:
175 debug_stream.write(b'SIGPIPE-HELPER: Server process terminated\n')
176 debug_stream.write(b'SIGPIPE-HELPER: Shut down\n')