From d0e8877b2801587e89a002f8c2ae579e109e3463 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:14:56 +0800 Subject: [PATCH 01/10] refactor: Update signature to accept generator --- src/thread/thread.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index 0b20fcd..3932ce2 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -32,7 +32,7 @@ class ParallelProcessing: ... HookFunction, ) from typing_extensions import Generic, ParamSpec -from typing import List, Callable, Optional, Union, Mapping, Sequence, Tuple +from typing import List, Callable, Optional, Union, Mapping, Sequence, Tuple, Generator Threads: set['Thread'] = set() @@ -387,7 +387,7 @@ def _wrap_function(self, function: TargetFunction) -> TargetFunction: @wraps(function) def wrapper( index: int, - data_chunk: Sequence[_Dataset_T], + data_chunk: Generator[_Dataset_T, None, None], *args: _Target_P.args, **kwargs: _Target_P.kwargs, ) -> List[_Target_T]: From c4eb1f52cf513f80a5df504b0d8e04e2df02ac8e Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:15:51 +0800 Subject: [PATCH 02/10] perf: Move from enumerate to incremental --- src/thread/thread.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index 3932ce2..0648f5d 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -392,10 +392,13 @@ def wrapper( **kwargs: _Target_P.kwargs, ) -> List[_Target_T]: computed: List[Data_Out] = [] - for i, data_entry in enumerate(data_chunk): + + i = 0 + for data_entry in data_chunk: v = function(data_entry, *args, **kwargs) computed.append(v) self._threads[index].progress = round((i + 1) / len(data_chunk), 5) + i += 1 self._completed += 1 if self._completed == len(self._threads): @@ -507,7 +510,7 @@ def start(self) -> None: i: v for i, v in self.overflow_kwargs.items() if i != 'name' and i != 'args' } - for i, data_chunk in enumerate(chunk_split(self.dataset, max_threads)): + i = 0 chunk_thread = Thread( target=self.function, args=[i, data_chunk, *parsed_args, *self.overflow_args], @@ -516,6 +519,7 @@ def start(self) -> None: ) self._threads.append(_ThreadWorker(chunk_thread, 0)) chunk_thread.start() + i += 1 # Handle abrupt exit From a86772e28941ecbf619e2df436e8ae9c041075e7 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:16:24 +0800 Subject: [PATCH 03/10] fix: Utilise parsed length --- src/thread/thread.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index 0648f5d..b266c04 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -387,6 +387,7 @@ def _wrap_function(self, function: TargetFunction) -> TargetFunction: @wraps(function) def wrapper( index: int, + length: int, data_chunk: Generator[_Dataset_T, None, None], *args: _Target_P.args, **kwargs: _Target_P.kwargs, @@ -397,7 +398,7 @@ def wrapper( for data_entry in data_chunk: v = function(data_entry, *args, **kwargs) computed.append(v) - self._threads[index].progress = round((i + 1) / len(data_chunk), 5) + self._threads[index].progress = round((i + 1) / length, 5) i += 1 self._completed += 1 From 8d4c937d19acea4f609a9c780166496c69e31cee Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:17:02 +0800 Subject: [PATCH 04/10] refactor: Update signature --- src/thread/utils/algorithm.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/thread/utils/algorithm.py b/src/thread/utils/algorithm.py index bc531b4..679195c 100644 --- a/src/thread/utils/algorithm.py +++ b/src/thread/utils/algorithm.py @@ -8,10 +8,10 @@ |_ b.py """ -from typing import List, Sequence, Any +from typing import List, Tuple -def chunk_split(dataset: Sequence[Any], number_of_chunks: int) -> List[List[Any]]: +def chunk_split(dataset_length: int, number_of_chunks: int) -> List[Tuple[int, int]]: """ Splits a dataset into balanced chunks From 99138e93cd8276f9e38601099ff9727c079e53fd Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:17:16 +0800 Subject: [PATCH 05/10] docs: Update docstring --- src/thread/utils/algorithm.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/thread/utils/algorithm.py b/src/thread/utils/algorithm.py index 679195c..d037fe3 100644 --- a/src/thread/utils/algorithm.py +++ b/src/thread/utils/algorithm.py @@ -21,13 +21,13 @@ def chunk_split(dataset_length: int, number_of_chunks: int) -> List[Tuple[int, i Parameters ---------- - :param dataset: This should be the dataset you want to split into chunks + :param dataset_length: This should be the length of the dataset you want to split into chunks :param number_of_chunks: The should be the number of chunks it will attempt to split into Returns ------- - :returns list[list[Any]]: The split dataset + :returns list[tuple[int, int]]: The chunked dataset slices Raises ------ From 3a4623c2d571785a768ff8545f2f66d0465e435f Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:17:42 +0800 Subject: [PATCH 06/10] refactor: Utilise parsed length --- src/thread/utils/algorithm.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/thread/utils/algorithm.py b/src/thread/utils/algorithm.py index d037fe3..60e1c37 100644 --- a/src/thread/utils/algorithm.py +++ b/src/thread/utils/algorithm.py @@ -33,17 +33,16 @@ def chunk_split(dataset_length: int, number_of_chunks: int) -> List[Tuple[int, i ------ AssertionError: The number of chunks specified is larger than the dataset size """ - length = len(dataset) assert ( - length >= number_of_chunks + dataset_length >= number_of_chunks ), 'The number of chunks specified is larger than the dataset size' - chunk_count = length // number_of_chunks - overflow = length % number_of_chunks + chunk_count = dataset_length // number_of_chunks + overflow = dataset_length % number_of_chunks i = 0 split = [] - while i < length: + while i < dataset_length: chunk_length = chunk_count + int(overflow > 0) b = i + chunk_length From 2506988bc30785a994db018c1f03bcd83752a501 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:18:09 +0800 Subject: [PATCH 07/10] refactor: Push tuple instead of list slice --- src/thread/utils/algorithm.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/thread/utils/algorithm.py b/src/thread/utils/algorithm.py index 60e1c37..0250abf 100644 --- a/src/thread/utils/algorithm.py +++ b/src/thread/utils/algorithm.py @@ -46,7 +46,7 @@ def chunk_split(dataset_length: int, number_of_chunks: int) -> List[Tuple[int, i chunk_length = chunk_count + int(overflow > 0) b = i + chunk_length - split.append(dataset[i:b]) + split.append((i, b)) overflow -= 1 i = b From 09e481792dc84deda7782832c3cc0b77bcb1a7ac Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:19:46 +0800 Subject: [PATCH 08/10] perf: Create and parse generator Improves memory efficiency and runtime Fixes #45 --- src/thread/thread.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index b266c04..060057d 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -512,9 +512,16 @@ def start(self) -> None: } i = 0 + for chunkStart, chunkEnd in chunk_split(len(self.dataset), max_threads): chunk_thread = Thread( target=self.function, - args=[i, data_chunk, *parsed_args, *self.overflow_args], + args=[ + i, + chunkEnd - chunkStart, + (self.dataset[x] for x in range(chunkStart, chunkEnd)), + *parsed_args, + *self.overflow_args, + ], name=name_format and name_format % i or None, **self.overflow_kwargs, ) From 6c83a23894343d051891f27d1ffb4c3b7737dd85 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:20:38 +0800 Subject: [PATCH 09/10] test: Add simple chunking tests --- tests/test_algorithm.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 tests/test_algorithm.py diff --git a/tests/test_algorithm.py b/tests/test_algorithm.py new file mode 100644 index 0000000..61df76f --- /dev/null +++ b/tests/test_algorithm.py @@ -0,0 +1,22 @@ +from src.thread.utils import algorithm + + +def test_chunking_1(): + assert algorithm.chunk_split(5, 1) == [(0, 5)] + + +def test_chunking_2(): + assert algorithm.chunk_split(5, 2) == [(0, 3), (3, 5)] + + +def test_chunking_3(): + assert algorithm.chunk_split(100, 8) == [ + (0, 13), + (13, 26), + (26, 39), + (39, 52), + (52, 64), + (64, 76), + (76, 88), + (88, 100), + ] From 3ba18116e853242ad1ddea4ac3a92ecca76ed849 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Mon, 4 Mar 2024 16:20:50 +0800 Subject: [PATCH 10/10] test: Add dynamic chunking tests --- tests/test_algorithm.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/test_algorithm.py b/tests/test_algorithm.py index 61df76f..9e5ed4d 100644 --- a/tests/test_algorithm.py +++ b/tests/test_algorithm.py @@ -1,3 +1,4 @@ +import random from src.thread.utils import algorithm @@ -20,3 +21,25 @@ def test_chunking_3(): (76, 88), (88, 100), ] + + +def test_chunking_dynamic(): + dataset_length = random.randint(400, int(10e6)) + thread_count = random.randint(2, 100) + + expected_chunk_length_low = dataset_length // thread_count + expected_chunk_high = dataset_length % thread_count + + i = 0 + heap = [] + while i < dataset_length: + chunk_length = expected_chunk_length_low + int(expected_chunk_high > 0) + b = i + chunk_length + + heap.append((i, b)) + expected_chunk_high -= 1 + i = b + + assert ( + algorithm.chunk_split(dataset_length, thread_count) == heap + ), f'\nLength: {dataset_length}\nThreads: {thread_count}\nExpected: {heap}\nActual: {algorithm.chunk_split(dataset_length, thread_count)}'