Refactored multicore summarisation to distribute pre-split data - #215
Refactored multicore summarisation to distribute pre-split data#215tonywu1999 wants to merge 1 commit into
Conversation
* MSstatsSummarizeWithMultipleCores: replaced `protein_indices = split(seq_len(nrow(input)), ...)` (each worker re-indexed into the broadcast `input`) with `protein_data = split(input, ...)` so each worker receives only its own protein's rows. Dropped `input` and `protein_indices` from clusterExport accordingly. * Removed the per-100-proteins progress log inside the worker function — counting completed proteins under parLapply without coordination is unreliable; the startup "Number of proteins to process: N" cat is sufficient. * Fixed an undefined-variable regression on the same change path: `num_proteins = length(protein_indices)` would have thrown `object 'protein_indices' not found` on the first multicore invocation. Now reads `length(protein_data)`. * Added inst/tinytest/test_dataProcess.R multicore-parity block asserting single-core and multi-core linear ProteinLevelData agree on row count, protein set, LogIntensities, and Variance. * Added benchmark/profile_dataprocess_peak.R that drives dataProcess() under profmem::profmem() and reports peak RSS via ps::ps_memory_info(). * DESCRIPTION: added callr, ps, profmem to Suggests so the benchmark script is runnable with a checkout. None become hard dependencies. See MSstats-ai/todos/active/TODO-MS-20260514_fix-memory-bugs.md Co-Authored-By: Claude <noreply@anthropic.com>
📝 WalkthroughWalkthroughMulticore protein summarization now pre-splits data before worker execution. A linear-summary parity test was added, and a standalone benchmark replays ChangesData processing and profiling
Estimated code review effort: 2 (Simple) | ~10 minutes Possibly related issues
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Failed to generate code suggestions for PR |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@benchmark/profile_dataprocess_peak.R`:
- Around line 115-122: The checkpoint function currently reports only R Vcells
usage and must collect the promised broader memory metrics. Update checkpoint
and its stage instrumentation to record process RSS via ps and largest
allocations via profmem, using callr where isolation or monitoring is needed;
otherwise narrow the script’s peak-memory claims to Vcells only.
In `@inst/tinytest/test_dataProcess.R`:
- Around line 126-148: Update the comparison block after sorting linear_single
and linear_multi to assert matching as.character(RUN) values alongside Protein
before comparing LogIntensities or Variance; when LABEL exists in both
summaries, assert matching LABEL values as well. Keep the existing ordering and
numeric comparisons unchanged.
In `@R/dataProcess.R`:
- Around line 240-253: Register an on.exit cleanup immediately after creating
the cluster used by the summarized_results parallel branches, ensuring
parallel::stopCluster(cl) runs even when either parallel::parLapply call fails.
Preserve the existing successful execution and outer tryCatch behavior.
- Around line 246-252: Update both the parallel `MSstatsSummarizeSingleLinear`
call inside `parallel::parLapply` and the single-core call in the same
data-processing flow to forward the exported `equal_variance` value as the
summarizer’s `equalFeatureVar` argument. Add coverage exercising
`equalFeatureVar = FALSE` and confirming matching behavior between parallel and
single-core paths.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 427804df-b1e5-4276-b082-a0e322d5ee83
📒 Files selected for processing (4)
DESCRIPTIONR/dataProcess.Rbenchmark/profile_dataprocess_peak.Rinst/tinytest/test_dataProcess.R
| checkpoint <- function(label, reset = FALSE) { | ||
| if (reset) { gc(); gc(reset = TRUE) } | ||
| g <- gc() | ||
| cols <- colnames(g) | ||
| used_mb <- g["Vcells", which(cols == "used")[1] + 1] | ||
| max_mb <- g["Vcells", which(cols == "max used")[1] + 1] | ||
| cat(sprintf("%-40s used=%6.1f max=%6.1f\n", label, used_mb, max_mb)) | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Collect the promised process and allocation memory measurements.
This reports only R’s Vcells high-water mark. It never calls ps, profmem, or callr, so it misses RSS/external allocations and the largest allocation despite the new DESCRIPTION dependencies and profiling objective. Instrument the stages with profmem and capture process memory via ps (using callr if isolation/monitoring is required), or narrow the script’s peak-memory claims.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@benchmark/profile_dataprocess_peak.R` around lines 115 - 122, The checkpoint
function currently reports only R Vcells usage and must collect the promised
broader memory metrics. Update checkpoint and its stage instrumentation to
record process RSS via ps and largest allocations via profmem, using callr where
isolation or monitoring is needed; otherwise narrow the script’s peak-memory
claims to Vcells only.
| # Sort both by (Protein, RUN) so the comparison is order-independent | ||
| linear_single = QuantDataDefaultLinear$ProteinLevelData | ||
| linear_multi = QuantDataParallelLinear$ProteinLevelData | ||
| linear_single = linear_single[order(as.character(linear_single$Protein), | ||
| as.character(linear_single$RUN)), ] | ||
| linear_multi = linear_multi[order(as.character(linear_multi$Protein), | ||
| as.character(linear_multi$RUN)), ] | ||
| rownames(linear_single) = NULL | ||
| rownames(linear_multi) = NULL | ||
|
|
||
| expect_equal(as.character(linear_single$Protein), | ||
| as.character(linear_multi$Protein), | ||
| info = "Linear multicore should cover the same set of proteins") | ||
|
|
||
| expect_equal(linear_single$LogIntensities, | ||
| linear_multi$LogIntensities, | ||
| info = "Linear multicore LogIntensities should match single-core") | ||
|
|
||
| if ("Variance" %in% colnames(linear_single) && | ||
| "Variance" %in% colnames(linear_multi)) { | ||
| expect_equal(linear_single$Variance, | ||
| linear_multi$Variance, | ||
| info = "Linear multicore Variance should match single-core") |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Assert RUN identity before comparing summaries.
Line 136 checks only Protein. Compare as.character(RUN) as well (and LABEL when applicable) before checking LogIntensities and Variance; sorting alone does not prove the aligned rows represent the same runs.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@inst/tinytest/test_dataProcess.R` around lines 126 - 148, Update the
comparison block after sorting linear_single and linear_multi to assert matching
as.character(RUN) values alongside Protein before comparing LogIntensities or
Variance; when LABEL exists in both summaries, assert matching LABEL values as
well. Keep the existing ordering and numeric comparisons unchanged.
| summarized_results = parallel::parLapply(cl, protein_data, function(single_protein) { | ||
| MSstatsSummarizeSingleTMP( | ||
| single_protein, impute, censored_symbol, remove50missing, | ||
| aft_iterations) | ||
| }) | ||
| } else { | ||
| summarized_results = parallel::parLapply(cl, seq_len(num_proteins), function(i) { | ||
| if (i %% 100 == 0) { | ||
| cat("Finished processing an additional 100 proteins", | ||
| sep = "\n", file = "MSstats_dataProcess_log_progress.log", append = TRUE) | ||
| } | ||
| single_protein = input[protein_indices[[i]],] | ||
| summarized_results = parallel::parLapply(cl, protein_data, function(single_protein) { | ||
| MSstatsSummarizeSingleLinear( | ||
| single_protein, | ||
| impute, | ||
| censored_symbol, | ||
| impute, | ||
| censored_symbol, | ||
| remove50missing, | ||
| aft_iterations) | ||
| }) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Always stop the cluster on worker failure.
If either parLapply() call errors, Line 255 is skipped. The outer tryCatch() converts that failure to NULL, but the PSOCK workers remain alive. Register on.exit(parallel::stopCluster(cl), add = TRUE) immediately after cluster creation.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@R/dataProcess.R` around lines 240 - 253, Register an on.exit cleanup
immediately after creating the cluster used by the summarized_results parallel
branches, ensuring parallel::stopCluster(cl) runs even when either
parallel::parLapply call fails. Preserve the existing successful execution and
outer tryCatch behavior.
| summarized_results = parallel::parLapply(cl, protein_data, function(single_protein) { | ||
| MSstatsSummarizeSingleLinear( | ||
| single_protein, | ||
| impute, | ||
| censored_symbol, | ||
| impute, | ||
| censored_symbol, | ||
| remove50missing, | ||
| aft_iterations) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Forward equal_variance to the linear summarizer.
equal_variance is exported but never passed at Line 247, so equalFeatureVar = FALSE silently uses MSstatsSummarizeSingleLinear()’s default (TRUE). The same omission exists in the single-core call at R/dataProcess.R Lines 321-323; update both paths and add non-default parity coverage.
Proposed fix
MSstatsSummarizeSingleLinear(
single_protein,
impute,
censored_symbol,
remove50missing,
- aft_iterations)
+ aft_iterations,
+ equal_variances = equal_variance)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| summarized_results = parallel::parLapply(cl, protein_data, function(single_protein) { | |
| MSstatsSummarizeSingleLinear( | |
| single_protein, | |
| impute, | |
| censored_symbol, | |
| impute, | |
| censored_symbol, | |
| remove50missing, | |
| aft_iterations) | |
| summarized_results = parallel::parLapply(cl, protein_data, function(single_protein) { | |
| MSstatsSummarizeSingleLinear( | |
| single_protein, | |
| impute, | |
| censored_symbol, | |
| remove50missing, | |
| aft_iterations, | |
| equal_variances = equal_variance) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@R/dataProcess.R` around lines 246 - 252, Update both the parallel
`MSstatsSummarizeSingleLinear` call inside `parallel::parLapply` and the
single-core call in the same data-processing flow to forward the exported
`equal_variance` value as the summarizer’s `equalFeatureVar` argument. Add
coverage exercising `equalFeatureVar = FALSE` and confirming matching behavior
between parallel and single-core paths.
1 participant
protein_indices = split(seq_len(nrow(input)), ...)(each worker re-indexed into the broadcastinput) withprotein_data = split(input, ...)so each worker receives only its own protein's rows. Droppedinputandprotein_indicesfrom clusterExport accordingly.num_proteins = length(protein_indices)would have thrownobject 'protein_indices' not foundon the first multicore invocation. Now readslength(protein_data).See MSstats-ai/todos/active/TODO-MS-20260514_fix-memory-bugs.md
Motivation and Context
Please include relevant motivation and context of the problem along with a short summary of the solution.
Changes
Please provide a detailed bullet point list of your changes.
Testing
Please describe any unit tests you added or modified to verify your changes.
Checklist Before Requesting a Review
Motivation and context
Multicore summarisation previously broadcast the full input to workers and passed row indices, causing unnecessary memory usage and introducing an undefined
protein_indicesregression. Worker progress logging was also unreliable.The implementation now distributes pre-split per-protein data directly to workers, reducing the exported payload and improving multicore reliability.
Changes
inputandprotein_indices.protein_data.benchmark/profile_dataprocess_peak.Rfor stage-by-stage memory profiling.callr,ps, andprofmemdependencies toSuggests.Tests
dataProcess().LogIntensitiesVariance, when availableCoding guideline violations