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

def broadcastToMiners(self, block):
        self.log.info("broadcasting block to miners: " + repr(block))
        coros = [peer.sendMinerBlock(block) for id, peer in self.peers.items() if peer.role == 'M']
        f = gather(*coros, loop=self.loop)

        loop = self.loop
        if loop.is_running():
            future = asyncio.run_coroutine_threadsafe(f, loop)
            res =  future.result(30)
        else:
            res = loop.run_until_complete(f)

        return res


    # send block to peers

def test_add_sync_utility(self):
        util = getUtility(IQueueUtility)
        var = []

        async def printHi():
            var.append('hola')

        context = self.layer.app['plone'].conn.root()
        v = AsyncMockView(context, self.layer.app['plone'].conn, printHi, self.layer.app)
        loop = asyncio.get_event_loop()
        future = asyncio.run_coroutine_threadsafe(util.add(v), loop)
        future2 = asyncio.run_coroutine_threadsafe(util.add(v), loop)
        total = future.result()
        total = future2.result()

        future = asyncio.run_coroutine_threadsafe(util._queue.join(), loop)
        total = future.result()  # noqa
        self.assertTrue('hola' in var)
        self.assertTrue(len(var) == 2)

def test_direct_call(self):
        """Calls the activity directly."""
        """Creates one trigger and an activity and triggers it."""
        foo = TriggerBase()
        bas = [None]
        s = Semaphore(0)

        @activity(foo)
        async def bar():
            bas[0] = "Triggered"
            s.release()

        asyncio.run_coroutine_threadsafe(bar(), self.loop)
        s.acquire()

        self.assertEqual(bas[0], "Triggered")

    # Activities outside of modules where only meant to be used during early stages of development. The are officially
    # not supported.

def test_parameters(self):
        """Triggers an activity and passes extra parameters."""
        bas = [None]
        foo = TriggerBase()

        s = Semaphore(0)

        @activity(foo, "arg", k="kwarg")
        async def bar(p, k):
            bas[0] = p + k
            s.release()

        asyncio.run_coroutine_threadsafe(foo.trigger(), self.loop)
        assert s.acquire(timeout=0.1)

        self.assertEqual(bas[0], "argkwarg")

def test_multiple_activity(self):
        bas = [None, None]
        foo = TriggerBase()

        s = Semaphore(0)

        @activity(foo)
        async def bar1():
            bas[0] = "bar1"

        @activity(foo)
        async def bar2():
            bas[1] = "bar2"
            await asyncio.sleep(0.01)
            s.release()

        asyncio.run_coroutine_threadsafe(foo.trigger(), self.loop)
        self.assertTrue(s.acquire(timeout=1))

        self.assertEqual(bas[0], "bar1")
        self.assertEqual(bas[1], "bar2")

def test_simple_descriptor_trigger(self):

        class Foo(ModuleBase):
            def __init__(self, s):
                super().__init__()
                self.bar = None
                self.s = s

            trigger = DescriptorClassTrigger(TriggerBase)

            @activity(trigger)
            async def activity(self):
                self.bar = "qwertyuiop"
                s.release()

        s = Semaphore(0)
        foo = Foo(s)
        asyncio.run_coroutine_threadsafe(foo.trigger.trigger(), self.loop)
        self.assertTrue(s.acquire(timeout=0.1))

        self.assertEqual(foo.bar, "qwertyuiop")

def update_regular(self, loop):
        """
        A regularly ran function that takes care of periodic tasks.
        Such as making cooldowns run out and checking if members still exist.
        :param loop:
        :return:
        """
        while self.run_timer:
            await asyncio.sleep(self.plugin_config["poll_every"])
            for k, t_guild in self.storage["members"].items():
                t_lst = []
                for k1, t_member in t_guild.items():
                    if t_member.needs_reinit:
                        t_member.reinit(self)
                    if t_member.member and t_member.guild.get_member(t_member.member.id):
                        t_future = asyncio.run_coroutine_threadsafe(t_member.update(self), loop=loop)
                        try:
                            t_future.result()
                        except Exception as e:
                            print(e)
                    else:
                        t_lst.append(k1)
                for k1 in t_lst:
                    t_guild.pop(k1)

def __init__(self, token, url, loop, no_wait=False):
        log.debug("Starting delegated client")
        self.loop = loop or get_event_loop()

        async def run_client(loop):
            self.session = DelegatedRawSocketAsexorClient(url, token, loop)
            try:
                await self.session.start()
            except Exception:
                log.exception("Cannot start client")
                await self.session.stop()
                raise

        fut = asyncio.run_coroutine_threadsafe(run_client(self.loop), self.loop)
        if not no_wait:
            fut.result(WAIT_TIMEOUT)

def __init__(self, token, ws_url, session_id=None,loop=None):
        self.loop = loop or get_event_loop()
        self._pending_tasks = {}
        log.info('Starting client')
        self.session = None
        self.session_id = session_id

        async def run_client(loop):
            self.session = AsexorClient(ws_url, token, session_id=self.session_id, loop=loop)
            self.session.subscribe(self.task_callback)
            try:
                await self.session.start()
            except Exception:
                await self.session.stop()
                raise

        fut = asyncio.run_coroutine_threadsafe(run_client(self.loop), self.loop)
        fut.result(WAIT_TIMEOUT)

def sendMsg(self,txt):
        #Do we need to send a new msg? Or just edit the old one
        if self.newMsg:
            #Lets hope this works
            #Eventually create a new msg, and give us the result at self.curMsg
            self.curMsg = asyncio.run_coroutine_threadsafe(self.bot.send_message(self.channel, txt), self.bot.loop)

            self.newMsg = False
        else:
            #First we need to check if the Msg has been sent yet, if not guess we just miss out on a msg
            if self.curMsg.done():
                #When I edit I only edit the original messege, so keep track of our additions
                self.curMsgTxt = self.curMsgTxt + " \n" + txt
                self.editMsg(self.curMsg.result(),self.curMsgTxt)
            else:
                print("Message not done yet")

def _dispatch_coroutine(co, loop, abort_future):
    '''Execute a coroutine in a loop's thread and block till completion.

    Wraps a co-routine function; calling the function causes the co-routine
    to be dispatched in the event loop's thread and blocks until that call completes.

    Waits for either the coroutine or abort_future to complete.
    abort_future provides the main event loop with a means of triggering a
    clean shutdown in the case of an exception.
    '''
    fut = asyncio.run_coroutine_threadsafe(co, loop)
    result = concurrent.futures.wait((fut, abort_future), return_when=concurrent.futures.FIRST_COMPLETED)
    result =  list(result.done)[0].result()
    if getattr(result, '__wrapped__', None) is None:
        # If the call retuned the wrapped contents of a _SyncProxy then return
        # the enclosing proxy instead to the sync caller
        wrapper = getattr(result, '__wrapper__', None)
        if wrapper is not None:
            result = wrapper
    return result

def send_messages(websocket):
    """this is executed by a dedicated thread. Gets messages from the outgoing_messages queue and sends them on the websocket"""
    # set the event loop because this is another thread
    asyncio.set_event_loop(loop)
    try:
        while True:
            msg = outgoing_messages.get()
            print(msg)
            # run the async function in synchronous context
            future = asyncio.run_coroutine_threadsafe(
                websocket.send(json.dumps(msg)), loop)
            future.result()
    except websockets.exceptions.ConnectionClosed as e:
        # this exception occurred because the connection was closed
        # put back the message in the queue
        # TODO find the way to put on the first place, without using deque that
        # does not block
        outgoing_messages.put(msg)

def target(self, fail=False, cancel=False, timeout=None,
               advance_coro=False):
        """Run add coroutine in the event loop."""
        coro = self.add(1, 2, fail=fail, cancel=cancel)
        future = asyncio.run_coroutine_threadsafe(coro, self.loop)
        if advance_coro:
            # this is for test_run_coroutine_threadsafe_task_factory_exception;
            # otherwise it spills errors and breaks **other** unittests, since
            # 'target' is interacting with threads.

            # With this call, `coro` will be advanced, so that
            # CoroWrapper.__del__ won't do anything when asyncio tests run
            # in debug mode.
            self.loop.call_soon_threadsafe(coro.send, None)
        try:
            return future.result(timeout)
        finally:
            future.done() or future.cancel()

def run(self):
        print("Starting CPU solver")
        s = Solver()

        while self.job == None or self.nonce1 == None:
            time.sleep(2)
            print(".", end='', flush=True)

        while not self._stop:
            nonce2 = self.increase_nonce()
            nonce2 = nonce2.rjust(32 - len(self.nonce1) - len(self.solver_nonce), b'\0')

            header = self.job.build_header(self.nonce1 + self.solver_nonce + nonce2)

            sol_cnt = s.find_solutions(header)
            self.counter(sol_cnt) # Increase counter for stats

            for i in range(sol_cnt):
                solution = b'\xfd\x40\x05' + s.get_solution(i)

                if self.job.is_valid(header, solution, self.job.target):
                    print("FOUND VALID SOLUTION!")
                    # asyncio.run_coroutine_threadsafe(self.on_share(self.job, self.solver_nonce + nonce2, solution), self.loop)
                    asyncio.async(self.on_share(self.job, self.solver_nonce + nonce2, solution), loop=self.loop)

def test_cancel_where_tasks(monitor, tn_client, loop):
    tn = tn_client

    async def sleeper(loop):
        await asyncio.sleep(100, loop=loop)  # xxx

    fut = asyncio.run_coroutine_threadsafe(sleeper(loop), loop=loop)
    # TODO: we should not rely on timeout
    time.sleep(0.1)

    task_ids = get_task_ids(loop)
    assert len(task_ids) > 0
    for t_id in task_ids:
        resp = execute(tn, 'where {}\n'.format(t_id))
        assert 'Task' in resp
        resp = execute(tn, 'cancel {}\n'.format(t_id))
        assert 'Cancel task' in resp
    fut.cancel()

def put_message(self, message):
        """
        :param message: protobuf generated validator_pb2.Message
        """
        if not self._ready_event.is_set():
            return

        with self._condition:
            self._condition.wait_for(
                lambda: self._event_loop is not None
                and self._send_queue is not None
            )

        asyncio.run_coroutine_threadsafe(
            self._put_message(message),
            self._event_loop)

def game_loop(asyncio_loop):
    print("Game loop thread id {}".format(threading.get_ident()))
    # a coroutine to run in main thread
    async def notify():
        print("Notify thread id {}".format(threading.get_ident()))
        await tick.acquire()
        tick.notify_all()
        tick.release()

    while 1:
        task = asyncio.run_coroutine_threadsafe(notify(), asyncio_loop)
        # blocking the thread
        sleep(1)
        # make sure the task has finished
        task.result()

def game_loop(asyncio_loop):
    # coroutine to run in main thread
    async def notify():
        await tick.acquire()
        tick.notify_all()
        tick.release()

    queue = Queue()

    # function to run in a different process
    def worker():
        while 1:
            print("doing heavy calculation in process {}".format(os.getpid()))
            sleep(1)
            queue.put("calculation result")

    Process(target=worker).start()

    while 1:
        # blocks this thread but not main thread with event loop
        result = queue.get()
        print("getting {} in process {}".format(result, os.getpid()))
        task = asyncio.run_coroutine_threadsafe(notify(), asyncio_loop)
        task.result()

def runcoro(async_function):
    """
    Runs an asynchronous function without needing to use await - useful for lambda

    Args:
        async_function (Coroutine): The asynchronous function to run
    """

    future = _asyncio.run_coroutine_threadsafe(async_function, client.loop)
    result = future.result()
    return result

def download_next_song(self, song):
        """Downloads the next song and starts playing it"""

        dl_ydl_opts = dict(ydl_opts)
        dl_ydl_opts["progress_hooks"] = [self.ytdl_progress_hook]
        dl_ydl_opts["outtmpl"] = self.output_format

        # Move the songs from the next cache to the current cache
        self.move_next_cache()

        self.state = 'ready'
        self.play_empty()
        # Download the file and create the stream
        with youtube_dl.YoutubeDL(dl_ydl_opts) as ydl:
            try:
                ydl.download([song])
            except DownloadStreamException:
                # This is a livestream, use the appropriate player
                future = asyncio.run_coroutine_threadsafe(self.create_stream_player(song, dl_ydl_opts), client.loop)
                try:
                    future.result()
                except Exception as e:
                    logger.exception(e)
                    self.vafter_ts()
                    return
            except PermissionError:
                # File is still in use, it'll get cleared next time
                pass
            except youtube_dl.utils.DownloadError as e:
                self.logger.exception(e)
                self.statuslog.error(e)
                self.vafter_ts()
                return
            except Exception as e:
                self.logger.exception(e)
                self.vafter_ts()
                return

def vafter_ts(self):
        """Function that is called after a song finishes playing"""
        logger.debug("Song finishing")
        future = asyncio.run_coroutine_threadsafe(self.vafter(), client.loop)
        try:
            future.result()
        except Exception as e:
            logger.exception(e)

def flush(self):
        try:
            asyncio.run_coroutine_threadsafe(self.usend_when_ready(), client.loop)
        except Exception as e:
            logger.exception(e)
            return

def stop(self):
        """Stop Modis and log it out of Discord."""
        self.button_toggle_text.set("Start Modis")
        self.state = "off"

        logger.info("Stopping Discord Modis")

        from ._client import client
        asyncio.run_coroutine_threadsafe(client.logout(), client.loop)
        self.status_bar.set_status(0)

def _job_watcher(obj, jobs_id, job_hash, job, session):
    asyncio.run_coroutine_threadsafe(
        obj._set_job(jobs_id, job_hash, {'status': 'running'}, session),
        session.loop
    ).result()
    start_time = datetime.now()

    try:
        if asyncio.iscoroutinefunction(job.func):
            result = asyncio.run_coroutine_threadsafe(job(), session.loop).result()
        else:
            executor = ThreadPoolExecutor(1)
            job = executor.submit(job)
            result = job.result()
            executor.shutdown()

    except Exception as error:
        result = {'name': error.__class__.__name__, 'message': str(error)}
        status = 'error'
        obj._logger.exception('From job {}:{}'.format(jobs_id, job_hash))

    else:
        status = 'done'

    end_time = datetime.now()
    time_info = {
        'start': str(start_time)[:-3],
        'end': str(end_time)[:-3],
        'elapsed': str(end_time - start_time)[:-3]
    }
    job_obj = {'status': status, 'result': result, 'time_info': time_info}

    asyncio.run_coroutine_threadsafe(
        obj._set_job(jobs_id, job_hash, job_obj, session),
        session.loop
    ).result()
    session.bind.close()
    session.close()

def submit(self, coro):
        return asyncio.run_coroutine_threadsafe(coro, self._loop)

def test_hello(self):
        async def hello(self):
            session = aiohttp.ClientSession()
            async with session.ws_connect(
                    'ws://localhost:{port}/plone/plone/@ws'.format(
                        port=TESTING_PORT),
                    headers={'AUTHORIZATION': 'Basic %s' % ADMIN_TOKEN}) as ws:
                # we should check version
                sending = {
                    'op': 'GET',
                    'value': '/'
                }
                ws.send_str(json.dumps(sending))
                async for msg in ws:
                    if msg.tp == aiohttp.WSMsgType.text:
                        message = json.loads(msg.data)
                        if 'op' in message and message['op'] == 'close':
                            await ws.close()
                            break  # noqa
                        else:
                            self.assertTrue(len(message['items']) == 0)
                            await ws.close()
                    elif msg.tp == aiohttp.WSMsgType.closed:
                        break  # noqa
                    elif msg.tp == aiohttp.WSMsgType.error:
                        break  # noqa
                return {}

        loop = asyncio.get_event_loop()
        future = asyncio.run_coroutine_threadsafe(hello(self), loop)
        result = future.result()  # noqa

def send_message(self, message):
        print(message)
        asyncio.run_coroutine_threadsafe(self.ws.send(message), self.loop)

def get_data(self, items_model, session):
        asyncio.run_coroutine_threadsafe(asyncio.sleep(0.5), session.loop).result()

        data = [{'item_key': '2|test2', 'value': 1},
                {'item_key': '1|test1', 'value': 3},
                {'item_key': '3|test3', 'value': 2}]
        data = map(ujson.dumps, data)
        data = '\n'.join(data)

        filename_prefix = 'top_seller'
        file_ = gzip.open(os.path.join(self._data_path, filename_prefix) + '-000000001.gz', 'wt')
        file_.write(data)
        file_.close()
        return {'lines_count': 3}

def run_coro(coro, session):
    if not asyncio.iscoroutine(coro):
        coro = _convert_future_to_coro(coro)

    if session.loop.is_running():
        return asyncio.run_coroutine_threadsafe(coro, session.loop).result()
    else:
        return session.loop.run_until_complete(coro)

def start(self):
        """
        Starts the clock
        :return: Returns an Asyncio.Future object.
        """
        self.running = True
        self.trigger_time = self.loop.time() + self.period
        return asyncio.run_coroutine_threadsafe(self.timer_callback(), loop=self.loop)

def start(self):
        """
        Starts the clock

        :return: Returns an Asyncio.Future object that completes after the first tick.
        """
        self.running = True
        return asyncio.run_coroutine_threadsafe(self.timer_callback(), loop=self.loop)

def flush_threadsafe(self, data):
        """
        Thread safe version of flush.

        :param data: Data to flush.
        """
        # ctlog.debug("OutputPort.flush_threadsafe({})".format(data))
        if self.channel is not None:
            loop = get_event_loop()
            asyncio.run_coroutine_threadsafe(self.channel.flush(data), loop)

def transmit_threadsafe(self, command_id, *args):
        """
        A thread safe version of send.

        :param command_id: Command id
        :param *args: Command arguments
        """
        asyncio.run_coroutine_threadsafe(self.transmit(command_id, *args), self.loop)

def test_simple_trigger(self):
        """Creates one trigger and an activity and triggers it."""
        foo = TriggerBase()
        bas = [None]
        s = Semaphore(0)

        @activity(foo)
        async def bar():
            bas[0] = "Triggered"
            s.release()

        asyncio.run_coroutine_threadsafe(foo.trigger(), self.loop)
        self.assertTrue(s.acquire(timeout=0.1))

        self.assertEqual(bas[0], "Triggered")

def test_decoder(self):
        """
        This test, tests whether the decoder is working properly by given it a raw package and checking whether the
        result matches.

        """

        async def run(q):
            try:
                decoder = Decoder()
                inp_data = [0, [1, 2, 3]]

                data = umsgpack.packb(inp_data)

                await decoder.digest(struct.pack(">I", len(data)))
                await decoder.digest(data)

                out_data = await decoder.get()

                assert inp_data == out_data
                q.put(None)
            except:
                q.put(sys.exc_info())

        q = Queue()
        loop = get_event_loop()
        asyncio.run_coroutine_threadsafe(run(q), loop)

        # If you get an Empty exception over here, it means the co-routine timed out.
        exc_info = q.get(timeout=1)
        if exc_info is not None:
            raise exc_info[1].with_traceback(exc_info[2])

def test_ping(self):
        async def run(q):
            try:
                connections = {}
                listener = Listener("127.0.0.1", 8888, connections)
                await listener.wait_until_started()

                client_connection = await Connection.from_host("127.0.0.1", 8888)

                # This one is optional. Any commands transmitting will wait for the connection to be ready anyways.
                await client_connection.wait_for_ready()

                await client_connection.ping()
                await client_connection.ping()

                q.put(None)
            except:
                q.put(sys.exc_info())

        q = Queue()
        loop = get_event_loop()
        asyncio.run_coroutine_threadsafe(run(q), loop)

        # If you get an Empty exception over here, it means the co-routine timed out.
        exc_info = q.get(timeout=1)
        if exc_info is not None:
            raise exc_info[1].with_traceback(exc_info[2])

def test_multiple_module_instances(self):

        class Foo(ModuleBase):
            trigger = DescriptorClassTrigger(TriggerBase)

            def __init__(self, s, bar):
                super().__init__()
                self.s = s
                self.bar = bar

            @activity(trigger)
            async def activity(self):
                self.bar.append(self)
                self.s.release()

        # TriggerBase each instance individually
        foo = []
        bar = []
        s = Semaphore(0)
        for i in range(5):
            foo.append(Foo(s, bar))
            asyncio.run_coroutine_threadsafe(foo[-1].trigger.trigger(), self.loop)
            self.assertTrue(s.acquire(timeout=0.1))
        self.assertListSameContent(bar, foo)

        # TODO: Do this test on it's own. This setup doesn't work since I changed all activities to coroutines.
        # TriggerBase all instances of the Foo module class
        # bar = []
        # asyncio.run_coroutine_threadsafe(Foo.trigger.trigger(), self.loop)
        # self.assertListSameContent(foo, bar)

def test_simple(self):
        class A(ModuleBase):
            op = DescriptorStatic(OutputPortDescriptorInstance)

            def __init__(self, channel_register):
                super().__init__(channel_register)
                self.op.subscribe()

            async def transmit(self):
                # print("transmitting:", self.op.channel_name)
                await self.op.flush("Some Data")

        class B(ModuleBase):
            ip = InputPortStatic(channel_name="op")

            def __init__(self, channel_register, semaphore):
                super().__init__(channel_register)
                self.semaphore = semaphore
                self.ip.subscribe()

            @activity(ip)
            async def foo(self, ip):
                assert ip == "Some Data"
                self.semaphore.release()

        semaphore = Semaphore(0)

        channel_register = ChannelRegister()

        a = A(channel_register)
        b = B(channel_register, semaphore)

        loop = event_loop.get()
        loop.set_debug(True)
        asyncio.run_coroutine_threadsafe(a.transmit(), loop=loop)
        assert semaphore.acquire(timeout=0.1)

def test_simple_input_output_ports(self):
        ujml_code = '<?xml version="1.0"?><ujml version="{}">'.format(uj_version) + '''
                       <f_stoff s="s"/>
                    </ujml>'''
        s = Semaphore(0)
        globs = {"s": s}
        ujml = from_string(ujml_code, globals=globs)
        loop = get_event_loop()
        asyncio.run_coroutine_threadsafe(ujml[0].transmit(), loop=loop)
        assert s.acquire(timeout=0.1)

def wait_for_result(coroutine):
    future = asyncio.run_coroutine_threadsafe(coroutine, loop)
    return future.result()

def cleanup():
    # This is to instruct chain.run() to exit, which will cause the event loop to stop.
    chain._should_stop.set()
    # This will block until the event loop has stopped.
    t.join()
    # The above was needed because the event loop stops when chain.run() returns and then
    # chain.stop() would never finish if we just ran it with run_coroutine_threadsafe().
    loop.run_until_complete(chain.stop())
    loop.close()

def send(self, msg: Message):
        if self._messages is not None:
            return asyncio.run_coroutine_threadsafe(self._messages.put(msg), self._loop)

def await_state(self, wait=None):
        asyncio.run_coroutine_threadsafe(self.join_messages(), self._loop).result(Maybe(wait) | short_timeout)
        return self.data

def close(self):
        #leave session
        async def leave():
            await self.session.stop()
        fut = asyncio.run_coroutine_threadsafe(leave(), self.loop)
        fut.result(WAIT_TIMEOUT)

def call_no_wait(self, user, role, method, *args, **kwargs):
        if not self.is_active():
            raise Exception('Missing or inactive sesssion')
        async def do_call():
            return await self.session.execute(user, role, method, *args, **kwargs)
        future = asyncio.run_coroutine_threadsafe(do_call(), self.loop)
        return future.result(MAX_TIMEOUT)

def close(self):
        #leave session
        async def leave():
            await self.session.stop()
            self._pending_tasks.clear()
        fut = asyncio.run_coroutine_threadsafe(leave(), self.loop)
        fut.result(WAIT_TIMEOUT)

def call_no_wait(self, method, *args, **kwargs):
        """Just schedules task, returns task id"""
        if not self.session or not self.session.active:
            raise Exception('Missing or inactive sesssion')
        async def do_call():
            task_id =  await self.session.execute(method, *args, **kwargs)
            self._pending_tasks[task_id] = Ignore
            return task_id
        future = asyncio.run_coroutine_threadsafe(do_call(), self.loop)
        return future.result(MAX_TIMEOUT)

def editMsg(self,msg,txt):
        asyncio.run_coroutine_threadsafe(self.bot.edit_message(msg,txt), self.bot.loop)

def removeMsg(self,msg):
        asyncio.run_coroutine_threadsafe(self.bot.delete_message(msg), self.bot.loop)

def stop(self):
        '''Cleaning shutdown the running loop and thread.'''
        if self._running:
            async def _stop():
                await self.coz_conn.shutdown()
                self.loop.call_soon(lambda: self.loop.stop())
            asyncio.run_coroutine_threadsafe(_stop(), self.loop).result()
            self.thread.join()
            self._running = False