This is an archive of the discontinued Mercurial Phabricator instance.

Differential D9020

sshpeer: avoid having a destructor that can block forks forever
AbandonedPublic

Authored by valentin.gatienbaron on Sep 14 2020, 11:07 AM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Summary

This is solution 2. that was described in the previous commit.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

durin42 added subscribers: indygreg, durin42.Oct 14 2020, 12:49 PM

So I think I actually prefer D9019 by a hair here. @indygreg do you have a preference?

valentin.gatienbaron added a comment.Oct 15 2020, 12:13 AM

I meant for this change to be in addition to D9019, rather than an alternative, for robustness as written in a bit more detail in the commit message of D9019.

Although looking at these couple of changes again, maybe instead I could close the peers explicitly in push/pull/incoming/outgoing/id. Such a change doesn't fix all deadlocks on its own, but along with D9019, it should be enough.

marmoute added a subscriber: marmoute.Jan 7 2021, 6:13 PM

Having thing explicitly close is probably better, why maybe a developer warning when this isn't the case ?

valentin.gatienbaron added a comment.EditedJan 25 2021, 11:49 AM

Closing the peer is what I was describing as solution 3 in D9019. But I think I was wrong in saying that it doesn't help with calls to logtoprocess in the middle of commands. I also just noticed that pull does that, in fact. I'll try and find some time to do that.

valentin.gatienbaron added a comment.Feb 15 2021, 4:43 PM

I submitted changes to close peers explicitly instead of doing this. It's definitely more invasive though.

Revision Contents
Changeset List

PathPackages
Mmercurial/sshpeer.py (27 lines)

Diff 25636

mercurial/sshpeer.py

# sshpeer.py - ssh repository proxy class for mercurial # sshpeer.py - ssh repository proxy class for mercurial
# #
# Copyright 2005, 2006 Matt Mackall <mpm@selenic.com> # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import os
import re import re
import uuid import uuid
from .i18n import _ from .i18n import _
from .pycompat import getattr from .pycompat import getattr
from . import ( from . import (
error, error,
pycompat, pycompat,
def close(self): def close(self):
return self._main.close() return self._main.close()
def flush(self): def flush(self):
return self._main.flush() return self._main.flush()
def _cleanuppipes(ui, pipei, pipeo, pipee): def _cleanuppipes(ui, pipei, pipeo, pipee, origpid):
"""Clean up pipes used by an SSH connection.""" """Clean up pipes used by an SSH connection."""
if pipeo: if pipeo:
pipeo.close() pipeo.close()
if pipei: if pipei:
pipei.close() pipei.close()
if pipee: if pipee:
# Try to read from the err descriptor until EOF. if origpid == os.getpid():
# Try to read from the err descriptor until EOF. But if we
# "leak" an ssh peer in a fork, just close without
# reading.
try: try:
for l in pipee: for l in pipee:
ui.status(_(b'remote: '), l) ui.status(_(b'remote: '), l)
except (IOError, ValueError): except (IOError, ValueError):
pass pass
pipee.close() pipee.close()
def _makeconnection(ui, sshcmd, args, remotecmd, path, sshenv=None): def _makeconnection(ui, sshcmd, args, remotecmd, path, sshenv=None):
"""Create an SSH connection to a server. """Create an SSH connection to a server.
Returns a tuple of (process, stdin, stdout, stderr) for the Returns a tuple of (process, stdin, stdout, stderr) for the
# stderr any time we perform I/O. # stderr any time we perform I/O.
if autoreadstderr: if autoreadstderr:
stdout = doublepipe(ui, util.bufferedinputpipe(stdout), stderr) stdout = doublepipe(ui, util.bufferedinputpipe(stdout), stderr)
stdin = doublepipe(ui, stdin, stderr) stdin = doublepipe(ui, stdin, stderr)
self._pipeo = stdin self._pipeo = stdin
self._pipei = stdout self._pipei = stdout
self._pipee = stderr self._pipee = stderr
self._origpid = os.getpid() # recorded to detect calls to fork()
self._caps = caps self._caps = caps
self._autoreadstderr = autoreadstderr self._autoreadstderr = autoreadstderr
# Commands that have a "framed" response where the first line of the # Commands that have a "framed" response where the first line of the
# response contains the length of that response. # response contains the length of that response.
_FRAMED_COMMANDS = { _FRAMED_COMMANDS = {
b'batch', b'batch',
} }
def _readerr(self): def _readerr(self):
_forwardoutput(self.ui, self._pipee) _forwardoutput(self.ui, self._pipee)
def _abort(self, exception): def _abort(self, exception):
self._cleanup() self._cleanup()
raise exception raise exception
def _cleanup(self): def _cleanup(self):
_cleanuppipes(self.ui, self._pipei, self._pipeo, self._pipee) _cleanuppipes(self.ui, self._pipei, self._pipeo, self._pipee,
self._origpid)
__del__ = _cleanup __del__ = _cleanup
def _sendrequest(self, cmd, args, framed=False): def _sendrequest(self, cmd, args, framed=False):
if self.ui.debugflag and self.ui.configbool( if self.ui.debugflag and self.ui.configbool(
b'devel', b'debug.peer-request' b'devel', b'debug.peer-request'
): ):
dbg = self.ui.debug dbg = self.ui.debug
``stdin``, ``stdout``, and ``stderr`` are the pipes connected to the ``stdin``, ``stdout``, and ``stderr`` are the pipes connected to the
SSH server's stdio handles. SSH server's stdio handles.
This function is factored out to allow creating peers that don't This function is factored out to allow creating peers that don't
actually spawn a new process. It is useful for starting SSH protocol actually spawn a new process. It is useful for starting SSH protocol
servers and clients via non-standard means, which can be useful for servers and clients via non-standard means, which can be useful for
testing. testing.
""" """
origgpid = os.getpid()
try: try:
protoname, caps = _performhandshake(ui, stdin, stdout, stderr) protoname, caps = _performhandshake(ui, stdin, stdout, stderr)
except Exception: except Exception:
_cleanuppipes(ui, stdout, stdin, stderr) _cleanuppipes(ui, stdout, stdin, stderr, origgpid)
raise raise
if protoname == wireprototypes.SSHV1: if protoname == wireprototypes.SSHV1:
return sshv1peer( return sshv1peer(
ui, ui,
path, path,
proc, proc,
stdin, stdin,
proc, proc,
stdin, stdin,
stdout, stdout,
stderr, stderr,
caps, caps,
autoreadstderr=autoreadstderr, autoreadstderr=autoreadstderr,
) )
else: else:
_cleanuppipes(ui, stdout, stdin, stderr) _cleanuppipes(ui, stdout, stdin, stderr, origgpid)
raise error.RepoError( raise error.RepoError(
_(b'unknown version of SSH protocol: %s') % protoname _(b'unknown version of SSH protocol: %s') % protoname
) )
def instance(ui, path, create, intents=None, createopts=None): def instance(ui, path, create, intents=None, createopts=None):
"""Create an SSH peer. """Create an SSH peer.