def check_router_credentials(self, routers: List[BaseIndustrialRouter]):
        """
        Check default credentials for a list of routers.

        :param routers: List of routers to be checked.
        """
        async def check_router_credentials_aio(routers):
            semaphore = Semaphore(100)
            futures = []
            for router in routers:
                if isinstance(router, self.__class__.router_cls):
                    futures.append(asyncio.ensure_future(self.check_default_password(router, semaphore=semaphore)))

            await asyncio.wait(futures, return_when=ALL_COMPLETED, )

        asyncio.run(check_router_credentials_aio(routers), debug=True) 

def cancellable_wait(self, *awaitables: Awaitable[_R], timeout: float = None) -> _R:
        futures = [asyncio.ensure_future(a, loop=self.loop) for a in awaitables + (self.wait(),)]
        try:
            done, pending = await asyncio.wait(
                futures,
                timeout=timeout,
                return_when=asyncio.FIRST_COMPLETED,
                loop=self.loop,
            )
        except CancelledError:
            for future in futures:
                future.cancel()
            raise
        for task in pending:
            task.cancel()
        await asyncio.wait(pending, return_when=asyncio.ALL_COMPLETED, loop=self.loop,)
        if not done:
            raise TimeoutError()
        if self.triggered_token is not None:
            for task in done:
                task.exception()
            raise OperationCancelled(f'Cancellation requested by {self.triggered_token} token')
        return done.pop().result() 

def run(self):
        while True:
            len_list_activity = len(self.list_activity)
            len_list_TV = len(self.list_TV)
            set_TV = set(self.list_TV)
            tasklist = []
            for i in set_TV:
                task = asyncio.ensure_future(bilibiliCilent.handle_1_room_TV(i))
                tasklist.append(task)
            if tasklist:
                await asyncio.wait(tasklist, return_when=asyncio.ALL_COMPLETED)
            else:
                pass

            del self.list_activity[:len_list_activity]
            del self.list_TV[:len_list_TV]
            if len_list_activity == 0 and len_list_TV == 0:
                await asyncio.sleep(1.1)
            else:
                await asyncio.sleep(1.0) 

def _resolve(self, hosts, resolver):
        futures = []
        resolve_results = []

        for each_host in hosts:
            each_future = resolver.query(each_host.target, "A")
            each_future.database_host = each_host
            futures.append(each_future)

            if len(futures) >= 10:
                (resolve_batch, _) = await asyncio.wait(
                    futures, return_when=asyncio.ALL_COMPLETED
                )
                resolve_results += resolve_batch
                futures = []

        if futures:
            (resolve_batch, _) = await asyncio.wait(
                futures, return_when=asyncio.ALL_COMPLETED
            )
            resolve_results += resolve_batch

        return resolve_results 

def save_as_csv(hourly_url: str, daily_url: str):
    tasks = [save_url_to_csv(hourly_url, '%Y-%m-%d %I-%p', 'coinbase-1h-btc-usd.csv'),
             save_url_to_csv(daily_url, '%Y-%m-%d', 'coinbase-1d-btc-usd.csv')]
    # also FIRST_EXCEPTION and ALL_COMPLETED (default)
    done, pending = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
    print('>> done: ', done)
    print('>> pending: ', pending)  # will be empty if using default return_when setting 

def sane_wait_for(futures, *, timeout):
    ensured = [
        asyncio.ensure_future(fut) for fut in futures
    ]
    done, pending = await asyncio.wait(ensured, timeout=timeout, return_when=asyncio.ALL_COMPLETED)

    if len(pending) != 0:
        raise asyncio.TimeoutError()

    return done 

def test_ensure_future_with_run_until_complete_with_wait(self):

        async def f():
            task = asyncio.ensure_future(self.coro())
            done, pending = self.loop.run_until_complete(
                asyncio.wait([task], return_when=asyncio.ALL_COMPLETED))
            task = done.pop()
            return task.result()

        result = self.loop.run_until_complete(f())
        self.assertEqual(result, 42) 

def async_run(self):
        """
        协程主函数
        """
        # 注册信号处理
        loop = asyncio.get_event_loop()
        for signame in {'SIGINT', 'SIGTERM'}:
            loop.add_signal_handler(
                getattr(signal, signame),
                functools.partial(self._exit, signame, loop))

        # 初始化context
        await audit_tools.context.Context().async_init()

        # 启动插件
        plugin_tasks = []
        for plugin_name in self.plugin_loaded:
            plugin_tasks.append(loop.create_task(
                self.plugin_loaded[plugin_name].async_run()))

        # 启动获取扫描结果队列的协程
        task_fetch_rasp_result = loop.create_task(self._fetch_from_queue())

        # 执行获取新扫描任务
        await self._fetch_new_scan()

        # 结束所有协程任务，reset共享内存
        task_fetch_rasp_result.cancel()
        await asyncio.wait({task_fetch_rasp_result})
        for task in plugin_tasks:
            task.cancel()
        await asyncio.wait(set(plugin_tasks), return_when=asyncio.ALL_COMPLETED)
        Communicator().reset_all_value() 

def close(self):
        packet = EndOfStream()
        packet.owner = self.component
        await self.send_packet(packet)
        await asyncio.wait([conn.queue.join() for conn in self.connections], return_when=asyncio.ALL_COMPLETED)
        self.open = False
        self.log.debug("Closing {}".format(self.name)) 

def testSendReceive(self):
        c1 = Component('c1')
        c1.outputs.add(OutputPort('a'))
        c2 = Component('c2')
        c2.outputs.add(OutputPort('b'))
        c3 = Component('c3')
        c3.inputs.add(InputPort('in1'))
        c3.inputs.add(InputPort('in2'))
        graph = Multigraph()
        graph.connect(c1.outputs['a'], c3.inputs['in1'])
        graph.connect(c2.outputs['b'], c3.inputs['in2'])
        # graph.set_initial_packet(c3.inputs['in1'], 6)
        async def send(messages):
            for m in messages:
                await asyncio.sleep(0.2)
                [asyncio.ensure_future(c1.outputs['a'].send_to_all(m)), asyncio.ensure_future(c2.outputs['b'].send_to_all(m))]

            asyncio.ensure_future(c1.outputs['a'].close())
            asyncio.ensure_future(c2.outputs['b'].close())
        async def receive():
            while True:
                print(c3.inputs['in1'].queue)
                res, doing = await asyncio.wait([c3.inputs['in1'].receive(),c3.inputs['in2'].receive()], return_when=asyncio.ALL_COMPLETED)
                print('done receiving')
                print(res.pop().result(),res.pop().result())
                # await asyncio.sleep(0)
        futures = [asyncio.ensure_future(send([1,2,3,4,5])),asyncio.ensure_future(receive())]
        loop = asyncio.get_event_loop()
        loop.run_until_complete(futures[1]) 

def run(self):
        while True:
            len_list_console = len(self.list_console)
            tasklist = []
            for i in self.list_console:
                if isinstance(i, list):
                    # 对10号单独简陋处理
                    for j in range(len(i[0])):
                        if isinstance(i[0][j], list):
                            i[0][j] = await i[0][j][1](*(i[0][j][0]))
                    task = asyncio.ensure_future(i[1](*i[0]))
                else:
                    task = asyncio.ensure_future(i())
                tasklist.append(task)
            if tasklist:
                try:
                    await asyncio.wait(tasklist, return_when=asyncio.ALL_COMPLETED)
                except Exception:
                    Printer().printer(traceback.format_exc(), "Error", "red")
                # print('本批次结束')
            else:
                # print('本批次轮空')
                pass

            if len_list_console == 0:
                await asyncio.sleep(1)
            else:
                self.lock.acquire()
                del self.list_console[:len_list_console]
                self.lock.release()
                await asyncio.sleep(0.3) 

def handle_1_room_TV(real_roomid):
    await asyncio.sleep(random.uniform(0, 1))
    result = await utils.check_room_true(real_roomid)
    if True in result:
        Printer().printer(f"检测到房间 {real_roomid} 的钓鱼操作", "Warning", "red")
    else:
        await bilibili().post_watching_history(real_roomid)
        response = await bilibili().get_giftlist_of_TV(real_roomid)
        json_response = await response.json(content_type=None)
        checklen = json_response['data']['gift']
        num = len(checklen)
        list_available_raffleid = []
        for j in range(0, num):
            raffleid = json_response['data']['gift'][j]['raffleId']
            if Statistics().check_TVlist(raffleid):
                type = json_response['data']['gift'][j]['type']
                time_wait = json_response['data']['gift'][j]['time_wait']
                time_limit = json_response['data']['gift'][j]['time']
                list_available_raffleid.append([type, raffleid, time_wait, time_limit])
        tasklist = []
        num_available = len(list_available_raffleid)
        for k in list_available_raffleid:
            task = asyncio.ensure_future(handle_1_TV_raffle(*k, num_available, real_roomid))
            tasklist.append(task)
        if tasklist:
            await asyncio.wait(tasklist, return_when=asyncio.ALL_COMPLETED) 

def cancel(self) -> None:
        for f in self._pending.values():
            f.cancel()
        if self._pending:
            await asyncio.wait(
                list(self._pending.values()), return_when=asyncio.ALL_COMPLETED
            ) 

def _wait(
        tasks: Tasks,
        *,
        timeout: Optional[float] = None,
        return_when: Any = asyncio.ALL_COMPLETED,
) -> Tuple[Set[asyncio_Task], Set[asyncio_Task]]:
    if not tasks:
        return set(), set()
    done, pending = await asyncio.wait(tasks, timeout=timeout, return_when=return_when)
    return cast(Set[asyncio_Task], done), cast(Set[asyncio_Task], pending) 

def get_nameservers(hosts, logger=None):
    nameservers = []
    top_domains = []

    for hostname in hosts:
        top_domains.append('.'.join(hostname.split('.')[-2:]))

    resolver = aiodns.DNSResolver(loop=asyncio.get_event_loop())

    for top_server_name in list(set(top_domains)):
        try:
            result = await resolver.query(top_server_name, "NS")
            nameservers += list(map(lambda x: x.host, result))
        except:
            pass

    if logger:
        logger.debug(
            "Scopes resolve, found NSes: {}".format(
                nameservers
            )
        )

    futures = []
    for ns in nameservers:
        each_future = resolver.query(ns, "A")
        futures.append(each_future)

    (done_futures, _) = await asyncio.wait(
        futures, return_when=asyncio.ALL_COMPLETED
    )

    nameservers_ips = ['8.8.8.8']

    while done_futures:
        each_future = done_futures.pop()

        try:
            result = each_future.result()
            nameservers_ips += list(map(lambda x: x.host, result))
        except:
            pass

    if logger:
        logger.debug(
            "Scopes resolve, resolved NSes: {}".format(
                nameservers_ips
            )
        )

    return list(set(nameservers_ips))