diff --git a/stdlib/3/multiprocessing/__init__.pyi b/stdlib/3/multiprocessing/__init__.pyi index 1acad6412775..59b82daf3b46 100644 --- a/stdlib/3/multiprocessing/__init__.pyi +++ b/stdlib/3/multiprocessing/__init__.pyi @@ -1,8 +1,12 @@ # Stubs for multiprocessing -from typing import Any, Callable, Iterable, Mapping, Optional, Dict, List, Union, TypeVar +from typing import ( + Any, Callable, ContextManager, Iterable, Mapping, Optional, Dict, List, + Union, TypeVar, +) from logging import Logger +from multiprocessing import pool from multiprocessing.context import BaseContext from multiprocessing.managers import SyncManager from multiprocessing.pool import AsyncResult @@ -12,11 +16,9 @@ import queue _T = TypeVar('_T') -class Lock(): +class Lock(ContextManager[Lock]): def acquire(self, block: bool = ..., timeout: int = ...) -> None: ... def release(self) -> None: ... - def __enter__(self) -> 'Lock': ... - def __exit__(self, exc_type, exc_value, tb) -> None: ... class Event(object): def __init__(self, *, ctx: BaseContext) -> None: ... @@ -25,54 +27,13 @@ class Event(object): def clear(self) -> None: ... def wait(self, timeout: Optional[int] = ...) -> bool: ... -class Pool(): - def __init__(self, processes: Optional[int] = ..., - initializer: Optional[Callable[..., None]] = ..., - initargs: Iterable[Any] = ..., - maxtasksperchild: Optional[int] = ..., - context: Optional[Any] = None) -> None: ... - def apply(self, - func: Callable[..., Any], - args: Iterable[Any] = ..., - kwds: Dict[str, Any]=...) -> Any: ... - def apply_async(self, - func: Callable[..., Any], - args: Iterable[Any] = ..., - kwds: Dict[str, Any] = ..., - callback: Optional[Callable[..., None]] = None, - error_callback: Optional[Callable[[BaseException], None]] = None) -> AsyncResult: ... - def map(self, - func: Callable[..., Any], - iterable: Iterable[Any] = ..., - chunksize: Optional[int] = ...) -> List[Any]: ... - def map_async(self, func: Callable[..., Any], - iterable: Iterable[Any] = ..., - chunksize: Optional[int] = ..., - callback: Optional[Callable[..., None]] = None, - error_callback: Optional[Callable[[BaseException], None]] = None) -> AsyncResult: ... - def imap(self, - func: Callable[..., Any], - iterable: Iterable[Any] = ..., - chunksize: Optional[int] = None) -> Iterable[Any]: ... - def imap_unordered(self, - func: Callable[..., Any], - iterable: Iterable[Any] = ..., - chunksize: Optional[int] = None) -> Iterable[Any]: ... - def starmap(self, - func: Callable[..., Any], - iterable: Iterable[Iterable[Any]] = ..., - chunksize: Optional[int] = None) -> List[Any]: ... - def starmap_async(self, - func: Callable[..., Any], - iterable: Iterable[Iterable[Any]] = ..., - chunksize: Optional[int] = ..., - callback: Optional[Callable[..., None]] = None, - error_callback: Optional[Callable[[BaseException], None]] = None) -> AsyncResult: ... - def close(self) -> None: ... - def terminate(self) -> None: ... - def join(self) -> None: ... - def __enter__(self) -> 'Pool': ... - def __exit__(self, exc_type, exc_val, exc_tb) -> None: ... +# N.B. This is generated at runtime by partially applying +# multiprocessing.context.BaseContext.Pool, so the two signatures should be +# identical (modulo self). +def Pool(processes: Optional[int] = ..., + initializer: Optional[Callable[..., Any]] = ..., + initargs: Iterable[Any] = ..., + maxtasksperchild: Optional[int] = ...) -> pool.Pool: ... class Process(): name: str diff --git a/stdlib/3/multiprocessing/context.pyi b/stdlib/3/multiprocessing/context.pyi index bd1eed7f69f4..07138bbb5f77 100644 --- a/stdlib/3/multiprocessing/context.pyi +++ b/stdlib/3/multiprocessing/context.pyi @@ -3,7 +3,9 @@ from logging import Logger import multiprocessing import sys -from typing import Any, Callable, Optional, List, Sequence, Tuple, Type, Union +from typing import ( + Any, Callable, Iterable, Optional, List, Sequence, Tuple, Type, Union, +) class ProcessError(Exception): ... @@ -49,13 +51,16 @@ class BaseContext(object): def JoinableQueue(self, maxsize: int = ...) -> Any: ... # TODO: change return to SimpleQueue once a stub exists in multiprocessing.queues def SimpleQueue(self) -> Any: ... + # N.B. This method is partially applied at runtime to generate + # multiprocessing.Pool, so the two signatures should be identical (modulo + # self). def Pool( self, processes: Optional[int] = ..., initializer: Optional[Callable[..., Any]] = ..., - initargs: Tuple = ..., + initargs: Iterable[Any] = ..., maxtasksperchild: Optional[int] = ... - ) -> multiprocessing.Pool: ... + ) -> multiprocessing.pool.Pool: ... # TODO: typecode_or_type param is a ctype with a base class of _SimpleCData or array.typecode Need to figure out # how to handle the ctype # TODO: change return to RawValue once a stub exists in multiprocessing.sharedctypes diff --git a/stdlib/3/multiprocessing/managers.pyi b/stdlib/3/multiprocessing/managers.pyi index c2bad5b332eb..74634e3aafbc 100644 --- a/stdlib/3/multiprocessing/managers.pyi +++ b/stdlib/3/multiprocessing/managers.pyi @@ -5,7 +5,8 @@ import queue import threading from typing import ( - Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, TypeVar + Any, Callable, ContextManager, Dict, Iterable, List, Mapping, Optional, + Sequence, TypeVar, ) _T = TypeVar('_T') @@ -16,13 +17,11 @@ class Namespace: ... _Namespace = Namespace -class BaseManager: +class BaseManager(ContextManager[BaseManager]): def register(self, typeid: str, callable: Any = ...) -> None: ... def shutdown(self) -> None: ... def start(self, initializer: Optional[Callable[..., Any]] = ..., initargs: Iterable[Any] = ...) -> None: ... - def __enter__(self) -> 'BaseManager': ... - def __exit__(self, exc_type, exc_value, tb) -> None: ... class SyncManager(BaseManager): def BoundedSemaphore(self, value: Any = ...) -> threading.BoundedSemaphore: ... diff --git a/stdlib/3/multiprocessing/pool.pyi b/stdlib/3/multiprocessing/pool.pyi index c3b789393521..bdfd476f9167 100644 --- a/stdlib/3/multiprocessing/pool.pyi +++ b/stdlib/3/multiprocessing/pool.pyi @@ -2,7 +2,12 @@ # NOTE: These are incomplete! -from typing import Any, Callable, Iterable, Mapping, Optional, Dict, List +from typing import ( + Any, Callable, ContextManager, Iterable, Mapping, Optional, Dict, List, + TypeVar, +) + +_T = TypeVar('_T', bound='Pool') class AsyncResult(): def get(self, timeout: float = ...) -> Any: ... @@ -10,10 +15,12 @@ class AsyncResult(): def ready(self) -> bool: ... def successful(self) -> bool: ... -class ThreadPool(): - def __init__(self, processes: Optional[int] = None, - initializer: Optional[Callable[..., Any]] = None, - initargs: Iterable[Any] = ...) -> None: ... +class Pool(ContextManager[Pool]): + def __init__(self, processes: Optional[int] = ..., + initializer: Optional[Callable[..., None]] = ..., + initargs: Iterable[Any] = ..., + maxtasksperchild: Optional[int] = ..., + context: Optional[Any] = None) -> None: ... def apply(self, func: Callable[..., Any], args: Iterable[Any] = ..., @@ -30,7 +37,7 @@ class ThreadPool(): chunksize: Optional[int] = None) -> List[Any]: ... def map_async(self, func: Callable[..., Any], iterable: Iterable[Any] = ..., - chunksize: Optional[Optional[int]] = None, + chunksize: Optional[int] = None, callback: Optional[Callable[..., None]] = None, error_callback: Optional[Callable[[BaseException], None]] = None) -> AsyncResult: ... def imap(self, @@ -54,5 +61,11 @@ class ThreadPool(): def close(self) -> None: ... def terminate(self) -> None: ... def join(self) -> None: ... - def __enter__(self) -> 'ThreadPool': ... - def __exit__(self, exc_type, exc_val, exc_tb) -> None: ... + def __enter__(self: _T) -> _T: ... + + +class ThreadPool(Pool, ContextManager[ThreadPool]): + + def __init__(self, processes: Optional[int] = None, + initializer: Optional[Callable[..., Any]] = None, + initargs: Iterable[Any] = ...) -> None: ...