Python eventlet 模块，event() 实例源码

def process_secgroup_after(self, resource, event, trigger, **kwargs):
        """Callback for handling security group/rule commit-complete events

        This is when we should tell other things that a change has
        happened and has been recorded permanently in the DB.
        """
        # In Liberty, this is the only callback that's called.
        # We use our own event names, which will identify AFTER_*
        # events as the right time to commit, so in this case we
        # simply call the commit function ourselves.

        # This is not perfect - since we're not committing in one
        # transaction we can commit the secgroup change but fail to
        # propagate it to the journal and from there  to etcd on a
        # crash.  It's all we can do for Liberty as it doesn't support
        # in-transaction precommit events.
        if not PRECOMMIT:
            self.process_secgroup_commit(resource, event, trigger, **kwargs)

        # Whatever the object that caused this, we've put something
        # in the journal and now need to nudge the communicator
        self.kick()

def __init__(self):
        super(EtcdAgentCommunicator, self).__init__()

        self.etcd_client = etcd.Client()  # TODO(ijw): give this args

        # We need certain directories to exist
        self.do_etcd_mkdir(LEADIN + '/state')
        self.do_etcd_mkdir(LEADIN + '/nodes')

        # TODO(ijw): .../state/<host> lists all known hosts, and they
        # heartbeat when they're functioning

        # Get the physnets the agents know about.  This is updated
        # periodically in the return thread below.
        self.physical_networks = set()
        self._find_physnets()

        self.db_q_ev = eventlet.event.Event()

        self.return_thread = eventlet.spawn(self._return_worker)
        self.forward_thread = eventlet.spawn(self._forward_worker)

def __init__(self):
            self._ev = eventlet.event.Event()
            self._cond = False

def _broadcast(self):
            self._ev.send()
            # Since eventlet Event doesn't allow multiple send() operations
            # on an event, re-create the underlying event.
            # Note: _ev.reset() is obsolete.
            self._ev = eventlet.event.Event()

def start_threads(self, resource, event, trigger):
        LOG.debug('Starting background threads for Neutron worker')
        self.return_thread = self.make_return_worker()
        self.forward_thread = self.make_forward_worker()

def __init__(self):
            self._ev = eventlet.event.Event()
            self._cond = False

def _broadcast(self):
            self._ev.send()
            # Since eventlet Event doesn't allow multiple send() operations
            # on an event, re-create the underlying event.
            # Note: _ev.reset() is obsolete.
            self._ev = eventlet.event.Event()

def _forward_worker(self):
        LOG.debug('forward worker begun')

        session = neutron_db_api.get_session()
        while True:
            try:
                def work(k, v):
                    LOG.debug('forward worker updating etcd key %s' % k)
                    if self.do_etcd_update(k, v):
                        return True
                    else:
                        os.sleep(1) # something went bad; breathe, in
                                    # case we end up in a tight loop
                        return False

                LOG.debug('forward worker reading journal')
                while db.journal_read(session, work):
                    pass
                LOG.debug('forward worker has emptied journal')

                # work queue is now empty.
                LOG.debug("ML2_VPP(%s): worker thread pausing" % self.__class__.__name__)
                # Wait to be kicked, or (in case of emergency) run every
                # few seconds in case another thread or process dumped
                # work and failed to process it
                try:
                    with eventlet.Timeout(PARANOIA_TIME) as t:
                        # Wait for kick
                        dummy = self.db_q_ev.wait()
                        # Clear the event - we will now process till
                        # we've run out of things in the backlog
                        # so any trigger lost in this gap is harmless
                        self.db_q_ev.reset()
                        LOG.debug("ML2_VPP(%s): worker thread kicked: %s" % (self.__class__.__name__, str(dummy)))
                except eventlet.Timeout:
                    LOG.debug("ML2_VPP(%s): worker thread suspicious of a long pause" % self.__class__.__name__)
                    pass
                LOG.debug("ML2_VPP(%s): worker thread active" % self.__class__.__name__)
            except Exception, e:
                # TODO(ijw): log exception properly
                LOG.error("problems in forward worker: %s", e)
                LOG.error(traceback.format_exc())
                # never quit
                #pass

    ######################################################################