def mocked_executor():
    """Context that patches the derived executor classes to return the same
    executor object. Also patches the future object returned by executor's
    submit()."""

    import importlib
    import concurrent.futures as futures
    import qiskit.providers.basicaer.basicaerjob as basicaerjob

    executor = unittest.mock.MagicMock(spec=futures.Executor)
    executor.submit.return_value = unittest.mock.MagicMock(spec=futures.Future)
    mock_options = {'return_value': executor, 'autospec': True}
    with patch.object(futures, 'ProcessPoolExecutor', **mock_options),\
            patch.object(futures, 'ThreadPoolExecutor', **mock_options):
        importlib.reload(basicaerjob)
        yield basicaerjob.BasicAerJob, executor 

def get_executor() -> Executor:
    """
    Get the current context's executor pool.

    Returns
    -------
    Executor

    Raises
    ------
    NotConnectedError
        If there is not a pool for this context
    """
    try:
        if _is_notebook:
            return executor.get(_jupyter_context["executor"])
        else:
            return executor.get()
    except (LookupError, KeyError):
        raise NotConnectedError 

def context(connection: Connection, executor_pool: Executor, redis_conn: StrictRedis):
    """
    Context manager which can be used to temporarily provide a connection, redis client
    and pool.

    Parameters
    ----------
    connection : Connection
        Connection which will be used within this context
    executor_pool : Executor
        Executor pool which will be used within this context
    redis_conn : StrictRedis
        Redis client which will be used within this context
    """
    db_token = db.set(connection)
    redis_token = redis_connection.set(redis_conn)
    executor_token = executor.set(executor_pool)
    try:
        yield
    finally:
        db.reset(db_token)
        redis_connection.reset(redis_token)
        executor.reset(executor_token) 

def _run(self):
        first_completed = concurrent.FIRST_COMPLETED

        if self._get_max_tasks() < 1:
            raise RuntimeError("Executor has no workers")

        try:
            while not self.goal(self.learner):
                futures = self._get_futures()
                done, _ = concurrent.wait(futures, return_when=first_completed)
                self._process_futures(done)
        finally:
            remaining = self._remove_unfinished()
            if remaining:
                concurrent.wait(remaining)
            self._cleanup() 

def _run(self):
        first_completed = asyncio.FIRST_COMPLETED

        if self._get_max_tasks() < 1:
            raise RuntimeError("Executor has no workers")

        try:
            while not self.goal(self.learner):
                futures = self._get_futures()
                done, _ = await asyncio.wait(
                    futures, return_when=first_completed, loop=self.ioloop
                )
                self._process_futures(done)
        finally:
            remaining = self._remove_unfinished()
            if remaining:
                await asyncio.wait(remaining)
            self._cleanup() 

def __init__(
        self,
        function: ElementRenderFunction,
        state_parameters: Optional[str],
        run_in_executor: Union[bool, Executor] = False,
    ):
        super().__init__()
        self._function = function
        signature, var_positional, var_keyword = _extract_signature(function)
        self._function_signature = signature
        self._function_var_positional_param = var_positional
        self._function_var_keyword_param = var_keyword
        self._layout: Optional["AbstractLayout"] = None
        self._cross_update_state: Dict[str, Any] = {}
        self._cross_update_parameters: List[str] = list(
            map(str.strip, (state_parameters or "").split(","))
        )
        self._state: Dict[str, Any] = {}
        self._state_updated: bool = False
        self._animation_futures: List[asyncio.Future[None]] = []
        self._run_in_executor = run_in_executor 

def call_in_executor(self, func: Callable, *args, executor: Union[Executor, str] = None,
                         **kwargs) -> Awaitable:
        """
        Call the given callable in an executor.

        :param func: the callable to call
        :param args: positional arguments to call the callable with
        :param executor: either an :class:`~concurrent.futures.Executor` instance, the resource
            name of one or ``None`` to use the event loop's default executor
        :param kwargs: keyword arguments to call the callable with
        :return: an awaitable that resolves to the return value of the call

        """
        assert check_argument_types()
        if isinstance(executor, str):
            executor = self.require_resource(Executor, executor)

        return asyncio_extras.call_in_executor(func, *args, executor=executor, **kwargs) 

def executor(self) -> Executor:
        """
        Return the concurrent.futures executor instance to use for this worker.

        Can be passed via the `executor` argument to `__init__` or set `use_threads=True` to use the Threaded executor.

        The worker will use a process based executor by default.

        :return: executor instance
        :rtype: concurrent.futures.Executor
        """
        if self._executor is None:
            self._executor = self._executor_class(max_workers=self.concurrency)
        return self._executor 

def __init__(self, delegate, executor):
        """
        Wrap the specified synchronous delegate instance, and submit() all method calls to the
        specified Executor instance.
        """
        self._delegate = delegate
        self._executor = executor 

def set_running_or_notify_cancel(self):
            """Mark the future as running or process any cancel notifications.

            Should only be used by Executor implementations and unit tests.

            If the future has been cancelled (cancel() was called and returned
            True) then any threads waiting on the future completing (though
            calls to as_completed() or wait()) are notified and False is
            returned.

            If the future was not cancelled then it is put in the running state
            (future calls to running() will return True) and True is returned.

            This method should be called by Executor implementations before
            executing the work associated with this future. If this method
            returns False then the work should not be executed.

            Returns:
                False if the Future was cancelled, True otherwise.

            Raises:
                RuntimeError: if this method was already called or if
                    set_result() or set_exception() was called.
            """
            with self._condition:
                if self._state == CANCELLED:
                    self._state = CANCELLED_AND_NOTIFIED
                    for waiter in self._waiters:
                        waiter.add_cancelled(self)
                    # self._condition.notify_all() is not necessary because
                    # self.cancel() triggers a notification.
                    return False
                elif self._state == PENDING:
                    self._state = RUNNING
                    return True
                else:
                    LOGGER.critical('Future %s in unexpected state: %s',
                                    id(self),
                                    self._state)
                    raise RuntimeError('Future in unexpected state') 

def set_result(self, result):
            """Sets the return value of work associated with the future.

            Should only be used by Executor implementations and unit tests.
            """
            with self._condition:
                self._result = result
                self._state = FINISHED
                for waiter in self._waiters:
                    waiter.add_result(self)
                self._condition.notify_all()
            self._invoke_callbacks() 

def set_exception(self, exception):
            """Sets the result of the future as being the given exception.

            Should only be used by Executor implementations and unit tests.
            """
            with self._condition:
                self._exception = exception
                self._state = FINISHED
                for waiter in self._waiters:
                    waiter.add_exception(self)
                self._condition.notify_all()
            self._invoke_callbacks() 

def shutdown(self, wait=True):
            """Clean-up the resources associated with the Executor.

            It is safe to call this method several times. Otherwise, no other
            methods can be called after this one.

            Args:
                wait: If True then shutdown will not return until all running
                    futures have finished executing and the resources used by
                    the executor have been reclaimed.
            """
            pass 

def bind_context(
    connection: Connection, executor_pool: Executor, redis_conn: StrictRedis
):
    """
    Set the current context's connection, executor and redis connection, replacing
    any that were previously set.

    Parameters
    ----------
    connection : Connection
        Connection to set
    executor_pool : Executor
        Executor to be the new pool
    redis_conn : StrictRedis
        Redis client

    """
    if _is_notebook:
        global _jupyter_context
        _jupyter_context["db"] = connection
        _jupyter_context["executor"] = executor_pool
        _jupyter_context["redis_connection"] = redis_conn
    else:
        db.set(connection)
        executor.set(executor_pool)
        redis_connection.set(redis_conn) 

def __init__(self, max_workers: Optional[int] = None):
        self._max_workers = (
            max_workers or multiprocessing.cpu_count()
        )  # type: int
        self._pool = None  # type: Optional[Executor] 

def submit(self, fn, *args, **kwargs):
    with self._shutdown_lock:
      if self._shutdown:
        raise RuntimeError('cannot schedule new futures after shutdown')

    f = futures.Future()
    t = threading.Thread(
        target=_worker, args=(f, fn, args, kwargs),
        name='Executor for %s args=%s kwargs=%s' % (fn, args, kwargs))
    t.start()
    return f 

def _ensure_executor(executor):
    if executor is None:
        executor = _default_executor()

    if isinstance(executor, concurrent.Executor):
        return executor
    elif with_ipyparallel and isinstance(executor, ipyparallel.Client):
        return executor.executor()
    elif with_distributed and isinstance(executor, distributed.Client):
        return executor.get_executor()
    else:
        raise TypeError(
            "Only a concurrent.futures.Executor, distributed.Client,"
            " or ipyparallel.Client can be used."
        ) 

def element(
    function: Callable[..., Any],
    *,
    state: Optional[str] = None,
    run_in_executor: Union[bool, Executor] = False,
) -> ElementConstructor:
    ... 

def element(
    *, state: Optional[str] = None, run_in_executor: Union[bool, Executor] = False
) -> Callable[[ElementRenderFunction], ElementConstructor]:
    ... 

def element(
    function: Optional[ElementRenderFunction] = None,
    state: Optional[str] = None,
    run_in_executor: Union[bool, Executor] = False,
) -> Callable[..., Any]:
    """A decorator for defining an :class:`Element`.

    Parameters:
        function:
            The function that will render a :term:`VDOM` model.
        state:
            A comma seperated string of function parameters that should be retained
            across updates unless explicitely changed when calling :meth:`Element.update`.
        run_in_executor:
            Whether or not to run the given ``function`` in a background thread. This is
            useful for long running and blocking operations that might prevent other
            elements from rendering in the meantime.
    """

    def setup(func: ElementRenderFunction) -> ElementConstructor:

        if not inspect.iscoroutinefunction(func):
            raise TypeError(f"Expected a coroutine function, not {func}")

        @wraps(func)
        def constructor(*args: Any, **kwargs: Any) -> Element:
            element = Element(func, state, run_in_executor)
            element.update(*args, **kwargs)
            return element

        return constructor

    if function is not None:
        return setup(function)
    else:
        return setup 

def threadpool(self, executor: Union[Executor, str] = None):
        """
        Return an asynchronous context manager that runs the block in a (thread pool) executor.

        :param executor: either an :class:`~concurrent.futures.Executor` instance, the resource
            name of one or ``None`` to use the event loop's default executor
        :return: an asynchronous context manager

        """
        assert check_argument_types()
        if isinstance(executor, str):
            executor = self.require_resource(Executor, executor)

        return asyncio_extras.threadpool(executor) 

def special_executor(context):
    executor = ThreadPoolExecutor(1)
    context.add_resource(executor, 'special', types=[Executor])
    yield executor
    executor.shutdown() 

def test_call_in_executor_explicit(self, context, use_resource_name):
        executor = ThreadPoolExecutor(1)
        context.add_resource(executor, types=[Executor])
        context.add_teardown_callback(executor.shutdown)
        executor_arg = 'default' if use_resource_name else executor
        worker_thread = await context.call_in_executor(current_thread, executor=executor_arg)
        assert worker_thread is not current_thread() 

def executor(arg: Union[Executor, str, Callable] = None):
    """
    Decorate a function so that it runs in an :class:`~concurrent.futures.Executor`.

    If a resource name is given, the first argument must be a :class:`~.Context`.

    Usage::

        @executor
        def should_run_in_executor():
            ...

    With a resource name::

        @executor('resourcename')
        def should_run_in_executor(ctx):
            ...

    :param arg: a callable to decorate, an :class:`~concurrent.futures.Executor` instance, the
        resource name of one or ``None`` to use the event loop's default executor
    :return: the wrapped function

    """
    def outer_wrapper(func: Callable):
        @wraps(func)
        def inner_wrapper(*args, **kwargs):
            try:
                ctx = next(arg for arg in args[:2] if isinstance(arg, Context))
            except StopIteration:
                raise RuntimeError('the first positional argument to {}() has to be a Context '
                                   'instance'.format(callable_name(func))) from None

            executor = ctx.require_resource(Executor, resource_name)
            return asyncio_extras.call_in_executor(func, *args, executor=executor, **kwargs)

        return inner_wrapper

    if isinstance(arg, str):
        resource_name = arg
        return outer_wrapper

    return asyncio_extras.threadpool(arg)