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

def spawn_after(seconds, *args, **kwargs):
        raise_error = kwargs.pop('raise_error', False)

        def _launch(func, *args, **kwargs):
            # Mimic gevent's default raise_error=False behaviour
            # by not propagating an exception to the joiner.
            try:
                return func(*args, **kwargs)
            except TaskExit:
                pass
            except:
                if raise_error:
                    raise
                # Log uncaught exception.
                # Note: this is an intentional divergence from gevent
                # behaviour; gevent silently ignores such exceptions.
                LOG.error('hub: uncaught exception: %s',
                          traceback.format_exc())

        return eventlet.spawn_after(seconds, _launch, *args, **kwargs)

def test_soon_spawn(self):
        result = []

        def func1():
            result.append("spawn")

        def func2():
            result.append("spawn_after")
            self.loop.stop()

        def schedule_greenthread():
            eventlet.spawn(func1)
            eventlet.spawn_after(0.010, func2)

        self.loop.call_soon(schedule_greenthread)
        self.loop.run_forever()
        self.assertEqual(result, ["spawn", "spawn_after"])

def init_gui(self):
        # Create the root window
        self.root = tkinter.Tk()
        self.root.geometry('350x75+490+550')
        self.root.title('Hybrid driver')

        # Create a label with status
        self.status = tkinter.StringVar()
        label = tkinter.Label(self.root, width=350, height=75,
                              textvariable=self.status)
        label.pack(fill=tkinter.BOTH, expand=1)

        # Bind key event handlers
        self.root.bind('<Left>', lambda e: self.turn_left())
        self.root.bind('<Right>', lambda e: self.turn_right())
        self.root.bind('<Up>', lambda e: self.speed_up())
        self.root.bind('<Down>', lambda e: self.slow_down())
        self.root.bind('<Key>', self.keydown)

        self.update_status()

        # Start UI loop
        eventlet.spawn_after(1, self.main_loop)

def init_gui(self):
        # Create the root window
        self.root = tkinter.Tk()
        self.root.geometry('350x75+490+550')
        self.root.title('SDC Live Trainer')

        # Create a label with status
        self.status = tkinter.StringVar()
        label = tkinter.Label(self.root, width=350, height=75,
                              textvariable=self.status)
        label.pack(fill=tkinter.BOTH, expand=1)

        # Bind key event handlers
        self.root.bind('<Left>', lambda e: self.turn_left())
        self.root.bind('<Right>', lambda e: self.turn_right())
        self.root.bind('<Up>', lambda e: self.speed_up())
        self.root.bind('<Down>', lambda e: self.slow_down())
        self.root.bind('<Key>', self.keydown)

        self.update_status()

        # Start UI loop
        eventlet.spawn_after(1, self.main_loop)

def init_gui(self):
        # Create the root window
        self.root = tkinter.Tk()
        self.root.geometry('350x75+490+550')
        self.root.title('Manual driver')

        # Create a label with status
        self.status = tkinter.StringVar()
        label = tkinter.Label(self.root, width=350, height=75,
                              textvariable=self.status)
        label.pack(fill=tkinter.BOTH, expand=1)

        # Bind key event handlers
        self.root.bind('<Left>', lambda e: self.turn_left())
        self.root.bind('<Right>', lambda e: self.turn_right())
        self.root.bind('<Up>', lambda e: self.speed_up())
        self.root.bind('<Down>', lambda e: self.slow_down())
        self.root.bind('<Key>', self.keydown)

        self.update_status()

        # Start UI loop
        eventlet.spawn_after(1, self.main_loop)

def spawn_after(seconds, *args, **kwargs):
        raise_error = kwargs.pop('raise_error', False)

        def _launch(func, *args, **kwargs):
            # Mimic gevent's default raise_error=False behaviour
            # by not propagating an exception to the joiner.
            try:
                return func(*args, **kwargs)
            except TaskExit:
                pass
            except BaseException as e:
                if raise_error:
                    raise e
                # Log uncaught exception.
                # Note: this is an intentional divergence from gevent
                # behaviour; gevent silently ignores such exceptions.
                LOG.error('hub: uncaught exception: %s',
                          traceback.format_exc())

        return eventlet.spawn_after(seconds, _launch, *args, **kwargs)

def install_update(self, old_flows, new_flows):
        self.no_of_received_msgs = 0
        self.current_start_time = time() * 1000
        new_msgs = self.handler.do_install_update(old_flows, new_flows, self.test_number, self.skip_deadlock)
        # self.log.info(new_msgs)
        self.current_sending_time = time() * 1000
        self.current_computation_time = self.current_sending_time - self.current_start_time
        self.log.debug("delay from ctrl to sw: %s" % global_vars.sw_to_ctrl_delays)

        pickled_msgs = []
        for new_msg in new_msgs:
            new_msg.computation_time_in_ctrl = self.current_computation_time
            str_message = pickle.dumps(new_msg, pickle.HIGHEST_PROTOCOL)
            pickled_msgs.append(str_message)

        c = 0
        for new_msg in new_msgs:
            # self.send_to_switch(new_msg, pickled_msgs[c])
            # latency = global_vars.sw_to_ctrl_delays[new_msg.dst_id]/1000
            eventlet.spawn_after(0, self.send_to_switch, new_msg, pickled_msgs[c])
            c += 1

def respond_with(gn_event_name=None, should_disconnect=False, emit_response=True):
    def factory(view_func):
        @wraps(view_func)
        def decorator(*args, **kwargs):
            tb = None
            try:
                status_code, data = view_func(*args, **kwargs)
            except Exception as e:
                environ.env.stats.incr(gn_event_name + '.exception')
                tb = traceback.format_exc()
                logger.error('%s: %s' % (gn_event_name, str(e)))
                environ.env.capture_exception(sys.exc_info())

                if should_disconnect and environ.env.config.get(ConfigKeys.DISCONNECT_ON_FAILED_LOGIN, False):
                    eventlet.spawn_after(seconds=1, func=_delayed_disconnect, sid=environ.env.request.sid)
                return 500, str(e)
            finally:
                if tb is not None:
                    logger.exception(tb)

            if status_code != 200:
                logger.warning('in decorator, status_code: %s, data: %s' % (status_code, str(data)))
                if should_disconnect and environ.env.config.get(ConfigKeys.DISCONNECT_ON_FAILED_LOGIN, False):
                    eventlet.spawn_after(seconds=1, func=_delayed_disconnect, sid=environ.env.request.sid)

            # in some cases the callback is enough
            if emit_response:
                response_message = environ.env.response_formatter(status_code, data)
                environ.env.emit(gn_event_name, response_message)

            return status_code, None
        return decorator
    return factory

def _run(self, application):
        LOG.debug("Running scrubber application")
        self.daemon_pool.spawn_n(application.run, self.event)
        eventlet.spawn_after(self.wakeup_time, self._run, application)
        LOG.debug("Next run scheduled in %s seconds", self.wakeup_time)

def suspect_deadlock_on_link(self, link):
        if len(link.to_adds) + len(link.to_adds_loop) == 0:
            return
        suspecting_start = time()
        self.suspecting_deadlocks[(link.src, link.dst)] = suspecting_start
        link.scheduling_mode = constants.SUSPECTING_LINK
        # self.log.info("can_violate_congestion: %s" % self.can_violate_congestion)
        # self.log.info("segments to be done %s" % self.segments_to_be_done)
        if not self.can_violate_congestion:
            eventlet.spawn_after(constants.SUSPECTING_TIME, self.suspecting_time_expire, (link.src, link.dst))
        else:
            self.suspecting_deadlock_for_this_test = True
            eventlet.spawn_after(constants.SUSPECTING_TIME_SKIP_DEADLOCK, self.skipping_deadlock)

def work_with_deadlock_link(self, link):
        update_infos = {}
        affected_links = deque([])
        if self.check_and_execute_all_updates_on_link(link, update_infos, affected_links) <= 0:
            eventlet.spawn_after(constants.SUSPECTING_TIME_SKIP_DEADLOCK, self.skipping_deadlock)
            # self.log.info("wait to violate congestion on link %d->%d: current_avail %f"
            #               % (link.src, link.dst, link.avail_cap))
            # self.log.info("segments to be done: %s" % self.segments_to_be_done)
            self.suspecting_deadlock_for_this_test = True
            return update_infos, affected_links
        elif link.scheduling_mode == constants.SPLITTING_LINK:
            eventlet.spawn_after(constants.SUSPECTING_TIME, self.work_with_deadlock_link, link)
        else:
            self.unsuspecting_link(link)
        return update_infos, affected_links

def delete_ephemeral_rooms(gn_env: GNEnvironment):
    from activitystreams import parse as as_parser

    if len(gn_env.config) == 0 or gn_env.config.get(ConfigKeys.TESTING, False):
        # assume we're testing
        return

    def delete():
        from dino import utils

        channel_dict = gn_env.db.get_channels()

        for channel_id, _ in channel_dict.items():
            rooms = gn_env.db.rooms_for_channel(channel_id)

            for room_id, room_info in rooms.items():
                short_id = room_id.split('-')[0]
                room_name = room_info['name']
                logger.debug('checking room %s: %s' % (room_id, room_name))

                users = gn_env.db.users_in_room(room_id)
                if len(users) > 0:
                    logger.debug('[%s] NOT removing room (%s), has % user(s) in it' % (short_id, room_name, len(users)))
                    continue

                if not room_info['ephemeral']:
                    logger.debug('[%s] NOT removing room (%s), not ephemeral' % (short_id, room_name))
                    continue

                logger.info('[%s] removing ephemeral room (%s)' % (short_id, room_name))

                try:
                    gn_env.db.get_room_name(room_id)
                except NoSuchRoomException:
                    logger.info('[%s] ephemeral room (%s) has already been removed' % (short_id, room_name))
                    continue

                activity = utils.activity_for_remove_room('0', 'server', room_id, room_name, 'empty ephemeral room')

                gn_env.db.remove_room(channel_id, room_id)
                gn_env.out_of_scope_emit(
                    'gn_room_removed', activity, broadcast=True, include_self=True, namespace='/ws')
                gn_env.observer.emit('on_remove_room', (activity, as_parser(activity)))

    import eventlet
    eventlet.spawn_after(seconds=5*60, func=delete)