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

Remove logging and leave it up to the user

Browse source
This commit is contained in:
Andras Schmelczer 2022-07-02 08:26:26 +02:00
parent 2e12947429
commit aa3131dcc1
9 changed files with 78 additions and 142 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/great_ai/utilities/parallel_map/get_config.py
View file

@ -16,7 +16,6 @@ def get_config(
input_values: Union[Sequence, Iterable],
chunk_size: Optional[int],
concurrency: Optional[int],
disable_logging: bool,
) -> ParallelMapConfiguration:
is_input_sequence = hasattr(input_values, "__len__")
@ -57,8 +56,4 @@ def get_config(
function_name=function.__name__,
)
if not disable_logging:
logger.info("Parallel map: configured ✅")
config.pretty_print()
return config

80
src/great_ai/utilities/parallel_map/manage_communication.py
View file

@ -1,9 +1,7 @@
import multiprocessing as mp
import queue
import traceback
from typing import Any, Dict, Iterable, TypeVar, Union
from tqdm.cli import tqdm
from typing import Dict, Iterable, TypeVar, Union
from ..chunk import chunk
from ..logger import get_logger
@ -16,7 +14,6 @@ V = TypeVar("V")
def manage_communication(
*,
tqdm_options: Dict[str, Any],
input_values: Iterable[T],
chunk_size: int,
input_queue: Union[mp.Queue, queue.Queue],
@ -24,54 +21,49 @@ def manage_communication(
unordered: bool,
ignore_exceptions: bool,
) -> Iterable[V]:
progress = tqdm(**tqdm_options)
chunks = iter(chunk(enumerate(input_values), chunk_size=chunk_size))
indexed_results: Dict[int, V] = {}
next_output_index = 0
read_input_length = 0
is_iteration_over = False
try:
while not is_iteration_over or next_output_index < read_input_length:
if not is_iteration_over:
try:
next_chunk = next(chunks)
input_queue.put(next_chunk)
read_input_length += len(next_chunk)
except StopIteration:
is_iteration_over = True
except Exception as e:
while not is_iteration_over or next_output_index < read_input_length:
if not is_iteration_over:
try:
next_chunk = next(chunks)
input_queue.put(next_chunk)
read_input_length += len(next_chunk)
except StopIteration:
is_iteration_over = True
except Exception as e:
if not ignore_exceptions:
raise e
else:
logger.error(
f"Exception {e} encountered in input, traceback:\n{traceback.format_exc()}"
)
try:
result_chunk = output_queue.get_nowait()
for index, value, exception in result_chunk:
if exception is not None:
e, tb = exception
if not ignore_exceptions:
raise e
else:
logger.error(
f"Exception {e} encountered in input, traceback:\n{traceback.format_exc()}"
f"Exception {e} encountered in worker, traceback:\n{tb}"
)
if unordered:
yield value
next_output_index += 1
else:
indexed_results[index] = value
try:
result_chunk = output_queue.get_nowait()
progress.update(len(result_chunk))
for index, value, exception in result_chunk:
if exception is not None:
e, tb = exception
if not ignore_exceptions:
raise e
else:
logger.error(
f"Exception {e} encountered in worker, traceback:\n{tb}"
)
if unordered:
yield value
next_output_index += 1
else:
indexed_results[index] = value
if not unordered:
while next_output_index in indexed_results:
yield indexed_results[next_output_index]
del indexed_results[next_output_index]
next_output_index += 1
except queue.Empty:
pass
finally:
progress.close()
if not unordered:
while next_output_index in indexed_results:
yield indexed_results[next_output_index]
del indexed_results[next_output_index]
next_output_index += 1
except queue.Empty:
pass

7
src/great_ai/utilities/parallel_map/manage_serial.py
View file

@ -1,7 +1,5 @@
import traceback
from typing import Any, Callable, Dict, Iterable, TypeVar
from tqdm.cli import tqdm
from typing import Callable, Iterable, TypeVar
from ..logger import get_logger
@ -14,12 +12,11 @@ V = TypeVar("V")
def manage_serial(
*,
function: Callable[[T], V],
tqdm_options: Dict[str, Any],
input_values: Iterable[T],
ignore_exceptions: bool,
) -> Iterable[V]:
try:
for v in tqdm(input_values, **tqdm_options):
for v in input_values:
try:
yield function(v)
except Exception as e:

46
src/great_ai/utilities/parallel_map/parallel_map.py
View file

@ -1,5 +1,5 @@
import multiprocessing as mp
from typing import Callable, Iterable, Optional, Sequence, TypeVar, overload
from typing import Callable, Iterable, Literal, Optional, Sequence, TypeVar, overload
from .get_config import get_config
from .manage_communication import manage_communication
@ -17,9 +17,8 @@ def parallel_map(
*,
chunk_size: Optional[int],
concurrency: Optional[int],
disable_logging: Optional[bool],
unordered: Optional[bool],
ignore_exceptions: Optional[bool],
ignore_exceptions: Optional[Literal[False]],
) -> Iterable[V]:
...
@ -31,20 +30,44 @@ def parallel_map(
*,
chunk_size: int,
concurrency: Optional[int],
disable_logging: Optional[bool],
unordered: Optional[bool],
ignore_exceptions: Optional[bool],
ignore_exceptions: Optional[Literal[False]],
) -> Iterable[V]:
...
@overload
def parallel_map(
function: Callable[[T], V],
input_values: Sequence[T],
*,
chunk_size: Optional[int],
concurrency: Optional[int],
unordered: Optional[bool],
ignore_exceptions: True,
) -> Iterable[Optional[V]]:
...
@overload
def parallel_map(
function: Callable[[T], V],
input_values: Iterable[T],
*,
chunk_size: int,
concurrency: Optional[int],
unordered: Optional[bool],
ignore_exceptions: True,
) -> Iterable[Optional[V]]:
...
def parallel_map(
function,
input_values,
*,
chunk_size=None,
concurrency=None,
disable_logging=False,
unordered=False,
ignore_exceptions=False,
):
@ -53,21 +76,11 @@ def parallel_map(
input_values=input_values,
chunk_size=chunk_size,
concurrency=concurrency,
disable_logging=disable_logging,
)
tqdm_options = dict(
desc=f"Parallel map {config.function_name}",
disable=disable_logging,
total=config.input_length,
miniters=1,
dynamic_ncols=True,
)
if config.concurrency == 1:
yield from manage_serial(
function=function,
tqdm_options=tqdm_options,
input_values=input_values,
ignore_exceptions=ignore_exceptions,
)
@ -98,7 +111,6 @@ def parallel_map(
try:
yield from manage_communication(
tqdm_options=tqdm_options,
input_values=input_values,
chunk_size=config.chunk_size,
input_queue=input_queue,

10
src/great_ai/utilities/parallel_map/parallel_map_configuration.py
View file

@ -14,13 +14,3 @@ class ParallelMapConfiguration(BaseModel):
input_length: Optional[int]
serialized_map_function: bytes
function_name: str
def pretty_print(self, prefix=" ⚙️ "):
logger.info(f"{prefix} concurrency: {self.concurrency}")
logger.info(f"{prefix} chunk size: {self.chunk_size}")
logger.info(
f"{prefix} chunk count: {self.chunk_count if self.chunk_count else 'unknown'}"
)
logger.info(
f"{prefix} function size: {len(self.serialized_map_function) / 1024:.0f} kB"
)

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

@ -18,7 +18,6 @@ def threaded_parallel_map(
*,
chunk_size: Optional[int],
concurrency: Optional[int],
disable_logging: Optional[bool],
unordered: Optional[bool],
ignore_exceptions: Optional[bool],
) -> Iterable[V]:
@ -32,7 +31,6 @@ def threaded_parallel_map(
*,
chunk_size: int,
concurrency: Optional[int],
disable_logging: Optional[bool],
unordered: Optional[bool],
ignore_exceptions: Optional[bool],
) -> Iterable[V]:
@ -45,7 +43,6 @@ def threaded_parallel_map(
*,
chunk_size=None,
concurrency=None,
disable_logging=False,
unordered=False,
ignore_exceptions=False,
):
@ -54,21 +51,11 @@ def threaded_parallel_map(
input_values=input_values,
chunk_size=chunk_size,
concurrency=concurrency,
disable_logging=disable_logging,
)
tqdm_options = dict(
desc=f"Threaded parallel map {config.function_name}",
disable=disable_logging,
total=config.input_length,
miniters=1,
dynamic_ncols=True,
)
if config.concurrency == 1:
yield from manage_serial(
function=function,
tqdm_options=tqdm_options,
input_values=input_values,
ignore_exceptions=ignore_exceptions,
)
@ -96,7 +83,6 @@ def threaded_parallel_map(
t.start()
yield from manage_communication(
tqdm_options=tqdm_options,
input_values=input_values,
chunk_size=config.chunk_size,
input_queue=input_queue,