procutil: move protectio/restoreio from commandserver

Some variants of this will be useful for stdio-based servers such as
sshserver.

mercurial/commandserver.py

@@ -1,565 +1,536 b''
 # commandserver.py - communicate with Mercurial's API over a pipe
 #
 #  Copyright Matt Mackall <[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 errno
 import gc
 import os
 import random
 import signal
 import socket
 import struct
 import traceback
 
 try:
     import selectors
     selectors.BaseSelector
 except ImportError:
     from .thirdparty import selectors2 as selectors
 
 from .i18n import _
 from . import (
     encoding,
     error,
     pycompat,
     util,
 )
 from .utils import (
     procutil,
 )
 
 logfile = None
 
 def log(*args):
     if not logfile:
         return
 
     for a in args:
         logfile.write(str(a))
 
     logfile.flush()
 
 class channeledoutput(object):
     """
     Write data to out in the following format:
 
     data length (unsigned int),
     data
     """
     def __init__(self, out, channel):
         self.out = out
         self.channel = channel
 
     @property
     def name(self):
         return '<%c-channel>' % self.channel
 
     def write(self, data):
         if not data:
             return
         # single write() to guarantee the same atomicity as the underlying file
         self.out.write(struct.pack('>cI', self.channel, len(data)) + data)
         self.out.flush()
 
     def __getattr__(self, attr):
         if attr in ('isatty', 'fileno', 'tell', 'seek'):
             raise AttributeError(attr)
         return getattr(self.out, attr)
 
 class channeledinput(object):
     """
     Read data from in_.
 
     Requests for input are written to out in the following format:
     channel identifier - 'I' for plain input, 'L' line based (1 byte)
     how many bytes to send at most (unsigned int),
 
     The client replies with:
     data length (unsigned int), 0 meaning EOF
     data
     """
 
     maxchunksize = 4 * 1024
 
     def __init__(self, in_, out, channel):
         self.in_ = in_
         self.out = out
         self.channel = channel
 
     @property
     def name(self):
         return '<%c-channel>' % self.channel
 
     def read(self, size=-1):
         if size < 0:
             # if we need to consume all the clients input, ask for 4k chunks
             # so the pipe doesn't fill up risking a deadlock
             size = self.maxchunksize
             s = self._read(size, self.channel)
             buf = s
             while s:
                 s = self._read(size, self.channel)
                 buf += s
 
             return buf
         else:
             return self._read(size, self.channel)
 
     def _read(self, size, channel):
         if not size:
             return ''
         assert size > 0
 
         # tell the client we need at most size bytes
         self.out.write(struct.pack('>cI', channel, size))
         self.out.flush()
 
         length = self.in_.read(4)
         length = struct.unpack('>I', length)[0]
         if not length:
             return ''
         else:
             return self.in_.read(length)
 
     def readline(self, size=-1):
         if size < 0:
             size = self.maxchunksize
             s = self._read(size, 'L')
             buf = s
             # keep asking for more until there's either no more or
             # we got a full line
             while s and s[-1] != '\n':
                 s = self._read(size, 'L')
                 buf += s
 
             return buf
         else:
             return self._read(size, 'L')
 
     def __iter__(self):
         return self
 
     def next(self):
         l = self.readline()
         if not l:
             raise StopIteration
         return l
 
     def __getattr__(self, attr):
         if attr in ('isatty', 'fileno', 'tell', 'seek'):
             raise AttributeError(attr)
         return getattr(self.in_, attr)
 
 class server(object):
     """
     Listens for commands on fin, runs them and writes the output on a channel
     based stream to fout.
     """
     def __init__(self, ui, repo, fin, fout):
         self.cwd = pycompat.getcwd()
 
         # developer config: cmdserver.log
         logpath = ui.config("cmdserver", "log")
         if logpath:
             global logfile
             if logpath == '-':
                 # write log on a special 'd' (debug) channel
                 logfile = channeledoutput(fout, 'd')
             else:
                 logfile = open(logpath, 'a')
 
         if repo:
             # the ui here is really the repo ui so take its baseui so we don't
             # end up with its local configuration
             self.ui = repo.baseui
             self.repo = repo
             self.repoui = repo.ui
         else:
             self.ui = ui
             self.repo = self.repoui = None
 
         self.cerr = channeledoutput(fout, 'e')
         self.cout = channeledoutput(fout, 'o')
         self.cin = channeledinput(fin, fout, 'I')
         self.cresult = channeledoutput(fout, 'r')
 
         self.client = fin
 
     def cleanup(self):
         """release and restore resources taken during server session"""
 
     def _read(self, size):
         if not size:
             return ''
 
         data = self.client.read(size)
 
         # is the other end closed?
         if not data:
             raise EOFError
 
         return data
 
     def _readstr(self):
         """read a string from the channel
 
         format:
         data length (uint32), data
         """
         length = struct.unpack('>I', self._read(4))[0]
         if not length:
             return ''
         return self._read(length)
 
     def _readlist(self):
         """read a list of NULL separated strings from the channel"""
         s = self._readstr()
         if s:
             return s.split('\0')
         else:
             return []
 
     def runcommand(self):
         """ reads a list of \0 terminated arguments, executes
         and writes the return code to the result channel """
         from . import dispatch  # avoid cycle
 
         args = self._readlist()
 
         # copy the uis so changes (e.g. --config or --verbose) don't
         # persist between requests
         copiedui = self.ui.copy()
         uis = [copiedui]
         if self.repo:
             self.repo.baseui = copiedui
             # clone ui without using ui.copy because this is protected
             repoui = self.repoui.__class__(self.repoui)
             repoui.copy = copiedui.copy # redo copy protection
             uis.append(repoui)
             self.repo.ui = self.repo.dirstate._ui = repoui
             self.repo.invalidateall()
 
         for ui in uis:
             ui.resetstate()
             # any kind of interaction must use server channels, but chg may
             # replace channels by fully functional tty files. so nontty is
             # enforced only if cin is a channel.
             if not util.safehasattr(self.cin, 'fileno'):
                 ui.setconfig('ui', 'nontty', 'true', 'commandserver')
 
         req = dispatch.request(args[:], copiedui, self.repo, self.cin,
                                self.cout, self.cerr)
 
         try:
             ret = (dispatch.dispatch(req) or 0) & 255 # might return None
             self.cresult.write(struct.pack('>i', int(ret)))
         finally:
             # restore old cwd
             if '--cwd' in args:
                 os.chdir(self.cwd)
 
     def getencoding(self):
         """ writes the current encoding to the result channel """
         self.cresult.write(encoding.encoding)
 
     def serveone(self):
         cmd = self.client.readline()[:-1]
         if cmd:
             handler = self.capabilities.get(cmd)
             if handler:
                 handler(self)
             else:
                 # clients are expected to check what commands are supported by
                 # looking at the servers capabilities
                 raise error.Abort(_('unknown command %s') % cmd)
 
         return cmd != ''
 
     capabilities = {'runcommand': runcommand,
                     'getencoding': getencoding}
 
     def serve(self):
         hellomsg = 'capabilities: ' + ' '.join(sorted(self.capabilities))
         hellomsg += '\n'
         hellomsg += 'encoding: ' + encoding.encoding
         hellomsg += '\n'
         hellomsg += 'pid: %d' % procutil.getpid()
         if util.safehasattr(os, 'getpgid'):
             hellomsg += '\n'
             hellomsg += 'pgid: %d' % os.getpgid(0)
 
         # write the hello msg in -one- chunk
         self.cout.write(hellomsg)
 
         try:
             while self.serveone():
                 pass
         except EOFError:
             # we'll get here if the client disconnected while we were reading
             # its request
             return 1
 
         return 0
 
-def _protectio(uin, uout):
-    """Duplicate streams and redirect original to null if (uin, uout) are
-    stdio
-
-    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()
-    newfiles = []
-    nullfd = os.open(os.devnull, os.O_RDWR)
-    for f, sysf, mode in [(uin, procutil.stdin, r'rb'),
-                          (uout, procutil.stdout, r'wb')]:
-        if f is sysf:
-            newfd = os.dup(f.fileno())
-            os.dup2(nullfd, f.fileno())
-            f = os.fdopen(newfd, mode)
-        newfiles.append(f)
-    os.close(nullfd)
-    return tuple(newfiles)
-
-def _restoreio(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()
-
 class pipeservice(object):
     def __init__(self, ui, repo, opts):
         self.ui = ui
         self.repo = repo
 
     def init(self):
         pass
 
     def run(self):
         ui = self.ui
         # redirect stdio to null device so that broken extensions or in-process
         # hooks will never cause corruption of channel protocol.
-        fin, fout = _protectio(ui.fin, ui.fout)
+        fin, fout = procutil.protectstdio(ui.fin, ui.fout)
         try:
             sv = server(ui, self.repo, fin, fout)
             return sv.serve()
         finally:
             sv.cleanup()
-            _restoreio(ui.fin, ui.fout, fin, fout)
+            procutil.restorestdio(ui.fin, ui.fout, fin, fout)
 
 def _initworkerprocess():
     # use a different process group from the master process, in order to:
     # 1. make the current process group no longer "orphaned" (because the
     #    parent of this process is in a different process group while
     #    remains in a same session)
     #    according to POSIX 2.2.2.52, orphaned process group will ignore
     #    terminal-generated stop signals like SIGTSTP (Ctrl+Z), which will
     #    cause trouble for things like ncurses.
     # 2. the client can use kill(-pgid, sig) to simulate terminal-generated
     #    SIGINT (Ctrl+C) and process-exit-generated SIGHUP. our child
     #    processes like ssh will be killed properly, without affecting
     #    unrelated processes.
     os.setpgid(0, 0)
     # change random state otherwise forked request handlers would have a
     # same state inherited from parent.
     random.seed()
 
 def _serverequest(ui, repo, conn, createcmdserver):
     fin = conn.makefile('rb')
     fout = conn.makefile('wb')
     sv = None
     try:
         sv = createcmdserver(repo, conn, fin, fout)
         try:
             sv.serve()
         # handle exceptions that may be raised by command server. most of
         # known exceptions are caught by dispatch.
         except error.Abort as inst:
             ui.warn(_('abort: %s\n') % inst)
         except IOError as inst:
             if inst.errno != errno.EPIPE:
                 raise
         except KeyboardInterrupt:
             pass
         finally:
             sv.cleanup()
     except: # re-raises
         # also write traceback to error channel. otherwise client cannot
         # see it because it is written to server's stderr by default.
         if sv:
             cerr = sv.cerr
         else:
             cerr = channeledoutput(fout, 'e')
         traceback.print_exc(file=cerr)
         raise
     finally:
         fin.close()
         try:
             fout.close()  # implicit flush() may cause another EPIPE
         except IOError as inst:
             if inst.errno != errno.EPIPE:
                 raise
 
 class unixservicehandler(object):
     """Set of pluggable operations for unix-mode services
 
     Almost all methods except for createcmdserver() are called in the main
     process. You can't pass mutable resource back from createcmdserver().
     """
 
     pollinterval = None
 
     def __init__(self, ui):
         self.ui = ui
 
     def bindsocket(self, sock, address):
         util.bindunixsocket(sock, address)
         sock.listen(socket.SOMAXCONN)
         self.ui.status(_('listening at %s\n') % address)
         self.ui.flush()  # avoid buffering of status message
 
     def unlinksocket(self, address):
         os.unlink(address)
 
     def shouldexit(self):
         """True if server should shut down; checked per pollinterval"""
         return False
 
     def newconnection(self):
         """Called when main process notices new connection"""
 
     def createcmdserver(self, repo, conn, fin, fout):
         """Create new command server instance; called in the process that
         serves for the current connection"""
         return server(self.ui, repo, fin, fout)
 
 class unixforkingservice(object):
     """
     Listens on unix domain socket and forks server per connection
     """
 
     def __init__(self, ui, repo, opts, handler=None):
         self.ui = ui
         self.repo = repo
         self.address = opts['address']
         if not util.safehasattr(socket, 'AF_UNIX'):
             raise error.Abort(_('unsupported platform'))
         if not self.address:
             raise error.Abort(_('no socket path specified with --address'))
         self._servicehandler = handler or unixservicehandler(ui)
         self._sock = None
         self._oldsigchldhandler = None
         self._workerpids = set()  # updated by signal handler; do not iterate
         self._socketunlinked = None
 
     def init(self):
         self._sock = socket.socket(socket.AF_UNIX)
         self._servicehandler.bindsocket(self._sock, self.address)
         if util.safehasattr(procutil, 'unblocksignal'):
             procutil.unblocksignal(signal.SIGCHLD)
         o = signal.signal(signal.SIGCHLD, self._sigchldhandler)
         self._oldsigchldhandler = o
         self._socketunlinked = False
 
     def _unlinksocket(self):
         if not self._socketunlinked:
             self._servicehandler.unlinksocket(self.address)
             self._socketunlinked = True
 
     def _cleanup(self):
         signal.signal(signal.SIGCHLD, self._oldsigchldhandler)
         self._sock.close()
         self._unlinksocket()
         # don't kill child processes as they have active clients, just wait
         self._reapworkers(0)
 
     def run(self):
         try:
             self._mainloop()
         finally:
             self._cleanup()
 
     def _mainloop(self):
         exiting = False
         h = self._servicehandler
         selector = selectors.DefaultSelector()
         selector.register(self._sock, selectors.EVENT_READ)
         while True:
             if not exiting and h.shouldexit():
                 # clients can no longer connect() to the domain socket, so
                 # we stop queuing new requests.
                 # for requests that are queued (connect()-ed, but haven't been
                 # accept()-ed), handle them before exit. otherwise, clients
                 # waiting for recv() will receive ECONNRESET.
                 self._unlinksocket()
                 exiting = True
             ready = selector.select(timeout=h.pollinterval)
             if not ready:
                 # only exit if we completed all queued requests
                 if exiting:
                     break
                 continue
             try:
                 conn, _addr = self._sock.accept()
             except socket.error as inst:
                 if inst.args[0] == errno.EINTR:
                     continue
                 raise
 
             pid = os.fork()
             if pid:
                 try:
                     self.ui.debug('forked worker process (pid=%d)\n' % pid)
                     self._workerpids.add(pid)
                     h.newconnection()
                 finally:
                     conn.close()  # release handle in parent process
             else:
                 try:
                     self._runworker(conn)
                     conn.close()
                     os._exit(0)
                 except:  # never return, hence no re-raises
                     try:
                         self.ui.traceback(force=True)
                     finally:
                         os._exit(255)
         selector.close()
 
     def _sigchldhandler(self, signal, frame):
         self._reapworkers(os.WNOHANG)
 
     def _reapworkers(self, options):
         while self._workerpids:
             try:
                 pid, _status = os.waitpid(-1, options)
             except OSError as inst:
                 if inst.errno == errno.EINTR:
                     continue
                 if inst.errno != errno.ECHILD:
                     raise
                 # no child processes at all (reaped by other waitpid()?)
                 self._workerpids.clear()
                 return
             if pid == 0:
                 # no waitable child processes
                 return
             self.ui.debug('worker process exited (pid=%d)\n' % pid)
             self._workerpids.discard(pid)
 
     def _runworker(self, conn):
         signal.signal(signal.SIGCHLD, self._oldsigchldhandler)
         _initworkerprocess()
         h = self._servicehandler
         try:
             _serverequest(self.ui, self.repo, conn, h.createcmdserver)
         finally:
             gc.collect()  # trigger __del__ since worker process uses os._exit

mercurial/utils/procutil.py

@@ -1,319 +1,348 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 imp
 import io
 import os
 import signal
 import subprocess
 import sys
 import tempfile
 import time
 
 from ..i18n import _
 
 from .. import (
     encoding,
     error,
     policy,
     pycompat,
 )
 
 osutil = policy.importmod(r'osutil')
 
 stderr = pycompat.stderr
 stdin = pycompat.stdin
 stdout = pycompat.stdout
 
 def isatty(fp):
     try:
         return fp.isatty()
     except AttributeError:
         return False
 
 # 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
 if isatty(stdout):
     stdout = os.fdopen(stdout.fileno(), r'wb', 1)
 
 if pycompat.iswindows:
     from .. import windows as platform
     stdout = platform.winstdout(stdout)
 else:
     from .. import posix as platform
 
 explainexit = platform.explainexit
 findexe = platform.findexe
 _gethgcmd = platform.gethgcmd
 getuser = platform.getuser
 getpid = os.getpid
 hidewindow = platform.hidewindow
 popen = platform.popen
 quotecommand = platform.quotecommand
 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 popen2(cmd, env=None, newlines=False):
     # 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(cmd, shell=True, bufsize=-1,
                          close_fds=closefds,
                          stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                          universal_newlines=newlines,
                          env=env)
     return p.stdin, p.stdout
 
 def popen3(cmd, env=None, newlines=False):
     stdin, stdout, stderr, p = popen4(cmd, env, newlines)
     return stdin, stdout, stderr
 
 def popen4(cmd, env=None, newlines=False, bufsize=-1):
     p = subprocess.Popen(cmd, shell=True, bufsize=bufsize,
                          close_fds=closefds,
                          stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE,
                          universal_newlines=newlines,
                          env=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(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 = tempfile.mkstemp(prefix='hg-filter-in-')
         fp = os.fdopen(infd, r'wb')
         fp.write(s)
         fp.close()
         outfd, outname = tempfile.mkstemp(prefix='hg-filter-out-')
         os.close(outfd)
         cmd = cmd.replace('INFILE', inname)
         cmd = cmd.replace('OUTFILE', outname)
         code = os.system(cmd)
         if pycompat.sysplatform == 'OpenVMS' and code & 1:
             code = 0
         if code:
             raise error.Abort(_("command '%s' failed: %s") %
                               (cmd, explainexit(code)))
         with open(outname, '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 = {
     'tempfile:': tempfilter,
     'pipe:': pipefilter,
 }
 
 def filter(s, cmd):
     "filter a string through a command that transforms its input to its output"
     for name, fn in _filtertable.iteritems():
         if cmd.startswith(name):
             return fn(s, cmd[len(name):].lstrip())
     return pipefilter(s, cmd)
 
 def mainfrozen():
     """return True if we are a frozen executable.
 
     The code supports py2exe (most common, Windows only) and tools/freeze
     (portable, not much used).
     """
     return (pycompat.safehasattr(sys, "frozen") or # new py2exe
             pycompat.safehasattr(sys, "importers") or # old py2exe
             imp.is_frozen(u"__main__")) # tools/freeze
 
 _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('HG')
         mainmod = sys.modules[r'__main__']
         if hg:
             _sethgexecutable(hg)
         elif mainfrozen():
             if getattr(sys, 'frozen', None) == 'macosx_app':
                 # Env variable set by py2app
                 _sethgexecutable(encoding.environ['EXECUTABLEPATH'])
             else:
                 _sethgexecutable(pycompat.sysexecutable)
         elif (os.path.basename(
             pycompat.fsencode(getattr(mainmod, '__file__', ''))) == 'hg'):
             _sethgexecutable(pycompat.fsencode(mainmod.__file__))
         else:
             exe = findexe('hg') or os.path.basename(sys.argv[0])
             _sethgexecutable(exe)
     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 to null if (uin, uout) are
+    stdio
+
+    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()
+    newfiles = []
+    nullfd = os.open(os.devnull, os.O_RDWR)
+    for f, sysf, mode in [(uin, stdin, r'rb'),
+                          (uout, stdout, r'wb')]:
+        if f is sysf:
+            newfd = os.dup(f.fileno())
+            os.dup2(nullfd, f.fileno())
+            f = os.fdopen(newfd, mode)
+        newfiles.append(f)
+    os.close(nullfd)
+    return tuple(newfiles)
+
+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 '0'
         if val is True:
             return '1'
         return pycompat.bytestr(val)
     env = dict(encoding.environ)
     if environ:
         env.update((k, py2shell(v)) for k, v in environ.iteritems())
     env['HG'] = hgexecutable()
     return 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
     cmd = quotecommand(cmd)
     env = shellenviron(environ)
     if out is None or isstdout(out):
         rc = subprocess.call(cmd, shell=True, close_fds=closefds,
                              env=env, cwd=cwd)
     else:
         proc = subprocess.Popen(cmd, shell=True, close_fds=closefds,
                                 env=env, cwd=cwd, stdout=subprocess.PIPE,
                                 stderr=subprocess.STDOUT)
         for line in iter(proc.stdout.readline, ''):
             out.write(line)
         proc.wait()
         rc = proc.returncode
     if pycompat.sysplatform == 'OpenVMS' and rc & 1:
         rc = 0
     return rc
 
 def gui():
     '''Are we running in a GUI?'''
     if pycompat.isdarwin:
         if '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("DISPLAY")
 
 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 mainfrozen():
         if getattr(sys, 'frozen', None) == 'macosx_app':
             # Env variable set by py2app
             return [encoding.environ['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)