Julian/JuliaMSI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Hot fix to the precalculation atomic variable accountability, and other base threads safety

Browse Source
This commit is contained in:
Pixelguy14 2026-03-12 12:51:16 -06:00
parent 3da0dcaae9
commit 8b6bed0ff5
3 changed files with 51 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
app.jl
View File

@ -1434,7 +1434,11 @@ end
# --- 2. Parameter Assembly with Validation --- # --- 2. Parameter Assembly with Validation ---
current_pipeline_step = "Configuring parameters..." current_pipeline_step = "Configuring parameters..."
println("DEBUG: Configuring parameters and validating enabled steps...") println("DEBUG: Configuring parameters and validating enabled steps...")
ref_peaks = Dict{Float64, String}(p["mz"] => p["label"] for p in reference_peaks_list) ref_peaks = Dict{Float64, String}(
parse(Float64, string(p["mz"])) => string(get(p, "label", ""))
for p in reference_peaks_list
if tryparse(Float64, string(p["mz"])) !== nothing
)
final_params = Dict{Symbol, Dict{Symbol, Any}}() final_params = Dict{Symbol, Dict{Symbol, Any}}()
validation_errors = String[] validation_errors = String[]
@ -1944,7 +1948,11 @@ end
try try
msg = "Recalculating suggestions..." msg = "Recalculating suggestions..."
ref_peaks = Dict(p["mz"] => p["label"] for p in reference_peaks_list) ref_peaks = Dict{Float64, String}(
parse(Float64, string(p["mz"])) => string(get(p, "label", ""))
for p in reference_peaks_list
if tryparse(Float64, string(p["mz"])) !== nothing
)
recommended_params = main_precalculation(msi_data, reference_peaks=ref_peaks) recommended_params = main_precalculation(msi_data, reference_peaks=ref_peaks)

66
src/MSIData.jl
View File

@ -277,9 +277,9 @@ mutable struct MSIData
# Buffer Pool for binary data operations # Buffer Pool for binary data operations
buffer_pool::SimpleBufferPool buffer_pool::SimpleBufferPool
# Pre-computed analytics/metadata - use Threads.Atomic for compatibility # Pre-computed analytics/metadata - use Base.Threads.Atomic for compatibility
global_min_mz::Threads.Atomic{Float64} global_min_mz::Base.Threads.Atomic{Float64}
global_max_mz::Threads.Atomic{Float64} global_max_mz::Base.Threads.Atomic{Float64}
spectrum_stats_df::Union{DataFrame, Nothing} spectrum_stats_df::Union{DataFrame, Nothing}
bloom_filters::Union{Vector{<:BloomFilter}, Nothing} bloom_filters::Union{Vector{<:BloomFilter}, Nothing}
analytics_ready::AtomicFlag analytics_ready::AtomicFlag
@ -289,7 +289,7 @@ mutable struct MSIData
obj = new(source, metadata, instrument_meta, dims, coordinate_map, obj = new(source, metadata, instrument_meta, dims, coordinate_map,
Dict(), [], cache_size, ReentrantLock(), Dict(), [], cache_size, ReentrantLock(),
SimpleBufferPool(), SimpleBufferPool(),
Threads.Atomic{Float64}(Inf), Threads.Atomic{Float64}(-Inf), Base.Threads.Atomic{Float64}(Inf), Base.Threads.Atomic{Float64}(-Inf),
nothing, nothing, AtomicFlag(), nothing) nothing, nothing, AtomicFlag(), nothing)
# Ensure file handles are closed when the object is garbage collected # Ensure file handles are closed when the object is garbage collected
@ -319,8 +319,8 @@ Sets the global m/z range for the MSIData object.
- `nothing` - `nothing`
""" """
function set_global_mz_range!(data::MSIData, min_mz::Float64, max_mz::Float64) function set_global_mz_range!(data::MSIData, min_mz::Float64, max_mz::Float64)
Threads.atomic_xchg!(data.global_min_mz, min_mz) Base.Threads.atomic_xchg!(data.global_min_mz, min_mz)
Threads.atomic_xchg!(data.global_max_mz, max_mz) Base.Threads.atomic_xchg!(data.global_max_mz, max_mz)
end end
""" """
@ -337,8 +337,8 @@ Gets the global m/z range for the MSIData object.
""" """
function get_global_mz_range(data::MSIData) function get_global_mz_range(data::MSIData)
# Atomic read using atomic_add! with zero # Atomic read using atomic_add! with zero
min_mz = Threads.atomic_add!(data.global_min_mz, 0.0) min_mz = Base.Threads.atomic_add!(data.global_min_mz, 0.0)
max_mz = Threads.atomic_add!(data.global_max_mz, 0.0) max_mz = Base.Threads.atomic_add!(data.global_max_mz, 0.0)
return (min_mz, max_mz) return (min_mz, max_mz)
end end
@ -896,8 +896,8 @@ function precompute_analytics(msi_data::MSIData)
num_spectra = length(msi_data.spectra_metadata) num_spectra = length(msi_data.spectra_metadata)
# Initialize thread-local variables for global stats # Initialize thread-local variables for global stats
thread_local_min_mz = [Inf for _ in 1:Threads.nthreads()] thread_local_min_mz = [Inf for _ in 1:Base.Threads.nthreads()]
thread_local_max_mz = [-Inf for _ in 1:Threads.nthreads()] thread_local_max_mz = [-Inf for _ in 1:Base.Threads.nthreads()]
# Initialize vectors for per-spectrum stats # Initialize vectors for per-spectrum stats
tics = Vector{Float64}(undef, num_spectra) tics = Vector{Float64}(undef, num_spectra)
@ -937,7 +937,7 @@ function precompute_analytics(msi_data::MSIData)
# Update thread-local global m/z range # Update thread-local global m/z range
local_min, local_max = extrema(mz) local_min, local_max = extrema(mz)
thread_id = Threads.threadid() thread_id = Base.Threads.threadid()
thread_local_min_mz[thread_id] = min(thread_local_min_mz[thread_id], local_min) thread_local_min_mz[thread_id] = min(thread_local_min_mz[thread_id], local_min)
thread_local_max_mz[thread_id] = max(thread_local_max_mz[thread_id], local_max) thread_local_max_mz[thread_id] = max(thread_local_max_mz[thread_id], local_max)
min_mzs[idx] = local_min min_mzs[idx] = local_min
@ -963,16 +963,16 @@ function precompute_analytics(msi_data::MSIData)
# Only update if we found valid ranges # Only update if we found valid ranges
if isfinite(g_min_mz) && isfinite(g_max_mz) && g_min_mz < g_max_mz if isfinite(g_min_mz) && isfinite(g_max_mz) && g_min_mz < g_max_mz
Threads.atomic_xchg!(msi_data.global_min_mz, g_min_mz) Base.Threads.atomic_xchg!(msi_data.global_min_mz, g_min_mz)
Threads.atomic_xchg!(msi_data.global_max_mz, g_max_mz) Base.Threads.atomic_xchg!(msi_data.global_max_mz, g_max_mz)
else else
# Fallback: calculate from first spectrum # Fallback: calculate from first spectrum
try try
if num_spectra > 0 if num_spectra > 0
mz, _ = GetSpectrum(msi_data, 1) mz, _ = GetSpectrum(msi_data, 1)
if !isempty(mz) if !isempty(mz)
Threads.atomic_xchg!(msi_data.global_min_mz, minimum(mz)) Base.Threads.atomic_xchg!(msi_data.global_min_mz, minimum(mz))
Threads.atomic_xchg!(msi_data.global_max_mz, maximum(mz)) Base.Threads.atomic_xchg!(msi_data.global_max_mz, maximum(mz))
end end
end end
catch e catch e
@ -1040,10 +1040,10 @@ function get_total_spectrum_imzml(msi_data::MSIData; num_bins::Int=2000, masked_
local global_min_mz, global_max_mz local global_min_mz, global_max_mz
if Threads.atomic_add!(msi_data.global_min_mz, 0.0) !== Inf if Base.Threads.atomic_add!(msi_data.global_min_mz, 0.0) !== Inf
println(" Using pre-computed m/z range.") println(" Using pre-computed m/z range.")
global_min_mz = Threads.atomic_add!(msi_data.global_min_mz, 0.0) global_min_mz = Base.Threads.atomic_add!(msi_data.global_min_mz, 0.0)
global_max_mz = Threads.atomic_add!(msi_data.global_max_mz, 0.0) global_max_mz = Base.Threads.atomic_add!(msi_data.global_max_mz, 0.0)
else else
# 1. First Pass: Find the global m/z range by reading the fast .ibd file # 1. First Pass: Find the global m/z range by reading the fast .ibd file
pass1_start_time = time_ns() pass1_start_time = time_ns()
@ -1077,14 +1077,14 @@ function get_total_spectrum_imzml(msi_data::MSIData; num_bins::Int=2000, masked_
min_mz = global_min_mz min_mz = global_min_mz
# Initialize thread-local intensity sums # Initialize thread-local intensity sums
thread_local_intensity_sums = [zeros(Float64, num_bins) for _ in 1:Threads.nthreads()] thread_local_intensity_sums = [zeros(Float64, num_bins) for _ in 1:Base.Threads.nthreads()]
thread_local_spectra_processed = [0 for _ in 1:Threads.nthreads()] thread_local_spectra_processed = [0 for _ in 1:Base.Threads.nthreads()]
# 3. Second Pass: Optimized binning # 3. Second Pass: Optimized binning
pass2_start_time = time_ns() pass2_start_time = time_ns()
println(" Pass 2: Summing intensities into $num_bins bins...") println(" Pass 2: Summing intensities into $num_bins bins...")
_iterate_spectra_fast(msi_data, masked_indices === nothing ? nothing : collect(masked_indices)) do idx, mz, intensity _iterate_spectra_fast(msi_data, masked_indices === nothing ? nothing : collect(masked_indices)) do idx, mz, intensity
thread_id = Threads.threadid() thread_id = Base.Threads.threadid()
thread_local_spectra_processed[thread_id] += 1 thread_local_spectra_processed[thread_id] += 1
if isempty(mz) if isempty(mz)
return return
@ -1109,7 +1109,7 @@ function get_total_spectrum_imzml(msi_data::MSIData; num_bins::Int=2000, masked_
# Combine thread-local results # Combine thread-local results
intensity_sum = zeros(Float64, num_bins) intensity_sum = zeros(Float64, num_bins)
num_spectra_processed = 0 num_spectra_processed = 0
for i in 1:Threads.nthreads() for i in 1:Base.Threads.nthreads()
intensity_sum .+= thread_local_intensity_sums[i] intensity_sum .+= thread_local_intensity_sums[i]
num_spectra_processed += thread_local_spectra_processed[i] num_spectra_processed += thread_local_spectra_processed[i]
end end
@ -1157,10 +1157,10 @@ function get_total_spectrum_mzml(msi_data::MSIData; num_bins::Int=2000, masked_i
local global_min_mz, global_max_mz local global_min_mz, global_max_mz
if Threads.atomic_add!(msi_data.global_min_mz, 0.0) !== Inf if Base.Threads.atomic_add!(msi_data.global_min_mz, 0.0) !== Inf
println(" Using pre-computed m/z range.") println(" Using pre-computed m/z range.")
global_min_mz = Threads.atomic_add!(msi_data.global_min_mz, 0.0) global_min_mz = Base.Threads.atomic_add!(msi_data.global_min_mz, 0.0)
global_max_mz = Threads.atomic_add!(msi_data.global_max_mz, 0.0) global_max_mz = Base.Threads.atomic_add!(msi_data.global_max_mz, 0.0)
else else
# --- Pass 1: Find m/z range and cache it --- # --- Pass 1: Find m/z range and cache it ---
pass1_start_time = time_ns() pass1_start_time = time_ns()
@ -1200,11 +1200,11 @@ function get_total_spectrum_mzml(msi_data::MSIData; num_bins::Int=2000, masked_i
min_mz = global_min_mz min_mz = global_min_mz
# Initialize thread-local intensity sums # Initialize thread-local intensity sums
thread_local_intensity_sums = [zeros(Float64, num_bins) for _ in 1:Threads.nthreads()] thread_local_intensity_sums = [zeros(Float64, num_bins) for _ in 1:Base.Threads.nthreads()]
thread_local_spectra_processed = [0 for _ in 1:Threads.nthreads()] thread_local_spectra_processed = [0 for _ in 1:Base.Threads.nthreads()]
_iterate_spectra_fast(msi_data, masked_indices === nothing ? nothing : collect(masked_indices)) do idx, mz, intensity _iterate_spectra_fast(msi_data, masked_indices === nothing ? nothing : collect(masked_indices)) do idx, mz, intensity
thread_id = Threads.threadid() thread_id = Base.Threads.threadid()
thread_local_spectra_processed[thread_id] += 1 thread_local_spectra_processed[thread_id] += 1
if isempty(mz) if isempty(mz)
return return
@ -1224,7 +1224,7 @@ function get_total_spectrum_mzml(msi_data::MSIData; num_bins::Int=2000, masked_i
# Combine thread-local results # Combine thread-local results
intensity_sum = zeros(Float64, num_bins) intensity_sum = zeros(Float64, num_bins)
num_spectra_processed = 0 num_spectra_processed = 0
for i in 1:Threads.nthreads() for i in 1:Base.Threads.nthreads()
intensity_sum .+= thread_local_intensity_sums[i] intensity_sum .+= thread_local_intensity_sums[i]
num_spectra_processed += thread_local_spectra_processed[i] num_spectra_processed += thread_local_spectra_processed[i]
end end
@ -1722,7 +1722,7 @@ end
A smart dispatcher for internal, high-performance iteration over spectra. A smart dispatcher for internal, high-performance iteration over spectra.
It automatically chooses between serial and parallel execution based on the It automatically chooses between serial and parallel execution based on the
number of available threads (`Threads.nthreads()`). number of available threads (`Base.Threads.nthreads()`).
# Arguments # Arguments
- `f`: A function to call for each spectrum, with the signature `f(index, mz, intensity)`. - `f`: A function to call for each spectrum, with the signature `f(index, mz, intensity)`.
@ -1745,7 +1745,7 @@ function _iterate_spectra_fast(f::Function, data::MSIData, indices_to_iterate::U
# Default to all indices if not specified # Default to all indices if not specified
all_indices = (indices_to_iterate === nothing) ? (1:length(data.spectra_metadata)) : indices_to_iterate all_indices = (indices_to_iterate === nothing) ? (1:length(data.spectra_metadata)) : indices_to_iterate
if Threads.nthreads() > 1 && length(all_indices) > 100 # Heuristic: only parallelize for enough work if Base.Threads.nthreads() > 1 && length(all_indices) > 100 # Heuristic: only parallelize for enough work
_iterate_spectra_fast_parallel(f, data, all_indices) _iterate_spectra_fast_parallel(f, data, all_indices)
else else
_iterate_spectra_fast_serial(f, data, all_indices) _iterate_spectra_fast_serial(f, data, all_indices)
@ -1794,11 +1794,11 @@ Each thread gets its own file handle, eliminating contention.
""" """
function _iterate_spectra_fast_parallel(f::Function, data::MSIData, indices::AbstractVector) function _iterate_spectra_fast_parallel(f::Function, data::MSIData, indices::AbstractVector)
# Split indices into chunks for each thread # Split indices into chunks for each thread
n_chunks = Threads.nthreads() n_chunks = Base.Threads.nthreads()
chunk_size = ceil(Int, length(indices) / n_chunks) chunk_size = ceil(Int, length(indices) / n_chunks)
chunks = collect(Iterators.partition(indices, chunk_size)) chunks = collect(Iterators.partition(indices, chunk_size))
Threads.@threads for chunk in chunks Base.Threads.@threads for chunk in chunks
# Each thread gets its own file handle based on source type # Each thread gets its own file handle based on source type
if data.source isa ImzMLSource if data.source isa ImzMLSource
local_handle = open(data.source.ibd_handle.path, "r") local_handle = open(data.source.ibd_handle.path, "r")

16
src/Precalculations.jl
View File

@ -81,12 +81,12 @@ function analyze_mass_accuracy(
println("Analyzing mass accuracy for $(length(spectrum_indices)) spectra...") println("Analyzing mass accuracy for $(length(spectrum_indices)) spectra...")
all_ppm_errors = Float64[] all_ppm_errors = Float64[]
total_matched_peaks = Atomic{Int}(0) total_matched_peaks = Base.Threads.Atomic{Int}(0)
total_spectra_processed = Atomic{Int}(0) total_spectra_processed = Base.Threads.Atomic{Int}(0)
results_lock = ReentrantLock() results_lock = ReentrantLock()
_iterate_spectra_fast(msi_data, spectrum_indices) do idx, mz, intensity _iterate_spectra_fast(msi_data, spectrum_indices) do idx, mz, intensity
atomic_add!(total_spectra_processed, 1) Base.Threads.atomic_add!(total_spectra_processed, 1)
if !validate_spectrum(mz, intensity) if !validate_spectrum(mz, intensity)
@warn "Spectrum $idx is invalid, skipping mass accuracy analysis for it." @warn "Spectrum $idx is invalid, skipping mass accuracy analysis for it."
return return
@ -111,7 +111,7 @@ function analyze_mass_accuracy(
lock(results_lock) do lock(results_lock) do
push!(all_ppm_errors, min_ppm_error) push!(all_ppm_errors, min_ppm_error)
end end
atomic_add!(total_matched_peaks, 1) Base.Threads.atomic_add!(total_matched_peaks, 1)
end end
end end
end end
@ -671,8 +671,8 @@ function analyze_peak_characteristics(msi_data::MSIData, instrument_analysis::Di
peak_counts = Int[] peak_counts = Int[]
fwhm_values = Float64[] fwhm_values = Float64[]
spectra_analyzed = Atomic{Int}(0) spectra_analyzed = Base.Threads.Atomic{Int}(0)
peaks_analyzed = Atomic{Int}(0) peaks_analyzed = Base.Threads.Atomic{Int}(0)
results_lock = ReentrantLock() results_lock = ReentrantLock()
_iterate_spectra_fast(msi_data, spectrum_indices) do idx, mz, intensity _iterate_spectra_fast(msi_data, spectrum_indices) do idx, mz, intensity
@ -680,7 +680,7 @@ function analyze_peak_characteristics(msi_data::MSIData, instrument_analysis::Di
return return
end end
atomic_add!(spectra_analyzed, 1) Base.Threads.atomic_add!(spectra_analyzed, 1)
meta = msi_data.spectra_metadata[idx] meta = msi_data.spectra_metadata[idx]
# Detect peaks with lower SNR threshold to find more peaks # Detect peaks with lower SNR threshold to find more peaks
@ -709,7 +709,7 @@ function analyze_peak_characteristics(msi_data::MSIData, instrument_analysis::Di
r2 = _fit_gaussian_and_r2(mz, intensity, peak_idx, 5) r2 = _fit_gaussian_and_r2(mz, intensity, peak_idx, 5)
push!(r_squared_values, r2) push!(r_squared_values, r2)
end end
atomic_add!(peaks_analyzed, 1) Base.Threads.atomic_add!(peaks_analyzed, 1)
if peaks_analyzed[] <= 3 if peaks_analyzed[] <= 3
println("DEBUG: Peak at m/z $(peak.mz), FWHM = $(fwhm_ppm) ppm, R² = $(r_squared_values[end])") println("DEBUG: Peak at m/z $(peak.mz), FWHM = $(fwhm_ppm) ppm, R² = $(r_squared_values[end])")