Python uvloop 模块，EventLoopPolicy() 实例源码

def get_loop_and_logger(self, test_type):
        self.route_data = {
            "type": "async",
            "path": "/hello-sync-private",
            "image": "iron/hello",
            "is_public": "false"
        }
        try:
            testloop = asyncio.get_event_loop()
        except Exception:
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
            testloop = asyncio.get_event_loop()

        logger = log.UnifiedLogger(
            log_to_console=False,
            filename=("./picasso-{}-tests-run-{}.log"
                      .format(test_type, datetime.datetime.now())),
            level="DEBUG").setup_logger(__package__)
        return testloop, logger

def run(self, app, host, port):
        asyncio.get_event_loop().close()
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

        loop = asyncio.get_event_loop()

        loop.add_signal_handler(signal.SIGQUIT, self.handle_exit, signal.SIGQUIT, None)
        loop.add_signal_handler(signal.SIGTERM, self.handle_exit, signal.SIGTERM, None)
        loop.add_signal_handler(signal.SIGINT, self.handle_exit, signal.SIGINT, None)
        loop.add_signal_handler(signal.SIGABRT, self.handle_exit, signal.SIGABRT, None)

        loop.create_task(self.create_server(loop, app, host, port))
        loop.create_task(self.tick(loop))

        logger.warning('Starting worker [{}] serving at: {}:{}'.format(os.getpid(), host, port))

        loop.run_forever()

def start_server(host='127.0.0.1', port=17000, db_name='db'):
    global db
    db = shelfdb.open(db_name)
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.get_event_loop()
    server = asyncio.start_server(handler, host, port, loop=loop)
    server = loop.run_until_complete(server)

    # Serve requests until Ctrl+C is pressed
    print('Serving on {}'.format(server.sockets[0].getsockname()))
    print('Database :', db_name)
    print('pid :', os.getpid())
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

    # Close the server
    server.close()
    db.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

def handle(self, *args, **options):
        if not name is 'nt':
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        loop = asyncio.get_event_loop()

        url_status_checker(loop=loop)
        feed_status_checker(loop=loop)
        clean_images_from_db(loop=loop)
        clean_images_from_folder(loop=loop)
        img_resizer(loop=loop)

        loop.close()

        make_category_thumbs()

        self.stdout.write(self.style.SUCCESS('Successfully done parsing jobs'))

def main():
    config_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.home() / '.config' / 'whereisit.toml'
    with open(config_path) as f:
        config = toml.load(f)

    import logging
    logging.getLogger('aiohttp.client').setLevel(logging.ERROR)

    db_path = Path.home() / '.local' / 'share' / config['database']['path']
    orm.sql_debug(config['database'].get('debug', False))
    database.bind("sqlite", str(db_path), create_db=True)
    database.generate_mapping(create_tables=True)

    with orm.db_session():
        orm.select(t for t in database.Tracking
                   if t.id not in list(config['trackings'].keys())).delete(bulk=True)

    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    with contextlib.closing(asyncio.get_event_loop()) as loop:
        tracker = Tracker(loop=loop, db=database, config=config)
        loop.create_task(tracker.run())
        loop.run_forever()

def web(wait):
    """
    Serve the application
    If the database doesn't already exist it will be created.
    """
    settings = Settings(sender_cls='app.worker.Sender')
    print(settings.to_string(True), flush=True)
    setup_logging(settings)

    logger.info('waiting for elasticsearch and redis to come up...')
    # give es a chance to come up fully, this just prevents lots of es errors, create_indices is itself lenient

    # skip wait as es and redis are generally already up and delay is causing missed requests
    # wait and sleep(4)
    _check_services_ready(settings)

    _elasticsearch_setup(settings)
    logger.info('starting server...')
    asyncio.get_event_loop().close()
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.get_event_loop()
    app = create_app(loop, settings)
    run_app(app, port=8000, loop=loop, print=lambda v: None, access_log=None)

def create_http_server(config):
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    import noir.router

    for service_name in config.services:
        importlib.import_module(service_name)

    srv = web.Server(
        functools.partial(server_handler, 
        (config.parse_request or default_parse_request, config.prepare_response or default_prepare_response)),
        tcp_keepalive=config.keep_alive,
        keepalive_timeout=config.keep_alive_timeout)

    loop = asyncio.get_event_loop()

    f = loop.create_server(srv, '0.0.0.0', config.port, reuse_port=True)
    t = loop.run_until_complete(f)

    logger.info('server on %s', t.sockets[0].getsockname())
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        loop.run_until_complete(srv.shutdown())
    loop.close()

def get_event_loop(debug=False):
    if not debug:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        asyncio.set_event_loop(uvloop.new_event_loop())
    return asyncio.get_event_loop()

def get_event_loop(self, debug=False):
        if not debug:
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
            asyncio.set_event_loop(uvloop.new_event_loop())
        AsyncIOMainLoop().install()
        return asyncio.get_event_loop()

def init_process(self):
        # Close any existing event loop before setting a
        # new policy.
        asyncio.get_event_loop().close()

        # Setup uvloop policy, so that every
        # asyncio.get_event_loop() will create an instance
        # of uvloop event loop.
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

        super().init_process()

def handle(self, *args, **options):
        if not name is 'nt':
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        loop = asyncio.get_event_loop()

        posts = Post.objects.all()
        loop.run_until_complete(asyncio.gather(*[processs_content(post=post) \
            for post in posts], return_exceptions=True))

        loop.close()

        self.stdout.write(self.style.SUCCESS('Successfully extracted keywords for posts'))

def handle(self, *args, **options):
        if not name is 'nt':
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        loop = asyncio.get_event_loop()

        feed_status_checker(loop=loop)
        parse_all_feeds(loop=loop)
        banner()
        title_cleaner_from_db(loop=loop)
        update_db_with_cleaned_content(loop=loop)

        loop.close()

        self.stdout.write(self.style.SUCCESS('Successfully done parsing jobs'))

def handle(self, *args, **options):
        if not name is 'nt':
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        loop = asyncio.get_event_loop()

        posts = Post.objects.all()
        loop.run_until_complete(asyncio.gather(*[processs_content(post=post) \
            for post in posts], return_exceptions=True))

        loop.close()

        self.stdout.write(self.style.SUCCESS('Successfully extracted meaning for posts'))

def main(use_uvloop):
    """Executes the main bot."""
    if use_uvloop:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    bot = init_bot()
    logger.info('Init: Initializing event loop')
    loop = asyncio.get_event_loop()
    logger.info('Init: Starting bot!')
    runbot(loop, bot)
    return bot.is_restart

def update_all():
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.get_event_loop()
    task = asyncio.ensure_future(update_all_books(loop=loop))
    loop.run_until_complete(task)
    return task.result() or None

def async_callback(func, **kwargs):
    """
    Call the asynchronous function
    :param func: a async function
    :param kwargs: params
    :return: result
    """
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.get_event_loop()
    task = asyncio.ensure_future(func(**kwargs))
    loop.run_until_complete(task)
    return task.result()

def serve(reuse_port=False):
    asyncio.get_event_loop().close()
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    def proto_factory():
        resolver = DictResolver(foo='http://localhost:8081')
        http_dispatcher = HttpDispatcher(resolver)
        # rabbit_dispatcher = RabbitDispatcher()
        return GatewayProtocol(loop, dispatcher=http_dispatcher)

    srv_coro = loop.create_server(proto_factory, '0.0.0.0', 8080,
                                  reuse_port=reuse_port)
    srv = loop.run_until_complete(srv_coro)
    print('Listening on: ', srv.sockets[0].getsockname())
    loop.add_signal_handler(SIGINT, loop.stop)
    loop.add_signal_handler(SIGTERM, loop.stop)

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    finally:
        srv.close()
        loop.run_until_complete(srv.wait_closed())
        loop.close()

def __init__(self,
                 task_list: List[Task]=None,
                 config: Config=Config,
                 stats: Stats=Stats,
                 executor: Executor=None,
                 debug: bool=False) -> None:
        '''Initialize Tasky and automatically start a list of tasks.
        One of the following methods must be called on the resulting objects
        to start the event loop: `run_forever()`, `run_until_complete()`, or
        `run_for_time()`.'''

        if uvloop:
            Log.debug('using uvloop event loop')
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

        self.loop = asyncio.new_event_loop()
        self.loop.add_signal_handler(signal.SIGINT, self.sigint)
        self.loop.add_signal_handler(signal.SIGTERM, self.sigterm)
        self.loop.set_exception_handler(self.exception)
        asyncio.set_event_loop(self.loop)

        if debug:
            Log.debug('enabling asyncio debug mode')
            self.loop.set_debug(True)

        self.all_tasks = {}
        self.running_tasks = set()
        self.initial_tasks = list(task_list)

        self.configuration = config
        self.stats = stats
        self.executor = executor

        self.monitor = False
        self.terminate_on_finish = False
        self.stop_attempts = 0

def loop() -> Generator:
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop

def event_loop_policy(self):
        policy = self.default('EVENT_LOOP_POLICY', None)
        if policy is None:
            return None
        if policy == 'uvloop':
            import uvloop
            return uvloop.EventLoopPolicy()
        raise self.Error('unknown event loop policy "{}"'.format(policy))

def install_uvevent_loop():  # pragma: no cover
    """Install uvloop as default event loop when available.

    See: http://magic.io/blog/uvloop-blazing-fast-python-networking/

    """
    try:
        import uvloop
    except ImportError:
        pass
    else:
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

def init_process(self):
        import uvloop

        # Close any existing event loop before setting a
        # new policy.
        asyncio.get_event_loop().close()

        # Setup uvloop policy, so that every
        # asyncio.get_event_loop() will create an instance
        # of uvloop event loop.
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

        super().init_process()

def setUpClass(cls):
        if os.environ.get('USE_UVLOOP'):
            import uvloop
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)
        cls.loop = loop

def init(config):
    log.debug("Loglevel set to %s", logging.getLevelName(log.getEffectiveLevel()))
    asyncio.set_event_loop(None)
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)

    app = create_app(loop, config)
    ssl_context = create_ssl_context(app['config'])
    run_app(app, loop, ssl_context=ssl_context)

def init_uvloop():
    import asyncio
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

def default_event_loop(request=None, config=None):
    if request is not None:
        config = request.config
    loop = config.getoption("--loop")
    if loop == 'uvloop':
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    else:
        loop = 'asyncio'
    return loop

def main():
    parser = argparse.ArgumentParser(description='holosocket server')
    parser.add_argument('-c', '--config', help='config file')
    parser.add_argument('-4', '--ipv4', action='store_true', help='ipv4 only')
    parser.add_argument('--debug', action='store_true', help='debug mode')

    args = parser.parse_args()

    if args.config:
        with open(args.config, 'r') as f:
            config = yaml.load(f, Loader=Loader)

    if args.debug:
        LOGGING_MODE = logging.DEBUG
    else:
        LOGGING_MODE = logging.INFO

    logging.basicConfig(
        level=LOGGING_MODE,
        format='{asctime} {levelname} {message}',
        datefmt='%Y-%m-%d %H:%M:%S',
        style='{')

    if args.ipv4:
        SERVER = config['server']

    else:
        SERVER = (config['server'], '::')

    SERVER_PORT = config['server_port']
    KEY = config['password']
    try:
        DNS = config['dns']
    except KeyError:
        DNS = None

    try:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        logging.info('uvloop mode')
    except ImportError:
        logging.info('pure asyncio mode')

    loop = asyncio.get_event_loop()
    server = Server(KEY, nameservers=DNS)
    coro = asyncio.start_server(server.handle, SERVER, SERVER_PORT, loop=loop)
    server = loop.run_until_complete(coro)

    try:
        loop.run_forever()

    except KeyboardInterrupt:
        pass

    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

def main():
    parser = argparse.ArgumentParser(description='holosocket local')
    parser.add_argument('-c', '--config', help='config file')
    parser.add_argument('--debug', action='store_true', help='debug mode')

    args = parser.parse_args()

    if args.config:
        with open(args.config, 'r') as f:
            config = yaml.load(f, Loader=Loader)

    if args.debug:
        LOGGING_MODE = logging.DEBUG
    else:
        LOGGING_MODE = logging.INFO

    logging.basicConfig(
        level=LOGGING_MODE,
        format='{asctime} {levelname} {message}',
        datefmt='%Y-%m-%d %H:%M:%S',
        style='{')

    SERVER = config['server']
    try:
        V6_SERVER = config['v6_server']
    except KeyError:
        V6_SERVER = None

    SERVER_PORT = config['server_port']
    LOCAL = config['local']
    PORT = config['local_port']
    KEY = config['password']

    try:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        logging.info('uvloop mode')
    except ImportError:
        logging.info('pure asyncio mode')

    loop = asyncio.get_event_loop()
    server = Server(SERVER, V6_SERVER, SERVER_PORT, KEY)
    coro = asyncio.start_server(server.handle, LOCAL, PORT, loop=loop)
    server = loop.run_until_complete(coro)

    try:
        loop.run_forever()

    except KeyboardInterrupt:
        pass

    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

def server(host, port, db_uri,
           keystone_endpoint,
           functions_url,
           log_level,
           log_file,
           debug,
           ):
    """
    Starts Picasso API service
    """
    logger = log.UnifiedLogger(
        log_to_console=True if not log_file else False,
        filename=None if not log_file else log_file,
        level=log_level).setup_logger(__package__)

    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.get_event_loop()

    parts = parse.urlparse(functions_url)

    fnclient = config.FunctionsClient(
        parts.hostname,
        api_port=parts.port,
        api_protocol=parts.scheme,
        api_version=parts.path[1:]
    )
    loop.run_until_complete(fnclient.ping(loop=loop))
    connection_pool = config.Connection(db_uri, loop=loop)

    config.Config(
        auth_url=keystone_endpoint,
        functions_client=fnclient,
        logger=logger,
        connection=connection_pool,
        event_loop=loop,
    )

    API(
        host=host, port=port, loop=loop,
        logger=logger, debug=debug
    ).apply_swagger(
        swagger_url="/api",
        description="Picasso API service docs",
        api_version="v1.0.0",
        title="Picasso API",
    ).initialize()

def run(self):
        """
        Run the program.  This is the main entrypoint to the magnate client
        """
        # Create the statedir if it doesn't exist
        if not os.path.exists(self.cfg['state_dir']):
            os.makedirs(self.cfg['state_dir'])

        twiggy_addon.dict_config(self.cfg['logging'])

        ui_plugins = load('magnate.ui', subclasses=UserInterface)
        for UIClass in ui_plugins:  #pylint: disable=invalid-name
            if UIClass.__module__.startswith('magnate.ui.{}'.format(self.cfg['ui_plugin'])):
                break
        else:
            print('Unknown user ui: {}'.format(self.cfg['ui_plugin']))
            return 1

        # Try using uvloop instead of the asyncio event loop
        if self.cfg['use_uvloop']:
            try:
                import uvloop
            except:
                print('Could not import uvloop.  Falling back on asyncio event loop')
            try:
                asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
            except:
                print('Could not set uvloop to be the event loop.  Falling back on asyncio event loop')

        loop = asyncio.get_event_loop()
        self.pubpen = PubPen(loop)
        self._setup_markets()
        self.dispatcher = Dispatcher(self, self.markets)

        # UIClass is always available because we'd have already returned (via
        # the for-else) if UIClass was not defined
        try:
            user_interface = UIClass(self.pubpen, self.cfg['ui_args']) #pylint: disable=undefined-loop-variable
            return user_interface.run()
        except Exception as e:
            log.trace('error').error('Exception raised while running the user interface')
            raise

def __init__(self, name, path=None,
                 template_inputs=None,
                 logger=None,
                 event_loop=None,
                 enable_rollback=False):
        """
        Represents AIOrchestra deployment context designed to
        manage deployment through its lifecycle

        :param name: deployment context name
        :type name: str
        :param path: path to TOSCA template
        :type path: str
        :param template_inputs: TOSCA template input parameters
        :type template_inputs: dict
        :param logger: python logger instance
        :param event_loop: asyncio or any compatible event loop
        :type event_loop: asyncio.Loop
        :param enable_rollback: weather to enable rollback on failure or not
        """
        self.__name = name
        self._tmplt = tosca_template.ToscaTemplate(
            path=path, a_file=True,
            parsed_params=template_inputs)
        self.__path = path
        self.origin_nodes = self._tmplt.graph.nodetemplates
        self.vertices = self._tmplt.graph.vertices
        self.inputs_definitions = self._tmplt.inputs
        self.__outputs = self._tmplt.outputs
        self.template_inputs = template_inputs if template_inputs else {}
        self.__status = self.PENDING
        if not logger:
            self.logger = log.UnifiedLogger(
                log_to_console=True,
                level="DEBUG").setup_logger(__name__)
        else:
            self.logger = logger
        if not event_loop:
            if uvloop:
                uv_loop = uvloop.new_event_loop()
                asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
                asyncio.set_event_loop(uv_loop)
            self.event_loop = asyncio.get_event_loop()
        else:
            self.event_loop = event_loop
        self.__orchestra_nodes = [node.OrchestraNode(self, origin_node)
                                  for origin_node in self.origin_nodes]
        self.__deployment_plan = None
        self.rollback_enabled = enable_rollback