def run_in_executor(self, executor, callback, *args):
        """Run callback in executor.

        If no executor is provided, the default executor will be used, which defers execution to
        a background thread.
        """
        self._logger.debug('Running callback {} with args {} in executor'.format(callback, args))
        if isinstance(callback, asyncio.Handle):
            assert not args
            assert not isinstance(callback, asyncio.TimerHandle)
            if callback._cancelled:
                f = asyncio.Future()
                f.set_result(None)
                return f
            callback, args = callback.callback, callback.args

        if executor is None:
            self._logger.debug('Using default executor')
            executor = self.__default_executor

        if executor is None:
            self._logger.debug('Creating default executor')
            executor = self.__default_executor = QThreadExecutor()

        return asyncio.wrap_future(executor.submit(callback, *args)) 

def _Compute(
      self,
      request: executor_pb2.ComputeRequest,
      context: grpc.ServicerContext,
  ) -> executor_pb2.ComputeResponse:
    """Asynchronous implemention of `Compute`."""
    py_typecheck.check_type(request, executor_pb2.ComputeRequest)
    try:
      value_id = str(request.value_ref.id)
      with self._lock:
        future_val = asyncio.wrap_future(self._values[value_id])
      val = await future_val
      result_val = await val.compute()
      val_type = val.type_signature
      value_proto, _ = executor_service_utils.serialize_value(
          result_val, val_type)
      return executor_pb2.ComputeResponse(value=value_proto)
    except (ValueError, TypeError) as err:
      _set_invalid_arg_err(context, err)
      return executor_pb2.ComputeResponse() 

def get(self):
        consumer_token = ConsumerToken(
            self.authenticator.client_id,
            self.authenticator.client_secret,
        )

        handshaker = Handshaker(
            self.authenticator.mw_index_url, consumer_token
        )

        redirect, request_token = await wrap_future(
            self.authenticator.executor.submit(handshaker.initiate)
        )

        self.set_secure_cookie(
            AUTH_REQUEST_COOKIE_NAME,
            jsonify(request_token),
            expires_days=1,
            path=url_path_join(self.base_url, 'hub', 'oauth_callback'),
            httponly=True)
        self.log.info('oauth redirect: %r', redirect)

        self.redirect(redirect) 

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 ask(self, dst, content=None, dst_node=None):
        """Send request and wait response"""
        if not dst_node:
            dst_node = self.registery.choice_dst_node(dst)
        msg = self.registery.create_message(
            dst=dst,
            is_ask=True,
            content=content,
            src=self.actor.name,
            dst_node=dst_node,
        )
        if msg.is_local:
            future = ThreadFuture()
            msg.future = future
            if self.actor.is_async:
                return asyncio.wrap_future(future)
            else:
                return future.result()
        else:
            return self._actor_client.ask(msg) 

def watch_build_pods(self):
        """Watch build pods

        Every build_cleanup_interval:
        - delete stopped build pods
        - delete running build pods older than build_max_age
        """
        while True:
            try:
                await asyncio.wrap_future(
                    self.executor.submit(
                        lambda: Build.cleanup_builds(
                            self.kube_client,
                            self.build_namespace,
                            self.build_max_age,
                        )
                    )
                )
            except Exception:
                app_log.exception("Failed to cleanup build pods")
            await asyncio.sleep(self.build_cleanup_interval) 

def _get_pods(self):
        """Get information about build and user pods"""
        app_log.info("Getting pod statistics")
        k8s = self.settings["kubernetes_client"]
        pool = self.settings["executor"]

        get_user_pods = asyncio.wrap_future(
            pool.submit(
                k8s.list_namespaced_pod,
                self.settings["build_namespace"],
                label_selector="app=jupyterhub,component=singleuser-server",
            )
        )

        get_build_pods = asyncio.wrap_future(
            pool.submit(
                k8s.list_namespaced_pod,
                self.settings["build_namespace"],
                label_selector="component=binderhub-build",
            )
        )

        return await asyncio.gather(get_user_pods, get_build_pods) 

def test_fetchall_prevents_sqlite_misuse(self):
        # test that calling fetchall sufficiently avoids the race
        attempts = 0

        def executemany_fetchall(query, params):
            self.db.executemany(query, params).fetchall()

        while attempts < self.max_misuse_attempts:
            f1 = asyncio.wrap_future(
                self.loop.run_in_executor(
                    self.executor, executemany_fetchall, "update test1 set val='derp' where id=?",
                    ((str(i),) for i in range(2))
                )
            )
            f2 = asyncio.wrap_future(
                self.loop.run_in_executor(
                    self.executor, executemany_fetchall, "update test2 set val='derp' where id=?",
                    ((str(i),) for i in range(2))
                )
            )
            attempts += 1
            await asyncio.gather(f1, f2) 

def _await_loopsafe(self, *coros_or_futures):
        """ 事件循环安全的异步等待。

        Args:
            *coros_or_futures: coroutine或future对象列表。

        Returns:
            返回coros_or_futures的返回结果列表。
        """
        current_loop = asyncio.get_running_loop()
        loop = self._loop
        if loop is None:
            loop = current_loop

        async def _execute_loop():
            with h.enter(self._handlers):
                r = await asyncio.gather(*coros_or_futures)
                return r
        fut = asyncio.run_coroutine_threadsafe(_execute_loop(), loop)
        result = await asyncio.wrap_future(fut)

        return result 

def connect(cls, path: Union[bytes, str], *args, **kwargs):
        db = cls()
        db.connection = await wrap_future(db.executor.submit(sqlite3.connect, path, *args, **kwargs))
        return db 

def executescript(self, script: str) -> Awaitable:
        return wrap_future(self.executor.submit(self.connection.executescript, script)) 

def execute_fetchall(self, sql: str, parameters: Iterable = None) -> Awaitable[Iterable[sqlite3.Row]]:
        parameters = parameters if parameters is not None else []
        def __fetchall(conn: sqlite3.Connection, *args, **kwargs):
            return conn.execute(*args, **kwargs).fetchall()
        return wrap_future(self.executor.submit(__fetchall, self.connection, sql, parameters)) 

def run(self, fun, *args, **kwargs) -> Awaitable:
        return wrap_future(self.executor.submit(self.__run_transaction, fun, *args, **kwargs)) 

def run_with_foreign_keys_disabled(self, fun, *args, **kwargs) -> Awaitable:
        return wrap_future(
            self.executor.submit(self.__run_transaction_with_foreign_keys_disabled, fun, *args, **kwargs)
        ) 

def test_submit(self):
        with warnings.catch_warnings(record=True) as w:
            expected = g(10, 9, z=8)
            f = self.pool.submit(g, 10, 9, z=8)
            actual = loop.run_until_complete(asyncio.wrap_future(f))
            self.assertEqual(actual, expected)

            self.assertNoWarnings(w) 

def test_submit_with_exception(self):
        with warnings.catch_warnings(record=True) as w:
            f = self.pool.submit(exc, 'Okay then')
            with self.assertRaises(RuntimeError):
                loop.run_until_complete(asyncio.wrap_future(f))

            self.assertNoWarnings(w) 

def CreateCall(
      self,
      request: executor_pb2.CreateCallRequest,
      context: grpc.ServicerContext,
  ) -> executor_pb2.CreateCallResponse:
    """Creates a call embedded in the executor."""
    py_typecheck.check_type(request, executor_pb2.CreateCallRequest)
    try:
      function_id = str(request.function_ref.id)
      argument_id = str(request.argument_ref.id)
      with self._lock:
        function_val = self._values[function_id]
        argument_val = self._values[argument_id] if argument_id else None

      async def _processing():
        function = await asyncio.wrap_future(function_val)
        argument = await asyncio.wrap_future(
            argument_val) if argument_val is not None else None
        return await self._executor.create_call(function, argument)

      coro = _processing()
      result_fut = self._run_coro_threadsafe_with_tracing(coro)
      result_id = str(uuid.uuid4())
      with self._lock:
        self._values[result_id] = result_fut
      return executor_pb2.CreateCallResponse(
          value_ref=executor_pb2.ValueRef(id=result_id))
    except (ValueError, TypeError) as err:
      _set_invalid_arg_err(context, err)
      return executor_pb2.CreateCallResponse() 

def CreateTuple(
      self,
      request: executor_pb2.CreateTupleRequest,
      context: grpc.ServicerContext,
  ) -> executor_pb2.CreateTupleResponse:
    """Creates a tuple embedded in the executor."""
    py_typecheck.check_type(request, executor_pb2.CreateTupleRequest)
    try:
      with self._lock:
        elem_futures = [self._values[e.value_ref.id] for e in request.element]
      elem_names = [
          str(elem.name) if elem.name else None for elem in request.element
      ]

      async def _processing():
        elem_values = await asyncio.gather(
            *[asyncio.wrap_future(v) for v in elem_futures])
        elements = list(zip(elem_names, elem_values))
        anon_tuple = anonymous_tuple.AnonymousTuple(elements)
        return await self._executor.create_tuple(anon_tuple)

      result_fut = self._run_coro_threadsafe_with_tracing(_processing())
      result_id = str(uuid.uuid4())
      with self._lock:
        self._values[result_id] = result_fut
      return executor_pb2.CreateTupleResponse(
          value_ref=executor_pb2.ValueRef(id=result_id))
    except (ValueError, TypeError) as err:
      _set_invalid_arg_err(context, err)
      return executor_pb2.CreateTupleResponse() 

def CreateSelection(
      self,
      request: executor_pb2.CreateSelectionRequest,
      context: grpc.ServicerContext,
  ) -> executor_pb2.CreateSelectionResponse:
    """Creates a selection embedded in the executor."""
    py_typecheck.check_type(request, executor_pb2.CreateSelectionRequest)
    try:
      with self._lock:
        source_fut = self._values[request.source_ref.id]

      async def _processing():
        source = await asyncio.wrap_future(source_fut)
        which_selection = request.WhichOneof('selection')
        if which_selection == 'name':
          coro = self._executor.create_selection(source, name=request.name)
        else:
          coro = self._executor.create_selection(source, index=request.index)
        return await coro

      result_fut = self._run_coro_threadsafe_with_tracing(_processing())
      result_id = str(uuid.uuid4())
      with self._lock:
        self._values[result_id] = result_fut
      return executor_pb2.CreateSelectionResponse(
          value_ref=executor_pb2.ValueRef(id=result_id))
    except (ValueError, TypeError) as err:
      _set_invalid_arg_err(context, err)
      return executor_pb2.CreateSelectionResponse() 

def _delegate_with_trace_ctx(coro, event_loop):
  coro_with_trace_ctx = tracing.wrap_coroutine_in_current_trace_context(coro)
  return asyncio.wrap_future(
      asyncio.run_coroutine_threadsafe(coro_with_trace_ctx, event_loop)) 

def authenticate(self, handler, data=None):
        consumer_token = ConsumerToken(
            self.client_id,
            self.client_secret,
        )

        handshaker = Handshaker(
            self.mw_index_url, consumer_token
        )
        request_token = dejsonify(handler.get_secure_cookie(AUTH_REQUEST_COOKIE_NAME))
        handler.clear_cookie(AUTH_REQUEST_COOKIE_NAME)
        access_token = await wrap_future(self.executor.submit(
            handshaker.complete, request_token, handler.request.query
        ))

        identity = await wrap_future(self.executor.submit(handshaker.identify, access_token))
        if identity and 'username' in identity:
            # this shouldn't be necessary anymore,
            # but keep for backward-compatibility
            return {
                'name': identity['username'].replace(' ', '_'),
                'auth_state': {
                    'ACCESS_TOKEN_KEY': access_token.key,
                    'ACCESS_TOKEN_SECRET': access_token.secret,
                    'MEDIAWIKI_USER_IDENTITY': identity,

                }
            }
        else:
            self.log.error("No username found in %s", identity) 

def __set_result(self, message, loop):
        concurrent = asyncio.run_coroutine_threadsafe(self.__upgrade_state(message), loop=loop)
        return asyncio.wrap_future(concurrent) 

def init(self, do_before_result: Callable[[], None]) -> asyncio.Future:
        self.to_async = futures.Future()
        do_before_result()
        return asyncio.wrap_future(self.to_async) 

def send_to_sync(self, value: Any, do_before_result: Callable[[], None]) -> Any:
        self.to_async = futures.Future()
        self.to_sync.set_result(value)
        do_before_result()
        return await asyncio.wrap_future(self.to_async) 

def test_wrap_future(self):

        def run(arg):
            return (arg, threading.get_ident())
        ex = concurrent.futures.ThreadPoolExecutor(1)
        f1 = ex.submit(run, 'oi')
        f2 = asyncio.wrap_future(f1, loop=self.loop)
        res, ident = self.loop.run_until_complete(f2)
        self.assertIsInstance(f2, asyncio.Future)
        self.assertEqual(res, 'oi')
        self.assertNotEqual(ident, threading.get_ident()) 

def test_wrap_future_future(self):
        f1 = asyncio.Future(loop=self.loop)
        f2 = asyncio.wrap_future(f1)
        self.assertIs(f1, f2) 

def test_wrap_future_use_global_loop(self, m_events):
        def run(arg):
            return (arg, threading.get_ident())
        ex = concurrent.futures.ThreadPoolExecutor(1)
        f1 = ex.submit(run, 'oi')
        f2 = asyncio.wrap_future(f1)
        self.assertIs(m_events.get_event_loop.return_value, f2._loop) 

def test_wrap_future_cancel(self):
        f1 = concurrent.futures.Future()
        f2 = asyncio.wrap_future(f1, loop=self.loop)
        f2.cancel()
        test_utils.run_briefly(self.loop)
        self.assertTrue(f1.cancelled())
        self.assertTrue(f2.cancelled()) 

def test_wrap_future_cancel2(self):
        f1 = concurrent.futures.Future()
        f2 = asyncio.wrap_future(f1, loop=self.loop)
        f1.set_result(42)
        f2.cancel()
        test_utils.run_briefly(self.loop)
        self.assertFalse(f1.cancelled())
        self.assertEqual(f1.result(), 42)
        self.assertTrue(f2.cancelled()) 

def handle_ask(self, request, data, dst, content_encoding):
        message_id = request.headers.get('actor-ask-id')
        src = request.headers.get('actor-ask-src')
        src_node = request.headers.get('actor-ask-src-node')
        dst_node = request.headers.get('actor-ask-dst-node')
        if not dst_node:
            dst_node = self.registery.current_node_name
        if not message_id:
            message_id = self.registery.generate_message_id()
        thread_future = ThreadFuture()
        msg = self.registery.create_message(
            id=message_id, content=data, is_ask=True,
            src=src, src_node=src_node, require_ack=False,
            dst=dst, dst_node=dst_node, future=thread_future,
        )
        try:
            self.queue.op_inbox(msg)
        except ActorStateError as ex:
            LOG.warning(ex)
            return Response(body=str(ex), status=400)
        result = await asyncio.wrap_future(thread_future)
        if result is None:
            return Response(status=204)
        result = ActorMessage.raw_encode(result, content_encoding=content_encoding)
        headers = {}
        if content_encoding == ContentEncoding.JSON:
            headers['content-type'] = 'application/json; charset=utf-8'
        else:
            headers['actor-content-encoding'] = content_encoding.value
        return Response(body=result, headers=headers)