andras/great-ai
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases 4 Packages Wiki Activity Actions

Debug stuck tests

Browse source
This commit is contained in:
Andras Schmelczer 2022-07-03 12:57:05 +02:00
parent c76342534a
commit e85edf4222
1 changed files with 104 additions and 104 deletions
Download patch file Download diff file Expand all files Collapse all files

208
src/great_ai/utilities/parallel_map/threaded_parallel_map.py
View file

@ -1,121 +1,121 @@
import queue # import queue
import threading # import threading
from typing import Callable, Iterable, Literal, Optional, Sequence, TypeVar, overload # from typing import Callable, Iterable, Literal, Optional, Sequence, TypeVar, overload
from .get_config import get_config # from .get_config import get_config
from .manage_communication import manage_communication # from .manage_communication import manage_communication
from .manage_serial import manage_serial # from .manage_serial import manage_serial
from .mapper_function import mapper_function # from .mapper_function import mapper_function
T = TypeVar("T") # T = TypeVar("T")
V = TypeVar("V") # V = TypeVar("V")
@overload # @overload
def threaded_parallel_map( # def threaded_parallel_map(
function: Callable[[T], V], # function: Callable[[T], V],
input_values: Sequence[T], # input_values: Sequence[T],
*, # *,
chunk_size: Optional[int], # chunk_size: Optional[int],
concurrency: Optional[int], # concurrency: Optional[int],
unordered: Optional[bool], # unordered: Optional[bool],
ignore_exceptions: Optional[Literal[False]], # ignore_exceptions: Optional[Literal[False]],
) -> Iterable[V]: # ) -> Iterable[V]:
... # ...
@overload # @overload
def threaded_parallel_map( # def threaded_parallel_map(
function: Callable[[T], V], # function: Callable[[T], V],
input_values: Iterable[T], # input_values: Iterable[T],
*, # *,
chunk_size: int, # chunk_size: int,
concurrency: Optional[int], # concurrency: Optional[int],
unordered: Optional[bool], # unordered: Optional[bool],
ignore_exceptions: Optional[Literal[False]], # ignore_exceptions: Optional[Literal[False]],
) -> Iterable[V]: # ) -> Iterable[V]:
... # ...
@overload # @overload
def threaded_parallel_map( # def threaded_parallel_map(
function: Callable[[T], V], # function: Callable[[T], V],
input_values: Sequence[T], # input_values: Sequence[T],
*, # *,
chunk_size: Optional[int], # chunk_size: Optional[int],
concurrency: Optional[int], # concurrency: Optional[int],
unordered: Optional[bool], # unordered: Optional[bool],
ignore_exceptions: True, # ignore_exceptions: True,
) -> Iterable[Optional[V]]: # ) -> Iterable[Optional[V]]:
... # ...
@overload # @overload
def threaded_parallel_map( # def threaded_parallel_map(
function: Callable[[T], V], # function: Callable[[T], V],
input_values: Iterable[T], # input_values: Iterable[T],
*, # *,
chunk_size: int, # chunk_size: int,
concurrency: Optional[int], # concurrency: Optional[int],
unordered: Optional[bool], # unordered: Optional[bool],
ignore_exceptions: True, # ignore_exceptions: True,
) -> Iterable[Optional[V]]: # ) -> Iterable[Optional[V]]:
... # ...
def threaded_parallel_map( # def threaded_parallel_map(
function, # function,
input_values, # input_values,
*, # *,
chunk_size=None, # chunk_size=None,
concurrency=None, # concurrency=None,
unordered=False, # unordered=False,
ignore_exceptions=False, # ignore_exceptions=False,
): # ):
config = get_config( # config = get_config(
function=function, # function=function,
input_values=input_values, # input_values=input_values,
chunk_size=chunk_size, # chunk_size=chunk_size,
concurrency=concurrency, # concurrency=concurrency,
) # )
if config.concurrency == 1: # if config.concurrency == 1:
yield from manage_serial( # yield from manage_serial(
function=function, # function=function,
input_values=input_values, # input_values=input_values,
ignore_exceptions=ignore_exceptions, # ignore_exceptions=ignore_exceptions,
) # )
input_queue = queue.Queue(config.concurrency * 2) # input_queue = queue.Queue(config.concurrency * 2)
output_queue = queue.Queue(config.concurrency * 2) # output_queue = queue.Queue(config.concurrency * 2)
should_stop = threading.Event() # should_stop = threading.Event()
threads = [ # threads = [
threading.Thread( # threading.Thread(
name=f"threaded_parallel_map_{config.function_name}_{i}", # name=f"threaded_parallel_map_{config.function_name}_{i}",
target=mapper_function, # target=mapper_function,
daemon=True, # daemon=True,
kwargs=dict( # kwargs=dict(
input_queue=input_queue, # input_queue=input_queue,
output_queue=output_queue, # output_queue=output_queue,
should_stop=should_stop, # should_stop=should_stop,
map_function=function, # map_function=function,
), # ),
) # )
for i in range(config.concurrency) # for i in range(config.concurrency)
] # ]
for t in threads: # for t in threads:
t.start() # t.start()
yield from manage_communication( # yield from manage_communication(
input_values=input_values, # input_values=input_values,
chunk_size=config.chunk_size, # chunk_size=config.chunk_size,
input_queue=input_queue, # input_queue=input_queue,
output_queue=output_queue, # output_queue=output_queue,
unordered=unordered, # unordered=unordered,
ignore_exceptions=ignore_exceptions, # ignore_exceptions=ignore_exceptions,
) # )
should_stop.set() # should_stop.set()
for t in threads: # for t in threads:
t.join() # t.join()