Skip to content

Fix/perf thread safe skip #161

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

Open
shrutu0929 wants to merge 3 commits into
base: main
Choose a base branch
from
+54 −43
Open

Fix/perf thread safe skip #161

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
00ad3a2
fix/perf: Thread-safe collection of semantic_skip_warnings in paralle…
Apr 9, 2026
9f20bb0
Merge branch 'main' into fix/perf-thread-safe-skip
shrutu0929 Apr 9, 2026
dea2f93
solve this error
Apr 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 26 additions & 17 deletions refactron/core/parallel.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
from typing import Any, Callable, Dict, List, Optional, Tuple

from refactron.core.analysis_result import FileAnalysisError
from refactron.core.models import FileMetrics
from refactron.core.models import FileMetrics, AnalysisSkipWarning
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai bot Apr 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Commit the formatter/linter output for this file.

Pre-commit is still red here: black and isort rewrite the file, and flake8 reports an overlong line in this changed signature/doc block.

As per coding guidelines, **/*.py: Use line length of 100 characters, enforced by black, isort, and flake8.

Also applies to: 58-75

🧰 Tools
🪛 GitHub Actions: Pre-commit

[error] pre-commit black hook id failed; files were modified by this hook (reformatted refactron/core/parallel.py). Commit the formatting changes.

[error] pre-commit isort hook id failed; files were modified by this hook (Fixing /home/runner/work/Refactron_lib/Refactron_lib/refactron/core/parallel.py). Commit the import/style changes.

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@refactron/core/parallel.py` at line 10, The file refactron.core.parallel has
uncommitted formatter/linter changes and an overlong signature/docblock (around
the block spanning lines 58–75); run and commit black and isort to normalize
imports and formatting, then break/rewrap the long function signature or
docstring text in the function or class defined in that 58–75 region so no line
exceeds 100 characters (ensure the function/method/class name in that block
retains its identifiers), then re-run flake8 to verify the overlong-line error
is resolved and commit the formatted file.


logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -58,20 +58,20 @@ def __init__(
def process_files(
self,
files: List[Path],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError]]],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError], Optional[AnalysisSkipWarning]]],
progress_callback: Optional[Callable[[int, int], None]] = None,
) -> Tuple[List[FileMetrics], List[FileAnalysisError]]:
) -> Tuple[List[FileMetrics], List[FileAnalysisError], List[AnalysisSkipWarning]]:
"""
Process multiple files in parallel.

Args:
files: List of file paths to process.
process_func: Function to process a single file. Should return
(FileMetrics, None) on success or (None, FileAnalysisError) on error.
(FileMetrics, None, skip_warn) on success or (None, FileAnalysisError, None) on error.
progress_callback: Optional callback for progress updates (completed, total).

Returns:
Tuple of (successful results, failed files).
Tuple of (successful results, failed files, skip warnings).
"""
if not self.enabled or len(files) <= 1:
# Process sequentially if disabled or only one file
Expand All @@ -86,20 +86,23 @@ def process_files(
def _process_sequential(
self,
files: List[Path],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError]]],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError], Optional[AnalysisSkipWarning]]],
progress_callback: Optional[Callable[[int, int], None]] = None,
) -> Tuple[List[FileMetrics], List[FileAnalysisError]]:
) -> Tuple[List[FileMetrics], List[FileAnalysisError], List[AnalysisSkipWarning]]:
"""Process files sequentially."""
results: List[FileMetrics] = []
errors: List[FileAnalysisError] = []
skips: List[AnalysisSkipWarning] = []

for i, file_path in enumerate(files):
try:
result, error = process_func(file_path)
result, error, skip = process_func(file_path)
if result is not None:
results.append(result)
if error is not None:
errors.append(error)
if skip is not None:
skips.append(skip)
except Exception as e:
logger.error(f"Unexpected error processing {file_path}: {e}", exc_info=True)
errors.append(
Expand All @@ -114,17 +117,18 @@ def _process_sequential(
if progress_callback:
progress_callback(i + 1, len(files))

return results, errors
return results, errors, skips

def _process_parallel_threads(
self,
files: List[Path],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError]]],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError], Optional[AnalysisSkipWarning]]],
progress_callback: Optional[Callable[[int, int], None]] = None,
) -> Tuple[List[FileMetrics], List[FileAnalysisError]]:
) -> Tuple[List[FileMetrics], List[FileAnalysisError], List[AnalysisSkipWarning]]:
"""Process files in parallel using threads."""
results: List[FileMetrics] = []
errors: List[FileAnalysisError] = []
skips: List[AnalysisSkipWarning] = []
completed = 0

with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
Expand All @@ -139,11 +143,13 @@ def _process_parallel_threads(
completed += 1

try:
result, error = future.result()
result, error, skip = future.result()
if result is not None:
results.append(result)
if error is not None:
errors.append(error)
if skip is not None:
skips.append(skip)
except Exception as e:
logger.error(f"Unexpected error processing {file_path}: {e}", exc_info=True)
recovery_msg = "Check the file for syntax errors or encoding issues"
Expand All @@ -159,14 +165,14 @@ def _process_parallel_threads(
if progress_callback:
progress_callback(completed, len(files))

return results, errors
return results, errors, skips

def _process_parallel_processes(
self,
files: List[Path],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError]]],
process_func: Callable[[Path], Tuple[Optional[FileMetrics], Optional[FileAnalysisError], Optional[AnalysisSkipWarning]]],
progress_callback: Optional[Callable[[int, int], None]] = None,
) -> Tuple[List[FileMetrics], List[FileAnalysisError]]:
) -> Tuple[List[FileMetrics], List[FileAnalysisError], List[AnalysisSkipWarning]]:
"""
Process files in parallel using processes.

Expand All @@ -176,6 +182,7 @@ def _process_parallel_processes(
"""
results: List[FileMetrics] = []
errors: List[FileAnalysisError] = []
skips: List[AnalysisSkipWarning] = []
completed = 0

try:
Expand All @@ -191,11 +198,13 @@ def _process_parallel_processes(
completed += 1

try:
result, error = future.result()
result, error, skip = future.result()
if result is not None:
results.append(result)
if error is not None:
errors.append(error)
if skip is not None:
skips.append(skip)
except Exception as e:
logger.error(f"Unexpected error processing {file_path}: {e}", exc_info=True)
recovery_msg = "Check the file for syntax errors or encoding issues"
Expand All @@ -216,7 +225,7 @@ def _process_parallel_processes(
logger.info("Falling back to sequential processing")
return self._process_sequential(files, process_func, progress_callback)

return results, errors
return results, errors, skips

def get_config(self) -> Dict[str, Any]:
"""
Expand Down
14 changes: 7 additions & 7 deletions refactron/core/refactron.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,17 +267,16 @@ def analyze(self, target: Union[str, Path]) -> AnalysisResult:
# Create a wrapper function for parallel processing
def process_file_wrapper(
file_path: Path,
) -> Tuple[Optional[FileMetrics], Optional[FileAnalysisError]]:
) -> Tuple[Optional[FileMetrics], Optional[FileAnalysisError], Optional[AnalysisSkipWarning]]:
try:
file_metrics, skip_warn = self._analyze_file(file_path)
if skip_warn is not None:
result.semantic_skip_warnings.append(skip_warn)
# Warnings are collected by ParallelProcessor.process_files return

# Update incremental tracker
if self.incremental_tracker.enabled:
self.incremental_tracker.update_file_state(file_path)

return file_metrics, None
return file_metrics, None, skip_warn
except AnalysisError as e:
logger.debug(f"Failed to analyze {file_path}: {e}")
error = FileAnalysisError(
Expand All @@ -286,7 +285,7 @@ def process_file_wrapper(
error_type=e.__class__.__name__,
recovery_suggestion=e.recovery_suggestion,
)
return None, error
return None, error, None
except Exception as e:
logger.error(f"Unexpected error analyzing {file_path}: {e}", exc_info=True)
error = FileAnalysisError(
Expand All @@ -295,16 +294,17 @@ def process_file_wrapper(
error_type=e.__class__.__name__,
recovery_suggestion="Check the file for syntax errors or encoding issues",
)
return None, error
return None, error, None

# Process files in parallel
file_metrics_list, error_list = self.parallel_processor.process_files(
file_metrics_list, error_list, skip_warnings = self.parallel_processor.process_files(
files,
process_file_wrapper,
)

result.file_metrics.extend(file_metrics_list)
result.failed_files.extend(error_list)
result.semantic_skip_warnings.extend(skip_warnings)
result.total_issues = sum(fm.issue_count for fm in file_metrics_list)
else:
# Sequential processing
Expand Down
35 changes: 18 additions & 17 deletions tests/test_config_management.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -903,16 +903,16 @@ def test_parallel_processor_sequential_and_thread_modes(tmp_path: Path) -> None:
def process_func(p: Path):
if p.name == "bad.py":
raise ValueError("boom")
return None, None
return None, None, None

p_seq = ParallelProcessor(max_workers=1, use_processes=False, enabled=True)
_, errors = p_seq.process_files(files, process_func)
_, errors, skips = p_seq.process_files(files, process_func)
assert p_seq.enabled is False
assert len(errors) == 1
assert isinstance(errors[0], FileAnalysisError)

p_thr = ParallelProcessor(max_workers=2, use_processes=False, enabled=True)
results, errors = p_thr.process_files(files, lambda p: (None, None))
results, errors, skips = p_thr.process_files(files, lambda p: (None, None, None))
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai bot Apr 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

RUF059: underscore unused unpacked return values in these tests.

Several tuple-unpacked variables are never read. Rename only the unused ones to underscore-prefixed names to clear Ruff warnings.

♻️ Suggested lint-safe updates 
-    results, errors, skips = p_thr.process_files(files, lambda p: (None, None, None))
+    results, errors, _skips = p_thr.process_files(files, lambda p: (None, None, None))

-        results, errors, skips = pp.process_files([Path("a.py")], raises_func)
+        _results, errors, _skips = pp.process_files([Path("a.py")], raises_func)

-        results, errors, skips = pp.process_files(files, success_func)
+        results, _errors, skips = pp.process_files(files, success_func)

-        results, errors, skips = pp.process_files([Path("a.py"), Path("b.py")], raises_func)
+        _results, errors, _skips = pp.process_files([Path("a.py"), Path("b.py")], raises_func)

-        results, errors, skips = pp.process_files([Path("a.py")], success_func)
+        results, _errors, _skips = pp.process_files([Path("a.py")], success_func)

-            results, errors, skips = pp.process_files([Path("a.py")], success_func)
+            results, _errors, _skips = pp.process_files([Path("a.py")], success_func)

Also applies to: 1229-1229, 1247-1247, 1253-1253, 1268-1268, 1278-1278

🧰 Tools
🪛 Ruff (0.15.9)

[warning] 915-915: Unpacked variable skips is never used

Prefix it with an underscore or any other dummy variable pattern

(RUF059)

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@tests/test_config_management.py` at line 915, The tuple-unpacked variables
from the call to p_thr.process_files (e.g., "results, errors, skips =
p_thr.process_files(...)" and similar unpackings at the other locations) include
values that are never used; rename the unused ones to underscore-prefixed names
(for example "_results", "_errors", or "_skips") so only the actually-read
variables remain non-underscored; update the specific unpackings that call
p_thr.process_files (and any equivalent test unpackings around the noted
locations) to use the underscore-prefixed names for unused values to silence the
RUF059 lint warnings.

assert results == []
assert errors == []
assert p_thr.get_config()["max_workers"] == 2
Expand Down Expand Up @@ -1171,11 +1171,11 @@ def make_error(path):


def success_func(p):
return make_metrics(p), None
return make_metrics(p), None, None


def error_func(p):
return None, make_error(p)
return None, make_error(p), None


def raises_func(p):
Expand Down Expand Up @@ -1210,23 +1210,23 @@ def test_get_config(self):
class TestSequentialProcessing:
def test_empty_files(self):
pp = ParallelProcessor(enabled=False)
results, errors = pp.process_files([], success_func)
assert results == [] and errors == []
results, errors, skips = pp.process_files([], success_func)
assert results == [] and errors == [] and skips == []

def test_single_file_success(self):
pp = ParallelProcessor(enabled=False)
files = [Path("a.py")]
results, errors = pp.process_files(files, success_func)
assert len(results) == 1 and len(errors) == 0
results, errors, skips = pp.process_files(files, success_func)
assert len(results) == 1 and len(errors) == 0 and len(skips) == 0

def test_single_file_error(self):
pp = ParallelProcessor(enabled=False)
results, errors = pp.process_files([Path("a.py")], error_func)
assert len(results) == 0 and len(errors) == 1
results, errors, skips = pp.process_files([Path("a.py")], error_func)
assert len(results) == 0 and len(errors) == 1 and len(skips) == 0

def test_single_file_exception(self):
pp = ParallelProcessor(enabled=False)
results, errors = pp.process_files([Path("a.py")], raises_func)
results, errors, skips = pp.process_files([Path("a.py")], raises_func)
assert len(errors) == 1

def test_progress_callback(self):
Expand All @@ -1244,12 +1244,13 @@ class TestThreadedProcessing:
def test_two_files_threads(self):
pp = ParallelProcessor(max_workers=2, use_processes=False, enabled=True)
files = [Path("a.py"), Path("b.py")]
results, errors = pp.process_files(files, success_func)
results, errors, skips = pp.process_files(files, success_func)
assert len(results) == 2
assert len(skips) == 0

def test_thread_error_handling(self):
pp = ParallelProcessor(max_workers=2, use_processes=False, enabled=True)
results, errors = pp.process_files([Path("a.py"), Path("b.py")], raises_func)
results, errors, skips = pp.process_files([Path("a.py"), Path("b.py")], raises_func)
assert len(errors) == 2

def test_thread_progress_callback(self):
Expand All @@ -1264,7 +1265,7 @@ def test_thread_progress_callback(self):

def test_single_file_goes_sequential(self):
pp = ParallelProcessor(max_workers=4, use_processes=False, enabled=True)
results, errors = pp.process_files([Path("a.py")], success_func)
results, errors, skips = pp.process_files([Path("a.py")], success_func)
assert len(results) == 1


Expand All @@ -1274,13 +1275,13 @@ def test_process_pool_falls_back_on_exception(self):
with patch(
"refactron.core.parallel.ProcessPoolExecutor", side_effect=Exception("spawn fail")
):
results, errors = pp.process_files([Path("a.py")], success_func)
results, errors, skips = pp.process_files([Path("a.py")], success_func)
assert len(results) == 1

def test_process_pool_success(self):
pp = ParallelProcessor(max_workers=2, use_processes=True, enabled=True)
mock_future = MagicMock()
mock_future.result.return_value = (make_metrics(Path("a.py")), None)
mock_future.result.return_value = (make_metrics(Path("a.py")), None, None)
mock_exec = MagicMock()
mock_exec.__enter__ = lambda s: s
mock_exec.__exit__ = MagicMock(return_value=False)
Expand Down
5 changes: 3 additions & 2 deletions tests/test_performance_optimization.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,11 +242,12 @@ def test_sequential_processing(self):

def process_func(file_path):
# Simulate processing
return None, None
return None, None, None

results, errors = processor.process_files(files, process_func)
results, errors, skips = processor.process_files(files, process_func)
assert len(results) == 0 # All return None
assert len(errors) == 0
assert len(skips) == 0


class TestMemoryProfiler:
Expand Down
Loading