This is an archive of the discontinued Mercurial Phabricator instance.

Differential D11087

sigpipe-remote: simply delegate pipe forwarding to subprocess we can kill
ClosedPublic

Authored by marmoute on Jul 12 2021, 12:02 AM.

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHG27ff81547d35: sigpipe-remote: simply delegate pipe forwarding to subprocess we can kill
Summary

Instead of using sophisticated logics with thread a non blocking pipes, we
simply spawn two new process in charge of reading the pipe and sending the
result to the client. When it is time to cut the pipe we violently kill them
without any remorse. This close the pipe regardless of any in progress os.read
call.

Ironically this is the very same things as what the initial shell setup was
doing, but in Python.

This makes the test pass run properly on Windows. This also reveal that the
Windows behavior is broken as the transaction is not properly rollback. However
this is an adventure for another time. Making the test behave properly was
enough effort.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

marmoute created this revision.Jul 12 2021, 12:02 AM
Alphare accepted this revision.Jul 13 2021, 11:34 AM
Alphare added a subscriber: Alphare.
Alphare added inline comments.
tests/test-transaction-rollback-on-sigpipe.t
37

nit: flaky (and yes I will point it out every time :D)

This revision is now accepted and ready to land.Jul 13 2021, 11:34 AM

Revision Contents
Changeset List

Diff 29301

tests/test-transaction-rollback-on-sigpipe.t

In the pretxnchangegroup hook, kill the PIDs recorded above to simulate ssh In the pretxnchangegroup hook, kill the PIDs recorded above to simulate ssh
disconnecting. Then exit nonzero, to force a transaction rollback. disconnecting. Then exit nonzero, to force a transaction rollback.
$ cat >remote/.hg/hgrc <<EOF $ cat >remote/.hg/hgrc <<EOF
> [hooks] > [hooks]
> pretxnchangegroup.00-break-things=sh "$RUNTESTDIR/testlib/wait-on-file" 10 "$SYNCFILE2" "$SYNCFILE1" > pretxnchangegroup.00-break-things=sh "$RUNTESTDIR/testlib/wait-on-file" 10 "$SYNCFILE2" "$SYNCFILE1"
> pretxnchangegroup.01-output-things=echo "some remote output to be forward to the closed pipe" > pretxnchangegroup.01-output-things=echo "some remote output to be forward to the closed pipe"
> pretxnchangegroup.02-output-things=echo "some more remote output"
> EOF > EOF
$ hg --cwd ./remote tip -T '{node|short}\n' $ hg --cwd ./remote tip -T '{node|short}\n'
000000000000 000000000000
$ cd local $ cd local
$ echo foo > foo ; hg commit -qAm "commit" $ echo foo > foo ; hg commit -qAm "commit"
$ hg push -e "\"$PYTHON\" \"$TESTDIR/dummyssh\"" --remotecmd "$remotecmd"
pushing to ssh://user@dummy/$TESTTMP/remote (use quiet to avoid flacky output from the server)
AlphareUnsubmitted
Not Done

nit: flaky (and yes I will point it out every time :D)

Alphare: nit: flaky (and yes I will point it out every time :D)
searching for changes
remote: adding changesets (py3 !) $ hg push --quiet -e "\"$PYTHON\" \"$TESTDIR/dummyssh\"" --remotecmd "$remotecmd"
remote: adding manifests (py3 !)
remote: adding file changes (py3 !)
remote: adding changesets (no-py3 no-chg !)
remote: adding manifests (no-py3 no-chg !)
remote: adding file changes (no-py3 no-chg !)
abort: stream ended unexpectedly (got 0 bytes, expected 4) abort: stream ended unexpectedly (got 0 bytes, expected 4)
[255] [255]
$ cat $SIGPIPE_REMOTE_DEBUG_FILE $ cat $SIGPIPE_REMOTE_DEBUG_FILE
SIGPIPE-HELPER: Starting SIGPIPE-HELPER: Starting
SIGPIPE-HELPER: Mercurial started
SIGPIPE-HELPER: Redirection in place SIGPIPE-HELPER: Redirection in place
SIGPIPE-HELPER: pipes closed in main
SIGPIPE-HELPER: SYNCFILE1 detected SIGPIPE-HELPER: SYNCFILE1 detected
SIGPIPE-HELPER: pipes closed SIGPIPE-HELPER: worker killed
SIGPIPE-HELPER: creating SYNCFILE2 SIGPIPE-HELPER: creating SYNCFILE2
SIGPIPE-HELPER: Shutting down SIGPIPE-HELPER: Shutting down
SIGPIPE-HELPER: Server process terminated SIGPIPE-HELPER: Server process terminated with status 255 (no-windows !)
SIGPIPE-HELPER: Server process terminated with status 1 (windows !)
SIGPIPE-HELPER: Shut down SIGPIPE-HELPER: Shut down
The remote should be left in a good state The remote should be left in a good state
$ hg --cwd ../remote tip -T '{node|short}\n' $ hg --cwd ../remote tip -T '{node|short}\n'
000000000000 000000000000
#if windows
XXX-Windows Broken behavior to be fixed
Behavior on Windows is broken and should be fixed. However this is a fairly
corner case situation and no data are being corrupted. This would affect
central repository being hosted on a Windows machine and accessed using ssh.
This was catch as we setup new CI for Windows. Making the test pass on Windows
was enough of a pain that fixing the behavior set aside for now. Dear and
honorable reader, feel free to fix it.
$ hg --cwd ../remote recover
rolling back interrupted transaction
(verify step skipped, run `hg verify` to check your repository content)
#else
$ hg --cwd ../remote recover $ hg --cwd ../remote recover
no interrupted transaction available no interrupted transaction available
[1] [1]
#endif

tests/testlib/sigpipe-remote.py

#!/usr/bin/env python3 #!/usr/bin/env python3
from __future__ import print_function from __future__ import print_function
import io import io
import os import os
import subprocess import subprocess
import sys import sys
import threading
import time import time
# 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 # 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
if isinstance(sys.stdout.buffer, io.BufferedWriter): if isinstance(sys.stdout.buffer, io.BufferedWriter):
print('SIGPIPE-HELPER: script need unbuffered output', file=sys.stderr) print('SIGPIPE-HELPER: script need unbuffered output', file=sys.stderr)
sys.exit(255) sys.exit(255)
def sysbytes(s): def sysbytes(s):
return s.encode('utf-8') return s.encode('utf-8')
def sysstr(s): def sysstr(s):
return s.decode('latin-1') return s.decode('latin-1')
piped_stdout = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC)
piped_stderr = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC)
stdout_writer = os.fdopen(piped_stdout[1], "rb")
stdout_reader = os.fdopen(piped_stdout[0], "rb")
stderr_writer = os.fdopen(piped_stderr[1], "rb")
stderr_reader = os.fdopen(piped_stderr[0], "rb")
debug_stream.write(b'SIGPIPE-HELPER: Starting\n') debug_stream.write(b'SIGPIPE-HELPER: Starting\n')
TESTLIB_DIR = os.path.dirname(sys.argv[0]) TESTLIB_DIR = os.path.dirname(sys.argv[0])
WAIT_SCRIPT = os.path.join(TESTLIB_DIR, 'wait-on-file') WAIT_SCRIPT = os.path.join(TESTLIB_DIR, 'wait-on-file')
hooks_cmd = '%s 10 %s %s' hooks_cmd = '%s 10 %s %s'
hooks_cmd %= ( hooks_cmd %= (
WAIT_SCRIPT, WAIT_SCRIPT,
SYNCFILE2, SYNCFILE2,
SYNCFILE1, SYNCFILE1,
) )
try:
cmd = ['hg'] cmd = ['hg']
cmd += sys.argv[1:] cmd += sys.argv[1:]
sub = subprocess.Popen( sub = subprocess.Popen(
cmd, cmd,
bufsize=0, bufsize=0,
close_fds=True, close_fds=True,
stdin=sys.stdin, stdin=sys.stdin,
stdout=stdout_writer, stdout=subprocess.PIPE,
stderr=stderr_writer, stderr=subprocess.PIPE,
) )
debug_stream.write(b'SIGPIPE-HELPER: Mercurial started\n') basedir = os.path.dirname(sys.argv[0])
worker = os.path.join(basedir, 'sigpipe-worker.py')
shut_down = threading.Event()
close_lock = threading.Lock()
cmd = [sys.executable, worker]
def _read(stream): stdout_worker = subprocess.Popen(
try: cmd,
return stream.read() bufsize=0,
except ValueError: close_fds=True,
# read on closed file stdin=sub.stdout,
return None stdout=sys.stdout,
stderr=sys.stderr,
)
def forward_stdout():
while not shut_down.is_set():
c = _read(stdout_reader)
while c is not None:
sys.stdout.buffer.write(c)
c = _read(stdout_reader)
time.sleep(0.001)
with close_lock:
if not stdout_reader.closed:
stdout_reader.close()
debug_stream.write(b'SIGPIPE-HELPER: stdout closed\n')
def forward_stderr():
while not shut_down.is_set():
c = _read(stderr_reader)
if c is not None:
sys.stderr.buffer.write(c)
c = _read(stderr_reader)
time.sleep(0.001)
with close_lock:
if not stderr_reader.closed:
stderr_reader.close()
debug_stream.write(b'SIGPIPE-HELPER: stderr closed\n')
stdout_thread = threading.Thread(target=forward_stdout, daemon=True)
stderr_thread = threading.Thread(target=forward_stderr, daemon=True)
try:
stdout_thread.start()
stderr_thread.start()
stderr_worker = subprocess.Popen(
cmd,
bufsize=0,
close_fds=True,
stdin=sub.stderr,
stdout=sys.stderr,
stderr=sys.stderr,
)
debug_stream.write(b'SIGPIPE-HELPER: Redirection in place\n') debug_stream.write(b'SIGPIPE-HELPER: Redirection in place\n')
os.close(sub.stdout.fileno())
os.close(sub.stderr.fileno())
debug_stream.write(b'SIGPIPE-HELPER: pipes closed in main\n')
try: try:
wait_file(sysbytes(SYNCFILE1)) wait_file(sysbytes(SYNCFILE1))
except RuntimeError as exc: except RuntimeError as exc:
msg = sysbytes(str(exc)) msg = sysbytes(str(exc))
debug_stream.write(b'SIGPIPE-HELPER: wait failed: %s\n' % msg) debug_stream.write(b'SIGPIPE-HELPER: wait failed: %s\n' % msg)
else: else:
debug_stream.write(b'SIGPIPE-HELPER: SYNCFILE1 detected\n') debug_stream.write(b'SIGPIPE-HELPER: SYNCFILE1 detected\n')
with close_lock: stdout_worker.kill()
if not stdout_reader.closed: stderr_worker.kill()
stdout_reader.close() stdout_worker.wait(10)
if not stderr_reader.closed: stderr_worker.wait(10)
stderr_reader.close() debug_stream.write(b'SIGPIPE-HELPER: worker killed\n')
sys.stdin.close()
debug_stream.write(b'SIGPIPE-HELPER: pipes closed\n')
debug_stream.write(b'SIGPIPE-HELPER: creating SYNCFILE2\n') debug_stream.write(b'SIGPIPE-HELPER: creating SYNCFILE2\n')
write_file(sysbytes(SYNCFILE2)) write_file(sysbytes(SYNCFILE2))
finally: finally:
debug_stream.write(b'SIGPIPE-HELPER: Shutting down\n') debug_stream.write(b'SIGPIPE-HELPER: Shutting down\n')
shut_down.set()
if not sys.stdin.closed: if not sys.stdin.closed:
sys.stdin.close() sys.stdin.close()
try: try:
sub.wait(timeout=30) sub.wait(timeout=30)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
msg = b'SIGPIPE-HELPER: Server process failed to terminate\n' msg = b'SIGPIPE-HELPER: Server process failed to terminate\n'
debug_stream.write(msg) debug_stream.write(msg)
sub.kill()
sub.wait()
msg = b'SIGPIPE-HELPER: Server process killed\n'
else: else:
debug_stream.write(b'SIGPIPE-HELPER: Server process terminated\n') msg = b'SIGPIPE-HELPER: Server process terminated with status %d\n'
msg %= sub.returncode
debug_stream.write(msg)
debug_stream.write(b'SIGPIPE-HELPER: Shut down\n') debug_stream.write(b'SIGPIPE-HELPER: Shut down\n')

tests/testlib/sigpipe-worker.py

  • This file was added.
PropertyOld ValueNew Value
File Modenull100755
#!/usr/bin/env python3
#
# This is literally `cat` but in python, one char at a time.
#
# see sigpipe-remote.py for details.
from __future__ import print_function
import io
import os
import sys
if isinstance(sys.stdout.buffer, io.BufferedWriter):
print('SIGPIPE-WORKER: script need unbuffered output', file=sys.stderr)
sys.exit(255)
while True:
c = os.read(sys.stdin.fileno(), 1)
os.write(sys.stdout.fileno(), c)