def block_check(loop):
    while True:
        try:
            time.sleep(1)
            future = asyncio.run_coroutine_threadsafe(asyncio.sleep(0), loop)
            blocked_for = 0
            while True:
                try:
                    future.result(1)
                    break
                except asyncio.TimeoutError:
                    blocked_for += 1
                    task = asyncio.current_task(loop)
                    buffer = io.StringIO()
                    task.print_stack(file=buffer)
                    buffer.seek(0)
                    log.warning("Event loop blocked for longer than %d seconds (%s)\n%s\n%s" % (
                        blocked_for,
                        str(task),
                        str(last_commands),
                        buffer.read()
                    ))
        except Exception:
            pass 

def _emit(self, record: logging.LogRecord) -> None:
        # JSON conversion based on Marsel Mavletkulov's json-log-formatter (MIT license)
        # https://github.com/marselester/json-log-formatter
        content = {
            name: value
            for name, value in record.__dict__.items()
            if name not in EXCLUDE_ATTRS
        }
        content["id"] = str(record.relativeCreated)
        content["msg"] = record.getMessage()
        content["time"] = datetime.fromtimestamp(record.created)

        if record.exc_info:
            content["exc_info"] = self.formatter.formatException(record.exc_info)

        for name, value in content.items():
            if isinstance(value, datetime):
                content[name] = value.astimezone().isoformat()
        asyncio.run_coroutine_threadsafe(self.send(content), loop=self.loop)
        self.lines.append(content) 

def run_async_coroutine(self, coroutine_to_run, timeout):
        """Start coroutine in dedicated thread and await its result with timeout"""
        start_time = time.time()
        coro_future = self.start_async_coroutine(coroutine_to_run)
        # run_coroutine_threadsafe returns future as concurrent.futures.Future() and not asyncio.Future
        # so, we can await it with timeout inside current thread
        try:
            coro_result = coro_future.result(timeout=timeout)
            self.logger.debug("scheduled {} returned {}".format(coroutine_to_run, coro_result))
            return coro_result
        except concurrent.futures.TimeoutError:
            passed = time.time() - start_time
            raise MolerTimeout(timeout=timeout,
                               kind="run_async_coroutine({})".format(coroutine_to_run),
                               passed_time=passed)
        except concurrent.futures.CancelledError:
            raise 

def __handler_get_tx_by_address(self, request, context):
        """Get Transaction by address

        :param request:
        :param context:
        :return:
        """
        params = json.loads(request.meta)
        address = params.pop('address', None)
        index = params.pop('index', None)

        if address is None or index is None:  # or params:
            return loopchain_pb2.Message(code=message_code.Response.fail_illegal_params)

        channel_stub = StubCollection().channel_stubs[request.channel]
        future = asyncio.run_coroutine_threadsafe(
            channel_stub.async_task().get_tx_by_address(address, index),
            self.peer_service.inner_service.loop
        )
        tx_list, next_index = future.result()
        tx_list_dumped = json.dumps(tx_list).encode(encoding=conf.PEER_DATA_ENCODING)

        return loopchain_pb2.Message(code=message_code.Response.success,
                                     meta=str(next_index),
                                     object=tx_list_dumped) 

def __get_status_cache(self, channel_name, time_in_seconds):
        """Cache status data.

        :param channel_name:
        :param time_in_seconds: An essential parameter for the `LRU cache` even if not used.

        :return:
        """
        try:
            channel_stub = StubCollection().channel_stubs[channel_name]
        except KeyError:
            raise ChannelStatusError(f"Invalid channel({channel_name})")

        if self.__status_cache is None:
            self.__status_cache = channel_stub.sync_task().get_status()
        else:
            future = asyncio.run_coroutine_threadsafe(
                channel_stub.async_task().get_status(),
                self.peer_service.inner_service.loop)
            future.add_done_callback(self.__set_status_cache)

        return self.__status_cache 

def Stop(self, request, context):
        """Peer를 중지시킨다

        :param request: 중지요청
        :param context:
        :return: 중지결과
        """
        if request is not None:
            utils.logger.info('Peer will stop... by: ' + request.reason)

        try:
            for channel_name in conf.CHANNEL_OPTION:
                channel_stub = StubCollection().channel_stubs[channel_name]
                asyncio.run_coroutine_threadsafe(channel_stub.async_task().stop(), self.peer_service.inner_service.loop)

            self.peer_service.p2p_server_stop()

        except Exception as e:
            utils.logger.debug("Score Service Already stop by other reason. %s", e)

        return loopchain_pb2.StopReply(status="0") 

def CreateTx(self, request, context):
        """make tx by client request and broadcast it to the network

        :param request:
        :param context:
        :return:
        """
        channel_name = request.channel or conf.LOOPCHAIN_DEFAULT_CHANNEL
        utils.logger.info(f"peer_outer_service::CreateTx request({request.data}), channel({channel_name})")

        channel_stub = StubCollection().channel_stubs[channel_name]
        result_hash = asyncio.run_coroutine_threadsafe(
            channel_stub.async_task().create_tx(request.data),
            self.peer_service.inner_service.loop
        ).result()

        return loopchain_pb2.CreateTxReply(
            response_code=message_code.Response.success,
            tx_hash=result_hash,
            more_info='') 

def GetPrecommitBlock(self, request, context):
        """Return the precommit bock.

        :param request:
        :param context:
        :return: loopchain.proto 의 PrecommitBlockReply 참고,
        """

        channel_name = conf.LOOPCHAIN_DEFAULT_CHANNEL if request.channel == '' else request.channel
        channel_stub = StubCollection().channel_stubs[channel_name]
        future = asyncio.run_coroutine_threadsafe(
            channel_stub.async_task().get_precommit_block(last_block_height=request.last_block_height),
            self.peer_service.inner_service.loop
        )
        response_code, response_message, block = future.result()

        return loopchain_pb2.PrecommitBlockReply(
            response_code=response_code, response_message=response_message, block=block) 

def GetInvokeResult(self, request, context):
        """get invoke result by tx_hash

        :param request: request.tx_hash = tx_hash
        :param context:
        :return: verify result
        """
        channel_name = conf.LOOPCHAIN_DEFAULT_CHANNEL if request.channel == '' else request.channel
        utils.logger.debug(f"peer_outer_service:GetInvokeResult in channel({channel_name})")

        channel_stub = StubCollection().channel_stubs[channel_name]
        future = asyncio.run_coroutine_threadsafe(
            channel_stub.async_task().get_invoke_result(request.tx_hash),
            self.peer_service.inner_service.loop
        )
        response_code, result = future.result()

        return loopchain_pb2.GetInvokeResultReply(response_code=response_code, result=result) 

def AnnounceUnconfirmedBlock(self, request, context):
        """Send the UnconfirmedBlock includes collected transactions to reps and request to verify it.

        :param request:
        :param context:
        :return:
        """
        channel_name = conf.LOOPCHAIN_DEFAULT_CHANNEL if request.channel == '' else request.channel
        channel_stub = StubCollection().channel_stubs[channel_name]

        try:
            round_ = request.round_
        except AttributeError:
            round_ = 0

        asyncio.run_coroutine_threadsafe(
            channel_stub.async_task().announce_unconfirmed_block(request.block, round_),
            self.peer_service.inner_service.loop
        )
        return loopchain_pb2.CommonReply(response_code=message_code.Response.success, message="success") 

def setup(hass, config):
    if (any(conf[CONF_TIME_AS] in (TZ_DEVICE_UTC, TZ_DEVICE_LOCAL)
           for conf in (config.get(DT_DOMAIN) or [])
           if conf[CONF_PLATFORM] == DOMAIN)):
        pkg = config[DOMAIN][CONF_TZ_FINDER]
        try:
            asyncio.run_coroutine_threadsafe(
                async_process_requirements(
                    hass, '{}.{}'.format(DOMAIN, DT_DOMAIN), [pkg]),
                hass.loop
            ).result()
        except RequirementsNotFound:
            _LOGGER.debug('Process requirements failed: %s', pkg)
            return False
        else:
            _LOGGER.debug('Process requirements suceeded: %s', pkg)

        if pkg.split('==')[0].strip().endswith('L'):
            from timezonefinderL import TimezoneFinder
        else:
            from timezonefinder import TimezoneFinder
        hass.data[DOMAIN] = TimezoneFinder()

    return True 

def init_drivers():
    """
    bring up all the drivers
    this function should be called by UMRManager
    :return:
    """
    config = UMRConfig.config.Driver

    for driver_name, driver_config in config.items():
        if driver_config.Base not in driver_class_lookup_table:
            logger.error(f'Base driver "{driver_config.Base}" not found')
            exit(-1)
        driver: BaseDriverMixin = driver_class_lookup_table[driver_config.Base](driver_name)
        driver.start()
        driver_lookup_table[driver_name] = driver

    loop = asyncio.get_event_loop()

    for driver_name in config.keys():
        asyncio.run_coroutine_threadsafe(__post_init(driver_name), loop) 

def handle_message(self, timestamp, source, groupID, message, attachments):
        """ Handle Signal message. It should be a command """

        logger.debug("Received Message {} {} {} {} {}".format(
            timestamp, message, groupID, message, attachments))

        if source in cfg.SIGNAL_CONTACTS:
            future = asyncio.run_coroutine_threadsafe(self.handle_command(message), self.alarm.work_loop)
            ret = future.result(10)

            m = "Signal {} : {}".format(source, ret)
            logger.info(m)
        else:
            m = "Signal {} (UNK): {}".format(source, message)
            logger.warning(m)

        self.send_message(m, EventLevel.INFO)
        ps.sendNotification(Notification(sender=self.name, message=m, level=EventLevel.INFO)) 

def handle_message(self, timestamp: str, source: str, message: str) -> None:
        """ Handle GSM message. It should be a command """

        logger.debug("Received: {} {} {}".format(
            timestamp, source, message))

        if source in cfg.GSM_CONTACTS:
            future = asyncio.run_coroutine_threadsafe(self.handle_command(message), self.alarm.work_loop)
            ret = future.result(10)

            m = "GSM {}: {}".format(source, ret)
            logger.info(m)
        else:
            m = "GSM {} (UNK): {}".format(source, message)
            logger.warning(m)

        self.send_message(m, EventLevel.INFO)
        ps.sendNotification(Notification(sender=self.name, message=m, level=EventLevel.INFO)) 

def main(self):
        print('start')
        async_q = self._queue.async_q
        main_loop = asyncio.get_event_loop()
        while not (self._stopped and async_q.empty()):

            try:
                event = self.queue.get_nowait()
            except asyncio.QueueEmpty:
                pass
            else:
                asyncio.run_coroutine_threadsafe(
                    self.event_hadler(event),
                    main_loop
                )
                async_q.task_done()
            await asyncio.sleep(0.0001) 

def run_in_order_threadsafe(awaitables, loop, timeout=0.5, block=True):
    """"Given a sequence of awaitables, schedule each threadsafe in order
    optionally blocking until completion.

    Returns a `concurrent.futures.Future` which can be used to wait on the
    result returned from the last awaitable. If `block` is `True` the final
    result will be waited on before returning control to the caller.
    """
    future = asyncio.run_coroutine_threadsafe(
        await_in_order(awaitables, loop, timeout),
        loop
    )

    if block:
        if not loop.is_running():
            result = loop.run_until_complete(
                asyncio.wrap_future(future, loop=loop))
            assert result is future.result()
        else:
            future.result(timeout)

    return future 

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 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 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_async(loop, coro):
    """Run an async function as a non-async function, blocking till it's done"""
    # When we bump minimum support to 3.7, use run()
    return asyncio.run_coroutine_threadsafe(coro, loop).result() 

def run_coroutine(coro):
    return asyncio.run_coroutine_threadsafe(coro, get_event_loop()) 

def write(self, data):
        ''' Write to serial port, called from UI thread '''
        if self.proto and async_main_loop:
            async_main_loop.call_soon_threadsafe(self._write, data)
            # asyncio.run_coroutine_threadsafe(self.proto.send_message, async_main_loop)
        else:
            self.log.warning('Comms: Cannot write to closed connection: ' + data)
            # self.app.main_window.async_display("<<< {}".format(data)) 

def write(self, data):
        ''' Write to serial port, called from UI thread '''
        if self.proto and async_main_loop:
            #self.log.debug('CommsNet: writing ' + data)
            async_main_loop.call_soon_threadsafe(self._write, data)
            #asyncio.run_coroutine_threadsafe(self.proto.send_message, async_main_loop)
        else:
            self.log.warning('CommsNet: Cannot write to closed connection: ' + data) 

def submit(self, fn, *args, **kwargs):
        coro = fn(*args, **kwargs)
        return asyncio.run_coroutine_threadsafe(coro, self._loop) 

def start_async_coroutine(self, coroutine_to_run):
        """Start coroutine in dedicated thread, don't await its result"""
        # we are scheduling to other thread (so, can't use asyncio.ensure_future() )
        self.logger.debug("scheduling {} into {}".format(coroutine_to_run, self.ev_loop))
        coro_future = asyncio.run_coroutine_threadsafe(coroutine_to_run, loop=self.ev_loop)
        return coro_future 

def run_coro(self, coro):
        if not self.loop.is_running():
            raise Exception("Loop is not running")
        future = asyncio.run_coroutine_threadsafe(coro, self.loop)
        return future.result() 

def run(self):
        while not self._stop_ev.wait(self.interval):
            if self._last_ack + self.heartbeat_timeout < time.perf_counter():
                log.warning("Shard ID %s has stopped responding to the gateway. Closing and restarting.", self.shard_id)
                coro = self.ws.close(4000)
                f = asyncio.run_coroutine_threadsafe(coro, loop=self.ws.loop)

                try:
                    f.result()
                except Exception:
                    pass
                finally:
                    self.stop()
                    return

            data = self.get_payload()
            log.debug(self.msg, data['d'])
            coro = self.ws.send_as_json(data)
            f = asyncio.run_coroutine_threadsafe(coro, loop=self.ws.loop)
            try:
                # block until sending is complete
                total = 0
                while True:
                    try:
                        f.result(10)
                        break
                    except concurrent.futures.TimeoutError:
                        total += 10
                        try:
                            frame = sys._current_frames()[self._main_thread_id]
                        except KeyError:
                            msg = self.block_msg
                        else:
                            stack = traceback.format_stack(frame)
                            msg = '%s\nLoop thread traceback (most recent call last):\n%s' % (self.block_msg, ''.join(stack))
                        log.warning(msg, total)

            except Exception:
                self.stop()
            else:
                self._last_send = time.perf_counter() 

def _speak(self, speaking):
        try:
            asyncio.run_coroutine_threadsafe(self.client.ws.speak(speaking), self.client.loop)
        except Exception as e:
            log.info("Speaking call in player failed: %s", e) 

def sync(coro, loop: asyncio.AbstractEventLoop):
        if asyncio.iscoroutine(coro):
            # Run async function in the loop and return the value or raise the exception
            return asyncio.run_coroutine_threadsafe(coro, loop=loop).result()

        return coro 

def sync(loop, func, *args, callback_timeout=None, **kwargs):
    """
    Run coroutine in loop running in separate thread.
    """

    e = threading.Event()
    main_tid = threading.get_ident()
    result = [None]
    error = [False]

    async def f():
        try:
            if main_tid == threading.get_ident():
                raise RuntimeError("sync() called from thread of running loop")
            await asyncio.sleep(0)
            thread_state.asynchronous = True
            future = func(*args, **kwargs)
            if callback_timeout is not None:
                future = asyncio.wait_for(future, callback_timeout)
            result[0] = await future
        except Exception:
            error[0] = sys.exc_info()
        finally:
            thread_state.asynchronous = False
            e.set()

    asyncio.run_coroutine_threadsafe(f(), loop=loop)
    if callback_timeout is not None:
        if not e.wait(callback_timeout):
            raise TimeoutError("timed out after %s s." % (callback_timeout,))
    else:
        while not e.is_set():
            e.wait(10)
    if error[0]:
        typ, exc, tb = error[0]
        raise exc.with_traceback(tb)
    else:
        return result[0]