Python concurrent.futures 模块，Executor() 实例源码

def __init__(
        self,
        REMOTE_IP,
        REMOTE_PORT,
        loop: asyncio.AbstractEventLoop() = None,
        executor: futures.Executor() = None
    ):
        self._input_list = []
        self._input_queue = asyncio.Queue()
        if loop:
            self.loop = loop
        else:
            self.loop = asyncio.get_event_loop()
        self.executor = executor
        self.REMOTE_IP = REMOTE_IP
        self.REMOTE_PORT = REMOTE_PORT

def __init__(
        self,
        device_list=[],
        connection_list=[],
        loop: asyncio.AbstractEventLoop() = None,
        executor: futures.Executor() = None
    ):
        if loop:
            self.loop = loop
        else:
            self.loop = asyncio.get_event_loop()
        if executor:
            self.executor = executor
        else:
            self.executor = futures.ProcessPoolExecutor()
        self.loop.set_default_executor(self.executor)
        self.device_list = device_list
        self.connection_list = connection_list

def open_async(file: Union[str, Path], *args, executor: Executor = None,
               **kwargs) -> AsyncFileWrapper:
    """
    Open a file and wrap it in an :class:`~AsyncFileWrapper`.

    Example::

        async def read_file_contents(path: str) -> bytes:
            async with open_async(path, 'rb') as f:
                return await f.read()

    The file wrapper can also be asynchronously iterated line by line::

        async def read_file_lines(path: str):
            async for line in open_async(path):
                print(line)

    :param file: the file path to open
    :param args: positional arguments to :func:`open`
    :param executor: the ``executor`` argument to :class:`~AsyncFileWrapper`
    :param kwargs: keyword arguments to :func:`open`
    :return: the wrapped file object

    """
    return AsyncFileWrapper(str(file), args, kwargs, executor)

def call_in_executor(func: Callable, *args, executor: Executor = None, **kwargs) -> Future:
    """
    Call the given callable in an executor.

    This is a nicer version of the following::

        get_event_loop().run_in_executor(executor, func, *args)

    If you need to pass keyword arguments named ``func`` or ``executor`` to the callable, use
    :func:`functools.partial` for that.

    :param func: a function
    :param args: positional arguments to call with
    :param executor: the executor to call the function in
    :param kwargs: keyword arguments to call with
    :return: a future that will resolve to the function call's return value

    """
    callback = partial(func, *args, **kwargs)
    return get_event_loop().run_in_executor(executor, callback)

def __init__(self, on_exit: ExitOption=ExitOption.ABANDON) -> None:
        """
        Init the state.

        Args:
            on_exit:  determines the behavior on exit when the executor is used as a context manager
                ExitOption.ABANDON:  this is the default.  All remaining threads are abandoned.
                ExitOption.NO_WAIT:   shuts down without waiting for remaining threads to complete.
                ExitOption.WAIT:     shuts down and waits for remaining threads to complete.  This is the
                    default behavior for a vanilla `futures.Executor`, but not for `ThreadExecutor`.

        Returns:  None
        Raises:  None
        Required Tests:  None
        """
        # shared state
        self._monitored_threads = {}  # type: dict
        self._monitor_thread = None  # type: Optional[threading.Thread]
        # lock for ^
        self._monitored_thread_lock = threading.RLock()
        # main thread state
        self._is_shutdown = False
        self._abandon = False
        self._on_exit = on_exit

def __exit__(self, exc_type: type, exc_val: Exception, exc_tb: Any) -> bool:
        """
        Exit the context manager.

        The default behavior of a `futures.Executor` is to shutdown and wait.  `ThreadExecutor`
        does whatever is specified in the `on_exit` parameter to `__init__`

        Args:
            exc_type, exc_val, exc_tb:  the exception data for any exception raised in the managed context, or None,
              if no exception was raised.

        Returns:
            a bool indicating whether or not the function handled the exception, if one was raised in the managed
            context.  If False, and there was an exception, the python runtime will raise the exception.
        Raises:  None
        Required Tests:  None
        """
        assert exc_type or exc_val or exc_tb or True  # Vulture
        self._exit_function(self._on_exit)()
        return False

def __init__(
        self,
        CALLBACK=None,
        BUF_SIZE=1,
        FILTER={'data'},
        WAIT_TIME=0.1,
        loop: asyncio.AbstractEventLoop()=None,
        executor: futures.Executor()=None,
        step_time=0.01,
        returnNone=False,
        count=0
        # Partially Timeout time stamps will be discarded on False
        # None will be returned for timeout data on True.
    ):
        self.WAIT_TIME = WAIT_TIME
        self.FILTER = FILTER
        self.BUF_SIZE = BUF_SIZE
        self.step_time = step_time
        self.buf_time_out = self.BUF_SIZE * self.WAIT_TIME * 2
        self._input_list = []
        self._output_list = []
        self._input_queue = asyncio.Queue()
        if loop:
            self.loop = loop
        else:
            self.loop = asyncio.get_event_loop()
        self.executor = executor
        if CALLBACK:
            self.CALLBACK = CALLBACK
        self.returnNone = returnNone
        self.count = count

def setNode(self, node: Node, items: List[Node]):
        self.node = node
        self.clear()
        self.nodes_to_items.clear()

        for ii, item in enumerate(filter(tz.identity, items)):  # filter null items -- TODO: find better place for filter
            self.insertCallListItem(ii, item)

        # NOTE: previously `setNode` involved submitting a job to an Executor
        # and appending the future to self.populate_futures. Currently
        # self.populate_futures is always empty.

def __init__(self,
                 task_list: List[Task]=None,
                 config: Config=Config,
                 stats: Stats=Stats,
                 executor: Executor=None,
                 debug: bool=False) -> None:
        '''Initialize Tasky and automatically start a list of tasks.
        One of the following methods must be called on the resulting objects
        to start the event loop: `run_forever()`, `run_until_complete()`, or
        `run_for_time()`.'''

        if uvloop:
            Log.debug('using uvloop event loop')
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

        self.loop = asyncio.new_event_loop()
        self.loop.add_signal_handler(signal.SIGINT, self.sigint)
        self.loop.add_signal_handler(signal.SIGTERM, self.sigterm)
        self.loop.set_exception_handler(self.exception)
        asyncio.set_event_loop(self.loop)

        if debug:
            Log.debug('enabling asyncio debug mode')
            self.loop.set_debug(True)

        self.all_tasks = {}
        self.running_tasks = set()
        self.initial_tasks = list(task_list)

        self.configuration = config
        self.stats = stats
        self.executor = executor

        self.monitor = False
        self.terminate_on_finish = False
        self.stop_attempts = 0

def execute(self, fn, *args, **kwargs) -> None:
        '''Execute an arbitrary function outside the event loop using
        a shared Executor.'''

        fn = functools.partial(fn, *args, **kwargs)
        return await self.loop.run_in_executor(self.executor, fn)

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 __init__(self, path: str, args: tuple, kwargs: dict, executor: Optional[Executor]) -> None:
        self._open_args = (path,) + args
        self._open_kwargs = kwargs
        self._executor = executor
        self._raw_file = None  # type: IOBase

def __init__(self, executor: Optional[Executor]) -> None:
        self.executor = executor
        self.exited = False

def threadpool(arg: Union[Executor, Callable] = None):
    """
    Return a decorator/asynchronous context manager that guarantees that the wrapped function or
    ``with`` block is run in the given executor.

    If no executor is given, the current event loop's default executor is used.
    Otherwise, the executor must be a PEP 3148 compliant thread pool executor.

    Callables wrapped with this must be used with ``await`` when called in the event loop thread.
    They can also be called in worker threads, just by omitting the ``await``.

    Example use as a decorator::

        @threadpool
        def this_runs_in_threadpool():
           return do_something_cpu_intensive()

        async def request_handler():
            result = await this_runs_in_threadpool()

    Example use as an asynchronous context manager::

        async def request_handler(in_url, out_url):
            page = await http_fetch(in_url)

            async with threadpool():
                data = transform_page(page)

            await http_post(out_url, page)

    :param arg: either a callable (when used as a decorator) or an executor in which to run the
        wrapped callable or the ``with`` block (when used as a context manager)

    """
    if callable(arg):
        # When used like @threadpool
        return _ThreadSwitcher(None)(arg)
    else:
        # When used like @threadpool(...) or async with threadpool(...)
        return _ThreadSwitcher(arg)

def test_default_exit() -> None:
    """Test Thread Executor default exit."""
    # When
    executor = ThreadExecutor()

    # Then
    utaw.assertEqual(executor._on_exit, ExitOption.ABANDON)  # pylint: disable=protected-access,no-member

def __init__(
        self,
        SERVER_IP='10.0.0.1',
        SERVER_TCP_PORT=4712,
        CLIENT_TCP_PORT='AUTO',
        SERVER_UDP_PORT=4713,
        CLIENT_UDP_PORT='AUTO',
        MODE='TCP',
        IDCODE=1,
        loop: asyncio.AbstractEventLoop() = None,
        executor: futures.Executor() = None,
        parser: Parser() = None,
        count=0
    ):
        self.IDCODE = IDCODE
        self.SERVER_IP = SERVER_IP
        self.SERVER_TCP_PORT = SERVER_TCP_PORT
        self.CLIENT_TCP_PORT = CLIENT_TCP_PORT
        self.SERVER_UDP_PORT = SERVER_UDP_PORT
        self.CLIENT_UDP_PORT = CLIENT_UDP_PORT
        self.MODE = MODE
        self.executor = executor
        if loop:
            self.loop = loop
        else:
            self.loop = asyncio.get_event_loop()
        if parser:
            self.parser = parser
        else:
            self.parser = Parser()
        self._output_list = []
        self.count = count
        if self.MODE == 'TCP':
            async def connect():
                self._count = count
                print('Connecting to:', self.SERVER_IP, '...')
                await self.loop.create_connection(
                    lambda: self._TCP(self),
                    host=self.SERVER_IP, port=self.SERVER_TCP_PORT)

            async def close():
                if self.cmd_transport:
                    self.cmd_transport.close()
        self.connect = connect
        self.close = close