Python asyncio 模块，SelectorEventLoop() 实例源码

def __init__(self, task_pool, loop=None):
        """
        :param onedrived.od_task.TaskPool task_pool:
        :param asyncio.SelectorEventLoop | None loop:
        """
        self._lock = threading.RLock()
        self.watch_descriptors = loosebidict()
        self.task_queue = []
        self.task_pool = task_pool
        self.notifier = _INotify()
        if loop is None:
            import asyncio
            self.loop = asyncio.get_event_loop()
        else:
            self.loop = loop
        self.loop.add_reader(self.notifier.fd, self.process_events)

def main():
    options.parse_command_line()

    if options.options.policy == 'select':
        selector = selectors.SelectSelector()
        loop = asyncio.SelectorEventLoop(selector)
        asyncio.set_event_loop(loop)

    AsyncIOMainLoop().install()

    setup_logging()

    app = TailSocketApplication()
    app.listen(options.options.port, address=options.options.ip)
    print("Starting server on http://{}:{}".format(
        options.options.ip, options.options.port))

    asyncio.get_event_loop().run_forever()

    return app

def main():
    """Main program.

    Parse arguments, set up event loop, run crawler, print report.
    """
    args = ARGS.parse_args()
    if not args.roots:
        print('Use --help for command line help')
        return

    log = Logger(args.level)

    if args.iocp:
        from asyncio.windows_events import ProactorEventLoop
        loop = ProactorEventLoop()
        asyncio.set_event_loop(loop)
    elif args.select:
        loop = asyncio.SelectorEventLoop()
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()

    roots = {fix_url(root) for root in args.roots}

    crawler = Crawler(log,
                      roots, exclude=args.exclude,
                      strict=args.strict,
                      max_redirect=args.max_redirect,
                      max_tries=args.max_tries,
                      max_tasks=args.max_tasks,
                      max_pool=args.max_pool,
                      )
    try:
        loop.run_until_complete(crawler.crawl())  # Crawler gonna crawl.
    except KeyboardInterrupt:
        sys.stderr.flush()
        print('\nInterrupted\n')
    finally:
        crawler.report()
        crawler.close()
        loop.close()

def create_event_loop(self):
            return asyncio.SelectorEventLoop()

def create_event_loop(self):
                return asyncio.SelectorEventLoop(
                    selectors.KqueueSelector())

            # kqueue doesn't support character devices (PTY) on Mac OS X older
            # than 10.9 (Maverick)

def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.EpollSelector())

def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.PollSelector())

    # Should always exist.

def setUp(self):
        self.loop = asyncio.SelectorEventLoop()
        self.set_event_loop(self.loop)

def setUp(self):
        self.loop = asyncio.SelectorEventLoop()
        self.set_event_loop(self.loop)

def main():
    """Main program.

    Parse arguments, set up event loop, run crawler, print report.
    """
    args = ARGS.parse_args()
    if not args.roots:
        print('Use --help for command line help')
        return

    log = Logger(args.level)

    if args.iocp:
        from asyncio.windows_events import ProactorEventLoop
        loop = ProactorEventLoop()
        asyncio.set_event_loop(loop)
    elif args.select:
        loop = asyncio.SelectorEventLoop()
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()

    roots = {fix_url(root) for root in args.roots}

    crawler = Crawler(log,
                      roots, exclude=args.exclude,
                      strict=args.strict,
                      max_redirect=args.max_redirect,
                      max_tries=args.max_tries,
                      max_tasks=args.max_tasks,
                      max_pool=args.max_pool,
                      )
    try:
        loop.run_until_complete(crawler.crawl())  # Crawler gonna crawl.
    except KeyboardInterrupt:
        sys.stderr.flush()
        print('\nInterrupted\n')
    finally:
        crawler.report()
        crawler.close()
        loop.close()

def _setup():
    selector = selectors.SelectSelector()
    loop = asyncio.SelectorEventLoop(selector)
    asyncio.set_event_loop(loop)

def broker(ctx, in_connect, in_bind, out_connect, out_bind):
    loop = LoopClass()
    context = azmq.Context(loop=loop)
    in_socket = context.socket(azmq.PULL)
    out_socket = context.socket(azmq.PUSH)

    if in_connect:
        click.echo("Incoming connecting to %s." % in_connect, err=True)
        in_socket.connect(in_connect)
    else:
        click.echo("Incoming binding on %s." % in_bind, err=True)
        in_socket.bind(in_bind)

    if out_connect:
        click.echo("Outgoing connecting to %s." % out_connect, err=True)
        out_socket.connect(out_connect)
    else:
        click.echo("Outgoing binding on %s." % out_bind, err=True)
        out_socket.bind(out_bind)

    async def run(context, in_socket, out_socket):
        click.echo("Broker started.", err=True)

        try:
            while True:
                msg = await in_socket.recv_multipart()
                await out_socket.send_multipart(msg)
        finally:
            click.echo("Broker stopped.", err=True)

    task = asyncio.ensure_future(
        run(context, in_socket, out_socket),
        loop=loop,
    )

    with allow_interruption((loop, context.close)):
        loop.run_until_complete(context.wait_closed())

    loop.close()

def event_loop():
    if sys.platform == 'win32':
        loop = asyncio.ProactorEventLoop()
    else:
        loop = asyncio.SelectorEventLoop()

    asyncio.set_event_loop(loop)

    yield loop

    tasks = [task for task in asyncio.Task.all_tasks() if not task.done()]

    if tasks:
        loop.run_until_complete(asyncio.wait_for(asyncio.wait(tasks), 5))

def check_event_loop():
    if sys.platform == 'win32' and isinstance(asyncio.get_event_loop(), asyncio.SelectorEventLoop):
        raise ValueError(
            'SelectorEventLoop is not supported on Windows, use asyncio.ProactorEventLoop instead.'
        )

def create_event_loop(self):
            return asyncio.SelectorEventLoop()

def create_event_loop(self):
                return asyncio.SelectorEventLoop(
                    selectors.KqueueSelector())

            # kqueue doesn't support character devices (PTY) on Mac OS X older
            # than 10.9 (Maverick)

def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.EpollSelector())

def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.PollSelector())

    # Should always exist.

def create_event_loop(self):
            return asyncio.SelectorEventLoop(selectors.SelectSelector())

def setUp(self):
        self.loop = asyncio.SelectorEventLoop()
        self.set_event_loop(self.loop)

def get_new_ioloop(self):
        """Override the creation of the IOLoop mimicking that of application.

        The result needs to be a Tornado IOLoop instance, we first configure
        the asyncio loop and then call IOLoop configure to use it.

        """
        if sys.platform == 'linux':
            selector = selectors.SelectSelector()
            loop = asyncio.SelectorEventLoop(selector)
            asyncio.set_event_loop(loop)

        IOLoop.configure('tornado.platform.asyncio.AsyncIOLoop')
        return IOLoop.current()

def safe_event_loop():
    """Fixture to fallback to `select` in Linux.

    """
    if sys.platform == 'linux':
        selector = selectors.SelectSelector()
        loop = asyncio.SelectorEventLoop(selector)
        asyncio.set_event_loop(loop)
        return loop

    return asyncio.get_event_loop_policy().new_event_loop()

def create_event_loop(self):
            return asyncio.SelectorEventLoop()

def create_event_loop(self):
                return asyncio.SelectorEventLoop(
                    selectors.KqueueSelector())

            # kqueue doesn't support character devices (PTY) on Mac OS X older
            # than 10.9 (Maverick)

def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.EpollSelector())

def create_event_loop(self):
                return asyncio.SelectorEventLoop(selectors.PollSelector())

    # Should always exist.

def create_event_loop(self):
            return asyncio.SelectorEventLoop(selectors.SelectSelector())

def setUp(self):
        self.loop = asyncio.SelectorEventLoop()
        self.set_event_loop(self.loop)