iframe-proxy | Sunbelt Computer Software

787 lines (653 loc) · 26.6 KB
# Copyright (C) 2007-2016 CEA/DAM
# Copyright (C) 2015-2016 Stephane Thiell <sthiell@stanford.edu>
# This file is part of ClusterShell.
# ClusterShell is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
# ClusterShell is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public
# License along with ClusterShell; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
Interface of underlying Task's Engine.
An Engine implements a loop your thread enters and uses to call event handlers
in response to incoming events (from workers, timers, etc.).
import errno
import heapq
import logging
import time
import traceback
LOGGER = logging.getLogger(__name__)
# Engine client fd I/O event interest bits
E_READ = 0x1
E_WRITE = 0x2
# Define epsilon value for time float arithmetic operations
EPSILON = 1.0e-3
# Special fanout value for unlimited
FANOUT_UNLIMITED = -1
# Special fanout value to use default Engine fanout
FANOUT_DEFAULT = None
class EngineException(Exception):
    Base engine exception.
class EngineAbortException(EngineException):
    Raised on user abort.
    def __init__(self, kill):
        EngineException.__init__(self)
        self.kill = kill
class EngineTimeoutException(EngineException):
    Raised when a timeout is encountered.
class EngineIllegalOperationError(EngineException):
    Error raised when an illegal operation has been performed.
class EngineAlreadyRunningError(EngineIllegalOperationError):
    Error raised when the engine is already running.
class EngineNotSupportedError(EngineException):
    Error raised when the engine mechanism is not supported.
    def __init__(self, engineid):
        EngineException.__init__(self)
        self.engineid = engineid
class EngineBaseTimer(object):
    Abstract class for ClusterShell's engine timer. Such a timer
    requires a relative fire time (delay) in seconds (as float), and
    supports an optional repeating interval in seconds (as float too).
    See EngineTimer for more information about ClusterShell timers.
    def __init__(self, fire_delay, interval=-1.0, autoclose=False):
        """
        Create a base timer.
        """
        # fire_delay is used for comparison between timers and MUST NOT be
        # None in Python 3 as comparison with float is not possible and could
        # lead to confusion anyway. If None is passed, fire_delay is now set
        # to -1 to avoid the timer to be armed in _EngineTimerQ.schedule().
        if fire_delay is None:
            self.fire_delay = -1.0
        else:
            self.fire_delay = fire_delay
        self.interval = interval
        self.autoclose = autoclose
        self._engine = None
        self._timercase = None
    def _set_engine(self, engine):
        """
        Bind to engine, called by Engine.
        """
        if self._engine:
            # A timer can be registered to only one engine at a time.
            raise EngineIllegalOperationError("Already bound to engine.")
        self._engine = engine
    def invalidate(self):
        """
        Invalidates a timer object, stopping it from ever firing again.
        """
        if self._engine:
            self._engine.timerq.invalidate(self)
            self._engine = None
    def is_valid(self):
        """
        Returns a boolean value that indicates whether an EngineTimer
        object is valid and able to fire.
        """
        return self._engine is not None
    def set_nextfire(self, fire_delay, interval=-1):
        """
        Set the next firing delay in seconds for an EngineTimer object.
        The optional parameter *interval* sets the firing interval
        of the timer. If not specified, the timer fires once and then
        is automatically invalidated.
        Time values are expressed in second using floating point
        values. Precision is implementation (and system) dependent.
        It is safe to call this method from the task owning this
        timer object, in any event handlers, anywhere.
        However, resetting a timer's next firing time may be a
        relatively expensive operation. It is more efficient to let
        timers autorepeat or to use this method from the timer's own
        event handler callback (ie. from its ev_timer).
        """
        if not self.is_valid():
            raise EngineIllegalOperationError("Operation on invalid timer.")
        self.fire_delay = fire_delay
        self.interval = interval
        self._engine.timerq.reschedule(self)
    def _fire(self):
        raise NotImplementedError("Derived classes must implement.")
class EngineTimer(EngineBaseTimer):
    Concrete class EngineTimer
    An EngineTimer object represents a timer bound to an engine that
    fires at a preset time in the future. Timers can fire either only
    once or repeatedly at fixed time intervals. Repeating timers can
    also have their next firing time manually adjusted.
    A timer is not a real-time mechanism; it fires when the task's
    underlying engine to which the timer has been added is running and
    able to check if the timer's firing time has passed.
    def __init__(self, fire_delay, interval, autoclose, handler):
        EngineBaseTimer.__init__(self, fire_delay, interval, autoclose)
        self.eh = handler
        assert self.eh is not None, "An event handler is needed for timer."
    def _fire(self):
        self.eh.ev_timer(self)
class _EngineTimerQ(object):
    class _EngineTimerCase(object):
        """
        Helper class that allows comparisons of fire times, to be easily used
        in an heapq.
        """
        def __init__(self, client):
            self.client = client
            self.client._timercase = self
            # arm timer (first time)
            assert self.client.fire_delay > -EPSILON
            self.fire_date = self.client.fire_delay + time.time()
        def __lt__(self, other):
            # NOTE: add @total_ordering decorator in Python 2.7+
            return self.fire_date < other.fire_date
        def __cmp__(self, other):
            # DEPRECATED: no longer used in Python 3
            return cmp(self.fire_date, other.fire_date)
        def arm(self, client):
            assert client is not None
            self.client = client
            self.client._timercase = self
            # setup next firing date
            time_current = time.time()
            if self.client.fire_delay > -EPSILON:
                self.fire_date = self.client.fire_delay + time_current
            else:
                interval = float(self.client.interval)
                assert interval > 0
                # Keep it simple: increase fire_date by interval even if
                # fire_date stays in the past, as in that case it's going to
                # fire again at next runloop anyway.
                self.fire_date += interval
                # Just print a debug message that could help detect issues
                # coming from a long-running timer handler.
                if self.fire_date < time_current:
                    LOGGER.debug("Warning: passed interval time for %r "
                                 "(long running event handler?)", self.client)
        def disarm(self):
            client = self.client
            client._timercase = None
            self.client = None
            return client
        def armed(self):
            return self.client is not None
    def __init__(self, engine):
        """
        Initializer.
        """
        self._engine = engine
        self.timers = []
        self.armed_count = 0
    def __len__(self):
        """
        Return the number of active timers.
        """
        return self.armed_count
    def schedule(self, client):
        """
        Insert and arm a client's timer.
        """
        # arm only if fire is set
        if client.fire_delay > -EPSILON:
            heapq.heappush(self.timers, _EngineTimerQ._EngineTimerCase(client))
            self.armed_count += 1
            if not client.autoclose:
                self._engine.evlooprefcnt += 1
    def reschedule(self, client):
        """
        Re-insert client's timer.
        """
        if client._timercase:
            self.invalidate(client)
            self._dequeue_disarmed()
            self.schedule(client)
    def invalidate(self, client):
        """
        Invalidate client's timer. Current implementation doesn't really remove
        the timer, but simply flags it as disarmed.
        """
        if not client._timercase:
            # if timer is being fire, invalidate its values
            client.fire_delay = -1.0
            client.interval = -1.0
            return
        if self.armed_count <= 0:
            raise ValueError("Engine client timer not found in timer queue")
        client._timercase.disarm()
        self.armed_count -= 1
        if not client.autoclose:
            self._engine.evlooprefcnt -= 1
    def _dequeue_disarmed(self):
        """
        Dequeue disarmed timers (sort of garbage collection).
        """
        while len(self.timers) > 0 and not self.timers[0].armed():
            heapq.heappop(self.timers)
    def fire_expired(self):
        """
        Remove expired timers from the queue and fire associated clients.
        """
        self._dequeue_disarmed()
        # Build a queue of expired timercases. Any expired (and still armed)
        # timer is fired, but only once per call.
        expired_timercases = []
        now = time.time()
        while self.timers and self.timers[0].fire_date <= now:
            expired_timercases.append(heapq.heappop(self.timers))
            self._dequeue_disarmed()
        for timercase in expired_timercases:
            # Be careful to recheck and skip any disarmed timers (eg. timer
            # could be invalidated from another timer's event handler)
            if not timercase.armed():
                continue
            # Disarm timer
            client = timercase.disarm()
            # Fire timer
            client.fire_delay = -1.0
            client._fire()
            # Rearm it if needed - Note: fire=0 is valid, interval=0 is not
            if client.fire_delay >= -EPSILON or client.interval > EPSILON:
                timercase.arm(client)
                heapq.heappush(self.timers, timercase)
            else:
                self.armed_count -= 1
                if not client.autoclose:
                    self._engine.evlooprefcnt -= 1
    def nextfire_delay(self):
        """
        Return next timer fire delay (relative time).
        """
        self._dequeue_disarmed()
        if len(self.timers) > 0:
            return max(0., self.timers[0].fire_date - time.time())
        return -1
    def clear(self):
        """
        Stop and clear all timers.
        """
        for timer in self.timers:
            if timer.armed():
                timer.client.invalidate()
        self.timers = []
        self.armed_count = 0
class Engine(object):
    Base class for ClusterShell Engines.
    Subclasses have to implement a runloop listening for client events.
    Subclasses that override other than "pure virtual methods" should call
    corresponding base class methods.
    identifier = "(none)"
    def __init__(self, info):
        """Initialize base class."""
        # take a reference on info dict
        self.info = info
        # and update engine id
        self.info['engine'] = self.identifier
        # keep track of all clients
        self._clients = set()
        self._ports = set()
        # keep track of the number of registered clients per worker
        # (this does not include ports)
        self._reg_stats = {}
        # keep track of registered file descriptors in a dict where keys
        # are fileno and values are (EngineClient, EngineClientStream) tuples
        self.reg_clifds = {}
        # fanout cache used to speed up client launch when fanout changed
        self._prev_fanout = 0    # fanout_diff != 0 the first time
        # Current loop iteration counter. It is the number of performed engine
        # loops in order to keep track of client registration epoch, so we can
        # safely process FDs by chunk and re-use FDs (see Engine._fd2client).
        self._current_loopcnt = 0
        # Current stream being processed
        self._current_stream = None
        # timer queue to handle both timers and clients timeout
        self.timerq = _EngineTimerQ(self)
        # reference count to the event loop (must include registered
        # clients and timers configured WITHOUT autoclose)
        self.evlooprefcnt = 0
        # running state
        self.running = False
        # runloop-has-exited flag
        self._exited = False
    def release(self):
        """Release engine-specific resources."""
        pass
    def clients(self):
        """Get a copy of clients set."""
        return self._clients.copy()
    def ports(self):
        """
        Get a copy of ports set.
        """
        return self._ports.copy()
    def _fd2client(self, fd):
        client, stream = self.reg_clifds.get(fd, (None, None))
        if client:
            if client._reg_epoch < self._current_loopcnt:
                return client, stream
            else:
                LOGGER.debug("_fd2client: ignoring just re-used FD %d",
                             stream.fd)
        return (None, None)
    def _can_register(self, client):
        assert not client.registered
        if not client.delayable or client.worker._fanout == FANOUT_UNLIMITED:
            return True
        elif client.worker._fanout is FANOUT_DEFAULT:
            return self._reg_stats.get('default', 0) < self.info['fanout']
        else:
            worker = client.worker
            return self._reg_stats.get(worker, 0) < worker._fanout
    def _update_reg_stats(self, client, offset):
        if client.worker._fanout is FANOUT_DEFAULT:
            key = 'default'
        else:
            key = client.worker
        self._reg_stats.setdefault(key, 0)
        self._reg_stats[key] += offset
    def add(self, client):
        """Add a client to engine."""
        # bind to engine
        client._set_engine(self)
        if client.delayable:
            # add to regular client set
            self._clients.add(client)
        else:
            # add to port set (non-delayable)
            self._ports.add(client)
        if self.running and self._can_register(client):
            # in-fly add if running
            self.register(client._start())
    def _remove(self, client, abort, did_timeout=False):
        """Remove a client from engine (subroutine)."""
        # be careful to also remove ports when engine has not started yet
        if client.registered or not client.delayable:
            if client.registered:
                self.unregister(client)
            # care should be taken to ensure correct closing flags
            client._close(abort=abort, timeout=did_timeout)
    def remove(self, client, abort=False, did_timeout=False):
        """
        Remove a client from engine. Does NOT aim to flush individual stream
        read buffers.
        """
        self._debug("REMOVE %s" % client)
        if client.delayable:
            self._clients.remove(client)
        else:
            self._ports.remove(client)
        self._remove(client, abort, did_timeout)
        # we just removed a client, so start pending client(s)
        self.start_clients()
    def remove_stream(self, client, stream):
        """
        Regular way to remove a client stream from engine, performing
        needed read flush as needed. If no more retainable stream
        remains for this client, this method automatically removes the
        entire client from engine.
        This function does nothing if the stream is not registered.
        """
        if stream.fd not in self.reg_clifds:
            LOGGER.debug("remove_stream: %s not registered", stream)
            return
        self.unregister_stream(client, stream)
        # _close_stream() will flush pending read buffers so may generate events
        client._close_stream(stream.name)
        # client may have been removed by previous events, if not check whether
        # some retained streams still remain
        if client in self._clients and not client.streams.retained():
            self.remove(client)
    def clear(self, did_timeout=False, clear_ports=False):
        """
        Remove all clients. Does not flush read buffers.
        Subclasses that override this method should call base class method.
        """
        all_clients = [self._clients]
        if clear_ports:
            all_clients.append(self._ports)
        for clients in all_clients:
            while len(clients) > 0:
                client = clients.pop()
                self._remove(client, True, did_timeout)
    def register(self, client):
        """
        Register an engine client. Subclasses that override this method
        should call base class method.
        """
        assert client in self._clients or client in self._ports
        assert not client.registered
        self._debug("REG %s (%s)(autoclose=%s)" % \
                (client.__class__.__name__, client.streams,
                 client.autoclose))
        client.registered = True
        client._reg_epoch = self._current_loopcnt
        if client.delayable:
            self._update_reg_stats(client, 1)
        # set interest event bits...
        for streams, ievent in ((client.streams.active_readers, E_READ),
                                (client.streams.active_writers, E_WRITE)):
            for stream in streams():
                self.reg_clifds[stream.fd] = client, stream
                stream.events |= ievent
                if not client.autoclose:
                    self.evlooprefcnt += 1
                self._register_specific(stream.fd, ievent)
        # start timeout timer
        self.timerq.schedule(client)
    def unregister_stream(self, client, stream):
        """Unregister a stream from a client."""
        self._debug("UNREG_STREAM stream=%s" % stream)
        assert stream is not None and stream.fd is not None
        assert stream.fd in self.reg_clifds, \
            "stream fd %d not registered" % stream.fd
        assert client.registered
        self._unregister_specific(stream.fd, stream.events & stream.evmask)
        self._debug("UNREG_STREAM unregistering stream fd %d (%d)" % \
                    (stream.fd, len(client.streams)))
        stream.events &= ~stream.evmask
        del self.reg_clifds[stream.fd]
        if not client.autoclose:
            self.evlooprefcnt -= 1
    def unregister(self, client):
        """Unregister a client"""
        # sanity check
        assert client.registered
        self._debug("UNREG %s (%s)" % (client.__class__.__name__, \
                client.streams))
        # remove timeout timer
        self.timerq.invalidate(client)
        # clear interest events...
        for streams, ievent in ((client.streams.active_readers, E_READ),
                                (client.streams.active_writers, E_WRITE)):
            for stream in streams():
                if stream.fd in self.reg_clifds:
                    self._unregister_specific(stream.fd, stream.events & ievent)
                    stream.events &= ~ievent
                    del self.reg_clifds[stream.fd]
                    if not client.autoclose:
                        self.evlooprefcnt -= 1
        client.registered = False
        if client.delayable:
            self._update_reg_stats(client, -1)
    def modify(self, client, sname, setmask, clearmask):
        """Modify the next loop interest events bitset for a client stream."""
        self._debug("MODEV set:0x%x clear:0x%x %s (%s)" % (setmask, clearmask,
        stream = client.streams[sname]
        stream.new_events &= ~clearmask
        stream.new_events |= setmask
        if self._current_stream is not stream:
            # modifying a non processing stream, apply new_events now
            self.set_events(client, stream)
    def _register_specific(self, fd, event):
        """Engine-specific register fd for event method."""
        raise NotImplementedError("Derived classes must implement.")
    def _unregister_specific(self, fd, ev_is_set):
        """Engine-specific unregister fd method."""
        raise NotImplementedError("Derived classes must implement.")
    def _modify_specific(self, fd, event, setvalue):
        """Engine-specific modify fd for event method."""
        raise NotImplementedError("Derived classes must implement.")
    def set_events(self, client, stream):
        """Set the active interest events bitset for a client stream."""
        self._debug("SETEV new_events:0x%x events:0x%x for %s[%s]" % \
            (stream.new_events, stream.events, client, stream.name))
        if not client.registered:
            LOGGER.debug("set_events: client %s[%s] not registered",
                         client, stream.name)
            return
        chgbits = stream.new_events ^ stream.events
        if chgbits == 0:
            return
        # configure interest events as appropriate
        for interest in (E_READ, E_WRITE):
            if chgbits & interest:
                assert stream.evmask & interest
                status = stream.new_events & interest
                self._modify_specific(stream.fd, interest, status)
                if status:
                    stream.events |= interest
                else:
                    stream.events &= ~interest
        stream.new_events = stream.events
    def set_reading(self, client, sname):
        """Set client reading state."""
        # listen for readable events
        self.modify(client, sname, E_READ, 0)
    def set_writing(self, client, sname):
        """Set client writing state."""
        # listen for writable events
        self.modify(client, sname, E_WRITE, 0)
    def add_timer(self, timer):
        """Add a timer instance to engine."""
        timer._set_engine(self)
        self.timerq.schedule(timer)
    def remove_timer(self, timer):
        """Remove engine timer from engine."""
        self.timerq.invalidate(timer)
    def fire_timers(self):
        """Fire expired timers for processing."""
        # Only fire timers if runloop is still retained
        if self.evlooprefcnt > 0:
            # Fire once any expired timers
            self.timerq.fire_expired()
    def start_ports(self):
        """Start and register all port clients."""
        # Ports are special, non-delayable engine clients
        for port in self._ports:
            if not port.registered:
                self._debug("START PORT %s" % port)
                self.register(port._start())
    def start_clients(self):
        """Start and register regular engine clients in respect of fanout."""
        # check if engine fanout has changed
        # NOTE: worker._fanout live changes not supported (see #323)
        fanout_diff = self.info['fanout'] - self._prev_fanout
        if fanout_diff:
            self._prev_fanout = self.info['fanout']
        for client in self._clients:
            if not client.registered and self._can_register(client):
                self._debug("START CLIENT %s" % client.__class__.__name__)
                self.register(client._start())
                # if first time or engine fanout has changed, we do a full scan
                if fanout_diff == 0:
                    # if engine fanout has not changed, we only start 1 client
    def run(self, timeout):
        """Run engine in calling thread."""
        # change to running state
        if self.running:
            raise EngineAlreadyRunningError()
        try:
            self.running = True
            # start port clients
            self.start_ports()
            # peek in ports for early pending messages
            self.snoop_ports()
            # start all other clients
            self.start_clients()
            # run loop until all clients and timers are removed
            self.runloop(timeout)
        except EngineTimeoutException:
            self.clear(did_timeout=True)
            raise
        except: # MUST use BaseException as soon as possible (py2.5+)
            # The game is over.
            exc_t, exc_val, exc_tb = sys.exc_info()
            try:
                # Close Engine clients
                self.clear()
            except:
                # self.clear() may still generate termination events that
                # may raises exceptions, overriding the other one above.
                # In the future, we should block new user events to avoid
                # that. Also, such cases could be better handled with
                # BaseException. For now, print a backtrace in debug to
                # help detect the problem.
                tbexc = traceback.format_exception(exc_t, exc_val, exc_tb)
                LOGGER.debug(''.join(tbexc))
                raise
            raise
        finally:
            # cleanup
            self.timerq.clear()
            self.running = False
            self._prev_fanout = 0
    def snoop_ports(self):
        """
        Peek in ports for possible early pending messages.
        This method simply tries to read port pipes in non-blocking mode.
        """
        # make a copy so that early messages on installed ports may
        # lead to new ports
        ports = self._ports.copy()
        for port in ports:
            try:
                port._handle_read('in')
            except (IOError, OSError) as ex:
                if ex.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
                    # no pending message
                    return
                # raise any other error
                raise
    def runloop(self, timeout):
        """Engine specific run loop. Derived classes must implement."""
        raise NotImplementedError("Derived classes must implement.")
    def abort(self, kill):
        """Abort runloop."""
        if self.running:
            raise EngineAbortException(kill)
        self.clear(clear_ports=kill)
    def exited(self):
        """Returns True if the engine has exited the runloop once."""
        return not self.running and self._exited
    def _debug(self, s):
        """library engine verbose debugging hook"""
        #LOGGER.debug(s)
        pass
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

Engine.py

Latest commit

History

Engine.py

File metadata and controls
