twisted/internet/process.py

   1 # -*- test-case-name: twisted.test.test_process -*-
   2 # Copyright (c) Twisted Matrix Laboratories.
   3 # See LICENSE for details.
   4
   5 """
   6 UNIX Process management.
   7
   8 Do NOT use this module directly - use reactor.spawnProcess() instead.
   9
  10 Maintainer: Itamar Shtull-Trauring
  11 """
  12
  13 # System Imports
  14 import gc, os, sys, stat, traceback, select, signal, errno
  15
  16 try:
  17     import pty
  18 except ImportError:
  19     pty = None
  20
  21 try:
  22     import fcntl, termios
  23 except ImportError:
  24     fcntl = None
  25
  26 from zope.interface import implements
  27
  28 from twisted.python import log, failure
  29 from twisted.python.util import switchUID
  30 from twisted.internet import fdesc, abstract, error
  31 from twisted.internet.main import CONNECTION_LOST, CONNECTION_DONE
  32 from twisted.internet._baseprocess import BaseProcess
  33 from twisted.internet.interfaces import IProcessTransport
  34
  35 # Some people were importing this, which is incorrect, just keeping it
  36 # here for backwards compatibility:
  37 ProcessExitedAlready = error.ProcessExitedAlready
  38
  39 reapProcessHandlers = {}
  40
  41 def reapAllProcesses():
  42     """
  43     Reap all registered processes.
  44     """
  45     for process in reapProcessHandlers.values():
  46         process.reapProcess()
  47
  48
  49 def registerReapProcessHandler(pid, process):
  50     """
  51     Register a process handler for the given pid, in case L{reapAllProcesses}
  52     is called.
  53
  54     @param pid: the pid of the process.
  55     @param process: a process handler.
  56     """
  57     if pid in reapProcessHandlers:
  58         raise RuntimeError("Try to register an already registered process.")
  59     try:
  60         auxPID, status = os.waitpid(pid, os.WNOHANG)
  61     except:
  62         log.msg('Failed to reap %d:' % pid)
  63         log.err()
  64         auxPID = None
  65     if auxPID:
  66         process.processEnded(status)
  67     else:
  68         # if auxPID is 0, there are children but none have exited
  69         reapProcessHandlers[pid] = process
  70
  71
  72 def unregisterReapProcessHandler(pid, process):
  73     """
  74     Unregister a process handler previously registered with
  75     L{registerReapProcessHandler}.
  76     """
  77     if not (pid in reapProcessHandlers
  78             and reapProcessHandlers[pid] == process):
  79         raise RuntimeError("Try to unregister a process not registered.")
  80     del reapProcessHandlers[pid]
  81
  82
  83 def detectLinuxBrokenPipeBehavior():
  84     """
  85     On some Linux version, write-only pipe are detected as readable. This
  86     function is here to check if this bug is present or not.
  87
  88     See L{ProcessWriter.doRead} for a more detailed explanation.
  89     """
  90     global brokenLinuxPipeBehavior
  91     r, w = os.pipe()
  92     os.write(w, 'a')
  93     reads, writes, exes = select.select([w], [], [], 0)
  94     if reads:
  95         # Linux < 2.6.11 says a write-only pipe is readable.
  96         brokenLinuxPipeBehavior = True
  97     else:
  98         brokenLinuxPipeBehavior = False
  99     os.close(r)
 100     os.close(w)
 101
 102 # Call at import time
 103 detectLinuxBrokenPipeBehavior()
 104
 105
 106 class ProcessWriter(abstract.FileDescriptor):
 107     """
 108     (Internal) Helper class to write into a Process's input pipe.
 109
 110     I am a helper which describes a selectable asynchronous writer to a
 111     process's input pipe, including stdin.
 112
 113     @ivar enableReadHack: A flag which determines how readability on this
 114         write descriptor will be handled.  If C{True}, then readability may
 115         indicate the reader for this write descriptor has been closed (ie,
 116         the connection has been lost).  If C{False}, then readability events
 117         are ignored.
 118     """
 119     connected = 1
 120     ic = 0
 121     enableReadHack = False
 122
 123     def __init__(self, reactor, proc, name, fileno, forceReadHack=False):
 124         """
 125         Initialize, specifying a Process instance to connect to.
 126         """
 127         abstract.FileDescriptor.__init__(self, reactor)
 128         fdesc.setNonBlocking(fileno)
 129         self.proc = proc
 130         self.name = name
 131         self.fd = fileno
 132
 133         if not stat.S_ISFIFO(os.fstat(self.fileno()).st_mode):
 134             # If the fd is not a pipe, then the read hack is never
 135             # applicable.  This case arises when ProcessWriter is used by
 136             # StandardIO and stdout is redirected to a normal file.
 137             self.enableReadHack = False
 138         elif forceReadHack:
 139             self.enableReadHack = True
 140         else:
 141             # Detect if this fd is actually a write-only fd. If it's
 142             # valid to read, don't try to detect closing via read.
 143             # This really only means that we cannot detect a TTY's write
 144             # pipe being closed.
 145             try:
 146                 os.read(self.fileno(), 0)
 147             except OSError:
 148                 # It's a write-only pipe end, enable hack
 149                 self.enableReadHack = True
 150
 151         if self.enableReadHack:
 152             self.startReading()
 153
 154     def fileno(self):
 155         """
 156         Return the fileno() of my process's stdin.
 157         """
 158         return self.fd
 159
 160     def writeSomeData(self, data):
 161         """
 162         Write some data to the open process.
 163         """
 164         rv = fdesc.writeToFD(self.fd, data)
 165         if rv == len(data) and self.enableReadHack:
 166             # If the send buffer is now empty and it is necessary to monitor
 167             # this descriptor for readability to detect close, try detecting
 168             # readability now.
 169             self.startReading()
 170         return rv
 171
 172     def write(self, data):
 173         self.stopReading()
 174         abstract.FileDescriptor.write(self, data)
 175
 176     def doRead(self):
 177         """
 178         The only way a write pipe can become "readable" is at EOF, because the
 179         child has closed it, and we're using a reactor which doesn't
 180         distinguish between readable and closed (such as the select reactor).
 181
 182         Except that's not true on linux < 2.6.11. It has the following
 183         characteristics: write pipe is completely empty => POLLOUT (writable in
 184         select), write pipe is not completely empty => POLLIN (readable in
 185         select), write pipe's reader closed => POLLIN|POLLERR (readable and
 186         writable in select)
 187
 188         That's what this funky code is for. If linux was not broken, this
 189         function could be simply "return CONNECTION_LOST".
 190
 191         BUG: We call select no matter what the reactor.
 192         If the reactor is pollreactor, and the fd is > 1024, this will fail.
 193         (only occurs on broken versions of linux, though).
 194         """
 195         if self.enableReadHack:
 196             if brokenLinuxPipeBehavior:
 197                 fd = self.fd
 198                 r, w, x = select.select([fd], [fd], [], 0)
 199                 if r and w:
 200                     return CONNECTION_LOST
 201             else:
 202                 return CONNECTION_LOST
 203         else:
 204             self.stopReading()
 205
 206     def connectionLost(self, reason):
 207         """
 208         See abstract.FileDescriptor.connectionLost.
 209         """
 210         # At least on OS X 10.4, exiting while stdout is non-blocking can
 211         # result in data loss.  For some reason putting the file descriptor
 212         # back into blocking mode seems to resolve this issue.
 213         fdesc.setBlocking(self.fd)
 214
 215         abstract.FileDescriptor.connectionLost(self, reason)
 216         self.proc.childConnectionLost(self.name, reason)
 217
 218
 219
 220 class ProcessReader(abstract.FileDescriptor):
 221     """
 222     ProcessReader
 223
 224     I am a selectable representation of a process's output pipe, such as
 225     stdout and stderr.
 226     """
 227     connected = 1
 228
 229     def __init__(self, reactor, proc, name, fileno):
 230         """
 231         Initialize, specifying a process to connect to.
 232         """
 233         abstract.FileDescriptor.__init__(self, reactor)
 234         fdesc.setNonBlocking(fileno)
 235         self.proc = proc
 236         self.name = name
 237         self.fd = fileno
 238         self.startReading()
 239
 240     def fileno(self):
 241         """
 242         Return the fileno() of my process's stderr.
 243         """
 244         return self.fd
 245
 246     def writeSomeData(self, data):
 247         # the only time this is actually called is after .loseConnection Any
 248         # actual write attempt would fail, so we must avoid that. This hack
 249         # allows us to use .loseConnection on both readers and writers.
 250         assert data == ""
 251         return CONNECTION_LOST
 252
 253     def doRead(self):
 254         """
 255         This is called when the pipe becomes readable.
 256         """
 257         return fdesc.readFromFD(self.fd, self.dataReceived)
 258
 259     def dataReceived(self, data):
 260         self.proc.childDataReceived(self.name, data)
 261
 262     def loseConnection(self):
 263         if self.connected and not self.disconnecting:
 264             self.disconnecting = 1
 265             self.stopReading()
 266             self.reactor.callLater(0, self.connectionLost,
 267                                    failure.Failure(CONNECTION_DONE))
 268
 269     def connectionLost(self, reason):
 270         """
 271         Close my end of the pipe, signal the Process (which signals the
 272         ProcessProtocol).
 273         """
 274         abstract.FileDescriptor.connectionLost(self, reason)
 275         self.proc.childConnectionLost(self.name, reason)
 276
 277
 278 class _BaseProcess(BaseProcess, object):
 279     """
 280     Base class for Process and PTYProcess.
 281     """
 282     status = None
 283     pid = None
 284
 285     def reapProcess(self):
 286         """
 287         Try to reap a process (without blocking) via waitpid.
 288
 289         This is called when sigchild is caught or a Process object loses its
 290         "connection" (stdout is closed) This ought to result in reaping all
 291         zombie processes, since it will be called twice as often as it needs
 292         to be.
 293
 294         (Unfortunately, this is a slightly experimental approach, since
 295         UNIX has no way to be really sure that your process is going to
 296         go away w/o blocking.  I don't want to block.)
 297         """
 298         try:
 299             try:
 300                 pid, status = os.waitpid(self.pid, os.WNOHANG)
 301             except OSError, e:
 302                 if e.errno == errno.ECHILD:
 303                     # no child process
 304                     pid = None
 305                 else:
 306                     raise
 307         except:
 308             log.msg('Failed to reap %d:' % self.pid)
 309             log.err()
 310             pid = None
 311         if pid:
 312             self.processEnded(status)
 313             unregisterReapProcessHandler(pid, self)
 314
 315
 316     def _getReason(self, status):
 317         exitCode = sig = None
 318         if os.WIFEXITED(status):
 319             exitCode = os.WEXITSTATUS(status)
 320         else:
 321             sig = os.WTERMSIG(status)
 322         if exitCode or sig:
 323             return error.ProcessTerminated(exitCode, sig, status)
 324         return error.ProcessDone(status)
 325
 326
 327     def signalProcess(self, signalID):
 328         """
 329         Send the given signal C{signalID} to the process. It'll translate a
 330         few signals ('HUP', 'STOP', 'INT', 'KILL', 'TERM') from a string
 331         representation to its int value, otherwise it'll pass directly the
 332         value provided
 333
 334         @type signalID: C{str} or C{int}
 335         """
 336         if signalID in ('HUP', 'STOP', 'INT', 'KILL', 'TERM'):
 337             signalID = getattr(signal, 'SIG%s' % (signalID,))
 338         if self.pid is None:
 339             raise ProcessExitedAlready()
 340         os.kill(self.pid, signalID)
 341
 342
 343     def _resetSignalDisposition(self):
 344         # The Python interpreter ignores some signals, and our child
 345         # process will inherit that behaviour. To have a child process
 346         # that responds to signals normally, we need to reset our
 347         # child process's signal handling (just) after we fork and
 348         # before we execvpe.
 349         for signalnum in range(1, signal.NSIG):
 350             if signal.getsignal(signalnum) == signal.SIG_IGN:
 351                 # Reset signal handling to the default
 352                 signal.signal(signalnum, signal.SIG_DFL)
 353
 354
 355     def _fork(self, path, uid, gid, executable, args, environment, **kwargs):
 356         """
 357         Fork and then exec sub-process.
 358
 359         @param path: the path where to run the new process.
 360         @type path: C{str}
 361         @param uid: if defined, the uid used to run the new process.
 362         @type uid: C{int}
 363         @param gid: if defined, the gid used to run the new process.
 364         @type gid: C{int}
 365         @param executable: the executable to run in a new process.
 366         @type executable: C{str}
 367         @param args: arguments used to create the new process.
 368         @type args: C{list}.
 369         @param environment: environment used for the new process.
 370         @type environment: C{dict}.
 371         @param kwargs: keyword arguments to L{_setupChild} method.
 372         """
 373         settingUID = (uid is not None) or (gid is not None)
 374         if settingUID:
 375             curegid = os.getegid()
 376             currgid = os.getgid()
 377             cureuid = os.geteuid()
 378             curruid = os.getuid()
 379             if uid is None:
 380                 uid = cureuid
 381             if gid is None:
 382                 gid = curegid
 383             # prepare to change UID in subprocess
 384             os.setuid(0)
 385             os.setgid(0)
 386
 387         collectorEnabled = gc.isenabled()
 388         gc.disable()
 389         try:
 390             self.pid = os.fork()
 391         except:
 392             # Still in the parent process
 393             if settingUID:
 394                 os.setregid(currgid, curegid)
 395                 os.setreuid(curruid, cureuid)
 396             if collectorEnabled:
 397                 gc.enable()
 398             raise
 399         else:
 400             if self.pid == 0: # pid is 0 in the child process
 401                 # do not put *ANY* code outside the try block. The child process
 402                 # must either exec or _exit. If it gets outside this block (due
 403                 # to an exception that is not handled here, but which might be
 404                 # handled higher up), there will be two copies of the parent
 405                 # running in parallel, doing all kinds of damage.
 406
 407                 # After each change to this code, review it to make sure there
 408                 # are no exit paths.
 409                 try:
 410                     # Stop debugging. If I am, I don't care anymore.
 411                     sys.settrace(None)
 412                     self._setupChild(**kwargs)
 413                     self._execChild(path, settingUID, uid, gid,
 414                                     executable, args, environment)
 415                 except:
 416                     # If there are errors, bail and try to write something
 417                     # descriptive to stderr.
 418                     # XXX: The parent's stderr isn't necessarily fd 2 anymore, or
 419                     #      even still available
 420                     # XXXX: however even libc assumes write(2, err) is a useful
 421                     #       thing to attempt
 422                     try:
 423                         stderr = os.fdopen(2, 'w')
 424                         stderr.write("Upon execvpe %s %s in environment %s\n:" %
 425                                      (executable, str(args),
 426                                       "id %s" % id(environment)))
 427                         traceback.print_exc(file=stderr)
 428                         stderr.flush()
 429                         for fd in range(3):
 430                             os.close(fd)
 431                     except:
 432                         pass # make *sure* the child terminates
 433                 # Did you read the comment about not adding code here?
 434                 os._exit(1)
 435
 436         # we are now in parent process
 437         if settingUID:
 438             os.setregid(currgid, curegid)
 439             os.setreuid(curruid, cureuid)
 440         if collectorEnabled:
 441             gc.enable()
 442         self.status = -1 # this records the exit status of the child
 443
 444     def _setupChild(self, *args, **kwargs):
 445         """
 446         Setup the child process. Override in subclasses.
 447         """
 448         raise NotImplementedError()
 449
 450     def _execChild(self, path, settingUID, uid, gid,
 451                    executable, args, environment):
 452         """
 453         The exec() which is done in the forked child.
 454         """
 455         if path:
 456             os.chdir(path)
 457         # set the UID before I actually exec the process
 458         if settingUID:
 459             switchUID(uid, gid)
 460         os.execvpe(executable, args, environment)
 461
 462     def __repr__(self):
 463         """
 464         String representation of a process.
 465         """
 466         return "<%s pid=%s status=%s>" % (self.__class__.__name__,
 467                                           self.pid, self.status)
 468
 469
 470 class _FDDetector(object):
 471     """
 472     This class contains the logic necessary to decide which of the available
 473     system techniques should be used to detect the open file descriptors for
 474     the current process. The chosen technique gets monkey-patched into the
 475     _listOpenFDs method of this class so that the detection only needs to occur
 476     once.
 477
 478     @ivars listdir: The implementation of listdir to use. This gets overwritten
 479         by the test cases.
 480     @ivars getpid: The implementation of getpid to use, returns the PID of the
 481         running process.
 482     @ivars openfile: The implementation of open() to use, by default the Python
 483         builtin.
 484     """
 485     # So that we can unit test this
 486     listdir = os.listdir
 487     getpid = os.getpid
 488     openfile = open
 489
 490     def __init__(self):
 491         self._implementations = [
 492             self._procFDImplementation, self._devFDImplementation,
 493             self._fallbackFDImplementation]
 494
 495
 496     def _listOpenFDs(self):
 497         """
 498         Return an iterable of file descriptors which I{may} be open in this
 499         process.
 500
 501         This will try to return the fewest possible descriptors without missing
 502         any.
 503         """
 504         self._listOpenFDs = self._getImplementation()
 505         return self._listOpenFDs()
 506
 507
 508     def _getImplementation(self):
 509         """
 510         Pick a method which gives correct results for C{_listOpenFDs} in this
 511         runtime environment.
 512
 513         This involves a lot of very platform-specific checks, some of which may
 514         be relatively expensive.  Therefore the returned method should be saved
 515         and re-used, rather than always calling this method to determine what it
 516         is.
 517
 518         See the implementation for the details of how a method is selected.
 519         """
 520         for impl in self._implementations:
 521             try:
 522                 before = impl()
 523             except:
 524                 continue
 525             try:
 526                 fp = self.openfile("/dev/null", "r")
 527                 after = impl()
 528             finally:
 529                 fp.close()
 530             if before != after:
 531                 return impl
 532         # If no implementation can detect the newly opened file above, then just
 533         # return the last one.  The last one should therefore always be one
 534         # which makes a simple static guess which includes all possible open
 535         # file descriptors, but perhaps also many other values which do not
 536         # correspond to file descriptors.  For example, the scheme implemented
 537         # by _fallbackFDImplementation is suitable to be the last entry.
 538         return impl
 539
 540
 541     def _devFDImplementation(self):
 542         """
 543         Simple implementation for systems where /dev/fd actually works.
 544         See: http://www.freebsd.org/cgi/man.cgi?fdescfs
 545         """
 546         dname = "/dev/fd"
 547         result = [int(fd) for fd in self.listdir(dname)]
 548         return result
 549
 550
 551     def _procFDImplementation(self):
 552         """
 553         Simple implementation for systems where /proc/pid/fd exists (we assume
 554         it works).
 555         """
 556         dname = "/proc/%d/fd" % (self.getpid(),)
 557         return [int(fd) for fd in self.listdir(dname)]
 558
 559
 560     def _fallbackFDImplementation(self):
 561         """
 562         Fallback implementation where either the resource module can inform us
 563         about the upper bound of how many FDs to expect, or where we just guess
 564         a constant maximum if there is no resource module.
 565
 566         All possible file descriptors from 0 to that upper bound are returned
 567         with no attempt to exclude invalid file descriptor values.
 568         """
 569         try:
 570             import resource
 571         except ImportError:
 572             maxfds = 1024
 573         else:
 574             # OS-X reports 9223372036854775808. That's a lot of fds to close.
 575             # OS-X should get the /dev/fd implementation instead, so mostly
 576             # this check probably isn't necessary.
 577             maxfds = min(1024, resource.getrlimit(resource.RLIMIT_NOFILE)[1])
 578         return range(maxfds)
 579
 580
 581
 582 detector = _FDDetector()
 583
 584 def _listOpenFDs():
 585     """
 586     Use the global detector object to figure out which FD implementation to
 587     use.
 588     """
 589     return detector._listOpenFDs()
 590
 591
 592 class Process(_BaseProcess):
 593     """
 594     An operating-system Process.
 595
 596     This represents an operating-system process with arbitrary input/output
 597     pipes connected to it. Those pipes may represent standard input,
 598     standard output, and standard error, or any other file descriptor.
 599
 600     On UNIX, this is implemented using fork(), exec(), pipe()
 601     and fcntl(). These calls may not exist elsewhere so this
 602     code is not cross-platform. (also, windows can only select
 603     on sockets...)
 604     """
 605     implements(IProcessTransport)
 606
 607     debug = False
 608     debug_child = False
 609
 610     status = -1
 611     pid = None
 612
 613     processWriterFactory = ProcessWriter
 614     processReaderFactory = ProcessReader
 615
 616     def __init__(self,
 617                  reactor, executable, args, environment, path, proto,
 618                  uid=None, gid=None, childFDs=None):
 619         """
 620         Spawn an operating-system process.
 621
 622         This is where the hard work of disconnecting all currently open
 623         files / forking / executing the new process happens.  (This is
 624         executed automatically when a Process is instantiated.)
 625
 626         This will also run the subprocess as a given user ID and group ID, if
 627         specified.  (Implementation Note: this doesn't support all the arcane
 628         nuances of setXXuid on UNIX: it will assume that either your effective
 629         or real UID is 0.)
 630         """
 631         if not proto:
 632             assert 'r' not in childFDs.values()
 633             assert 'w' not in childFDs.values()
 634         _BaseProcess.__init__(self, proto)
 635
 636         self.pipes = {}
 637         # keys are childFDs, we can sense them closing
 638         # values are ProcessReader/ProcessWriters
 639
 640         helpers = {}
 641         # keys are childFDs
 642         # values are parentFDs
 643
 644         if childFDs is None:
 645             childFDs = {0: "w", # we write to the child's stdin
 646                         1: "r", # we read from their stdout
 647                         2: "r", # and we read from their stderr
 648                         }
 649
 650         debug = self.debug
 651         if debug: print "childFDs", childFDs
 652
 653         _openedPipes = []
 654         def pipe():
 655             r, w = os.pipe()
 656             _openedPipes.extend([r, w])
 657             return r, w
 658
 659         # fdmap.keys() are filenos of pipes that are used by the child.
 660         fdmap = {} # maps childFD to parentFD
 661         try:
 662             for childFD, target in childFDs.items():
 663                 if debug: print "[%d]" % childFD, target
 664                 if target == "r":
 665                     # we need a pipe that the parent can read from
 666                     readFD, writeFD = pipe()
 667                     if debug: print "readFD=%d, writeFD=%d" % (readFD, writeFD)
 668                     fdmap[childFD] = writeFD     # child writes to this
 669                     helpers[childFD] = readFD    # parent reads from this
 670                 elif target == "w":
 671                     # we need a pipe that the parent can write to
 672                     readFD, writeFD = pipe()
 673                     if debug: print "readFD=%d, writeFD=%d" % (readFD, writeFD)
 674                     fdmap[childFD] = readFD      # child reads from this
 675                     helpers[childFD] = writeFD   # parent writes to this
 676                 else:
 677                     assert type(target) == int, '%r should be an int' % (target,)
 678                     fdmap[childFD] = target      # parent ignores this
 679             if debug: print "fdmap", fdmap
 680             if debug: print "helpers", helpers
 681             # the child only cares about fdmap.values()
 682
 683             self._fork(path, uid, gid, executable, args, environment, fdmap=fdmap)
 684         except:
 685             map(os.close, _openedPipes)
 686             raise
 687
 688         # we are the parent process:
 689         self.proto = proto
 690
 691         # arrange for the parent-side pipes to be read and written
 692         for childFD, parentFD in helpers.items():
 693             os.close(fdmap[childFD])
 694
 695             if childFDs[childFD] == "r":
 696                 reader = self.processReaderFactory(reactor, self, childFD,
 697                                         parentFD)
 698                 self.pipes[childFD] = reader
 699
 700             if childFDs[childFD] == "w":
 701                 writer = self.processWriterFactory(reactor, self, childFD,
 702                                         parentFD, forceReadHack=True)
 703                 self.pipes[childFD] = writer
 704
 705         try:
 706             # the 'transport' is used for some compatibility methods
 707             if self.proto is not None:
 708                 self.proto.makeConnection(self)
 709         except:
 710             log.err()
 711
 712         # The reactor might not be running yet.  This might call back into
 713         # processEnded synchronously, triggering an application-visible
 714         # callback.  That's probably not ideal.  The replacement API for
 715         # spawnProcess should improve upon this situation.
 716         registerReapProcessHandler(self.pid, self)
 717
 718
 719     def _setupChild(self, fdmap):
 720         """
 721         fdmap[childFD] = parentFD
 722
 723         The child wants to end up with 'childFD' attached to what used to be
 724         the parent's parentFD. As an example, a bash command run like
 725         'command 2>&1' would correspond to an fdmap of {0:0, 1:1, 2:1}.
 726         'command >foo.txt' would be {0:0, 1:os.open('foo.txt'), 2:2}.
 727
 728         This is accomplished in two steps::
 729
 730             1. close all file descriptors that aren't values of fdmap.  This
 731                means 0 .. maxfds (or just the open fds within that range, if
 732                the platform supports '/proc/<pid>/fd').
 733
 734             2. for each childFD::
 735
 736                  - if fdmap[childFD] == childFD, the descriptor is already in
 737                    place.  Make sure the CLOEXEC flag is not set, then delete
 738                    the entry from fdmap.
 739
 740                  - if childFD is in fdmap.values(), then the target descriptor
 741                    is busy. Use os.dup() to move it elsewhere, update all
 742                    fdmap[childFD] items that point to it, then close the
 743                    original. Then fall through to the next case.
 744
 745                  - now fdmap[childFD] is not in fdmap.values(), and is free.
 746                    Use os.dup2() to move it to the right place, then close the
 747                    original.
 748         """
 749
 750         debug = self.debug_child
 751         if debug:
 752             errfd = sys.stderr
 753             errfd.write("starting _setupChild\n")
 754
 755         destList = fdmap.values()
 756         for fd in _listOpenFDs():
 757             if fd in destList:
 758                 continue
 759             if debug and fd == errfd.fileno():
 760                 continue
 761             try:
 762                 os.close(fd)
 763             except:
 764                 pass
 765
 766         # at this point, the only fds still open are the ones that need to
 767         # be moved to their appropriate positions in the child (the targets
 768         # of fdmap, i.e. fdmap.values() )
 769
 770         if debug: print >>errfd, "fdmap", fdmap
 771         childlist = fdmap.keys()
 772         childlist.sort()
 773
 774         for child in childlist:
 775             target = fdmap[child]
 776             if target == child:
 777                 # fd is already in place
 778                 if debug: print >>errfd, "%d already in place" % target
 779                 fdesc._unsetCloseOnExec(child)
 780             else:
 781                 if child in fdmap.values():
 782                     # we can't replace child-fd yet, as some other mapping
 783                     # still needs the fd it wants to target. We must preserve
 784                     # that old fd by duping it to a new home.
 785                     newtarget = os.dup(child) # give it a safe home
 786                     if debug: print >>errfd, "os.dup(%d) -> %d" % (child,
 787                                                                    newtarget)
 788                     os.close(child) # close the original
 789                     for c, p in fdmap.items():
 790                         if p == child:
 791                             fdmap[c] = newtarget # update all pointers
 792                 # now it should be available
 793                 if debug: print >>errfd, "os.dup2(%d,%d)" % (target, child)
 794                 os.dup2(target, child)
 795
 796         # At this point, the child has everything it needs. We want to close
 797         # everything that isn't going to be used by the child, i.e.
 798         # everything not in fdmap.keys(). The only remaining fds open are
 799         # those in fdmap.values().
 800
 801         # Any given fd may appear in fdmap.values() multiple times, so we
 802         # need to remove duplicates first.
 803
 804         old = []
 805         for fd in fdmap.values():
 806             if not fd in old:
 807                 if not fd in fdmap.keys():
 808                     old.append(fd)
 809         if debug: print >>errfd, "old", old
 810         for fd in old:
 811             os.close(fd)
 812
 813         self._resetSignalDisposition()
 814
 815
 816     def writeToChild(self, childFD, data):
 817         self.pipes[childFD].write(data)
 818
 819     def closeChildFD(self, childFD):
 820         # for writer pipes, loseConnection tries to write the remaining data
 821         # out to the pipe before closing it
 822         # if childFD is not in the list of pipes, assume that it is already
 823         # closed
 824         if childFD in self.pipes:
 825             self.pipes[childFD].loseConnection()
 826
 827     def pauseProducing(self):
 828         for p in self.pipes.itervalues():
 829             if isinstance(p, ProcessReader):
 830                 p.stopReading()
 831
 832     def resumeProducing(self):
 833         for p in self.pipes.itervalues():
 834             if isinstance(p, ProcessReader):
 835                 p.startReading()
 836
 837     # compatibility
 838     def closeStdin(self):
 839         """
 840         Call this to close standard input on this process.
 841         """
 842         self.closeChildFD(0)
 843
 844     def closeStdout(self):
 845         self.closeChildFD(1)
 846
 847     def closeStderr(self):
 848         self.closeChildFD(2)
 849
 850     def loseConnection(self):
 851         self.closeStdin()
 852         self.closeStderr()
 853         self.closeStdout()
 854
 855     def write(self, data):
 856         """
 857         Call this to write to standard input on this process.
 858
 859         NOTE: This will silently lose data if there is no standard input.
 860         """
 861         if 0 in self.pipes:
 862             self.pipes[0].write(data)
 863
 864     def registerProducer(self, producer, streaming):
 865         """
 866         Call this to register producer for standard input.
 867
 868         If there is no standard input producer.stopProducing() will
 869         be called immediately.
 870         """
 871         if 0 in self.pipes:
 872             self.pipes[0].registerProducer(producer, streaming)
 873         else:
 874             producer.stopProducing()
 875
 876     def unregisterProducer(self):
 877         """
 878         Call this to unregister producer for standard input."""
 879         if 0 in self.pipes:
 880             self.pipes[0].unregisterProducer()
 881
 882     def writeSequence(self, seq):
 883         """
 884         Call this to write to standard input on this process.
 885
 886         NOTE: This will silently lose data if there is no standard input.
 887         """
 888         if 0 in self.pipes:
 889             self.pipes[0].writeSequence(seq)
 890
 891
 892     def childDataReceived(self, name, data):
 893         self.proto.childDataReceived(name, data)
 894
 895
 896     def childConnectionLost(self, childFD, reason):
 897         # this is called when one of the helpers (ProcessReader or
 898         # ProcessWriter) notices their pipe has been closed
 899         os.close(self.pipes[childFD].fileno())
 900         del self.pipes[childFD]
 901         try:
 902             self.proto.childConnectionLost(childFD)
 903         except:
 904             log.err()
 905         self.maybeCallProcessEnded()
 906
 907     def maybeCallProcessEnded(self):
 908         # we don't call ProcessProtocol.processEnded until:
 909         #  the child has terminated, AND
 910         #  all writers have indicated an error status, AND
 911         #  all readers have indicated EOF
 912         # This insures that we've gathered all output from the process.
 913         if self.pipes:
 914             return
 915         if not self.lostProcess:
 916             self.reapProcess()
 917             return
 918         _BaseProcess.maybeCallProcessEnded(self)
 919
 920
 921
 922 class PTYProcess(abstract.FileDescriptor, _BaseProcess):
 923     """
 924     An operating-system Process that uses PTY support.
 925     """
 926     implements(IProcessTransport)
 927
 928     status = -1
 929     pid = None
 930
 931     def __init__(self, reactor, executable, args, environment, path, proto,
 932                  uid=None, gid=None, usePTY=None):
 933         """
 934         Spawn an operating-system process.
 935
 936         This is where the hard work of disconnecting all currently open
 937         files / forking / executing the new process happens.  (This is
 938         executed automatically when a Process is instantiated.)
 939
 940         This will also run the subprocess as a given user ID and group ID, if
 941         specified.  (Implementation Note: this doesn't support all the arcane
 942         nuances of setXXuid on UNIX: it will assume that either your effective
 943         or real UID is 0.)
 944         """
 945         if pty is None and not isinstance(usePTY, (tuple, list)):
 946             # no pty module and we didn't get a pty to use
 947             raise NotImplementedError(
 948                 "cannot use PTYProcess on platforms without the pty module.")
 949         abstract.FileDescriptor.__init__(self, reactor)
 950         _BaseProcess.__init__(self, proto)
 951
 952         if isinstance(usePTY, (tuple, list)):
 953             masterfd, slavefd, ttyname = usePTY
 954         else:
 955             masterfd, slavefd = pty.openpty()
 956             ttyname = os.ttyname(slavefd)
 957
 958         try:
 959             self._fork(path, uid, gid, executable, args, environment,
 960                        masterfd=masterfd, slavefd=slavefd)
 961         except:
 962             if not isinstance(usePTY, (tuple, list)):
 963                 os.close(masterfd)
 964                 os.close(slavefd)
 965             raise
 966
 967         # we are now in parent process:
 968         os.close(slavefd)
 969         fdesc.setNonBlocking(masterfd)
 970         self.fd = masterfd
 971         self.startReading()
 972         self.connected = 1
 973         self.status = -1
 974         try:
 975             self.proto.makeConnection(self)
 976         except:
 977             log.err()
 978         registerReapProcessHandler(self.pid, self)
 979
 980     def _setupChild(self, masterfd, slavefd):
 981         """
 982         Setup child process after fork() but before exec().
 983         """
 984         os.close(masterfd)
 985         if hasattr(termios, 'TIOCNOTTY'):
 986             try:
 987                 fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
 988             except OSError:
 989                 pass
 990             else:
 991                 try:
 992                     fcntl.ioctl(fd, termios.TIOCNOTTY, '')
 993                 except:
 994                     pass
 995                 os.close(fd)
 996
 997         os.setsid()
 998
 999         if hasattr(termios, 'TIOCSCTTY'):
1000             fcntl.ioctl(slavefd, termios.TIOCSCTTY, '')
1001
1002         for fd in range(3):
1003             if fd != slavefd:
1004                 os.close(fd)
1005
1006         os.dup2(slavefd, 0) # stdin
1007         os.dup2(slavefd, 1) # stdout
1008         os.dup2(slavefd, 2) # stderr
1009
1010         for fd in _listOpenFDs():
1011             if fd > 2:
1012                 try:
1013                     os.close(fd)
1014                 except:
1015                     pass
1016
1017         self._resetSignalDisposition()
1018
1019
1020     # PTYs do not have stdin/stdout/stderr. They only have in and out, just
1021     # like sockets. You cannot close one without closing off the entire PTY.
1022     def closeStdin(self):
1023         pass
1024
1025     def closeStdout(self):
1026         pass
1027
1028     def closeStderr(self):
1029         pass
1030
1031     def doRead(self):
1032         """
1033         Called when my standard output stream is ready for reading.
1034         """
1035         return fdesc.readFromFD(
1036             self.fd,
1037             lambda data: self.proto.childDataReceived(1, data))
1038
1039     def fileno(self):
1040         """
1041         This returns the file number of standard output on this process.
1042         """
1043         return self.fd
1044
1045     def maybeCallProcessEnded(self):
1046         # two things must happen before we call the ProcessProtocol's
1047         # processEnded method. 1: the child process must die and be reaped
1048         # (which calls our own processEnded method). 2: the child must close
1049         # their stdin/stdout/stderr fds, causing the pty to close, causing
1050         # our connectionLost method to be called. #2 can also be triggered
1051         # by calling .loseConnection().
1052         if self.lostProcess == 2:
1053             _BaseProcess.maybeCallProcessEnded(self)
1054
1055     def connectionLost(self, reason):
1056         """
1057         I call this to clean up when one or all of my connections has died.
1058         """
1059         abstract.FileDescriptor.connectionLost(self, reason)
1060         os.close(self.fd)
1061         self.lostProcess += 1
1062         self.maybeCallProcessEnded()
1063
1064     def writeSomeData(self, data):
1065         """
1066         Write some data to the open process.
1067         """
1068         return fdesc.writeToFD(self.fd, data)