Python tornado 模块，ioloop() 实例源码

def exit_test():
    global periodic_checker
    if periodic_checker:
        periodic_checker.stop()
    os.kill(rolld_proc.pid, signal.SIGTERM)
    os.kill(nginx_proc.pid, signal.SIGTERM)
    # IOLoop.instance().add_timeout(time.time() + 5, partial(sys.exit, 0))
    # check if we have zombies left
    try:
        lines = subprocess.check_output('ps auxw | grep python | grep app.py | grep -v grep', shell=True)
        print lines
        assert len(lines) == 0
    except subprocess.CalledProcessError as grepexc:
        # grep shouldnt find anything so exit code should be 1
        if grepexc.returncode == 1:
            pass
        else:
            raise
    # if everything is fine, just stop our ioloop now.
    IOLoop.current().stop()

def install():
    """set the tornado IOLoop instance with the pyzmq IOLoop.

    After calling this function, tornado's IOLoop.instance() and pyzmq's
    IOLoop.instance() will return the same object.

    An assertion error will be raised if tornado's IOLoop has been initialized
    prior to calling this function.
    """
    from tornado import ioloop
    # check if tornado's IOLoop is already initialized to something other
    # than the pyzmq IOLoop instance:
    assert (not ioloop.IOLoop.initialized()) or \
        ioloop.IOLoop.instance() is IOLoop.instance(), "tornado IOLoop already initialized"

    if tornado_version >= (3,):
        # tornado 3 has an official API for registering new defaults, yay!
        ioloop.IOLoop.configure(ZMQIOLoop)
    else:
        # we have to set the global instance explicitly
        ioloop.IOLoop._instance = IOLoop.instance()

def main():
    # create an ioloop, do the above, then stop
    import time
    import _thread
    start = time.time()

    def _thread():
        ioloop.IOLoop.current().run_sync(communicate)

    for _ in range(5):
        _thread.start_new_thread(_thread, ())

    while 1:
        pass

    end = time.time()
    print((end-start))

def quit(self):
        self.ioloop.stop()
        logger.info("scheduler exiting...")

def install():
    """set the tornado IOLoop instance with the pyzmq IOLoop.

    After calling this function, tornado's IOLoop.instance() and pyzmq's
    IOLoop.instance() will return the same object.

    An assertion error will be raised if tornado's IOLoop has been initialized
    prior to calling this function.
    """
    from tornado import ioloop
    # check if tornado's IOLoop is already initialized to something other
    # than the pyzmq IOLoop instance:
    assert (not ioloop.IOLoop.initialized()) or \
        ioloop.IOLoop.instance() is IOLoop.instance(), "tornado IOLoop already initialized"

    if tornado_version >= (3,):
        # tornado 3 has an official API for registering new defaults, yay!
        ioloop.IOLoop.configure(ZMQIOLoop)
    else:
        # we have to set the global instance explicitly
        ioloop.IOLoop._instance = IOLoop.instance()

def install():
    """set the tornado IOLoop instance with the pyzmq IOLoop.

    After calling this function, tornado's IOLoop.instance() and pyzmq's
    IOLoop.instance() will return the same object.

    An assertion error will be raised if tornado's IOLoop has been initialized
    prior to calling this function.
    """
    from tornado import ioloop
    # check if tornado's IOLoop is already initialized to something other
    # than the pyzmq IOLoop instance:
    assert (not ioloop.IOLoop.initialized()) or \
        ioloop.IOLoop.instance() is IOLoop.instance(), "tornado IOLoop already initialized"

    if tornado_version >= (3,):
        # tornado 3 has an official API for registering new defaults, yay!
        ioloop.IOLoop.configure(ZMQIOLoop)
    else:
        # we have to set the global instance explicitly
        ioloop.IOLoop._instance = IOLoop.instance()

def install():
    """set the tornado IOLoop instance with the pyzmq IOLoop.

    After calling this function, tornado's IOLoop.instance() and pyzmq's
    IOLoop.instance() will return the same object.

    An assertion error will be raised if tornado's IOLoop has been initialized
    prior to calling this function.
    """
    from tornado import ioloop
    # check if tornado's IOLoop is already initialized to something other
    # than the pyzmq IOLoop instance:
    assert (not ioloop.IOLoop.initialized()) or \
        ioloop.IOLoop.instance() is IOLoop.instance(), "tornado IOLoop already initialized"

    if tornado_version >= (3,):
        # tornado 3 has an official API for registering new defaults, yay!
        ioloop.IOLoop.configure(ZMQIOLoop)
    else:
        # we have to set the global instance explicitly
        ioloop.IOLoop._instance = IOLoop.instance()

def install():
    """set the tornado IOLoop instance with the pyzmq IOLoop.

    After calling this function, tornado's IOLoop.instance() and pyzmq's
    IOLoop.instance() will return the same object.

    An assertion error will be raised if tornado's IOLoop has been initialized
    prior to calling this function.
    """
    from tornado import ioloop
    # check if tornado's IOLoop is already initialized to something other
    # than the pyzmq IOLoop instance:
    assert (not ioloop.IOLoop.initialized()) or \
        ioloop.IOLoop.instance() is IOLoop.instance(), "tornado IOLoop already initialized"

    if tornado_version >= (3,):
        # tornado 3 has an official API for registering new defaults, yay!
        ioloop.IOLoop.configure(ZMQIOLoop)
    else:
        # we have to set the global instance explicitly
        ioloop.IOLoop._instance = IOLoop.instance()

def shutdown(ioloop, server):
    ''' ??server

    :param server: tornado.httpserver.HTTPServer
    '''
    logging.info(
        "HTTP interpreter service will shutdown in %ss...", 1)
    server.stop()

    deadline = time.time() + 1

    def stop_loop():
        ''' ????loop
        '''
        now = time.time()
        if now < deadline and (ioloop._callbacks or ioloop._timeouts):
            ioloop.add_timeout(now + 1, stop_loop)
        else:
            # ?????? callback ? timeout ?
            ioloop.stop()
            logging.info('Shutdown!')

    stop_loop()

def main():
    ''' main ??
    '''
    # ?? search_engin_server
    ioloop = tornado.ioloop.IOLoop.instance()
    server = tornado.httpserver.HTTPServer(Application(), xheaders=True)
    server.listen(options.port)

    def sig_handler(sig, _):
        ''' ??????
        '''
        logging.warn("Caught signal: %s", sig)
        shutdown(ioloop, server)

    signal.signal(signal.SIGTERM, sig_handler)
    signal.signal(signal.SIGINT, sig_handler)
    ioloop.start()

def web(port = 23456,
        via_cli = False,
        ):
    """
    Bind Tornado server to specified port.    
    """

    print ('BINDING',port)

    try:
        tornado.options.parse_command_line()
        http_server = HTTPServer(Application(),
                                 xheaders=True,
                                 )
        http_server.bind(port)
        http_server.start(16) # Forks multiple sub-processes
        tornado.ioloop.IOLoop.instance().set_blocking_log_threshold(0.5)
        IOLoop.instance().start()

    except KeyboardInterrupt:
        print 'Exit'

    print ('WEB_STARTED')

def _when_complete(self, result, callback):
        try:
            if result is None:
                callback()
            elif isinstance(result, Future):
                if result.done():
                    if result.result() is not None:
                        raise ValueError('Expected None, got %r' % result.result())
                    callback()
                else:
                    # Delayed import of IOLoop because it's not available
                    # on app engine
                    from tornado.ioloop import IOLoop
                    IOLoop.current().add_future(
                        result, functools.partial(self._when_complete,
                                                  callback=callback))
            else:
                raise ValueError("Expected Future or None, got %r" % result)
        except Exception as e:
            self._handle_request_exception(e)

def __init__(self, handlers, application, queue=None, ioloop=None, listener_id=None):
        """
        handlers: list of TaskHandler classes
        application: a toshi.web.Application
        queue: the name of the subscribe channel to use for the tasks
        """

        if queue is None:
            queue = TASK_QUEUE_CHANNEL_NAME
        self.listener_id = listener_id

        self.application = application

        self.ioloop = ioloop or tornado.ioloop.IOLoop.current()
        self.queue_name = queue
        self._task_handlers = {}
        for handler, *optionals in handlers:
            if optionals:
                optionals = optionals[0]
            else:
                optionals = None
            self.add_task_handler(handler, optionals)
        self._tasks = {}
        self._running_tasks = {}
        self._shutdown_task_dispatch = False

def start(self):
        """Start the application and the ioloop"""

        self.log.info("Starting server with options:")
        for trait_name in self._command_line_config.trait_names():
            self.log.info("{}: {}".format(
                trait_name,
                getattr(self._command_line_config, trait_name)
                )
            )
        self.log.info("Listening for connections on {}:{}".format(
            self.command_line_config.ip,
            self.command_line_config.port))

        self.listen(self.command_line_config.port)

        tornado.ioloop.IOLoop.current().start()

    # Private

def __getattr__(self, attr):
        if not attr.startswith('async_'):
            raise AttributeError
        method_name = attr[6:]

        if hasattr(self, method_name + '_buscallback'):
            method = None
            bus_callback = getattr(self, method_name + '_buscallback')
        else:
            method = getattr(self._agent.bus, method_name)

            def bus_callback(method, callback, *args):
                results = method(self._agent, *args)
                self._agent.ioloop.add_callback(callback, results)
                # dbus-specific - indicates this method should only be called
                # once
                return False

        def _async(callback, *args):
            self._agent.bus.busthread_call(bus_callback, method, callback,
                                           *args)
        return _async

def __init__(self, ioloop, channel):
        """
        Initialize the Worker in the main thread.
        :param ioloop: The main tornado ioloop.
        :param channel:  The channel being listened.
        """
        self.jobs = []
        self.ioloop = ioloop  # Main tornado ioloop
        self.channel = channel  # Channel to listen and send messages to
        self.results = []
        self.tasks = 0
        self.running_jobs = []
        self.jobs_futures = []
        self.executor = ThreadPoolExecutor(
            max_workers=self.MAX_BATCH_OPERATIONS)

        self.init_cache()
        self.execute_evaluation_loop()
        # Receive a kick in the nuts when somebody talks
        io_loop.add_handler(conn.fileno(), self.receive(), io_loop.READ)

        # Always listen before talking
        print("Attached database listener")
        self.listen()
        self.talk("work!")

def main():
    tornado.options.parse_command_line()
    app = Application()
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)

    LOGGER.info('[server.main] Starting server on http://127.0.0.1:%s', options.port)

    try:
        LOGGER.info("\n[server.main] Server Started.\n")

        tornado.ioloop.IOLoop.current().start()

    except KeyboardInterrupt:
        LOGGER.error('\n[server.main] EXCEPTION KEYBOARDINTERRUPT INITIATED\n')
        LOGGER.info("[server.main] Stopping Server....")
        LOGGER.info('[server.main] closing all websocket connections objects and corresponsding mqtt client objects')
        LOGGER.info('Stopping Tornado\'s main iolooop')

        # Stopping main thread's ioloop, not to be confused with current thread's ioloop
        # which is ioloop.IOLoop.current()
        tornado.ioloop.IOLoop.instance().stop()

        LOGGER.info("\n[server.main] Server Stopped.")

def main():
    tornado.options.parse_command_line()
    app = Application()
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)

    LOGGER.info('[server.main] Starting server on http://127.0.0.1:%s', options.port)

    try:
        LOGGER.info("\n[server.main] Server Started.\n")

        tornado.ioloop.IOLoop.current().start()

    except KeyboardInterrupt:
        LOGGER.error('\n[server.main] EXCEPTION KEYBOARDINTERRUPT INITIATED\n')
        LOGGER.info("[server.main] Stopping Server....")
        LOGGER.info('[server.main] closing all websocket connections objects and corresponsding mqtt client objects')
        LOGGER.info('Stopping Tornado\'s main iolooop')

        # Stopping main thread's ioloop, not to be confused with current thread's ioloop
        # which is ioloop.IOLoop.current()
        tornado.ioloop.IOLoop.instance().stop()

        LOGGER.info("\n[server.main] Server Stopped.")

def backport_close(self, all_fds=False):
        """backport IOLoop.close to 3.0 from 3.1 (supports fd.close() method)"""
        from zmq.eventloop.minitornado.ioloop import PollIOLoop as mini_loop
        return mini_loop.close.__get__(self)(all_fds)

def run():

    logging.basicConfig(format="%(asctime)s %(levelname)s - %(message)s", level=logging.DEBUG)
    log = logging.getLogger("txn-service")
    log.info("Setting up argparse")
    parser = argparse.ArgumentParser(description='Process some integers.', prog='python -m blockchain')
    parser.add_argument('-p', '--port', default=8000)
    parser.add_argument('--debug', default=True, action="store_true")
    parser.add_argument('--private-key', dest="private_key", required=True, help="ECDSA private key for signing")
    parser.add_argument('--public-key', dest="public_key", required=True, help="ECDSA private key for signing")

    log.info("Parsing arguments")
    args = parser.parse_args()

    hdlrs = [
        (r"^/transaction$", TransactionHandler),
        (r"^/transaction/(.*)", TransactionHandler),
    ]

    log.info("Creating new tornado.web.Application")
    application = TransactionService(hdlrs,
        log = log,
        **vars(args))

    log.info("Starting transaction service on port %s" % args.port)
    application.listen(args.port)
    tornado.ioloop.IOLoop.current().start()

def run():

    logging.basicConfig(format="%(asctime)s %(levelname)s - %(message)s", level = logging.DEBUG)
    log = logging.getLogger("txn-service")
    log.info("Setting up argparse")
    parser = argparse.ArgumentParser(description='Process query info.', prog='python -m blockchain')
    parser.add_argument('-p', '--port', default = 8080)
    parser.add_argument('--debug', default = True, action = "store_true")

    log.info("Parsing arguments")
    args = parser.parse_args()

    query_hdlrs = [
        (r"^/transaction", QueryHandler),
        (r"^/transaction/(.*)", QueryHandler),
        (r"^/verification$", BlockVerificationHandler),
        (r"^/verification/(.*)", BlockVerificationHandler),
    ]

    log.info("Creating new tornado.web.Application")
    application = QueryService(query_hdlrs,
        log = log,
        **vars(args))

    log.info("Starting query service on port %s" % args.port)
    application.listen(args.port)
    tornado.ioloop.IOLoop.current().start()

def main():
    define(name='port', default=8000, type=int, help='run on the given port')
    tornado.options.parse_command_line()
    logger.info('================ spider web server has started ================ ')
    logger.info('       server start at port -> {}, debug mode = {}'.format(options.port, constants.DEBUG))
    app = make_web_app()
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()

def main():
    app = make_app(debug=True)
    app.listen(4000)
    print 'listening on port 4000'
    tornado.ioloop.IOLoop.current().start()

def get2(self, *args, **kwargs):
        client = tornado.httpclient.AsyncHTTPClient(max_clients=100)
        client.fetch('http://localhost:8888/blocking', callback=self.on_response)

        # ???asynchronous???
#         future = client.fetch('http://localhost:8888/blocking')
#         tornado.ioloop.IOLoop.current().add_future(future, callback=self.on_response)

        # future??
        # future?add_done_callback??????ioloop?future????????????
        # ??????add_done_callback????(?????)
#         future.add_done_callback(lambda x: future.set_result(x.result()))

def main():
    init_logging()
    io_loop = tornado.ioloop.IOLoop.instance()
    c1 = TCPClient("127.0.0.1", 8001, io_loop)
    c2 = TCPClient("127.0.0.1", 8001, io_loop)
    c1.connect()
    c2.connect()
    c2.set_shutdown()
    logging.info("**********************start ioloop******************")
    io_loop.start()

def main():
    # create an ioloop, do the above, then stop
    import time
    start = time.time()
    for _ in range(10000):
        ioloop.IOLoop.current().run_sync(communicate)
    end = time.time()
    print((end - start))

def init_one(self, ioloop, fetcher, processor,
                 result_worker=None, interactive=False):
        self.ioloop = ioloop
        self.fetcher = fetcher
        self.processor = processor
        self.result_worker = result_worker
        self.interactive = interactive
        self.running_task = 0

def send_task(self, task, force=True):
        if self.fetcher.http_client.free_size() <= 0:
            if force:
                self._send_buffer.appendleft(task)
            else:
                raise self.outqueue.Full
        self.ioloop.add_future(self.do_task(task), lambda x: x.result())

def run(self):
        import tornado.ioloop
        tornado.ioloop.PeriodicCallback(self.run_once, 100,
                                        io_loop=self.ioloop).start()
        self.ioloop.start()

def main():
    app = make_app(debug=True)
    app.listen(4000)
    print 'listening on port 4000'
    tornado.ioloop.IOLoop.current().start()

def backport_close(self, all_fds=False):
        """backport IOLoop.close to 3.0 from 3.1 (supports fd.close() method)"""
        from zmq.eventloop.minitornado.ioloop import PollIOLoop as mini_loop
        return mini_loop.close.__get__(self)(all_fds)

def backport_close(self, all_fds=False):
        """backport IOLoop.close to 3.0 from 3.1 (supports fd.close() method)"""
        from zmq.eventloop.minitornado.ioloop import PollIOLoop as mini_loop
        return mini_loop.close.__get__(self)(all_fds)

def backport_close(self, all_fds=False):
        """backport IOLoop.close to 3.0 from 3.1 (supports fd.close() method)"""
        from zmq.eventloop.minitornado.ioloop import PollIOLoop as mini_loop
        return mini_loop.close.__get__(self)(all_fds)

def backport_close(self, all_fds=False):
        """backport IOLoop.close to 3.0 from 3.1 (supports fd.close() method)"""
        from zmq.eventloop.minitornado.ioloop import PollIOLoop as mini_loop
        return mini_loop.close.__get__(self)(all_fds)

def backport_close(self, all_fds=False):
        """backport IOLoop.close to 3.0 from 3.1 (supports fd.close() method)"""
        from zmq.eventloop.minitornado.ioloop import PollIOLoop as mini_loop
        return mini_loop.close.__get__(self)(all_fds)

def backport_close(self, all_fds=False):
        """backport IOLoop.close to 3.0 from 3.1 (supports fd.close() method)"""
        from zmq.eventloop.minitornado.ioloop import PollIOLoop as mini_loop
        return mini_loop.close.__get__(self)(all_fds)

def add_interval_timer(seconds, func, *args, **kwargs):
    timer = tornado.ioloop.PeriodicCallback(partial(func, *args, **kwargs), seconds * 1000, IOLoop.instance())
    timer.start()
    return timer

def __init__(self, io_loop=None):
        if not io_loop:
            io_loop = tornado.ioloop.IOLoop.instance()
        self._io_loop = io_loop
        self._readers = {}
        self._writers = {}
        self._fds = {} # a map of fd to a (reader, writer) tuple
        self._delayedCalls = {}
        self._running = False
        self._closed = False
        PosixReactorBase.__init__(self)

    # IReactorTime

def __init__(self):
        # always use a new ioloop
        super(_TestReactor, self).__init__(IOLoop())

def install(io_loop=None):
    """
    Install the Tornado reactor.
    """
    if not io_loop:
        io_loop = tornado.ioloop.IOLoop.instance()
    reactor = TornadoReactor(io_loop)
    from twisted.internet.main import installReactor
    installReactor(reactor)
    return reactor

def main(port):
    #train()
    load_grocery()
    tornado.options.parse_command_line()
    print "start on port %s..." % port
    http_server = tornado.httpserver.HTTPServer(Application(), xheaders=True)
    http_server.listen(port)
    #tornado.autoreload.start()
    tornado.ioloop.IOLoop.instance().start()

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/connect", ClientSocket),
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/connect", ClientSocket),
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/connect", ClientSocket),
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/connect", ClientSocket),
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/connect", ClientSocket),
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def make_app():
    """Create and return the main Tornado web application.

    It will listen on the port assigned via `app.listen(port)`,
    and will run on Tornado's main ioloop,
    which can be started with `tornado.ioloop.IOLoop.current().start()`.
    """
    return tornado.web.Application([
        (r"/connect", ClientSocket),
        (r"/(.*)", tornado.web.StaticFileHandler, {
            "path": "client",
            "default_filename": "index.html"
        }),
    ], debug=True)

def __init__(self, io_loop=None):
        if not io_loop:
            io_loop = tornado.ioloop.IOLoop.instance()
        self._io_loop = io_loop
        self._readers = {}
        self._writers = {}
        self._fds = {} # a map of fd to a (reader, writer) tuple
        self._delayedCalls = {}
        self._running = False
        self._closed = False
        PosixReactorBase.__init__(self)

    # IReactorTime