Python gevent 模块，kill() 实例源码

def test_greenlet(self):
        """??????Greenlet????"""
        class MyGreenlet(gevent.Greenlet):
            def __init__(self, message, n):
                super(MyGreenlet, self).__init__()
                self.message = message
                self.n = n

            def _run(self):
                print(self.message)
                gevent.sleep(self.n)

        g1 = MyGreenlet("Hi there111!", 1)
        g1.start()
        g2 = MyGreenlet("Hi there222!", 2)
        g2.start()
        gevent.joinall([g1, g2])

    # def test_shutdown(self):
    #     def run_forever():
    #         _log.info('run_forever start..')
    #         gevent.sleep(1000)
    #     gevent.signal(signal.SIGQUIT, gevent.kill)
    #     thread = gevent.spawn(run_forever)
    #     thread.join()

def test_pool(self):
        """?????"""
        class SocketPool(object):

            def __init__(self):
                self.pool = Pool(1000)
                self.pool.start()

            def listen(self, socket):
                while True:
                    socket.recv()

            def add_handler(self, socket):
                if self.pool.full():
                    raise Exception("At maximum pool size")
                else:
                    self.pool.spawn(self.listen, socket)

            def shutdown(self):
                self.pool.kill()

def concurrent_find(func, params, **kw):
    timeout = kw.pop("concurrent_timeout", None)
    with async(func, list(params), **kw) as futures:
        future = None
        try:
            for future in futures.as_completed(timeout=timeout):
                if not future.exception() and future.result():
                    futures.kill()
                    return future.result()
            else:
                if future:
                    return future.result()
        except FutureTimeoutError as exc:
            if not timeout:
                # ??
                raise
            futures.kill()
            _logger.warning("Concurrent future timed out (%s)", exc)

def cli(log_level, live):
    logging.basicConfig(
        filename='arbloop.log',
        format='[%(asctime)s] [%(levelname)s] %(message)s',
        level=log_level
    )
    logging.info('Warming up traders ...')

    gevent.signal(signal.SIGQUIT, gevent.kill)

    workers = []
    for product in config.TRADER_PRODUCTS or []:
        trader = Trader(product=product, live=live)
        workers.append(
            gevent.spawn(trader.trade)
        )
    gevent.joinall(workers)

def on_open(self):
        if self.ws.protocol is None or (
                GRAPHQL_SUBSCRIPTIONS not in self.ws.protocol):
            self.ws.close(1002)

        def keep_alive_callback():
            if not self.ws.closed:
                self.send_keep_alive()
            else:
                gevent.kill(keep_alive_timer)

        if self.keep_alive:
            keep_alive_timer = gevent.spawn(self.timer, keep_alive_callback,
                                            self.keep_alive)

def main():
    """
    ?????
    """
    redis_client.delete(REDIS_VISITED_URLS)
    start_user_url = gen_user_page_url(START_USER)

    gevent.signal(signal.SIGQUIT, gevent.kill)
    request_api([start_user_url], async_get, parse_user)

def clear_gevent_instance(self,connection):
        print "clear gevent instance"
        if connection in self.connectList:
            sessionList = self.connectList[connection]['session']
            #??????
            for sessionId in sessionList:
                gevent.kill(sessionList[sessionId]['instance'])
            #??????
            sendInstance = self.connectList[connection]['sendInstance']
            recvInstance = self.connectList[connection]['recvInstance']
            connection.close()
            del self.connectList[connection]
            gevent.kill(sendInstance)
            gevent.kill(recvInstance)

def stop(self):
        if self.g:
            gevent.kill(self.g)

def async_raise_in_main_thread(exc, use_concurrent_loop=True):
    """
    Uses a unix signal to raise an exception to be raised in the main thread.
    """

    from plumbum import local
    pid = os.getpid()
    if not REGISTERED_SIGNAL:
        raise NotInitialized()

    # sometimes the signal isn't caught by the main-thread, so we should try a few times (WEKAPP-14543)
    def do_signal(raised_exc):
        global LAST_ERROR
        if LAST_ERROR is not raised_exc:
            _logger.debug("MainThread took the exception - we're done here")
            if use_concurrent_loop:
                raiser.stop()
            return

        _logger.info("Raising %s in main thread", type(LAST_ERROR))
        local.cmd.kill("-%d" % REGISTERED_SIGNAL, pid)

    if use_concurrent_loop:
        raiser = concurrent(do_signal, raised_exc=exc, loop=True, sleep=30, daemon=True, throw=False)
        raiser.start()
    else:
        do_signal(exc)

def _rimt(exc):
        _logger.info('YELLOW<<killing main thread greenlet>>')
        main_thread_greenlet = threading.main_thread()._greenlet
        orig_throw = main_thread_greenlet.throw

        # we must override "throw" method so exception will be raised with the original traceback
        def throw(*args):
            if len(args) == 1:
                ex = args[0]
                return orig_throw(ex.__class__, ex, ex.__traceback__)
            return orig_throw(*args)
        main_thread_greenlet.throw = throw
        gevent.kill(main_thread_greenlet, exc)
        _logger.debug('exiting the thread that failed')
        raise exc

def kill(wait=False):
        nonlocal killed
        futures.cancel()
        if executor:
            executor.shutdown(wait=wait)
        killed = True

def kill_this_process(graceful=False):
    from plumbum import local
    pid = os.getpid()
    if graceful:
        flag = '-HUP'
    else:
        flag = '-9'
    local.cmd.kill(flag, pid)