def _init_asyncio_patch():
    """
    Select compatible event loop for Tornado 5+.
    As of Python 3.8, the default event loop on Windows is `proactor`,
    however Tornado requires the old default "selector" event loop.
    As Tornado has decided to leave this to users to set, MkDocs needs
    to set it. See https://github.com/tornadoweb/tornado/issues/2608.
    """
    if sys.platform.startswith("win") and sys.version_info >= (3, 8):
        import asyncio
        try:
            from asyncio import WindowsSelectorEventLoopPolicy
        except ImportError:
            pass  # Can't assign a policy which doesn't exist.
        else:
            if not isinstance(asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy):
                asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy()) 

def main():
    args = parse_args()
    if args.uvloop:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

    loop = asyncio.get_event_loop()
    task = loop.create_task(Benchmark(args).bench_simple())
    task.add_done_callback(lambda _, loop=loop: loop.stop())

    def signal_hndl(_task=task):
        _task.cancel()
    loop.add_signal_handler(signal.SIGTERM, signal_hndl)
    loop.add_signal_handler(signal.SIGINT, signal_hndl)

    try:
        loop.run_forever()
    finally:
        loop.close()
        if not task.cancelled():
            task.result() 

def main():
    args = parse_args()
    if args.uvloop:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

    loop = asyncio.get_event_loop()
    task = loop.create_task(Benchmark(args).bench_simple())
    task.add_done_callback(lambda _, loop=loop: loop.stop())

    def signal_hndl(_task=task):
        _task.cancel()
    loop.add_signal_handler(signal.SIGTERM, signal_hndl)
    loop.add_signal_handler(signal.SIGINT, signal_hndl)

    try:
        loop.run_forever()
    finally:
        loop.close()
        if not task.cancelled():
            task.result() 

def init(self, path_app_root, path_data):
        log.initialize()

        asyncio.set_event_loop_policy(tornado.platform.asyncio.AnyThreadEventLoopPolicy())

        cfg = tp_cfg()
        cfg.app_path = path_app_root
        cfg.static_path = os.path.join(path_app_root, 'static')
        cfg.template_path = os.path.join(path_app_root, 'view')
        cfg.res_path = os.path.join(path_app_root, 'res')

        cfg.data_path = path_data
        cfg.cfg_path = os.path.join(path_data, 'etc')
        cfg.log_path = os.path.join(path_data, 'log')

        self._cfg_file = os.path.join(cfg.cfg_path, 'web.ini')
        if not cfg.load(self._cfg_file):
            return False

        return True 

def asyncio_worker(
    config: Config, sockets: Optional[Sockets] = None, shutdown_event: Optional[EventType] = None
) -> None:
    app = load_application(config.application_path)

    shutdown_trigger = None
    if shutdown_event is not None:
        shutdown_trigger = partial(check_multiprocess_shutdown_event, shutdown_event, asyncio.sleep)

    if config.workers > 1 and platform.system() == "Windows":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())  # type: ignore

    _run(
        partial(worker_serve, app, config, sockets=sockets),
        debug=config.debug,
        shutdown_trigger=shutdown_trigger,
    ) 

def uvloop_worker(
    config: Config, sockets: Optional[Sockets] = None, shutdown_event: Optional[EventType] = None
) -> None:
    try:
        import uvloop
    except ImportError as error:
        raise Exception("uvloop is not installed") from error
    else:
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

    app = load_application(config.application_path)

    shutdown_trigger = None
    if shutdown_event is not None:
        shutdown_trigger = partial(check_multiprocess_shutdown_event, shutdown_event, asyncio.sleep)

    _run(
        partial(worker_serve, app, config, sockets=sockets),
        debug=config.debug,
        shutdown_trigger=shutdown_trigger,
    ) 

def event_loop(request):
    """Overwrite `pytest_asyncio` eventloop to fix Windows issue.

    Default implementation causes `NotImplementedError` on Windows with
    Python 3.8, because they changed default eventloop in 3.8.

    NOTE: We do the same thing in the `example/settings.py` because it
    imports (and fails) before we have a chance to invoke this fixture.
    So, we could avoid adding this fixture, but I feel it is better to
    keep the proper solution here as well.

    """
    if sys.platform == "win32" and sys.version_info.minor >= 8:
        asyncio.set_event_loop_policy(
            asyncio.WindowsSelectorEventLoopPolicy()  # pylint: disable=no-member
        )
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close() 

def test_it_logs_messages(self):
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        loop = asyncio.get_event_loop()

        async def test():
            reader = asyncio.StreamReader(loop=loop)
            protocol = asyncio.StreamReaderProtocol(reader)

            transport, _ = await loop.connect_read_pipe(
                lambda: protocol, self.read_pipe
            )

            logger = Logger.with_default_handlers()
            await logger.info("Xablau")

            logged_content = await reader.readline()
            self.assertEqual(logged_content, b"Xablau\n")

            transport.close()
            await logger.shutdown()

        loop.run_until_complete(test()) 

def main() -> int:
    print("Hello, World!")

    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

    loop = asyncio.get_event_loop()
    # Each client connection will create a new protocol instance
    coro = loop.create_server(RedisProtocol, "127.0.0.1", 7878)
    server = loop.run_until_complete(coro)

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

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

    return 0 

def _init_asyncio_patch():
    """
    Select compatible event loop for Tornado 5+.

    As of Python 3.8, the default event loop on Windows is `proactor`,
    however Tornado requires the old default "selector" event loop.
    As Tornado has decided to leave this to users to set, MkDocs needs
    to set it. See https://github.com/tornadoweb/tornado/issues/2608.
    """
    if sys.platform.startswith("win") and sys.version_info >= (3, 8):
        import asyncio
        try:
            from asyncio import WindowsSelectorEventLoopPolicy
        except ImportError:
            pass  # Can't assign a policy which doesn't exist.
        else:
            if not isinstance(asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy):
                asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy()) 

def main():
    args = parse_args()
    if args.uvloop:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

    loop = asyncio.get_event_loop()
    task = loop.create_task(Benchmark(args).bench_simple())
    task.add_done_callback(lambda _, loop=loop: loop.stop())

    def signal_hndl(_task=task):
        _task.cancel()
    loop.add_signal_handler(signal.SIGTERM, signal_hndl)
    loop.add_signal_handler(signal.SIGINT, signal_hndl)

    try:
        loop.run_forever()
    finally:
        loop.close()
        if not task.cancelled():
            task.result() 

def main():
    args = parse_args()
    if args.uvloop:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

    loop = asyncio.get_event_loop()
    task = loop.create_task(Benchmark(args).bench_simple())
    task.add_done_callback(lambda _, loop=loop: loop.stop())

    def signal_hndl(_task=task):
        _task.cancel()
    loop.add_signal_handler(signal.SIGTERM, signal_hndl)
    loop.add_signal_handler(signal.SIGINT, signal_hndl)

    try:
        loop.run_forever()
    finally:
        loop.close()
        if not task.cancelled():
            task.result() 

def __init__(self,instance,host="localhost",port=None,autoreload=False):
        super(WebServer, self).__init__()
        self.app=None
        self.instance=instance
        self.host=host
        self.autoreload=autoreload

        if port is not None:
            self.port = port

        while not isFree("localhost", self.port):
            self.port += 1

        self.instance._webserver=(self.host,self.port)

        try: # https://bugs.python.org/issue37373 FIX: tornado/py3.8 on windows
            if sys.platform == 'win32':
                asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
        except:
            pass 

def test_get_event_loop_returns_running_loop(self):
        class Policy(asyncio.DefaultEventLoopPolicy):
            def get_event_loop(self):
                raise NotImplementedError

        loop = None

        old_policy = asyncio.get_event_loop_policy()
        try:
            asyncio.set_event_loop_policy(Policy())
            loop = asyncio.new_event_loop()
            self.assertIs(asyncio._get_running_loop(), None)

            async def func():
                self.assertIs(asyncio.get_event_loop(), loop)
                self.assertIs(asyncio._get_running_loop(), loop)

            loop.run_until_complete(func())
        finally:
            asyncio.set_event_loop_policy(old_policy)
            if loop is not None:
                loop.close()

        self.assertIs(asyncio._get_running_loop(), None) 

def __init__(self, verbosity=0, max_tasks=512):
        """Constructor.

        Args:
            verbosity: set output verbosity: 0 (default) is none, 3 is debug
            max_tasks: the maximum number of tasks asyncio will queue (default 512)
        """
        self.tasks = []
        self.errors = []
        self.fqdn = []
        self.ignore_hosts = []
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        self.loop = asyncio.get_event_loop()
        self.resolver = aiodns.DNSResolver(loop=self.loop, rotate=True)
        self.sem = asyncio.BoundedSemaphore(max_tasks)
        self.max_tasks = max_tasks
        self.verbosity = verbosity
        self.logger = ConsoleLogger(verbosity) 

def check(self, src, value):
        policy = asyncio.get_event_loop_policy()
        loop = policy.new_event_loop()
        asyncio.set_event_loop_policy(ForbiddenEventLoopPolicy())
        try:
            loop.run_until_complete(self._check(src, value, loop))
        finally:
            loop.close()
            asyncio.set_event_loop_policy(policy) 

def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

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

        self.__started = False
        self.__tasks = {}
        self.__procs = {}
        self.loop = asyncio.get_event_loop() 

def pytest_configure(config):
    bins = config.getoption('--redis-server')[:]
    cmd = 'which redis-server'
    if not bins:
        with os.popen(cmd) as pipe:
            path = pipe.read().rstrip()
        assert path, (
            "There is no redis-server on your computer."
            " Please install it first")
        REDIS_SERVERS[:] = [path]
    else:
        REDIS_SERVERS[:] = bins

    VERSIONS.update({srv: _read_server_version(srv)
                     for srv in REDIS_SERVERS})
    assert VERSIONS, ("Expected to detect redis versions", REDIS_SERVERS)

    class DynamicFixturePlugin:
        @pytest.fixture(scope='session',
                        params=REDIS_SERVERS,
                        ids=format_version)
        def server_bin(self, request):
            """Common for start_server and start_sentinel
            server bin path parameter.
            """
            return request.param
    config.pluginmanager.register(DynamicFixturePlugin(), 'server-bin-fixture')

    if config.getoption('--uvloop'):
        try:
            import uvloop
        except ImportError:
            raise RuntimeError(
                "Can not import uvloop, make sure it is installed")
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) 

def compat_event_loop():
    """OS agnostic context manager for an event loop."""
    if sys.platform.startswith("win"):
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

    event_loop = asyncio.get_event_loop()

    if event_loop.is_closed():
        event_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(event_loop)

    yield event_loop

    event_loop.close() 

def use_loud_event_loop():
    from moler.asyncio_runner import LoudEventLoopPolicy
    loud_policy = LoudEventLoopPolicy()
    asyncio.set_event_loop_policy(loud_policy) 

def tearDown(self):
        _SCENARIO.clear()
        _FIXTURES.clear()
        _FIXTURES.update(self.oldsetup)
        asyncio.set_event_loop_policy(self.policy) 

def _start_tornado(self):
        if os.name == 'nt' and sys.version_info > (3, 7):
            import asyncio
            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
        log.info('Starting Tornado server on %s', _readable_listen_address(self.listen_address, self.listen_port))

        # Max Buffersize set to 200MB            )
        http_server = HTTPServer(WSGIContainer(self.app),
                                 max_buffer_size=209700000,
                                 ssl_options=self.ssl_args)
        http_server.listen(self.listen_port, self.listen_address)
        self.wsgiserver = IOLoop.current()
        self.wsgiserver.start()
        # wait for stop signal
        self.wsgiserver.close(True) 

def setUp(self):
        # Trigger a cleanup of the mapping so we start with a clean slate.
        AsyncIOLoop().close()
        # If we don't clean up after ourselves other tests may fail on
        # py34.
        self.orig_policy = asyncio.get_event_loop_policy()
        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy()) 

def tearDown(self):
        asyncio.get_event_loop().close()
        asyncio.set_event_loop_policy(self.orig_policy) 

def tearDown(self):
        asyncio.set_event_loop_policy(self.orig_policy)
        self.executor.shutdown() 

def run_policy_test(self, accessor, expected_type):
        # With the default policy, non-main threads don't get an event
        # loop.
        self.assertRaises(
            (RuntimeError, AssertionError), self.executor.submit(accessor).result
        )
        # Set the policy and we can get a loop.
        asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())
        self.assertIsInstance(self.executor.submit(accessor).result(), expected_type)
        # Clean up to silence leak warnings. Always use asyncio since
        # IOLoop doesn't (currently) close the underlying loop.
        self.executor.submit(lambda: asyncio.get_event_loop().close()).result() 

def try_setup_uvloop():
    if config.SOCKS5_HOST and config.SOCKS5_PORT:
        # socks mode is not compatible with uvloop
        return
    try:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        print_err("Found uvloop, using it for optimal performance")
    except ImportError:
        pass 

def main():
    if WINDOWS:
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

    try:
        asyncio.run(amain())
    except (KeyboardInterrupt, SystemExit) as e:
        rootlogger.info('Received %r', e) 

def run_worker(cls, cls_args, sockname):
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    with devmode.CoverageConfig.enable_coverage_if_requested():
        asyncio.run(worker(cls, cls_args, sockname)) 

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