chgserver: backport py3 buffered I/O workarounds from procutil

I've recently switched to new machine and I found chg's stdout is fully
buffered.

Even though chg server is a daemon process, it inherits the environment
where the chg client originally forked the server. This means the server's
stdout might have been wrapped by LineBufferedWrapper. That's why we need
to do wrap/unwrap in both ways.

The "if" condition in _restoreio() looks weird, but I'm not willing to
clean things up because stdio behavior is fundamentally different between
py2 and py3, and py2 support will be dropped anyway.

mercurial/chgserver.py

@@ -1,741 +1,753 b''
 # chgserver.py - command server extension for cHg
 #
 # Copyright 2011 Yuya Nishihara <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 """command server extension for cHg
 
 'S' channel (read/write)
     propagate ui.system() request to client
 
 'attachio' command
     attach client's stdio passed by sendmsg()
 
 'chdir' command
     change current directory
 
 'setenv' command
     replace os.environ completely
 
 'setumask' command (DEPRECATED)
 'setumask2' command
     set umask
 
 'validate' command
     reload the config and check if the server is up to date
 
 Config
 ------
 
 ::
 
   [chgserver]
   # how long (in seconds) should an idle chg server exit
   idletimeout = 3600
 
   # whether to skip config or env change checks
   skiphash = False
 """
 
 from __future__ import absolute_import
 
 import inspect
 import os
 import re
 import socket
 import stat
 import struct
 import time
 
 from .i18n import _
 from .pycompat import (
     getattr,
     setattr,
 )
 
 from . import (
     commandserver,
     encoding,
     error,
     extensions,
     node,
     pycompat,
     util,
 )
 
 from .utils import (
     hashutil,
     procutil,
     stringutil,
 )
 
 
 def _hashlist(items):
     """return sha1 hexdigest for a list"""
     return node.hex(hashutil.sha1(stringutil.pprint(items)).digest())
 
 
 # sensitive config sections affecting confighash
 _configsections = [
     b'alias',  # affects global state commands.table
     b'diff-tools',  # affects whether gui or not in extdiff's uisetup
     b'eol',  # uses setconfig('eol', ...)
     b'extdiff',  # uisetup will register new commands
     b'extensions',
     b'fastannotate',  # affects annotate command and adds fastannonate cmd
     b'merge-tools',  # affects whether gui or not in extdiff's uisetup
     b'schemes',  # extsetup will update global hg.schemes
 ]
 
 _configsectionitems = [
     (b'commands', b'show.aliasprefix'),  # show.py reads it in extsetup
 ]
 
 # sensitive environment variables affecting confighash
 _envre = re.compile(
     br'''\A(?:
                     CHGHG
                     |HG(?:DEMANDIMPORT|EMITWARNINGS|MODULEPOLICY|PROF|RCPATH)?
                     |HG(?:ENCODING|PLAIN).*
                     |LANG(?:UAGE)?
                     |LC_.*
                     |LD_.*
                     |PATH
                     |PYTHON.*
                     |TERM(?:INFO)?
                     |TZ
                     )\Z''',
     re.X,
 )
 
 
 def _confighash(ui):
     """return a quick hash for detecting config/env changes
 
     confighash is the hash of sensitive config items and environment variables.
 
     for chgserver, it is designed that once confighash changes, the server is
     not qualified to serve its client and should redirect the client to a new
     server. different from mtimehash, confighash change will not mark the
     server outdated and exit since the user can have different configs at the
     same time.
     """
     sectionitems = []
     for section in _configsections:
         sectionitems.append(ui.configitems(section))
     for section, item in _configsectionitems:
         sectionitems.append(ui.config(section, item))
     sectionhash = _hashlist(sectionitems)
     # If $CHGHG is set, the change to $HG should not trigger a new chg server
     if b'CHGHG' in encoding.environ:
         ignored = {b'HG'}
     else:
         ignored = set()
     envitems = [
         (k, v)
         for k, v in pycompat.iteritems(encoding.environ)
         if _envre.match(k) and k not in ignored
     ]
     envhash = _hashlist(sorted(envitems))
     return sectionhash[:6] + envhash[:6]
 
 
 def _getmtimepaths(ui):
     """get a list of paths that should be checked to detect change
 
     The list will include:
     - extensions (will not cover all files for complex extensions)
     - mercurial/__version__.py
     - python binary
     """
     modules = [m for n, m in extensions.extensions(ui)]
     try:
         from . import __version__
 
         modules.append(__version__)
     except ImportError:
         pass
     files = []
     if pycompat.sysexecutable:
         files.append(pycompat.sysexecutable)
     for m in modules:
         try:
             files.append(pycompat.fsencode(inspect.getabsfile(m)))
         except TypeError:
             pass
     return sorted(set(files))
 
 
 def _mtimehash(paths):
     """return a quick hash for detecting file changes
 
     mtimehash calls stat on given paths and calculate a hash based on size and
     mtime of each file. mtimehash does not read file content because reading is
     expensive. therefore it's not 100% reliable for detecting content changes.
     it's possible to return different hashes for same file contents.
     it's also possible to return a same hash for different file contents for
     some carefully crafted situation.
 
     for chgserver, it is designed that once mtimehash changes, the server is
     considered outdated immediately and should no longer provide service.
 
     mtimehash is not included in confighash because we only know the paths of
     extensions after importing them (there is imp.find_module but that faces
     race conditions). We need to calculate confighash without importing.
     """
 
     def trystat(path):
         try:
             st = os.stat(path)
             return (st[stat.ST_MTIME], st.st_size)
         except OSError:
             # could be ENOENT, EPERM etc. not fatal in any case
             pass
 
     return _hashlist(pycompat.maplist(trystat, paths))[:12]
 
 
 class hashstate(object):
     """a structure storing confighash, mtimehash, paths used for mtimehash"""
 
     def __init__(self, confighash, mtimehash, mtimepaths):
         self.confighash = confighash
         self.mtimehash = mtimehash
         self.mtimepaths = mtimepaths
 
     @staticmethod
     def fromui(ui, mtimepaths=None):
         if mtimepaths is None:
             mtimepaths = _getmtimepaths(ui)
         confighash = _confighash(ui)
         mtimehash = _mtimehash(mtimepaths)
         ui.log(
             b'cmdserver',
             b'confighash = %s mtimehash = %s\n',
             confighash,
             mtimehash,
         )
         return hashstate(confighash, mtimehash, mtimepaths)
 
 
 def _newchgui(srcui, csystem, attachio):
     class chgui(srcui.__class__):
         def __init__(self, src=None):
             super(chgui, self).__init__(src)
             if src:
                 self._csystem = getattr(src, '_csystem', csystem)
             else:
                 self._csystem = csystem
 
         def _runsystem(self, cmd, environ, cwd, out):
             # fallback to the original system method if
             #  a. the output stream is not stdout (e.g. stderr, cStringIO),
             #  b. or stdout is redirected by protectfinout(),
             # because the chg client is not aware of these situations and
             # will behave differently (i.e. write to stdout).
             if (
                 out is not self.fout
                 or not util.safehasattr(self.fout, b'fileno')
                 or self.fout.fileno() != procutil.stdout.fileno()
                 or self._finoutredirected
             ):
                 return procutil.system(cmd, environ=environ, cwd=cwd, out=out)
             self.flush()
             return self._csystem(cmd, procutil.shellenviron(environ), cwd)
 
         def _runpager(self, cmd, env=None):
             self._csystem(
                 cmd,
                 procutil.shellenviron(env),
                 type=b'pager',
                 cmdtable={b'attachio': attachio},
             )
             return True
 
     return chgui(srcui)
 
 
 def _loadnewui(srcui, args, cdebug):
     from . import dispatch  # avoid cycle
 
     newui = srcui.__class__.load()
     for a in [b'fin', b'fout', b'ferr', b'environ']:
         setattr(newui, a, getattr(srcui, a))
     if util.safehasattr(srcui, b'_csystem'):
         newui._csystem = srcui._csystem
 
     # command line args
     options = dispatch._earlyparseopts(newui, args)
     dispatch._parseconfig(newui, options[b'config'])
 
     # stolen from tortoisehg.util.copydynamicconfig()
     for section, name, value in srcui.walkconfig():
         source = srcui.configsource(section, name)
         if b':' in source or source == b'--config' or source.startswith(b'$'):
             # path:line or command line, or environ
             continue
         newui.setconfig(section, name, value, source)
 
     # load wd and repo config, copied from dispatch.py
     cwd = options[b'cwd']
     cwd = cwd and os.path.realpath(cwd) or None
     rpath = options[b'repository']
     path, newlui = dispatch._getlocal(newui, rpath, wd=cwd)
 
     extensions.populateui(newui)
     commandserver.setuplogging(newui, fp=cdebug)
     if newui is not newlui:
         extensions.populateui(newlui)
         commandserver.setuplogging(newlui, fp=cdebug)
 
     return (newui, newlui)
 
 
 class channeledsystem(object):
     """Propagate ui.system() request in the following format:
 
     payload length (unsigned int),
     type, '\0',
     cmd, '\0',
     cwd, '\0',
     envkey, '=', val, '\0',
     ...
     envkey, '=', val
 
     if type == 'system', waits for:
 
     exitcode length (unsigned int),
     exitcode (int)
 
     if type == 'pager', repetitively waits for a command name ending with '\n'
     and executes it defined by cmdtable, or exits the loop if the command name
     is empty.
     """
 
     def __init__(self, in_, out, channel):
         self.in_ = in_
         self.out = out
         self.channel = channel
 
     def __call__(self, cmd, environ, cwd=None, type=b'system', cmdtable=None):
         args = [type, cmd, os.path.abspath(cwd or b'.')]
         args.extend(b'%s=%s' % (k, v) for k, v in pycompat.iteritems(environ))
         data = b'\0'.join(args)
         self.out.write(struct.pack(b'>cI', self.channel, len(data)))
         self.out.write(data)
         self.out.flush()
 
         if type == b'system':
             length = self.in_.read(4)
             (length,) = struct.unpack(b'>I', length)
             if length != 4:
                 raise error.Abort(_(b'invalid response'))
             (rc,) = struct.unpack(b'>i', self.in_.read(4))
             return rc
         elif type == b'pager':
             while True:
                 cmd = self.in_.readline()[:-1]
                 if not cmd:
                     break
                 if cmdtable and cmd in cmdtable:
                     cmdtable[cmd]()
                 else:
                     raise error.Abort(_(b'unexpected command: %s') % cmd)
         else:
             raise error.ProgrammingError(b'invalid S channel type: %s' % type)
 
 
 _iochannels = [
     # server.ch, ui.fp, mode
     (b'cin', b'fin', 'rb'),
     (b'cout', b'fout', 'wb'),
     (b'cerr', b'ferr', 'wb'),
 ]
 
 
 class chgcmdserver(commandserver.server):
     def __init__(
         self, ui, repo, fin, fout, sock, prereposetups, hashstate, baseaddress
     ):
         super(chgcmdserver, self).__init__(
             _newchgui(ui, channeledsystem(fin, fout, b'S'), self.attachio),
             repo,
             fin,
             fout,
             prereposetups,
         )
         self.clientsock = sock
         self._ioattached = False
         self._oldios = []  # original (self.ch, ui.fp, fd) before "attachio"
         self.hashstate = hashstate
         self.baseaddress = baseaddress
         if hashstate is not None:
             self.capabilities = self.capabilities.copy()
             self.capabilities[b'validate'] = chgcmdserver.validate
 
     def cleanup(self):
         super(chgcmdserver, self).cleanup()
         # dispatch._runcatch() does not flush outputs if exception is not
         # handled by dispatch._dispatch()
         self.ui.flush()
         self._restoreio()
         self._ioattached = False
 
     def attachio(self):
         """Attach to client's stdio passed via unix domain socket; all
         channels except cresult will no longer be used
         """
         # tell client to sendmsg() with 1-byte payload, which makes it
         # distinctive from "attachio\n" command consumed by client.read()
         self.clientsock.sendall(struct.pack(b'>cI', b'I', 1))
         clientfds = util.recvfds(self.clientsock.fileno())
         self.ui.log(b'chgserver', b'received fds: %r\n', clientfds)
 
         ui = self.ui
         ui.flush()
         self._saveio()
         for fd, (cn, fn, mode) in zip(clientfds, _iochannels):
             assert fd > 0
             fp = getattr(ui, fn)
             os.dup2(fd, fp.fileno())
             os.close(fd)
             if self._ioattached:
                 continue
             # reset buffering mode when client is first attached. as we want
             # to see output immediately on pager, the mode stays unchanged
             # when client re-attached. ferr is unchanged because it should
             # be unbuffered no matter if it is a tty or not.
             if fn == b'ferr':
                 newfp = fp
+            elif pycompat.ispy3:
+                # On Python 3, the standard library doesn't offer line-buffered
+                # binary streams, so wrap/unwrap it.
+                if fp.isatty():
+                    newfp = procutil.make_line_buffered(fp)
+                else:
+                    newfp = procutil.unwrap_line_buffered(fp)
             else:
-                # make it line buffered explicitly because the default is
-                # decided on first write(), where fout could be a pager.
+                # Python 2 uses the I/O streams provided by the C library, so
+                # make it line-buffered explicitly. Otherwise the default would
+                # be decided on first write(), where fout could be a pager.
                 if fp.isatty():
                     bufsize = 1  # line buffered
                 else:
                     bufsize = -1  # system default
                 newfp = os.fdopen(fp.fileno(), mode, bufsize)
+            if newfp is not fp:
                 setattr(ui, fn, newfp)
             setattr(self, cn, newfp)
 
         self._ioattached = True
         self.cresult.write(struct.pack(b'>i', len(clientfds)))
 
     def _saveio(self):
         if self._oldios:
             return
         ui = self.ui
         for cn, fn, _mode in _iochannels:
             ch = getattr(self, cn)
             fp = getattr(ui, fn)
             fd = os.dup(fp.fileno())
             self._oldios.append((ch, fp, fd))
 
     def _restoreio(self):
         if not self._oldios:
             return
         nullfd = os.open(os.devnull, os.O_WRONLY)
         ui = self.ui
         for (ch, fp, fd), (cn, fn, mode) in zip(self._oldios, _iochannels):
             newfp = getattr(ui, fn)
-            # close newfp while it's associated with client; otherwise it
-            # would be closed when newfp is deleted
-            if newfp is not fp:
+            # On Python 2, newfp and fp may be separate file objects associated
+            # with the same fd, so we must close newfp while it's associated
+            # with the client. Otherwise the new associated fd would be closed
+            # when newfp gets deleted. On Python 3, newfp is just a wrapper
+            # around fp even if newfp is not fp, so deleting newfp is safe.
+            if not (pycompat.ispy3 or newfp is fp):
                 newfp.close()
             # restore original fd: fp is open again
             try:
-                if newfp is fp and 'w' in mode:
+                if (pycompat.ispy3 or newfp is fp) and 'w' in mode:
                     # Discard buffered data which couldn't be flushed because
                     # of EPIPE. The data should belong to the current session
                     # and should never persist.
                     os.dup2(nullfd, fp.fileno())
                     fp.flush()
                 os.dup2(fd, fp.fileno())
             except OSError as err:
                 # According to issue6330, running chg on heavy loaded systems
                 # can lead to EBUSY. [man dup2] indicates that, on Linux,
                 # EBUSY comes from a race condition between open() and dup2().
                 # However it's not clear why open() race occurred for
                 # newfd=stdin/out/err.
                 self.ui.log(
                     b'chgserver',
                     b'got %s while duplicating %s\n',
                     stringutil.forcebytestr(err),
                     fn,
                 )
             os.close(fd)
             setattr(self, cn, ch)
             setattr(ui, fn, fp)
         os.close(nullfd)
         del self._oldios[:]
 
     def validate(self):
         """Reload the config and check if the server is up to date
 
         Read a list of '\0' separated arguments.
         Write a non-empty list of '\0' separated instruction strings or '\0'
         if the list is empty.
         An instruction string could be either:
             - "unlink $path", the client should unlink the path to stop the
               outdated server.
             - "redirect $path", the client should attempt to connect to $path
               first. If it does not work, start a new server. It implies
               "reconnect".
             - "exit $n", the client should exit directly with code n.
               This may happen if we cannot parse the config.
             - "reconnect", the client should close the connection and
               reconnect.
         If neither "reconnect" nor "redirect" is included in the instruction
         list, the client can continue with this server after completing all
         the instructions.
         """
         from . import dispatch  # avoid cycle
 
         args = self._readlist()
         try:
             self.ui, lui = _loadnewui(self.ui, args, self.cdebug)
         except error.ParseError as inst:
             dispatch._formatparse(self.ui.warn, inst)
             self.ui.flush()
             self.cresult.write(b'exit 255')
             return
         except error.Abort as inst:
             self.ui.error(_(b"abort: %s\n") % inst.message)
             if inst.hint:
                 self.ui.error(_(b"(%s)\n") % inst.hint)
             self.ui.flush()
             self.cresult.write(b'exit 255')
             return
         newhash = hashstate.fromui(lui, self.hashstate.mtimepaths)
         insts = []
         if newhash.mtimehash != self.hashstate.mtimehash:
             addr = _hashaddress(self.baseaddress, self.hashstate.confighash)
             insts.append(b'unlink %s' % addr)
             # mtimehash is empty if one or more extensions fail to load.
             # to be compatible with hg, still serve the client this time.
             if self.hashstate.mtimehash:
                 insts.append(b'reconnect')
         if newhash.confighash != self.hashstate.confighash:
             addr = _hashaddress(self.baseaddress, newhash.confighash)
             insts.append(b'redirect %s' % addr)
         self.ui.log(b'chgserver', b'validate: %s\n', stringutil.pprint(insts))
         self.cresult.write(b'\0'.join(insts) or b'\0')
 
     def chdir(self):
         """Change current directory
 
         Note that the behavior of --cwd option is bit different from this.
         It does not affect --config parameter.
         """
         path = self._readstr()
         if not path:
             return
         self.ui.log(b'chgserver', b"chdir to '%s'\n", path)
         os.chdir(path)
 
     def setumask(self):
         """Change umask (DEPRECATED)"""
         # BUG: this does not follow the message frame structure, but kept for
         # backward compatibility with old chg clients for some time
         self._setumask(self._read(4))
 
     def setumask2(self):
         """Change umask"""
         data = self._readstr()
         if len(data) != 4:
             raise ValueError(b'invalid mask length in setumask2 request')
         self._setumask(data)
 
     def _setumask(self, data):
         mask = struct.unpack(b'>I', data)[0]
         self.ui.log(b'chgserver', b'setumask %r\n', mask)
         util.setumask(mask)
 
     def runcommand(self):
         # pager may be attached within the runcommand session, which should
         # be detached at the end of the session. otherwise the pager wouldn't
         # receive EOF.
         globaloldios = self._oldios
         self._oldios = []
         try:
             return super(chgcmdserver, self).runcommand()
         finally:
             self._restoreio()
             self._oldios = globaloldios
 
     def setenv(self):
         """Clear and update os.environ
 
         Note that not all variables can make an effect on the running process.
         """
         l = self._readlist()
         try:
             newenv = dict(s.split(b'=', 1) for s in l)
         except ValueError:
             raise ValueError(b'unexpected value in setenv request')
         self.ui.log(b'chgserver', b'setenv: %r\n', sorted(newenv.keys()))
 
         encoding.environ.clear()
         encoding.environ.update(newenv)
 
     capabilities = commandserver.server.capabilities.copy()
     capabilities.update(
         {
             b'attachio': attachio,
             b'chdir': chdir,
             b'runcommand': runcommand,
             b'setenv': setenv,
             b'setumask': setumask,
             b'setumask2': setumask2,
         }
     )
 
     if util.safehasattr(procutil, b'setprocname'):
 
         def setprocname(self):
             """Change process title"""
             name = self._readstr()
             self.ui.log(b'chgserver', b'setprocname: %r\n', name)
             procutil.setprocname(name)
 
         capabilities[b'setprocname'] = setprocname
 
 
 def _tempaddress(address):
     return b'%s.%d.tmp' % (address, os.getpid())
 
 
 def _hashaddress(address, hashstr):
     # if the basename of address contains '.', use only the left part. this
     # makes it possible for the client to pass 'server.tmp$PID' and follow by
     # an atomic rename to avoid locking when spawning new servers.
     dirname, basename = os.path.split(address)
     basename = basename.split(b'.', 1)[0]
     return b'%s-%s' % (os.path.join(dirname, basename), hashstr)
 
 
 class chgunixservicehandler(object):
     """Set of operations for chg services"""
 
     pollinterval = 1  # [sec]
 
     def __init__(self, ui):
         self.ui = ui
         self._idletimeout = ui.configint(b'chgserver', b'idletimeout')
         self._lastactive = time.time()
 
     def bindsocket(self, sock, address):
         self._inithashstate(address)
         self._checkextensions()
         self._bind(sock)
         self._createsymlink()
         # no "listening at" message should be printed to simulate hg behavior
 
     def _inithashstate(self, address):
         self._baseaddress = address
         if self.ui.configbool(b'chgserver', b'skiphash'):
             self._hashstate = None
             self._realaddress = address
             return
         self._hashstate = hashstate.fromui(self.ui)
         self._realaddress = _hashaddress(address, self._hashstate.confighash)
 
     def _checkextensions(self):
         if not self._hashstate:
             return
         if extensions.notloaded():
             # one or more extensions failed to load. mtimehash becomes
             # meaningless because we do not know the paths of those extensions.
             # set mtimehash to an illegal hash value to invalidate the server.
             self._hashstate.mtimehash = b''
 
     def _bind(self, sock):
         # use a unique temp address so we can stat the file and do ownership
         # check later
         tempaddress = _tempaddress(self._realaddress)
         util.bindunixsocket(sock, tempaddress)
         self._socketstat = os.stat(tempaddress)
         sock.listen(socket.SOMAXCONN)
         # rename will replace the old socket file if exists atomically. the
         # old server will detect ownership change and exit.
         util.rename(tempaddress, self._realaddress)
 
     def _createsymlink(self):
         if self._baseaddress == self._realaddress:
             return
         tempaddress = _tempaddress(self._baseaddress)
         os.symlink(os.path.basename(self._realaddress), tempaddress)
         util.rename(tempaddress, self._baseaddress)
 
     def _issocketowner(self):
         try:
             st = os.stat(self._realaddress)
             return (
                 st.st_ino == self._socketstat.st_ino
                 and st[stat.ST_MTIME] == self._socketstat[stat.ST_MTIME]
             )
         except OSError:
             return False
 
     def unlinksocket(self, address):
         if not self._issocketowner():
             return
         # it is possible to have a race condition here that we may
         # remove another server's socket file. but that's okay
         # since that server will detect and exit automatically and
         # the client will start a new server on demand.
         util.tryunlink(self._realaddress)
 
     def shouldexit(self):
         if not self._issocketowner():
             self.ui.log(
                 b'chgserver', b'%s is not owned, exiting.\n', self._realaddress
             )
             return True
         if time.time() - self._lastactive > self._idletimeout:
             self.ui.log(b'chgserver', b'being idle too long. exiting.\n')
             return True
         return False
 
     def newconnection(self):
         self._lastactive = time.time()
 
     def createcmdserver(self, repo, conn, fin, fout, prereposetups):
         return chgcmdserver(
             self.ui,
             repo,
             fin,
             fout,
             conn,
             prereposetups,
             self._hashstate,
             self._baseaddress,
         )
 
 
 def chgunixservice(ui, repo, opts):
     # CHGINTERNALMARK is set by chg client. It is an indication of things are
     # started by chg so other code can do things accordingly, like disabling
     # demandimport or detecting chg client started by chg client. When executed
     # here, CHGINTERNALMARK is no longer useful and hence dropped to make
     # environ cleaner.
     if b'CHGINTERNALMARK' in encoding.environ:
         del encoding.environ[b'CHGINTERNALMARK']
     # Python3.7+ "coerces" the LC_CTYPE environment variable to a UTF-8 one if
     # it thinks the current value is "C". This breaks the hash computation and
     # causes chg to restart loop.
     if b'CHGORIG_LC_CTYPE' in encoding.environ:
         encoding.environ[b'LC_CTYPE'] = encoding.environ[b'CHGORIG_LC_CTYPE']
         del encoding.environ[b'CHGORIG_LC_CTYPE']
     elif b'CHG_CLEAR_LC_CTYPE' in encoding.environ:
         if b'LC_CTYPE' in encoding.environ:
             del encoding.environ[b'LC_CTYPE']
         del encoding.environ[b'CHG_CLEAR_LC_CTYPE']
 
     if repo:
         # one chgserver can serve multiple repos. drop repo information
         ui.setconfig(b'bundle', b'mainreporoot', b'', b'repo')
     h = chgunixservicehandler(ui)
     return commandserver.unixforkingservice(ui, repo=None, opts=opts, handler=h)

mercurial/utils/procutil.py

@@ -1,780 +1,787 b''
 # procutil.py - utility for managing processes and executable environment
 #
 #  Copyright 2005 K. Thananchayan <[e-mail]>
 #  Copyright 2005-2007 Matt Mackall <[e-mail]>
 #  Copyright 2006 Vadim Gelfer <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import contextlib
 import errno
 import io
 import os
 import signal
 import subprocess
 import sys
 import threading
 import time
 
 from ..i18n import _
 from ..pycompat import (
     getattr,
     open,
 )
 
 from .. import (
     encoding,
     error,
     policy,
     pycompat,
 )
 
 # Import like this to keep import-checker happy
 from ..utils import resourceutil
 
 osutil = policy.importmod('osutil')
 
 if pycompat.iswindows:
     from .. import windows as platform
 else:
     from .. import posix as platform
 
 
 def isatty(fp):
     try:
         return fp.isatty()
     except AttributeError:
         return False
 
 
 class LineBufferedWrapper(object):
     def __init__(self, orig):
         self.orig = orig
 
     def __getattr__(self, attr):
         return getattr(self.orig, attr)
 
     def write(self, s):
         orig = self.orig
         res = orig.write(s)
         if s.endswith(b'\n'):
             orig.flush()
         return res
 
 
 io.BufferedIOBase.register(LineBufferedWrapper)
 
 
 def make_line_buffered(stream):
     if pycompat.ispy3 and not isinstance(stream, io.BufferedIOBase):
         # On Python 3, buffered streams can be expected to subclass
         # BufferedIOBase. This is definitively the case for the streams
         # initialized by the interpreter. For unbuffered streams, we don't need
         # to emulate line buffering.
         return stream
     if isinstance(stream, LineBufferedWrapper):
         return stream
     return LineBufferedWrapper(stream)
 
 
+def unwrap_line_buffered(stream):
+    if isinstance(stream, LineBufferedWrapper):
+        assert not isinstance(stream.orig, LineBufferedWrapper)
+        return stream.orig
+    return stream
+
+
 class WriteAllWrapper(object):
     def __init__(self, orig):
         self.orig = orig
 
     def __getattr__(self, attr):
         return getattr(self.orig, attr)
 
     def write(self, s):
         write1 = self.orig.write
         m = memoryview(s)
         total_to_write = len(s)
         total_written = 0
         while total_written < total_to_write:
             total_written += write1(m[total_written:])
         return total_written
 
 
 io.IOBase.register(WriteAllWrapper)
 
 
 def _make_write_all(stream):
     assert pycompat.ispy3
     if isinstance(stream, WriteAllWrapper):
         return stream
     if isinstance(stream, io.BufferedIOBase):
         # The io.BufferedIOBase.write() contract guarantees that all data is
         # written.
         return stream
     # In general, the write() method of streams is free to write only part of
     # the data.
     return WriteAllWrapper(stream)
 
 
 if pycompat.ispy3:
     # Python 3 implements its own I/O streams.
     # TODO: .buffer might not exist if std streams were replaced; we'll need
     # a silly wrapper to make a bytes stream backed by a unicode one.
     stdin = sys.stdin.buffer
     stdout = _make_write_all(sys.stdout.buffer)
     stderr = _make_write_all(sys.stderr.buffer)
     if pycompat.iswindows:
         # Work around Windows bugs.
         stdout = platform.winstdout(stdout)
         stderr = platform.winstdout(stderr)
     if isatty(stdout):
         # The standard library doesn't offer line-buffered binary streams.
         stdout = make_line_buffered(stdout)
 else:
     # Python 2 uses the I/O streams provided by the C library.
     stdin = sys.stdin
     stdout = sys.stdout
     stderr = sys.stderr
     if pycompat.iswindows:
         # Work around Windows bugs.
         stdout = platform.winstdout(stdout)
         stderr = platform.winstdout(stderr)
     if isatty(stdout):
         if pycompat.iswindows:
             # The Windows C runtime library doesn't support line buffering.
             stdout = make_line_buffered(stdout)
         else:
             # glibc determines buffering on first write to stdout - if we
             # replace a TTY destined stdout with a pipe destined stdout (e.g.
             # pager), we want line buffering.
             stdout = os.fdopen(stdout.fileno(), 'wb', 1)
 
 
 findexe = platform.findexe
 _gethgcmd = platform.gethgcmd
 getuser = platform.getuser
 getpid = os.getpid
 hidewindow = platform.hidewindow
 readpipe = platform.readpipe
 setbinary = platform.setbinary
 setsignalhandler = platform.setsignalhandler
 shellquote = platform.shellquote
 shellsplit = platform.shellsplit
 spawndetached = platform.spawndetached
 sshargs = platform.sshargs
 testpid = platform.testpid
 
 try:
     setprocname = osutil.setprocname
 except AttributeError:
     pass
 try:
     unblocksignal = osutil.unblocksignal
 except AttributeError:
     pass
 
 closefds = pycompat.isposix
 
 
 def explainexit(code):
     """return a message describing a subprocess status
     (codes from kill are negative - not os.system/wait encoding)"""
     if code >= 0:
         return _(b"exited with status %d") % code
     return _(b"killed by signal %d") % -code
 
 
 class _pfile(object):
     """File-like wrapper for a stream opened by subprocess.Popen()"""
 
     def __init__(self, proc, fp):
         self._proc = proc
         self._fp = fp
 
     def close(self):
         # unlike os.popen(), this returns an integer in subprocess coding
         self._fp.close()
         return self._proc.wait()
 
     def __iter__(self):
         return iter(self._fp)
 
     def __getattr__(self, attr):
         return getattr(self._fp, attr)
 
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_value, exc_tb):
         self.close()
 
 
 def popen(cmd, mode=b'rb', bufsize=-1):
     if mode == b'rb':
         return _popenreader(cmd, bufsize)
     elif mode == b'wb':
         return _popenwriter(cmd, bufsize)
     raise error.ProgrammingError(b'unsupported mode: %r' % mode)
 
 
 def _popenreader(cmd, bufsize):
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=bufsize,
         close_fds=closefds,
         stdout=subprocess.PIPE,
     )
     return _pfile(p, p.stdout)
 
 
 def _popenwriter(cmd, bufsize):
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=bufsize,
         close_fds=closefds,
         stdin=subprocess.PIPE,
     )
     return _pfile(p, p.stdin)
 
 
 def popen2(cmd, env=None):
     # Setting bufsize to -1 lets the system decide the buffer size.
     # The default for bufsize is 0, meaning unbuffered. This leads to
     # poor performance on Mac OS X: http://bugs.python.org/issue4194
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=-1,
         close_fds=closefds,
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
         env=tonativeenv(env),
     )
     return p.stdin, p.stdout
 
 
 def popen3(cmd, env=None):
     stdin, stdout, stderr, p = popen4(cmd, env)
     return stdin, stdout, stderr
 
 
 def popen4(cmd, env=None, bufsize=-1):
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=bufsize,
         close_fds=closefds,
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         env=tonativeenv(env),
     )
     return p.stdin, p.stdout, p.stderr, p
 
 
 def pipefilter(s, cmd):
     '''filter string S through command CMD, returning its output'''
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         close_fds=closefds,
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
     )
     pout, perr = p.communicate(s)
     return pout
 
 
 def tempfilter(s, cmd):
     '''filter string S through a pair of temporary files with CMD.
     CMD is used as a template to create the real command to be run,
     with the strings INFILE and OUTFILE replaced by the real names of
     the temporary files generated.'''
     inname, outname = None, None
     try:
         infd, inname = pycompat.mkstemp(prefix=b'hg-filter-in-')
         fp = os.fdopen(infd, 'wb')
         fp.write(s)
         fp.close()
         outfd, outname = pycompat.mkstemp(prefix=b'hg-filter-out-')
         os.close(outfd)
         cmd = cmd.replace(b'INFILE', inname)
         cmd = cmd.replace(b'OUTFILE', outname)
         code = system(cmd)
         if pycompat.sysplatform == b'OpenVMS' and code & 1:
             code = 0
         if code:
             raise error.Abort(
                 _(b"command '%s' failed: %s") % (cmd, explainexit(code))
             )
         with open(outname, b'rb') as fp:
             return fp.read()
     finally:
         try:
             if inname:
                 os.unlink(inname)
         except OSError:
             pass
         try:
             if outname:
                 os.unlink(outname)
         except OSError:
             pass
 
 
 _filtertable = {
     b'tempfile:': tempfilter,
     b'pipe:': pipefilter,
 }
 
 
 def filter(s, cmd):
     """filter a string through a command that transforms its input to its
     output"""
     for name, fn in pycompat.iteritems(_filtertable):
         if cmd.startswith(name):
             return fn(s, cmd[len(name) :].lstrip())
     return pipefilter(s, cmd)
 
 
 _hgexecutable = None
 
 
 def hgexecutable():
     """return location of the 'hg' executable.
 
     Defaults to $HG or 'hg' in the search path.
     """
     if _hgexecutable is None:
         hg = encoding.environ.get(b'HG')
         mainmod = sys.modules['__main__']
         if hg:
             _sethgexecutable(hg)
         elif resourceutil.mainfrozen():
             if getattr(sys, 'frozen', None) == 'macosx_app':
                 # Env variable set by py2app
                 _sethgexecutable(encoding.environ[b'EXECUTABLEPATH'])
             else:
                 _sethgexecutable(pycompat.sysexecutable)
         elif (
             not pycompat.iswindows
             and os.path.basename(getattr(mainmod, '__file__', '')) == 'hg'
         ):
             _sethgexecutable(pycompat.fsencode(mainmod.__file__))
         else:
             _sethgexecutable(
                 findexe(b'hg') or os.path.basename(pycompat.sysargv[0])
             )
     return _hgexecutable
 
 
 def _sethgexecutable(path):
     """set location of the 'hg' executable"""
     global _hgexecutable
     _hgexecutable = path
 
 
 def _testfileno(f, stdf):
     fileno = getattr(f, 'fileno', None)
     try:
         return fileno and fileno() == stdf.fileno()
     except io.UnsupportedOperation:
         return False  # fileno() raised UnsupportedOperation
 
 
 def isstdin(f):
     return _testfileno(f, sys.__stdin__)
 
 
 def isstdout(f):
     return _testfileno(f, sys.__stdout__)
 
 
 def protectstdio(uin, uout):
     """Duplicate streams and redirect original if (uin, uout) are stdio
 
     If uin is stdin, it's redirected to /dev/null. If uout is stdout, it's
     redirected to stderr so the output is still readable.
 
     Returns (fin, fout) which point to the original (uin, uout) fds, but
     may be copy of (uin, uout). The returned streams can be considered
     "owned" in that print(), exec(), etc. never reach to them.
     """
     uout.flush()
     fin, fout = uin, uout
     if _testfileno(uin, stdin):
         newfd = os.dup(uin.fileno())
         nullfd = os.open(os.devnull, os.O_RDONLY)
         os.dup2(nullfd, uin.fileno())
         os.close(nullfd)
         fin = os.fdopen(newfd, 'rb')
     if _testfileno(uout, stdout):
         newfd = os.dup(uout.fileno())
         os.dup2(stderr.fileno(), uout.fileno())
         fout = os.fdopen(newfd, 'wb')
     return fin, fout
 
 
 def restorestdio(uin, uout, fin, fout):
     """Restore (uin, uout) streams from possibly duplicated (fin, fout)"""
     uout.flush()
     for f, uif in [(fin, uin), (fout, uout)]:
         if f is not uif:
             os.dup2(f.fileno(), uif.fileno())
             f.close()
 
 
 def shellenviron(environ=None):
     """return environ with optional override, useful for shelling out"""
 
     def py2shell(val):
         """convert python object into string that is useful to shell"""
         if val is None or val is False:
             return b'0'
         if val is True:
             return b'1'
         return pycompat.bytestr(val)
 
     env = dict(encoding.environ)
     if environ:
         env.update((k, py2shell(v)) for k, v in pycompat.iteritems(environ))
     env[b'HG'] = hgexecutable()
     return env
 
 
 if pycompat.iswindows:
 
     def shelltonative(cmd, env):
         return platform.shelltocmdexe(  # pytype: disable=module-attr
             cmd, shellenviron(env)
         )
 
     tonativestr = encoding.strfromlocal
 else:
 
     def shelltonative(cmd, env):
         return cmd
 
     tonativestr = pycompat.identity
 
 
 def tonativeenv(env):
     '''convert the environment from bytes to strings suitable for Popen(), etc.
     '''
     return pycompat.rapply(tonativestr, env)
 
 
 def system(cmd, environ=None, cwd=None, out=None):
     '''enhanced shell command execution.
     run with environment maybe modified, maybe in different dir.
 
     if out is specified, it is assumed to be a file-like object that has a
     write() method. stdout and stderr will be redirected to out.'''
     try:
         stdout.flush()
     except Exception:
         pass
     env = shellenviron(environ)
     if out is None or isstdout(out):
         rc = subprocess.call(
             tonativestr(cmd),
             shell=True,
             close_fds=closefds,
             env=tonativeenv(env),
             cwd=pycompat.rapply(tonativestr, cwd),
         )
     else:
         proc = subprocess.Popen(
             tonativestr(cmd),
             shell=True,
             close_fds=closefds,
             env=tonativeenv(env),
             cwd=pycompat.rapply(tonativestr, cwd),
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
         )
         for line in iter(proc.stdout.readline, b''):
             out.write(line)
         proc.wait()
         rc = proc.returncode
     if pycompat.sysplatform == b'OpenVMS' and rc & 1:
         rc = 0
     return rc
 
 
 _is_gui = None
 
 
 def _gui():
     '''Are we running in a GUI?'''
     if pycompat.isdarwin:
         if b'SSH_CONNECTION' in encoding.environ:
             # handle SSH access to a box where the user is logged in
             return False
         elif getattr(osutil, 'isgui', None):
             # check if a CoreGraphics session is available
             return osutil.isgui()
         else:
             # pure build; use a safe default
             return True
     else:
         return pycompat.iswindows or encoding.environ.get(b"DISPLAY")
 
 
 def gui():
     global _is_gui
     if _is_gui is None:
         _is_gui = _gui()
     return _is_gui
 
 
 def hgcmd():
     """Return the command used to execute current hg
 
     This is different from hgexecutable() because on Windows we want
     to avoid things opening new shell windows like batch files, so we
     get either the python call or current executable.
     """
     if resourceutil.mainfrozen():
         if getattr(sys, 'frozen', None) == 'macosx_app':
             # Env variable set by py2app
             return [encoding.environ[b'EXECUTABLEPATH']]
         else:
             return [pycompat.sysexecutable]
     return _gethgcmd()
 
 
 def rundetached(args, condfn):
     """Execute the argument list in a detached process.
 
     condfn is a callable which is called repeatedly and should return
     True once the child process is known to have started successfully.
     At this point, the child process PID is returned. If the child
     process fails to start or finishes before condfn() evaluates to
     True, return -1.
     """
     # Windows case is easier because the child process is either
     # successfully starting and validating the condition or exiting
     # on failure. We just poll on its PID. On Unix, if the child
     # process fails to start, it will be left in a zombie state until
     # the parent wait on it, which we cannot do since we expect a long
     # running process on success. Instead we listen for SIGCHLD telling
     # us our child process terminated.
     terminated = set()
 
     def handler(signum, frame):
         terminated.add(os.wait())
 
     prevhandler = None
     SIGCHLD = getattr(signal, 'SIGCHLD', None)
     if SIGCHLD is not None:
         prevhandler = signal.signal(SIGCHLD, handler)
     try:
         pid = spawndetached(args)
         while not condfn():
             if (pid in terminated or not testpid(pid)) and not condfn():
                 return -1
             time.sleep(0.1)
         return pid
     finally:
         if prevhandler is not None:
             signal.signal(signal.SIGCHLD, prevhandler)
 
 
 @contextlib.contextmanager
 def uninterruptible(warn):
     """Inhibit SIGINT handling on a region of code.
 
     Note that if this is called in a non-main thread, it turns into a no-op.
 
     Args:
       warn: A callable which takes no arguments, and returns True if the
             previous signal handling should be restored.
     """
 
     oldsiginthandler = [signal.getsignal(signal.SIGINT)]
     shouldbail = []
 
     def disabledsiginthandler(*args):
         if warn():
             signal.signal(signal.SIGINT, oldsiginthandler[0])
             del oldsiginthandler[0]
         shouldbail.append(True)
 
     try:
         try:
             signal.signal(signal.SIGINT, disabledsiginthandler)
         except ValueError:
             # wrong thread, oh well, we tried
             del oldsiginthandler[0]
         yield
     finally:
         if oldsiginthandler:
             signal.signal(signal.SIGINT, oldsiginthandler[0])
         if shouldbail:
             raise KeyboardInterrupt
 
 
 if pycompat.iswindows:
     # no fork on Windows, but we can create a detached process
     # https://msdn.microsoft.com/en-us/library/windows/desktop/ms684863.aspx
     # No stdlib constant exists for this value
     DETACHED_PROCESS = 0x00000008
     # Following creation flags might create a console GUI window.
     # Using subprocess.CREATE_NEW_CONSOLE might helps.
     # See https://phab.mercurial-scm.org/D1701 for discussion
     _creationflags = (
         DETACHED_PROCESS
         | subprocess.CREATE_NEW_PROCESS_GROUP  # pytype: disable=module-attr
     )
 
     def runbgcommand(
         script,
         env,
         shell=False,
         stdout=None,
         stderr=None,
         ensurestart=True,
         record_wait=None,
         stdin_bytes=None,
     ):
         '''Spawn a command without waiting for it to finish.'''
         # we can't use close_fds *and* redirect stdin. I'm not sure that we
         # need to because the detached process has no console connection.
 
         try:
             stdin = None
             if stdin_bytes is not None:
                 stdin = pycompat.unnamedtempfile()
                 stdin.write(stdin_bytes)
                 stdin.flush()
                 stdin.seek(0)
 
             p = subprocess.Popen(
                 tonativestr(script),
                 shell=shell,
                 env=tonativeenv(env),
                 close_fds=True,
                 creationflags=_creationflags,
                 stdin=stdin,
                 stdout=stdout,
                 stderr=stderr,
             )
             if record_wait is not None:
                 record_wait(p.wait)
         finally:
             if stdin is not None:
                 stdin.close()
 
 
 else:
 
     def runbgcommand(
         cmd,
         env,
         shell=False,
         stdout=None,
         stderr=None,
         ensurestart=True,
         record_wait=None,
         stdin_bytes=None,
     ):
         '''Spawn a command without waiting for it to finish.
 
 
         When `record_wait` is not None, the spawned process will not be fully
         detached and the `record_wait` argument will be called with a the
         `Subprocess.wait` function for the spawned process.  This is mostly
         useful for developers that need to make sure the spawned process
         finished before a certain point. (eg: writing test)'''
         if pycompat.isdarwin:
             # avoid crash in CoreFoundation in case another thread
             # calls gui() while we're calling fork().
             gui()
 
         # double-fork to completely detach from the parent process
         # based on http://code.activestate.com/recipes/278731
         if record_wait is None:
             pid = os.fork()
             if pid:
                 if not ensurestart:
                     # Even though we're not waiting on the child process,
                     # we still must call waitpid() on it at some point so
                     # it's not a zombie/defunct. This is especially relevant for
                     # chg since the parent process won't die anytime soon.
                     # We use a thread to make the overhead tiny.
                     def _do_wait():
                         os.waitpid(pid, 0)
 
                     t = threading.Thread(target=_do_wait)
                     t.daemon = True
                     t.start()
                     return
                 # Parent process
                 (_pid, status) = os.waitpid(pid, 0)
                 if os.WIFEXITED(status):
                     returncode = os.WEXITSTATUS(status)
                 else:
                     returncode = -(os.WTERMSIG(status))
                 if returncode != 0:
                     # The child process's return code is 0 on success, an errno
                     # value on failure, or 255 if we don't have a valid errno
                     # value.
                     #
                     # (It would be slightly nicer to return the full exception info
                     # over a pipe as the subprocess module does.  For now it
                     # doesn't seem worth adding that complexity here, though.)
                     if returncode == 255:
                         returncode = errno.EINVAL
                     raise OSError(
                         returncode,
                         b'error running %r: %s'
                         % (cmd, os.strerror(returncode)),
                     )
                 return
 
         returncode = 255
         try:
             if record_wait is None:
                 # Start a new session
                 os.setsid()
             # connect stdin to devnull to make sure the subprocess can't
             # muck up that stream for mercurial.
             if stdin_bytes is None:
                 stdin = open(os.devnull, b'r')
             else:
                 stdin = pycompat.unnamedtempfile()
                 stdin.write(stdin_bytes)
                 stdin.flush()
                 stdin.seek(0)
 
             if stdout is None:
                 stdout = open(os.devnull, b'w')
             if stderr is None:
                 stderr = open(os.devnull, b'w')
 
             p = subprocess.Popen(
                 cmd,
                 shell=shell,
                 env=env,
                 close_fds=True,
                 stdin=stdin,
                 stdout=stdout,
                 stderr=stderr,
             )
             if record_wait is not None:
                 record_wait(p.wait)
             returncode = 0
         except EnvironmentError as ex:
             returncode = ex.errno & 0xFF
             if returncode == 0:
                 # This shouldn't happen, but just in case make sure the
                 # return code is never 0 here.
                 returncode = 255
         except Exception:
             returncode = 255
         finally:
             # mission accomplished, this child needs to exit and not
             # continue the hg process here.
             stdin.close()
             if record_wait is None:
                 os._exit(returncode)