-"""A pub/sub Bus for managing
+"""A pub/sub Bus for managing states.
-This package is completely standalone, depending only on the stdlib.
-A Bus object is used to contain and manage site-wide behavior:
-daemonization, socket server start/stop, process reload, signal handling,
-drop privileges, PID file management, logging for all of these,
-In addition, a Bus object provides a place for frameworks
-to register code that runs in response to site-wide events (like
-process start and stop), or which controls or otherwise interacts with
-the site-wide components mentioned above. For example, a framework which
-uses file-based templates would add known template filenames to an
-Ideally, a Bus object will be flexible enough to be useful in a variety
-of invocation scenarios:
- 1. The deployer starts a site from the command line via a
- framework-neutral deployment script; applications from multiple frameworks
- are mixed in a single site. Command-line arguments and configuration
- files are used to define site-wide components such as a socket server,
- component graphs, autoreload behavior, signal handling, etc.
- 2. The deployer starts a site via some other process, such as Apache;
- applications from multiple frameworks are mixed in a single site.
- Autoreload and signal handling (from Python at least) are disabled.
- 3. The deployer starts a site via a framework-specific mechanism;
- for example, when running tests, exploring tutorials, or deploying
- single applications from a single framework. The framework controls
- which site-wide components are enabled as it sees fit.
+A Bus object is used to contain and manage behavior for any system
+of diverse components. A Bus object provides a place for frameworks
+to register code that runs in response to events, or which controls
+or otherwise interacts with the components.
The Bus object in this package uses topic-based publish-subscribe
-messaging to accomplish all this. A few topic channels are built in
-('start', 'stop', 'exit', 'graceful', 'log', and 'main'). Frameworks and
-site containers are free to define their own. If a message is sent to a
+messaging to accomplish all this. Frameworks and site containers
+are free to define their own channels. If a message is sent to a
channel that has not been defined or has no listeners, there is no effect.
-In general, there should only ever be a single Bus object per process.
-Frameworks and site containers share a single Bus object by publishing
-messages and subscribing listeners.
-The Bus object works as a finite state machine which models the current
-state of the process. Bus methods move it from one state to another;
-those methods then publish to subscribed listeners on the channel for
- STOPPING --> STOPPED --> EXITING -> X
import traceback as _traceback
-# Here I save the value of os.getcwd(), which, if I am imported early enough,
-# will be the directory from which the startup script was run. This is needed
-# by _do_execv(), to change back to the original directory before execv()ing a
-# new process. This is a defense against the application having changed the
-# current working directory (which could make sys.executable "not found" if
-# sys.executable is a relative-path, and/or cause other problems).
-_startup_cwd = os.getcwd()
- """Exception raised when errors occur in a listener during Bus.publish().
+ """Exception raised when errors occur in a listener during .publish()."""
def __init__(self, *args, **kwargs):
-# Use a flag to indicate the state of the bus.
+ def __init__(self, name):
- return "states.%s" % self.name
+ return "State(%s)" % repr(self.name)
- def __setattr__(self, key, value):
- if isinstance(value, self.State):
- object.__setattr__(self, key, value)
-states.STOPPED = states.State()
-states.STARTING = states.State()
-states.STARTED = states.State()
-states.STOPPING = states.State()
-states.EXITING = states.State()
- max_files = os.sysconf('SC_OPEN_MAX')
+ """An object with enumerated state attributes."""
- """Process state-machine and messenger.
+ """State machine and pub/sub messenger.
All listeners for a given channel are guaranteed to be called even
if others at the same channel fail. Each failure is logged, but
- max_cloexec_files = max_files
+ def __init__(self, states=None, state=None, channels=None):
+ if isinstance(states, StateEnum):
+ self.states = StateEnum()
+ setattr(self.states, s, State(s))
- self.state = states.STOPPED
- [(channel, set()) for channel
- in ('start', 'stop', 'exit', 'graceful', 'log', 'main')])
+ state = getattr(self.states, state)
+ self.listeners = dict((channel, set()) for channel in channels)
"""Discard all subscribed callbacks."""
- self.state = states.STOPPED
# Use items() as a snapshot instead of while+pop so that callers
# can be slightly lax in subscribing new listeners while the old
# ones are being removed.
- """An atexit handler which asserts the Bus is not running."""
- if self.state != states.EXITING:
- "The main thread is exiting, but the Bus is in the %r state; "
- "shutting it down automatically now. You must either call "
- "bus.block() after start(), or call bus.exit() before the "
- "main thread exits." % self.state, RuntimeWarning)
- """Start all services."""
- self.state = states.STARTING
- self.log('Bus STARTING')
- self.state = states.STARTED
- self.log('Bus STARTED')
- except (KeyboardInterrupt, SystemExit):
- self.log("Shutting down due to error in start listener:",
- level=40, traceback=True)
- e_info = sys.exc_info()
- # Any stop/exit errors will be logged inside publish().
- # Re-raise the original error
- """Stop all services and prepare to exit the process."""
- self.state = states.EXITING
- self.log('Bus EXITING')
- # This isn't strictly necessary, but it's better than seeing
- # "Waiting for child threads to terminate..." and then nothing.
- # This method is often called asynchronously (whether thread,
- # signal handler, console handler, or atexit handler), so we
- # can't just let exceptions propagate out unhandled.
- # Assume it's been logged and just die.
- os._exit(70) # EX_SOFTWARE
- if exitstate == states.STARTING:
- # exit() was called before start() finished, possibly due to
- # Ctrl-C because a start listener got stuck. In this case,
- # we could get stuck in a loop where Ctrl-C never exits the
- # process, so we just call os.exit here.
- os._exit(70) # EX_SOFTWARE
- """Restart the process (may close connections).
- This method does not restart the process from the calling thread;
- instead, it stops the bus and asks the main thread to call execv.
- """Advise all services to reload."""
- self.log('Bus graceful')
- def block(self, interval=0.1):
- """Wait for the EXITING state, KeyboardInterrupt or SystemExit.
- This function is intended to be called only by the main thread.
- After waiting for the EXITING state, it also waits for all threads
- to terminate, and then calls os.execv if self.execv is True. This
- design allows another thread to call bus.restart, yet have the main
- thread perform the actual execv call (required on some platforms).
- self.wait(states.EXITING, interval=interval, channel='main')
- except (KeyboardInterrupt, IOError):
- # The time.sleep call might raise
- # "IOError: [Errno 4] Interrupted function call" on KBInt.
- self.log('Keyboard Interrupt: shutting down bus')
- self.log('SystemExit raised: shutting down bus')
- # Waiting for ALL child threads to finish is necessary on OS X.
- # See http://www.cherrypy.org/ticket/581.
- # It's also good to let them all shut down before allowing
- # the main thread to call atexit handlers.
- # See http://www.cherrypy.org/ticket/751.
- self.log("Waiting for child threads to terminate...")
- for t in threading.enumerate():
- if t != threading.currentThread() and t.isAlive():
- # Note that any dummy (external) threads are always daemonic.
- self.log("Waiting for thread %s." % t.getName())
def wait(self, state, interval=0.1, channel=None):
"""Poll for the given state(s) at intervals; publish to channel."""
if isinstance(state, (tuple, list)):
+ _states = state
+ _states = [state]
- while self.state not in _states:
+ while self.state not in _states:
- """Re-execute the current process.
- This must be called from the main thread, because certain platforms
- (OS X) don't allow execv to be called in a child thread very well.
- self.log('Re-spawning %s' % ' '.join(args))
- if sys.platform[:4] == 'java':
- from _systemrestart import SystemRestart
- args.insert(0, sys.executable)
- if sys.platform == 'win32':
- args = ['"%s"' % arg for arg in args]
- if self.max_cloexec_files:
- os.execv(sys.executable, args)
- def _set_cloexec(self):
- """Set the CLOEXEC flag on all open files (except stdin/out/err).
- If self.max_cloexec_files is an integer (the default), then on
- platforms which support it, it represents the max open files setting
- for the operating system. This function will be called just before
- the process is restarted via os.execv() to prevent open files
- from persisting into the new process.
- Set self.max_cloexec_files to 0 to disable this behavior.
- for fd in range(3, self.max_cloexec_files): # skip stdin/out/err
- flags = fcntl.fcntl(fd, fcntl.F_GETFD)
- fcntl.fcntl(fd, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)
- """Stop all services."""
- self.state = states.STOPPING
- self.log('Bus STOPPING')
- self.state = states.STOPPED
- self.log('Bus STOPPED')
- def start_with_callback(self, func, args=None, kwargs=None):
- """Start 'func' in a new thread T, then start self (and return T)."""
- def _callback(func_, *a, **kw):
- t = threading.Thread(target=_callback, args=args, kwargs=kwargs)
- t.setName('Bus Callback ' + t.getName())
def log(self, msg="", level=20, traceback=False):
"""Log the given message. Append the last traceback if requested."""