10943311926

Committed 19 Sep 2024 02:41PM UTC coverage: 98.89%. Remained the same

Build # 10943311926

Build Type

Pull #1062

github

Committed by

web-flow

Commit Message

Merge 1e881487b into 86c59ead1

Pull Request Pull Request #1062: Prepare for release v0.2.3

Run Details

1903 of 1910 branches covered (99.63%)

Branch coverage included in aggregate %.

8695 of 8807 relevant lines covered (98.73%)

0.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.22

/src/basilisp/lang/futures.py

from concurrent.futures import Future as _Future  # noqa # pylint: disable=unused-import
from concurrent.futures import ProcessPoolExecutor as _ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor as _ThreadPoolExecutor
from concurrent.futures import TimeoutError as _TimeoutError
from typing import Callable, Optional, TypeVar

import attr
from typing_extensions import ParamSpec

from basilisp.lang.interfaces import IBlockingDeref

T = TypeVar("T")
P = ParamSpec("P")


@attr.frozen(eq=True, repr=False)
class Future(IBlockingDeref[T]):
    _future: "_Future[T]"

    def __repr__(self):  # pragma: no cover
        return self._future.__repr__()

    def cancel(self) -> bool:
        return self._future.cancel()

    def cancelled(self) -> bool:
        return self._future.cancelled()

    def deref(
        self, timeout: Optional[float] = None, timeout_val: Optional[T] = None
    ) -> Optional[T]:
        try:
            return self._future.result(timeout=timeout)
        except _TimeoutError:
            return timeout_val

    def done(self) -> bool:
        return self._future.done()

    @property
    def is_realized(self) -> bool:
        return self.done()

    # Pass `Future.result(timeout=...)` through so `Executor.map(...)` can
    # still work with this Future wrapper.
    def result(self, timeout: Optional[float] = None) -> T:
        return self._future.result(timeout=timeout)


# Basilisp's standard Future executor is the `ThreadPoolExecutor`, but since
# it is set via a dynamic variable, it can be rebound using the binding macro.
# Callers may wish to use a process pool if they have CPU bound work.


class ProcessPoolExecutor(_ProcessPoolExecutor):  # pragma: no cover
    def __init__(self, max_workers: Optional[int] = None):
        super().__init__(max_workers=max_workers)

    # pylint: disable=arguments-differ
    def submit(  # type: ignore[override]
        self, fn: Callable[P, T], *args: P.args, **kwargs: P.kwargs
    ) -> "Future[T]":
        return Future(super().submit(fn, *args, **kwargs))


class ThreadPoolExecutor(_ThreadPoolExecutor):
    def __init__(
        self,
        max_workers: Optional[int] = None,
        thread_name_prefix: str = "basilisp-futures",
    ):
        super().__init__(max_workers=max_workers, thread_name_prefix=thread_name_prefix)

    # pylint: disable=arguments-differ
    def submit(  # type: ignore[override]
        self, fn: Callable[P, T], *args: P.args, **kwargs: P.kwargs
    ) -> "Future[T]":
        return Future(super().submit(fn, *args, **kwargs))

1	from concurrent.futures import Future as _Future # noqa # pylint: disable=unused-import	1✔
2	from concurrent.futures import ProcessPoolExecutor as _ProcessPoolExecutor	1✔
3	from concurrent.futures import ThreadPoolExecutor as _ThreadPoolExecutor	1✔
4	from concurrent.futures import TimeoutError as _TimeoutError	1✔
5	from typing import Callable, Optional, TypeVar	1✔
6
7	import attr	1✔
8	from typing_extensions import ParamSpec	1✔
9
10	from basilisp.lang.interfaces import IBlockingDeref	1✔
11
12	T = TypeVar("T")	1✔
13	P = ParamSpec("P")	1✔
14
15
16	@attr.frozen(eq=True, repr=False)	1✔
17	class Future(IBlockingDeref[T]):	1✔
18	_future: "_Future[T]"	1✔
19
20	def __repr__(self): # pragma: no cover
21	return self._future.__repr__()
22
23	def cancel(self) -> bool:	1✔
24	return self._future.cancel()	1✔
25
26	def cancelled(self) -> bool:	1✔
27	return self._future.cancelled()	1✔
28
29	def deref(	1✔
30	self, timeout: Optional[float] = None, timeout_val: Optional[T] = None
31	) -> Optional[T]:
32	try:	1✔
33	return self._future.result(timeout=timeout)	1✔
34	except _TimeoutError:	1✔
35	return timeout_val	1✔
36
37	def done(self) -> bool:	1✔
38	return self._future.done()	1✔
39
40	@property	1✔
41	def is_realized(self) -> bool:	1✔
42	return self.done()	1✔
43
44	# Pass `Future.result(timeout=...)` through so `Executor.map(...)` can
45	# still work with this Future wrapper.
46	def result(self, timeout: Optional[float] = None) -> T:	1✔
47	return self._future.result(timeout=timeout)	×
48
49
50	# Basilisp's standard Future executor is the `ThreadPoolExecutor`, but since
51	# it is set via a dynamic variable, it can be rebound using the binding macro.
52	# Callers may wish to use a process pool if they have CPU bound work.
53
54
55	class ProcessPoolExecutor(_ProcessPoolExecutor): # pragma: no cover
56	def __init__(self, max_workers: Optional[int] = None):
57	super().__init__(max_workers=max_workers)
58
59	# pylint: disable=arguments-differ
60	def submit( # type: ignore[override]
61	self, fn: Callable[P, T], args: P.args, *kwargs: P.kwargs
62	) -> "Future[T]":
63	return Future(super().submit(fn, args, *kwargs))
64
65
66	class ThreadPoolExecutor(_ThreadPoolExecutor):	1✔
67	def __init__(	1✔
68	self,
69	max_workers: Optional[int] = None,
70	thread_name_prefix: str = "basilisp-futures",
71	):
72	super().__init__(max_workers=max_workers, thread_name_prefix=thread_name_prefix)	1✔
73
74	# pylint: disable=arguments-differ
75	def submit( # type: ignore[override]	1✔
76	self, fn: Callable[P, T], args: P.args, *kwargs: P.kwargs
77	) -> "Future[T]":
78	return Future(super().submit(fn, args, *kwargs))	1✔

basilisp-lang / basilisp / 10943311926

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous