Added initial rework of the concurrent.futures module

stdlib/concurrent/futures/_base.pyi

@@ -1,22 +1,9 @@
 import sys
 import threading
 from abc import abstractmethod
+from collections.abc import Container, Iterable, Iterator, Mapping, Sequence, Set
 from logging import Logger
-from typing import (
-    Any,
-    Callable,
-    Container,
-    Generic,
-    Iterable,
-    Iterator,
-    List,
-    Optional,
-    Protocol,
-    Sequence,
-    Set,
-    TypeVar,
-    overload,
-)
+from typing import Any, Callable, Generic, Optional, Protocol, TypeVar, overload
 
 if sys.version_info >= (3, 9):
     from types import GenericAlias
@@ -29,6 +16,8 @@ RUNNING: str
 CANCELLED: str
 CANCELLED_AND_NOTIFIED: str
 FINISHED: str
+_FUTURE_STATES: Sequence[str]
+_STATE_TO_DESCRIPTION_MAP: Mapping[str, str]
 LOGGER: Logger
 
 class Error(Exception): ...
@@ -99,7 +88,7 @@ def wait(fs: Iterable[Future[_T]], timeout: Optional[float] = ..., return_when:
 
 class _Waiter:
     event: threading.Event
-    finished_futures: List[Future[Any]]
+    finished_futures: Sequence[Future[Any]]
     def __init__(self) -> None: ...
     def add_result(self, future: Future[Any]) -> None: ...
     def add_exception(self, future: Future[Any]) -> None: ...

stdlib/concurrent/futures/process.pyi

@@ -1,9 +1,119 @@
+import multiprocessing as mp
+import multiprocessing.connection as mpconn
+import multiprocessing.context as mpcont
+import multiprocessing.queues as mpq
 import sys
-from typing import Any, Callable, Optional, Tuple
+import threading
+import weakref
+from collections.abc import Generator, Iterable, Mapping, MutableMapping, MutableSequence
+from concurrent.futures import _base
+from types import TracebackType
+from typing import Any, Callable, Optional, Tuple, Union
 
-from ._base import Executor
+_threads_wakeups: weakref.WeakKeyDictionary
+_global_shutdown: bool
 
-EXTRA_QUEUED_CALLS: Any
+class _ThreadWakeup:
+    _closed: bool
+    _reader: mpconn.Connection
+    _writer: mpconn.Connection
+    def __init__(self) -> None: ...
+    def close(self) -> None: ...
+    def wakeup(self) -> None: ...
+    def clear(self) -> None: ...
+
+def _python_exit() -> None: ...
+
+EXTRA_QUEUED_CALLS: int
+
+_MAX_WINDOWS_WORKERS: int
+
+class _RemoteTraceback(Exception):
+    tb: str
+    def __init__(self, tb: TracebackType) -> None: ...
+    def __str__(self) -> str: ...
+
+class _ExceptionWithTraceback:
+    exc: BaseException
+    tb: TracebackType
+    def __init__(self, exc: BaseException, tb: TracebackType) -> None: ...
+    def __reduce__(self) -> Union[str, Tuple[Any, ...]]: ...
+
+def _rebuild_exc(exc, tb) -> Exception: ...
+
+class _WorkItem(object):
+    future: _base.Future
+    fn: Callable[..., Any]
+    args: Iterable[Any]
+    kwargs: Mapping[str, Any]
+    def __init__(self, future: _base.Future, fn: Callable[..., Any], args: Iterable[Any], kwargs: Mapping[str, Any]) -> None: ...
+
+class _ResultItem(object):
+    work_id: int
+    exception: Exception
+    result: Any
+    def __init__(self, work_id: int, exception: Optional[Exception] = ..., result: Optional[Any] = ...) -> None: ...
+
+class _CallItem(object):
+    work_id: int
+    fn: Callable[..., Any]
+    args: Iterable[Any]
+    kwargs: Mapping[str, Any]
+    def __init__(self, work_id: int, fn: Callable[..., Any], args: Iterable[Any], kwargs: Mapping[str, Any]) -> None: ...
+
+class _SafeQueue(mpq.Queue):
+    pending_work_items: MutableMapping[int, _WorkItem]
+    shutdown_lock: threading.Lock
+    thread_wakeup: _ThreadWakeup
+    def __init__(
+        self,
+        max_size: Optional[int] = ...,
+        *,
+        ctx: mpcont.BaseContext,
+        pending_work_items: MutableMapping[int, _WorkItem],
+        shutdown_lock: threading.Lock,
+        thread_wakeup: _ThreadWakeup,
+    ) -> None: ...
+    def _on_queue_feeder_error(self, e, obj) -> None: ...
+
+def _get_chunks(*iterables: Any, chunksize: int) -> Generator[Tuple[Any], None, None]: ...
+def _process_chunk(fn: Callable[..., Any], chunk: Tuple[Any, None, None]) -> Generator[Any, None, None]: ...
+def _sendback_result(
+    result_queue: mpq.SimpleQueue[_WorkItem], work_id: int, result: Optional[Any] = ..., exception: Optional[Exception] = ...
+) -> None: ...
+def _process_worker(
+    call_queue: mpq.Queue[_CallItem],
+    result_queue: mpq.SimpleQueue[_ResultItem],
+    initializer: Optional[Callable[..., None]],
+    initargs: Tuple[Any, ...],
+) -> None: ...
+
+class _ExecutorManagerThread(threading.Thread):
+    thread_wakeup: _ThreadWakeup
+    shutdown_lock: threading.Lock
+    executor_reference: weakref.ref
+    processes: MutableMapping[int, mpcont.Process]
+    call_queue: mpq.Queue[_CallItem]
+    result_queue: mpq.SimpleQueue[_ResultItem]
+    work_ids_queue: mpq.Queue[int]
+    pending_work_items: MutableMapping[int, _WorkItem]
+    def __init__(self, executor: ProcessPoolExecutor) -> None: ...
+    def run(self) -> None: ...
+    def add_call_item_to_queue(self) -> None: ...
+    def wait_result_broken_or_wakeup(self) -> tuple[Any, bool, str]: ...
+    def process_result_item(self, result_item: Any) -> None: ...
+    def is_shutting_down(self) -> bool: ...
+    def terminate_broken(self, cause: str) -> None: ...
+    def flag_executor_shutting_down(self) -> None: ...
+    def shutdown_workers(self) -> None: ...
+    def join_executor_internals(self) -> None: ...
+    def get_n_children_alive(self) -> int: ...
+
+_system_limits_checked: bool
+_system_limited: Optional[bool]
+
+def _check_system_limits() -> None: ...
+def _chain_from_iterable_of_lists(iterable: Iterable[MutableSequence[Any]]) -> Any: ...
 
 if sys.version_info >= (3, 7):
     from ._base import BrokenExecutor
@@ -12,17 +122,31 @@ if sys.version_info >= (3, 7):
 else:
     class BrokenProcessPool(RuntimeError): ...
 
-if sys.version_info >= (3, 7):
-    from multiprocessing.context import BaseContext
-    class ProcessPoolExecutor(Executor):
+class ProcessPoolExecutor(_base.Executor):
+    _mp_context: Optional[mpcont.BaseContext]
+    _initializer: Optional[Callable[..., None]] = ...
+    _initargs: Tuple[Any, ...] = ...
+    _executor_manager_thread: _ThreadWakeup
+    _processes: MutableMapping[int, mpcont.Process]
+    _shutdown_thread: bool
+    _shutdown_lock: threading.Lock
+    _idle_worker_semaphore: threading.Semaphore
+    _broken: bool
+    _queue_count: int
+    _pending_work_items: MutableMapping[int, _WorkItem]
+    _cancel_pending_futures: bool
+    _executor_manager_thread_wakeup: _ThreadWakeup
+    _result_queue: mpq.SimpleQueue
+    _work_ids: mpq.Queue
+    if sys.version_info >= (3, 7):
         def __init__(
             self,
             max_workers: Optional[int] = ...,
-            mp_context: Optional[BaseContext] = ...,
+            mp_context: Optional[mpcont.BaseContext] = ...,
             initializer: Optional[Callable[..., None]] = ...,
             initargs: Tuple[Any, ...] = ...,
         ) -> None: ...
-
-else:
-    class ProcessPoolExecutor(Executor):
+    else:
         def __init__(self, max_workers: Optional[int] = ...) -> None: ...
+    def _start_executor_manager_thread(self) -> None: ...
+    def _adjust_process_count(self) -> None: ...

stdlib/concurrent/futures/thread.pyi

@@ -1,23 +1,66 @@
 import queue
 import sys
-from typing import Any, Callable, Generic, Iterable, Mapping, Optional, Tuple, TypeVar
+import threading
+import weakref
+from collections.abc import Iterable, Mapping, Set
+from concurrent.futures import _base
+from typing import Any, Callable, Optional, Tuple
 
-from ._base import Executor, Future
+_threads_queues: weakref.WeakKeyDictionary
+_shutdown: bool
+_global_shutdown_lock: threading.Lock
 
-if sys.version_info >= (3, 7):
-    from ._base import BrokenExecutor
-    class BrokenThreadPool(BrokenExecutor): ...
+def _python_exit() -> None: ...
 
 if sys.version_info >= (3, 9):
     from types import GenericAlias
 
-_S = TypeVar("_S")
+class _WorkItem(object):
+    future: _base.Future[Any]
+    fn: Callable[..., Any]
+    args: Iterable[Any]
+    kwargs: Mapping[str, Any]
+    def __init__(
+        self, future: _base.Future[Any], fn: Callable[..., Any], args: Iterable[Any], kwargs: Mapping[str, Any]
+    ) -> None: ...
+    def run(self) -> None: ...
+    if sys.version_info >= (3, 9):
+        def __class_getitem__(cls, item: Any) -> GenericAlias: ...
+
+if sys.version_info >= (3, 7):
+    def _worker(
+        executor_reference: weakref.ref,
+        work_queue: queue.SimpleQueue[Any],
+        initializer: Optional[Callable[..., None]],
+        initargs: Tuple[Any, ...],
+    ) -> None: ...
+
+else:
+    def _worker(
+        executor_reference: weakref.ref,
+        work_queue: queue.Queue[Any],
+        initializer: Optional[Callable[..., None]],
+        initargs: Tuple[Any, ...],
+    ) -> None: ...
 
-class ThreadPoolExecutor(Executor):
+if sys.version_info >= (3, 7):
+    from ._base import BrokenExecutor
+    class BrokenThreadPool(BrokenExecutor): ...
+
+class ThreadPoolExecutor(_base.Executor):
+    _max_workers: int
+    _idle_semaphore: threading.Semaphore
+    _threads: Set[threading.Thread]
+    _broken: bool
+    _shutdown: bool
+    _shutdown_lock: threading.Lock
+    _thread_name_prefix: Optional[str] = ...
+    _initializer: Optional[Callable[..., None]] = ...
+    _initargs: Tuple[Any, ...] = ...
     if sys.version_info >= (3, 7):
-        _work_queue: queue.SimpleQueue[Any]
+        _work_queue: queue.SimpleQueue[_WorkItem]
     else:
-        _work_queue: queue.Queue[Any]
+        _work_queue: queue.Queue[_WorkItem]
     if sys.version_info >= (3, 7):
         def __init__(
             self,
@@ -28,13 +71,5 @@ class ThreadPoolExecutor(Executor):
         ) -> None: ...
     else:
         def __init__(self, max_workers: Optional[int] = ..., thread_name_prefix: str = ...) -> None: ...
-
-class _WorkItem(Generic[_S]):
-    future: Future[_S]
-    fn: Callable[..., _S]
-    args: Iterable[Any]
-    kwargs: Mapping[str, Any]
-    def __init__(self, future: Future[_S], fn: Callable[..., _S], args: Iterable[Any], kwargs: Mapping[str, Any]) -> None: ...
-    def run(self) -> None: ...
-    if sys.version_info >= (3, 9):
-        def __class_getitem__(cls, item: Any) -> GenericAlias: ...
+    def _adjust_thread_count(self) -> None: ...
+    def _initializer_failed(self) -> None: ...