monitor/cherrypy/process/plugins.py

   1 """Site services for use with a Web Site Process Bus."""
   2
   3 import os
   4 import re
   5 import signal as _signal
   6 import sys
   7 import time
   8 import threading
   9
  10 from cherrypy._cpcompat import basestring, get_daemon, get_thread_ident, ntob, set
  11
  12 # _module__file__base is used by Autoreload to make
  13 # absolute any filenames retrieved from sys.modules which are not
  14 # already absolute paths.  This is to work around Python's quirk
  15 # of importing the startup script and using a relative filename
  16 # for it in sys.modules.
  17 #
  18 # Autoreload examines sys.modules afresh every time it runs. If an application
  19 # changes the current directory by executing os.chdir(), then the next time
  20 # Autoreload runs, it will not be able to find any filenames which are
  21 # not absolute paths, because the current directory is not the same as when the
  22 # module was first imported.  Autoreload will then wrongly conclude the file has
  23 # "changed", and initiate the shutdown/re-exec sequence.
  24 # See ticket #917.
  25 # For this workaround to have a decent probability of success, this module
  26 # needs to be imported as early as possible, before the app has much chance
  27 # to change the working directory.
  28 _module__file__base = os.getcwd()
  29
  30
  31 class SimplePlugin(object):
  32     """Plugin base class which auto-subscribes methods for known channels."""
  33
  34     bus = None
  35     """A :class:`Bus <cherrypy.process.wspbus.Bus>`, usually cherrypy.engine."""
  36
  37     def __init__(self, bus):
  38         self.bus = bus
  39
  40     def subscribe(self):
  41         """Register this object as a (multi-channel) listener on the bus."""
  42         for channel in self.bus.listeners:
  43             # Subscribe self.start, self.exit, etc. if present.
  44             method = getattr(self, channel, None)
  45             if method is not None:
  46                 self.bus.subscribe(channel, method)
  47
  48     def unsubscribe(self):
  49         """Unregister this object as a listener on the bus."""
  50         for channel in self.bus.listeners:
  51             # Unsubscribe self.start, self.exit, etc. if present.
  52             method = getattr(self, channel, None)
  53             if method is not None:
  54                 self.bus.unsubscribe(channel, method)
  55
  56
  57
  58 class SignalHandler(object):
  59     """Register bus channels (and listeners) for system signals.
  60
  61     You can modify what signals your application listens for, and what it does
  62     when it receives signals, by modifying :attr:`SignalHandler.handlers`,
  63     a dict of {signal name: callback} pairs. The default set is::
  64
  65         handlers = {'SIGTERM': self.bus.exit,
  66                     'SIGHUP': self.handle_SIGHUP,
  67                     'SIGUSR1': self.bus.graceful,
  68                    }
  69
  70     The :func:`SignalHandler.handle_SIGHUP`` method calls
  71     :func:`bus.restart()<cherrypy.process.wspbus.Bus.restart>`
  72     if the process is daemonized, but
  73     :func:`bus.exit()<cherrypy.process.wspbus.Bus.exit>`
  74     if the process is attached to a TTY. This is because Unix window
  75     managers tend to send SIGHUP to terminal windows when the user closes them.
  76
  77     Feel free to add signals which are not available on every platform. The
  78     :class:`SignalHandler` will ignore errors raised from attempting to register
  79     handlers for unknown signals.
  80     """
  81
  82     handlers = {}
  83     """A map from signal names (e.g. 'SIGTERM') to handlers (e.g. bus.exit)."""
  84
  85     signals = {}
  86     """A map from signal numbers to names."""
  87
  88     for k, v in vars(_signal).items():
  89         if k.startswith('SIG') and not k.startswith('SIG_'):
  90             signals[v] = k
  91     del k, v
  92
  93     def __init__(self, bus):
  94         self.bus = bus
  95         # Set default handlers
  96         self.handlers = {'SIGTERM': self.bus.exit,
  97                          'SIGHUP': self.handle_SIGHUP,
  98                          'SIGUSR1': self.bus.graceful,
  99                          }
 100
 101         if sys.platform[:4] == 'java':
 102             del self.handlers['SIGUSR1']
 103             self.handlers['SIGUSR2'] = self.bus.graceful
 104             self.bus.log("SIGUSR1 cannot be set on the JVM platform. "
 105                          "Using SIGUSR2 instead.")
 106             self.handlers['SIGINT'] = self._jython_SIGINT_handler
 107
 108         self._previous_handlers = {}
 109
 110     def _jython_SIGINT_handler(self, signum=None, frame=None):
 111         # See http://bugs.jython.org/issue1313
 112         self.bus.log('Keyboard Interrupt: shutting down bus')
 113         self.bus.exit()
 114
 115     def subscribe(self):
 116         """Subscribe self.handlers to signals."""
 117         for sig, func in self.handlers.items():
 118             try:
 119                 self.set_handler(sig, func)
 120             except ValueError:
 121                 pass
 122
 123     def unsubscribe(self):
 124         """Unsubscribe self.handlers from signals."""
 125         for signum, handler in self._previous_handlers.items():
 126             signame = self.signals[signum]
 127
 128             if handler is None:
 129                 self.bus.log("Restoring %s handler to SIG_DFL." % signame)
 130                 handler = _signal.SIG_DFL
 131             else:
 132                 self.bus.log("Restoring %s handler %r." % (signame, handler))
 133
 134             try:
 135                 our_handler = _signal.signal(signum, handler)
 136                 if our_handler is None:
 137                     self.bus.log("Restored old %s handler %r, but our "
 138                                  "handler was not registered." %
 139                                  (signame, handler), level=30)
 140             except ValueError:
 141                 self.bus.log("Unable to restore %s handler %r." %
 142                              (signame, handler), level=40, traceback=True)
 143
 144     def set_handler(self, signal, listener=None):
 145         """Subscribe a handler for the given signal (number or name).
 146
 147         If the optional 'listener' argument is provided, it will be
 148         subscribed as a listener for the given signal's channel.
 149
 150         If the given signal name or number is not available on the current
 151         platform, ValueError is raised.
 152         """
 153         if isinstance(signal, basestring):
 154             signum = getattr(_signal, signal, None)
 155             if signum is None:
 156                 raise ValueError("No such signal: %r" % signal)
 157             signame = signal
 158         else:
 159             try:
 160                 signame = self.signals[signal]
 161             except KeyError:
 162                 raise ValueError("No such signal: %r" % signal)
 163             signum = signal
 164
 165         prev = _signal.signal(signum, self._handle_signal)
 166         self._previous_handlers[signum] = prev
 167
 168         if listener is not None:
 169             self.bus.log("Listening for %s." % signame)
 170             self.bus.subscribe(signame, listener)
 171
 172     def _handle_signal(self, signum=None, frame=None):
 173         """Python signal handler (self.set_handler subscribes it for you)."""
 174         signame = self.signals[signum]
 175         self.bus.log("Caught signal %s." % signame)
 176         self.bus.publish(signame)
 177
 178     def handle_SIGHUP(self):
 179         """Restart if daemonized, else exit."""
 180         if os.isatty(sys.stdin.fileno()):
 181             # not daemonized (may be foreground or background)
 182             self.bus.log("SIGHUP caught but not daemonized. Exiting.")
 183             self.bus.exit()
 184         else:
 185             self.bus.log("SIGHUP caught while daemonized. Restarting.")
 186             self.bus.restart()
 187
 188
 189 try:
 190     import pwd, grp
 191 except ImportError:
 192     pwd, grp = None, None
 193
 194
 195 class DropPrivileges(SimplePlugin):
 196     """Drop privileges. uid/gid arguments not available on Windows.
 197
 198     Special thanks to Gavin Baker: http://antonym.org/node/100.
 199     """
 200
 201     def __init__(self, bus, umask=None, uid=None, gid=None):
 202         SimplePlugin.__init__(self, bus)
 203         self.finalized = False
 204         self.uid = uid
 205         self.gid = gid
 206         self.umask = umask
 207
 208     def _get_uid(self):
 209         return self._uid
 210     def _set_uid(self, val):
 211         if val is not None:
 212             if pwd is None:
 213                 self.bus.log("pwd module not available; ignoring uid.",
 214                              level=30)
 215                 val = None
 216             elif isinstance(val, basestring):
 217                 val = pwd.getpwnam(val)[2]
 218         self._uid = val
 219     uid = property(_get_uid, _set_uid,
 220         doc="The uid under which to run. Availability: Unix.")
 221
 222     def _get_gid(self):
 223         return self._gid
 224     def _set_gid(self, val):
 225         if val is not None:
 226             if grp is None:
 227                 self.bus.log("grp module not available; ignoring gid.",
 228                              level=30)
 229                 val = None
 230             elif isinstance(val, basestring):
 231                 val = grp.getgrnam(val)[2]
 232         self._gid = val
 233     gid = property(_get_gid, _set_gid,
 234         doc="The gid under which to run. Availability: Unix.")
 235
 236     def _get_umask(self):
 237         return self._umask
 238     def _set_umask(self, val):
 239         if val is not None:
 240             try:
 241                 os.umask
 242             except AttributeError:
 243                 self.bus.log("umask function not available; ignoring umask.",
 244                              level=30)
 245                 val = None
 246         self._umask = val
 247     umask = property(_get_umask, _set_umask,
 248         doc="""The default permission mode for newly created files and directories.
 249
 250         Usually expressed in octal format, for example, ``0644``.
 251         Availability: Unix, Windows.
 252         """)
 253
 254     def start(self):
 255         # uid/gid
 256         def current_ids():
 257             """Return the current (uid, gid) if available."""
 258             name, group = None, None
 259             if pwd:
 260                 name = pwd.getpwuid(os.getuid())[0]
 261             if grp:
 262                 group = grp.getgrgid(os.getgid())[0]
 263             return name, group
 264
 265         if self.finalized:
 266             if not (self.uid is None and self.gid is None):
 267                 self.bus.log('Already running as uid: %r gid: %r' %
 268                              current_ids())
 269         else:
 270             if self.uid is None and self.gid is None:
 271                 if pwd or grp:
 272                     self.bus.log('uid/gid not set', level=30)
 273             else:
 274                 self.bus.log('Started as uid: %r gid: %r' % current_ids())
 275                 if self.gid is not None:
 276                     os.setgid(self.gid)
 277                     os.setgroups([])
 278                 if self.uid is not None:
 279                     os.setuid(self.uid)
 280                 self.bus.log('Running as uid: %r gid: %r' % current_ids())
 281
 282         # umask
 283         if self.finalized:
 284             if self.umask is not None:
 285                 self.bus.log('umask already set to: %03o' % self.umask)
 286         else:
 287             if self.umask is None:
 288                 self.bus.log('umask not set', level=30)
 289             else:
 290                 old_umask = os.umask(self.umask)
 291                 self.bus.log('umask old: %03o, new: %03o' %
 292                              (old_umask, self.umask))
 293
 294         self.finalized = True
 295     # This is slightly higher than the priority for server.start
 296     # in order to facilitate the most common use: starting on a low
 297     # port (which requires root) and then dropping to another user.
 298     start.priority = 77
 299
 300
 301 class Daemonizer(SimplePlugin):
 302     """Daemonize the running script.
 303
 304     Use this with a Web Site Process Bus via::
 305
 306         Daemonizer(bus).subscribe()
 307
 308     When this component finishes, the process is completely decoupled from
 309     the parent environment. Please note that when this component is used,
 310     the return code from the parent process will still be 0 if a startup
 311     error occurs in the forked children. Errors in the initial daemonizing
 312     process still return proper exit codes. Therefore, if you use this
 313     plugin to daemonize, don't use the return code as an accurate indicator
 314     of whether the process fully started. In fact, that return code only
 315     indicates if the process succesfully finished the first fork.
 316     """
 317
 318     def __init__(self, bus, stdin='/dev/null', stdout='/dev/null',
 319                  stderr='/dev/null'):
 320         SimplePlugin.__init__(self, bus)
 321         self.stdin = stdin
 322         self.stdout = stdout
 323         self.stderr = stderr
 324         self.finalized = False
 325
 326     def start(self):
 327         if self.finalized:
 328             self.bus.log('Already deamonized.')
 329
 330         # forking has issues with threads:
 331         # http://www.opengroup.org/onlinepubs/000095399/functions/fork.html
 332         # "The general problem with making fork() work in a multi-threaded
 333         #  world is what to do with all of the threads..."
 334         # So we check for active threads:
 335         if threading.activeCount() != 1:
 336             self.bus.log('There are %r active threads. '
 337                          'Daemonizing now may cause strange failures.' %
 338                          threading.enumerate(), level=30)
 339
 340         # See http://www.erlenstar.demon.co.uk/unix/faq_2.html#SEC16
 341         # (or http://www.faqs.org/faqs/unix-faq/programmer/faq/ section 1.7)
 342         # and http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/66012
 343
 344         # Finish up with the current stdout/stderr
 345         sys.stdout.flush()
 346         sys.stderr.flush()
 347
 348         # Do first fork.
 349         try:
 350             pid = os.fork()
 351             if pid == 0:
 352                 # This is the child process. Continue.
 353                 pass
 354             else:
 355                 # This is the first parent. Exit, now that we've forked.
 356                 self.bus.log('Forking once.')
 357                 os._exit(0)
 358         except OSError:
 359             # Python raises OSError rather than returning negative numbers.
 360             exc = sys.exc_info()[1]
 361             sys.exit("%s: fork #1 failed: (%d) %s\n"
 362                      % (sys.argv[0], exc.errno, exc.strerror))
 363
 364         os.setsid()
 365
 366         # Do second fork
 367         try:
 368             pid = os.fork()
 369             if pid > 0:
 370                 self.bus.log('Forking twice.')
 371                 os._exit(0) # Exit second parent
 372         except OSError:
 373             exc = sys.exc_info()[1]
 374             sys.exit("%s: fork #2 failed: (%d) %s\n"
 375                      % (sys.argv[0], exc.errno, exc.strerror))
 376
 377         os.chdir("/")
 378         os.umask(0)
 379
 380         si = open(self.stdin, "r")
 381         so = open(self.stdout, "a+")
 382         se = open(self.stderr, "a+")
 383
 384         # os.dup2(fd, fd2) will close fd2 if necessary,
 385         # so we don't explicitly close stdin/out/err.
 386         # See http://docs.python.org/lib/os-fd-ops.html
 387         os.dup2(si.fileno(), sys.stdin.fileno())
 388         os.dup2(so.fileno(), sys.stdout.fileno())
 389         os.dup2(se.fileno(), sys.stderr.fileno())
 390
 391         self.bus.log('Daemonized to PID: %s' % os.getpid())
 392         self.finalized = True
 393     start.priority = 65
 394
 395
 396 class PIDFile(SimplePlugin):
 397     """Maintain a PID file via a WSPBus."""
 398
 399     def __init__(self, bus, pidfile):
 400         SimplePlugin.__init__(self, bus)
 401         self.pidfile = pidfile
 402         self.finalized = False
 403
 404     def start(self):
 405         pid = os.getpid()
 406         if self.finalized:
 407             self.bus.log('PID %r already written to %r.' % (pid, self.pidfile))
 408         else:
 409             open(self.pidfile, "wb").write(ntob("%s" % pid, 'utf8'))
 410             self.bus.log('PID %r written to %r.' % (pid, self.pidfile))
 411             self.finalized = True
 412     start.priority = 70
 413
 414     def exit(self):
 415         try:
 416             os.remove(self.pidfile)
 417             self.bus.log('PID file removed: %r.' % self.pidfile)
 418         except (KeyboardInterrupt, SystemExit):
 419             raise
 420         except:
 421             pass
 422
 423
 424 class PerpetualTimer(threading._Timer):
 425     """A responsive subclass of threading._Timer whose run() method repeats.
 426
 427     Use this timer only when you really need a very interruptible timer;
 428     this checks its 'finished' condition up to 20 times a second, which can
 429     results in pretty high CPU usage
 430     """
 431
 432     def run(self):
 433         while True:
 434             self.finished.wait(self.interval)
 435             if self.finished.isSet():
 436                 return
 437             try:
 438                 self.function(*self.args, **self.kwargs)
 439             except Exception:
 440                 self.bus.log("Error in perpetual timer thread function %r." %
 441                              self.function, level=40, traceback=True)
 442                 # Quit on first error to avoid massive logs.
 443                 raise
 444
 445
 446 class BackgroundTask(threading.Thread):
 447     """A subclass of threading.Thread whose run() method repeats.
 448
 449     Use this class for most repeating tasks. It uses time.sleep() to wait
 450     for each interval, which isn't very responsive; that is, even if you call
 451     self.cancel(), you'll have to wait until the sleep() call finishes before
 452     the thread stops. To compensate, it defaults to being daemonic, which means
 453     it won't delay stopping the whole process.
 454     """
 455
 456     def __init__(self, interval, function, args=[], kwargs={}):
 457         threading.Thread.__init__(self)
 458         self.interval = interval
 459         self.function = function
 460         self.args = args
 461         self.kwargs = kwargs
 462         self.running = False
 463
 464     def cancel(self):
 465         self.running = False
 466
 467     def run(self):
 468         self.running = True
 469         while self.running:
 470             time.sleep(self.interval)
 471             if not self.running:
 472                 return
 473             try:
 474                 self.function(*self.args, **self.kwargs)
 475             except Exception:
 476                 self.bus.log("Error in background task thread function %r." %
 477                              self.function, level=40, traceback=True)
 478                 # Quit on first error to avoid massive logs.
 479                 raise
 480
 481     def _set_daemon(self):
 482         return True
 483
 484
 485 class Monitor(SimplePlugin):
 486     """WSPBus listener to periodically run a callback in its own thread."""
 487
 488     callback = None
 489     """The function to call at intervals."""
 490
 491     frequency = 60
 492     """The time in seconds between callback runs."""
 493
 494     thread = None
 495     """A :class:`BackgroundTask<cherrypy.process.plugins.BackgroundTask>` thread."""
 496
 497     def __init__(self, bus, callback, frequency=60, name=None):
 498         SimplePlugin.__init__(self, bus)
 499         self.callback = callback
 500         self.frequency = frequency
 501         self.thread = None
 502         self.name = name
 503
 504     def start(self):
 505         """Start our callback in its own background thread."""
 506         if self.frequency > 0:
 507             threadname = self.name or self.__class__.__name__
 508             if self.thread is None:
 509                 self.thread = BackgroundTask(self.frequency, self.callback)
 510                 self.thread.bus = self.bus
 511                 self.thread.setName(threadname)
 512                 self.thread.start()
 513                 self.bus.log("Started monitor thread %r." % threadname)
 514             else:
 515                 self.bus.log("Monitor thread %r already started." % threadname)
 516     start.priority = 70
 517
 518     def stop(self):
 519         """Stop our callback's background task thread."""
 520         if self.thread is None:
 521             self.bus.log("No thread running for %s." % self.name or self.__class__.__name__)
 522         else:
 523             if self.thread is not threading.currentThread():
 524                 name = self.thread.getName()
 525                 self.thread.cancel()
 526                 if not get_daemon(self.thread):
 527                     self.bus.log("Joining %r" % name)
 528                     self.thread.join()
 529                 self.bus.log("Stopped thread %r." % name)
 530             self.thread = None
 531
 532     def graceful(self):
 533         """Stop the callback's background task thread and restart it."""
 534         self.stop()
 535         self.start()
 536
 537
 538 class Autoreloader(Monitor):
 539     """Monitor which re-executes the process when files change.
 540
 541     This :ref:`plugin<plugins>` restarts the process (via :func:`os.execv`)
 542     if any of the files it monitors change (or is deleted). By default, the
 543     autoreloader monitors all imported modules; you can add to the
 544     set by adding to ``autoreload.files``::
 545
 546         cherrypy.engine.autoreload.files.add(myFile)
 547
 548     If there are imported files you do *not* wish to monitor, you can adjust the
 549     ``match`` attribute, a regular expression. For example, to stop monitoring
 550     cherrypy itself::
 551
 552         cherrypy.engine.autoreload.match = r'^(?!cherrypy).+'
 553
 554     Like all :class:`Monitor<cherrypy.process.plugins.Monitor>` plugins,
 555     the autoreload plugin takes a ``frequency`` argument. The default is
 556     1 second; that is, the autoreloader will examine files once each second.
 557     """
 558
 559     files = None
 560     """The set of files to poll for modifications."""
 561
 562     frequency = 1
 563     """The interval in seconds at which to poll for modified files."""
 564
 565     match = '.*'
 566     """A regular expression by which to match filenames."""
 567
 568     def __init__(self, bus, frequency=1, match='.*'):
 569         self.mtimes = {}
 570         self.files = set()
 571         self.match = match
 572         Monitor.__init__(self, bus, self.run, frequency)
 573
 574     def start(self):
 575         """Start our own background task thread for self.run."""
 576         if self.thread is None:
 577             self.mtimes = {}
 578         Monitor.start(self)
 579     start.priority = 70
 580
 581     def sysfiles(self):
 582         """Return a Set of sys.modules filenames to monitor."""
 583         files = set()
 584         for k, m in sys.modules.items():
 585             if re.match(self.match, k):
 586                 if hasattr(m, '__loader__') and hasattr(m.__loader__, 'archive'):
 587                     f = m.__loader__.archive
 588                 else:
 589                     f = getattr(m, '__file__', None)
 590                     if f is not None and not os.path.isabs(f):
 591                         # ensure absolute paths so a os.chdir() in the app doesn't break me
 592                         f = os.path.normpath(os.path.join(_module__file__base, f))
 593                 files.add(f)
 594         return files
 595
 596     def run(self):
 597         """Reload the process if registered files have been modified."""
 598         for filename in self.sysfiles() | self.files:
 599             if filename:
 600                 if filename.endswith('.pyc'):
 601                     filename = filename[:-1]
 602
 603                 oldtime = self.mtimes.get(filename, 0)
 604                 if oldtime is None:
 605                     # Module with no .py file. Skip it.
 606                     continue
 607
 608                 try:
 609                     mtime = os.stat(filename).st_mtime
 610                 except OSError:
 611                     # Either a module with no .py file, or it's been deleted.
 612                     mtime = None
 613
 614                 if filename not in self.mtimes:
 615                     # If a module has no .py file, this will be None.
 616                     self.mtimes[filename] = mtime
 617                 else:
 618                     if mtime is None or mtime > oldtime:
 619                         # The file has been deleted or modified.
 620                         self.bus.log("Restarting because %s changed." % filename)
 621                         self.thread.cancel()
 622                         self.bus.log("Stopped thread %r." % self.thread.getName())
 623                         self.bus.restart()
 624                         return
 625
 626
 627 class ThreadManager(SimplePlugin):
 628     """Manager for HTTP request threads.
 629
 630     If you have control over thread creation and destruction, publish to
 631     the 'acquire_thread' and 'release_thread' channels (for each thread).
 632     This will register/unregister the current thread and publish to
 633     'start_thread' and 'stop_thread' listeners in the bus as needed.
 634
 635     If threads are created and destroyed by code you do not control
 636     (e.g., Apache), then, at the beginning of every HTTP request,
 637     publish to 'acquire_thread' only. You should not publish to
 638     'release_thread' in this case, since you do not know whether
 639     the thread will be re-used or not. The bus will call
 640     'stop_thread' listeners for you when it stops.
 641     """
 642
 643     threads = None
 644     """A map of {thread ident: index number} pairs."""
 645
 646     def __init__(self, bus):
 647         self.threads = {}
 648         SimplePlugin.__init__(self, bus)
 649         self.bus.listeners.setdefault('acquire_thread', set())
 650         self.bus.listeners.setdefault('start_thread', set())
 651         self.bus.listeners.setdefault('release_thread', set())
 652         self.bus.listeners.setdefault('stop_thread', set())
 653
 654     def acquire_thread(self):
 655         """Run 'start_thread' listeners for the current thread.
 656
 657         If the current thread has already been seen, any 'start_thread'
 658         listeners will not be run again.
 659         """
 660         thread_ident = get_thread_ident()
 661         if thread_ident not in self.threads:
 662             # We can't just use get_ident as the thread ID
 663             # because some platforms reuse thread ID's.
 664             i = len(self.threads) + 1
 665             self.threads[thread_ident] = i
 666             self.bus.publish('start_thread', i)
 667
 668     def release_thread(self):
 669         """Release the current thread and run 'stop_thread' listeners."""
 670         thread_ident = get_thread_ident()
 671         i = self.threads.pop(thread_ident, None)
 672         if i is not None:
 673             self.bus.publish('stop_thread', i)
 674
 675     def stop(self):
 676         """Release all threads and run all 'stop_thread' listeners."""
 677         for thread_ident, i in self.threads.items():
 678             self.bus.publish('stop_thread', i)
 679         self.threads.clear()
 680     graceful = stop
 681