GSK-1895 New worker pool #1478

Hartorn · 2023-10-13T01:25:17Z

Description

Re-implement a worker pool, with possibility for timeout

Also, had capacity to capture all loging for executed task

I removed python deps and related code, because it's was a bother for me under windows for testing.

Related Issue

Type of Change

📚 Examples / docs / tutorials / dependencies update
🔧 Bug fix (non-breaking change which fixes an issue)
🥂 Improvement (non-breaking change which improves an existing feature)
🚀 New feature (non-breaking change which adds functionality)
💥 Breaking change (fix or feature that would cause existing functionality to change)
🔐 Security fix

Checklist

I've read the CODE_OF_CONDUCT.md document.
I've read the CONTRIBUTING.md guide.
I've updated the code style using make codestyle.
I've written tests for all new methods and classes that I created.
I've written the docstring in Google format for all the methods and classes that I used.

giskard/utils/worker_pool.py

giskard/ml_worker/testing/registry/udf_repository.py

andreybavt

A couple of general comments:

the queue names are prefixed with _ but being referenced from outside of executor, for example: executor._pending_tasks_queue. I think it's ok if they become public, at least we're using them this way anyway
There are still untested corner cases, like "Future has been cancelled already". It'll be great to increase code coverage even more considering that this is a core part of the ML Worker

giskard/ml_worker/testing/registry/udf_repository.py

giskard/utils/worker_pool.py

andreybavt · 2023-10-19T14:18:10Z

giskard/utils/worker_pool.py

+ while not self._running_tasks_queue.empty():
+ try:
+ self._running_tasks_queue.get_nowait()
+ except BaseException:


should we log them at least?

Yeah, probably. Plus I should reduce the exeception scope

andreybavt · 2023-10-19T14:52:17Z

giskard/utils/worker_pool.py

+ return exit_codes
+
+
+def _results_thread(


From what I understand you're waking up frequently to check if the whole pool is in one of the FINAL_STATES or not. Wouldn't it be better to call executor._tasks_results.get(block=True, timeout=2) with a larger timeout and after catching Empty exception check the the pool is in one of the FINAL_STATES?
In this case there'll be less busy waiting.

Yeah I guess we could. I'm not a big fan of using try except as an if, but it may be better in this case.

andreybavt · 2023-10-19T15:17:26Z

giskard/utils/worker_pool.py

+ future.set_result(result.result)
+ else:
+ # TODO(Bazire): improve to get Traceback
+ future.set_exception(RuntimeError(result.exception))


why can't it by directly future.set_exception(result.exception) ?

Exception is a string in this case (the exception + traceback), because it needs to be pickable, and exception are linked to many objects + have circular references

andreybavt · 2023-10-19T15:17:55Z

giskard/utils/worker_pool.py

+ try:
+ del executor._futures_mapping[result.id]
+ except BaseException:
+ pass


same as above, we need to know if something goes not as planned

Yup, I will add some warning logs

andreybavt · 2023-10-19T15:22:10Z

giskard/utils/worker_pool.py

+ future = executor._futures_mapping.get(task.id)
+ if future is not None and future.set_running_or_notify_cancel():
+ executor._running_tasks_queue.put(task)
+ elif future is not None:


you're checking for future is not None twice, I think it'll be simple to have this check first and then a second if to check if the future is canceled or not under it

andreybavt · 2023-10-19T15:24:07Z

giskard/utils/worker_pool.py

+ # Mapping of the running tasks and worker pids
+ self._running_process: Dict[str, str] = self._manager.dict()
+ # Mapping of the running tasks and worker pids
+ self._with_timeout_tasks: List[Tuple[str, float]] = []


nitpick: can we use a dataclass here instead of a Tuple? It'll be more maintainable sice we won't have to remember what the parameters were

Actually, I created the dataclass and forgot to use it 😆 😿
Done !

giskard/utils/worker_pool.py

andreybavt · 2023-10-19T15:47:00Z

also let's get Sonar code smells fixed too (at least the ones not related to BaseException)

andreybavt · 2023-10-19T15:57:03Z

giskard/utils/worker_pool.py

+ self._pending_tasks_queue: SimpleQueue[GiskardTask] = self._mp_context.SimpleQueue()
+ # Queue with tasks to run
+ # As in ProcessPool, add one more to avoid idling process
+ self._running_tasks_queue: Queue[GiskardTask] = self._mp_context.Queue(maxsize=self._nb_workers + 1)


the way you use it, it should be rather Queue[Optional[GiskardTask]]

linear · 2023-10-19T16:01:21Z

GSK-1895 Improve worker parallelism: create a cancellable process pool

Hartorn

I took in account all comment.
Also, I improved the threads, especially the killer one.

I added a quich health_check too

giskard/utils/worker_pool.py

Hartorn · 2023-10-20T08:31:28Z

giskard/utils/worker_pool.py

+ # Mapping of the running tasks and worker pids
+ self._running_process: Dict[str, str] = self._manager.dict()
+ # Mapping of the running tasks and worker pids
+ self._with_timeout_tasks: List[Tuple[str, float]] = []


Actually, I created the dataclass and forgot to use it 😆 😿
Done !

giskard/utils/worker_pool.py

Hartorn · 2023-10-20T08:31:45Z

giskard/utils/worker_pool.py

+ self._pending_tasks_queue: SimpleQueue[GiskardTask] = self._mp_context.SimpleQueue()
+ # Queue with tasks to run
+ # As in ProcessPool, add one more to avoid idling process
+ self._running_tasks_queue: Queue[GiskardTask] = self._mp_context.Queue(maxsize=self._nb_workers + 1)


Hartorn · 2023-10-20T08:32:44Z

giskard/utils/worker_pool.py

+ while not self._running_tasks_queue.empty():
+ try:
+ self._running_tasks_queue.get_nowait()
+ except BaseException:


Hartorn · 2023-10-20T08:58:14Z

giskard/utils/worker_pool.py

+ return exit_codes
+
+
+def _results_thread(


Hartorn · 2023-10-20T08:58:22Z

giskard/utils/worker_pool.py

+ try:
+ del executor._futures_mapping[result.id]
+ except BaseException:
+ pass


Hartorn · 2023-10-20T08:58:43Z

giskard/utils/worker_pool.py

+ try:
+ del executor._futures_mapping[result.id]
+ except BaseException:
+ pass


Inokinoki

LGTM

# Conflicts: # .github/workflows/build-python.yml # pdm.lock # pyproject.toml

sonarcloud · 2023-10-25T15:03:54Z

Kudos, SonarCloud Quality Gate passed!

0 Bugs
0 Vulnerabilities
0 Security Hotspots
4 Code Smells

0.0% Coverage
0.0% Duplication

Hartorn requested review from andreybavt and Inokinoki October 13, 2023 01:25

Hartorn self-assigned this Oct 13, 2023

Hartorn force-pushed the new-worker-pool branch 7 times, most recently from bd58989 to a3b8bde Compare October 16, 2023 14:01

Hartorn changed the title ~~WIP: New worker pool~~ New worker pool Oct 16, 2023

Hartorn marked this pull request as ready for review October 16, 2023 15:32

Hartorn requested a review from a team as a code owner October 16, 2023 15:32

Inokinoki reviewed Oct 18, 2023

View reviewed changes

giskard/utils/worker_pool.py Outdated Show resolved Hide resolved

giskard/ml_worker/testing/registry/udf_repository.py Outdated Show resolved Hide resolved

andreybavt suggested changes Oct 19, 2023

View reviewed changes

andreybavt reviewed Oct 19, 2023

View reviewed changes

giskard/utils/worker_pool.py Show resolved Hide resolved

andreybavt reviewed Oct 19, 2023

View reviewed changes

andreybavt changed the title ~~New worker pool~~ GSK-1895 New worker pool Oct 19, 2023

Hartorn force-pushed the new-worker-pool branch from df6558d to 9e4784a Compare October 20, 2023 10:57

Hartorn commented Oct 20, 2023

View reviewed changes

Hartorn force-pushed the new-worker-pool branch from 2cf31fd to 0976146 Compare October 20, 2023 13:36

Inokinoki reviewed Oct 23, 2023

View reviewed changes

Hartorn force-pushed the new-worker-pool branch 5 times, most recently from 227cd39 to ed598a6 Compare October 23, 2023 15:52

Creating new worker pool

18565be

Hartorn and others added 10 commits October 23, 2023 18:11

Try to fix tests for CI

788f9f4

Removing git dependency

8dc5601

Repair tests with multiprocess (issue with coverage)

d9a853e

Adding logs capture in worker pool

4e126eb

Fix lock file

9c116e2

Improve threads + taking in account review

dfd6d1b

Making process manipulation safer

65af850

Adding some more logs and warm up at start for the worker

28d8b47

Improve code quality

82057d7

Marking analytics thread as daemon

62c72c4

Hartorn force-pushed the new-worker-pool branch from ed598a6 to 62c72c4 Compare October 23, 2023 16:23

Hartorn added 3 commits October 23, 2023 18:52

Minor cleaning

a873dae

Improving forced shutdown

b93067b

Re-working process worker

c53635d

Hartorn force-pushed the new-worker-pool branch from 3b2db1c to c53635d Compare October 23, 2023 17:53

Making safe_get safer

331b11e

Hartorn force-pushed the new-worker-pool branch from f7b8730 to 331b11e Compare October 23, 2023 19:42

Inokinoki and others added 5 commits October 24, 2023 12:56

Merge branch 'main' into new-worker-pool

91cb65d

Comment code

c50be07

Skipping tests for windows

626f69e

Merge branch 'main' into new-worker-pool

25eb2b9

Merge branch 'main' into new-worker-pool

8c04bfd

andreybavt approved these changes Oct 25, 2023

View reviewed changes

andreybavt added 4 commits October 25, 2023 16:27

Merge branch 'main' into new-worker-pool

cfebc73

# Conflicts: # .github/workflows/build-python.yml # pdm.lock # pyproject.toml

update lockfile

cc630bf

update lockfile

f4eb8bf

fix build

1e6cc6e

andreybavt merged commit e759a68 into main Oct 25, 2023
13 checks passed

andreybavt deleted the new-worker-pool branch October 25, 2023 15:56

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

GSK-1895 New worker pool #1478

GSK-1895 New worker pool #1478

Hartorn commented Oct 13, 2023 •

edited

Loading

andreybavt left a comment

andreybavt Oct 19, 2023

Hartorn Oct 19, 2023

Hartorn Oct 20, 2023

andreybavt Oct 19, 2023

Hartorn Oct 19, 2023

Hartorn Oct 20, 2023

andreybavt Oct 19, 2023

Hartorn Oct 19, 2023

andreybavt Oct 19, 2023

Hartorn Oct 19, 2023

Hartorn Oct 20, 2023

andreybavt Oct 19, 2023

Hartorn Oct 19, 2023

andreybavt Oct 19, 2023

Hartorn Oct 20, 2023

andreybavt commented Oct 19, 2023 •

edited

Loading

andreybavt Oct 19, 2023

Hartorn Oct 20, 2023

linear bot commented Oct 19, 2023

Hartorn left a comment

Hartorn Oct 20, 2023

Hartorn Oct 20, 2023

Hartorn Oct 20, 2023

Hartorn Oct 20, 2023

Hartorn Oct 20, 2023

Hartorn Oct 20, 2023

Inokinoki left a comment

sonarcloud bot commented Oct 25, 2023

GSK-1895 New worker pool #1478

GSK-1895 New worker pool #1478

Conversation

Hartorn commented Oct 13, 2023 • edited Loading

Description

Related Issue

Type of Change

Checklist

andreybavt left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andreybavt commented Oct 19, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

linear bot commented Oct 19, 2023

Hartorn left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Inokinoki left a comment

Choose a reason for hiding this comment

sonarcloud bot commented Oct 25, 2023

Hartorn commented Oct 13, 2023 •

edited

Loading

andreybavt commented Oct 19, 2023 •

edited

Loading