Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Avoid hanging forever after a parallel job was killed #7834

Merged
merged 10 commits into from Dec 12, 2022
Merged

Avoid hanging forever after a parallel job was killed #7834

Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
fc1baf7
Replace multiprocessing.pool with concurrent.futures.ProcessPoolExecu…
daniel-wer Nov 23, 2022
d5e17d8
Fix initargs parameter to be a tuple.
daniel-wer Nov 24, 2022
2657772
add whatsnew fragment
daniel-wer Nov 24, 2022
2e47049
Add a test that an error in the initializer for the parallel jobs, do…
daniel-wer Nov 28, 2022
236e5e9
Increase timeout for parallel initializer test to avoid flakiness.
daniel-wer Dec 5, 2022
61b8342
Guard concurrent.futures imports, because they are not available in E…
daniel-wer Dec 6, 2022
a52cb6c
Check availability of concurrent.futures module instead of multiproce…
daniel-wer Dec 8, 2022
30af13a
Small fixes
DanielNoord Dec 10, 2022
e164bea
Update tests/test_check_parallel.py
daniel-wer Dec 12, 2022
69729db
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Dec 12, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
12 changes: 5 additions & 7 deletions pylint/lint/parallel.py
View file
Open in desktop
Expand Up @@ -8,6 +8,7 @@
import warnings
from collections import defaultdict
from collections.abc import Iterable, Sequence
from concurrent.futures import ProcessPoolExecutor
from typing import TYPE_CHECKING, Any

import dill
Expand Down Expand Up @@ -137,9 +138,9 @@ def check_parallel(
# is identical to the linter object here. This is required so that
# a custom PyLinter object can be used.
initializer = functools.partial(_worker_initialize, arguments=arguments)
with multiprocessing.Pool(
jobs, initializer=initializer, initargs=[dill.dumps(linter)]
) as pool:
with ProcessPoolExecutor(
max_workers=jobs, initializer=initializer, initargs=(dill.dumps(linter),)
) as executor:
linter.open()
all_stats = []
all_mapreduce_data: defaultdict[
Expand All @@ -158,7 +159,7 @@ def check_parallel(
stats,
msg_status,
mapreduce_data,
) in pool.imap_unordered(_worker_check_single_file, files):
) in executor.map(_worker_check_single_file, files):
linter.file_state.base_name = base_name
linter.file_state._is_base_filestate = False
linter.set_current_module(module, file_path)
Expand All @@ -168,8 +169,5 @@ def check_parallel(
all_mapreduce_data[worker_idx].append(mapreduce_data)
linter.msg_status |= msg_status

DanielNoord marked this conversation as resolved.
Show resolved Hide resolved
pool.close()
pool.join()

_merge_mapreduce_data(linter, all_mapreduce_data)
linter.stats = merge_stats([linter.stats] + all_stats)