From 0b26cfae11ec68391e47946b2bd362ce45602a69 Mon Sep 17 00:00:00 2001 From: Noah Cylich Date: Tue, 9 Jun 2026 17:33:35 -0700 Subject: [PATCH] cactus-kernels: bounds & robustness in conv/DSP/WAV/thread-pool conv.cpp: drop the incorrect Apple depthwise fast path and guard L --- cactus-kernels/src/conv.cpp | 67 ++-------------------------------- cactus-kernels/src/dsp.cpp | 7 ++++ cactus-kernels/src/threading.h | 56 ++++++++++++++++++---------- cactus-kernels/src/wav.h | 19 +++++++++- 4 files changed, 65 insertions(+), 84 deletions(-) diff --git a/cactus-kernels/src/conv.cpp b/cactus-kernels/src/conv.cpp index a90b5d057..7a93d1e39 100644 --- a/cactus-kernels/src/conv.cpp +++ b/cactus-kernels/src/conv.cpp @@ -18,71 +18,12 @@ constexpr size_t T_TILE_F16 = 2; constexpr size_t ACCELERATE_K_THRESHOLD = 32; constexpr size_t ACCELERATE_L_THRESHOLD = 128; -#ifdef __APPLE__ -static void conv1d_causal_depthwise_f16_accelerate( - const __fp16* input, - const __fp16* weight, - __fp16* output, - size_t N, size_t L, size_t C, size_t K, size_t dilation) -{ - const size_t in_bs = L * C; - const size_t out_bs = L * C; - CactusThreading::parallel_for_2d(N, C, CactusThreading::Thresholds::ATTENTION, [&](size_t n, size_t c) { - const __fp16* Xb = input + n * in_bs; - __fp16* Yb = output + n * out_bs; - const __fp16* Wc = weight + c * K; - - if (dilation == 1) { - - size_t i = 0; - for(; i + 7 < L; i += 8){ - float16x8_t acc = vdupq_n_f16(0.0f); - - for (size_t k = 0; k < K; ++k) { - float16x8_t input_vec = vld1q_f16(&Xb[(i + k) * C + c]); - float16x8_t weight_vec = vdupq_n_f16(Wc[k]); - acc = vfmaq_f16(acc, input_vec, weight_vec); - } - vst1q_f16(&Yb[i * C + c], acc); - } - - for (; i < L; ++i) { - float acc = 0.0f; - for (size_t k = 0; k < K; ++k) { - acc += float(Xb[(i + k) * C + c]) * float(Wc[k]); - } - Yb[i * C + c] = (__fp16)acc; - } - } else { - - for (size_t i = 0; i < L; ++i) { - float acc = 0.0f; - - for (size_t k = 0; k < K; ++k) { - size_t idx = i + k * dilation; - acc += float(Xb[idx * C + c]) * float(Wc[k]); - } - - Yb[i * C + c] = (__fp16)acc; - } - } - }); -} -#endif - void cactus_conv1d_causal_depthwise_f16( const __fp16* input, const __fp16* weight, __fp16* output, size_t N, size_t L, size_t C, size_t K, size_t dilation) { -#ifdef __APPLE__ - if (K >= ACCELERATE_K_THRESHOLD && L >= ACCELERATE_L_THRESHOLD) { - conv1d_causal_depthwise_f16_accelerate(input, weight, output, N, L, C, K, dilation); - return; - } -#endif - const size_t in_bs = L * C; const size_t out_bs = L * C; @@ -316,7 +257,7 @@ static void conv1d_f16_accelerate( size_t K, size_t stride ){ - const size_t out_len = ((L - K) / stride) + 1; + const size_t out_len = (L < K) ? 0 : ((L - K) / stride) + 1; const size_t in_bs = C_in * L; const size_t out_bs = C_out * out_len; @@ -378,7 +319,7 @@ static void conv1d_f16_neon( size_t K, size_t stride ){ - const size_t out_len = ((L - K) / stride) + 1; + const size_t out_len = (L < K) ? 0 : ((L - K) / stride) + 1; const size_t in_bs = C_in * L; const size_t out_bs = C_out * out_len; @@ -448,7 +389,7 @@ static void conv1d_f16_gemm( size_t K, size_t stride ) { - const size_t out_len = (L - K) / stride + 1; + const size_t out_len = (L < K) ? 0 : (L - K) / stride + 1; const size_t col_K = C_in * K; std::vector W_f32(C_out * col_K); @@ -791,7 +732,7 @@ void cactus_stft_f16( size_t K, size_t stride, size_t num_fft_bins ) { - const size_t out_len = ((L - K) / stride) + 1; + const size_t out_len = (L < K) ? 0 : ((L - K) / stride) + 1; const size_t in_bs = C_in * L; const size_t out_bs = 2 * num_fft_bins * out_len; diff --git a/cactus-kernels/src/dsp.cpp b/cactus-kernels/src/dsp.cpp index 13d3fc6fe..3bd20d26a 100644 --- a/cactus-kernels/src/dsp.cpp +++ b/cactus-kernels/src/dsp.cpp @@ -452,6 +452,9 @@ void cactus_compute_spectrogram_f32( padded_waveform.assign(padded_length, 0.0f); if (std::strcmp(pad_mode, "reflect") == 0) { + if (waveform_length <= pad_length) { + throw std::invalid_argument("reflect padding requires waveform_length > frame_length/2"); + } for (size_t i = 0; i < pad_length; i++) padded_waveform[i] = waveform[pad_length - i]; std::copy(waveform, waveform + waveform_length, padded_waveform.data() + pad_length); for (size_t i = 0; i < pad_length; i++) padded_waveform[pad_length + waveform_length + i] = waveform[waveform_length - 2 - i]; @@ -465,6 +468,10 @@ void cactus_compute_spectrogram_f32( input_length = padded_length; } + if (input_length < frame_length) { + throw std::invalid_argument("waveform shorter than frame_length"); + } + const size_t num_frames = 1 + (input_length - frame_length) / hop_length; const size_t num_frequency_bins = (actual_fft_length / 2) + 1; diff --git a/cactus-kernels/src/threading.h b/cactus-kernels/src/threading.h index e5795329f..d4f9af30c 100644 --- a/cactus-kernels/src/threading.h +++ b/cactus-kernels/src/threading.h @@ -386,15 +386,25 @@ namespace CactusThreading { private: static constexpr size_t MAX_WORKERS = 16; + struct Batch { + std::mutex mutex; + std::condition_variable cv; + size_t remaining; + explicit Batch(size_t count) : remaining(count) {} + }; + + static std::shared_ptr& current_batch() { + static thread_local std::shared_ptr batch; + return batch; + } + std::vector workers; std::deque> tasks; std::mutex mutex; std::condition_variable work_available; - std::condition_variable work_done; bool stop{false}; - std::atomic pending_tasks{0}; size_t num_workers_; void worker_thread() { @@ -415,17 +425,21 @@ namespace CactusThreading { } task(); + } + } - if (pending_tasks.fetch_sub(1, std::memory_order_acq_rel) == 1) { - std::lock_guard lock(mutex); - work_done.notify_one(); - } + static void finish_batch(const std::shared_ptr& batch) { + size_t left; + { + std::lock_guard lock(batch->mutex); + left = --batch->remaining; } + if (left == 0) batch->cv.notify_all(); } public: explicit ThreadPool(size_t num_threads = std::thread::hardware_concurrency()) - : stop(false), pending_tasks(0) { + : stop(false) { num_workers_ = std::min(num_threads, MAX_WORKERS); if (num_workers_ == 0) num_workers_ = 1; @@ -477,7 +491,6 @@ namespace CactusThreading { { std::lock_guard lock(mutex); - pending_tasks.fetch_add(1, std::memory_order_relaxed); tasks.emplace_back([task](){ (*task)(); }); } work_available.notify_one(); @@ -487,35 +500,37 @@ namespace CactusThreading { template void enqueue_batch(size_t total_work, F task_func) { - if (total_work == 0) return; + if (total_work == 0) { current_batch().reset(); return; } const size_t num_tasks = std::min(num_workers_, total_work); const size_t per_worker = total_work / num_tasks; const size_t remainder = total_work % num_tasks; + auto batch = std::make_shared(num_tasks); + current_batch() = batch; + { std::lock_guard lock(mutex); - pending_tasks.fetch_add(num_tasks, std::memory_order_relaxed); - for (size_t w = 0; w < num_tasks; ++w) { size_t start = w * per_worker + std::min(w, remainder); size_t end = start + per_worker + (w < remainder ? 1 : 0); - tasks.emplace_back([=]() { task_func(start, end); }); + tasks.emplace_back([=]() { task_func(start, end); finish_batch(batch); }); } } work_available.notify_all(); } void wait_all() { - std::unique_lock lock(mutex); - work_done.wait(lock, [this] { - return pending_tasks.load(std::memory_order_acquire) == 0; - }); + auto batch = current_batch(); + if (!batch) return; + current_batch().reset(); + std::unique_lock lock(batch->mutex); + batch->cv.wait(lock, [&] { return batch->remaining == 0; }); } template void enqueue_n_threads(size_t total_work, size_t num_threads, F task_func) { - if (total_work == 0 || num_threads == 0) return; + if (total_work == 0 || num_threads == 0) { current_batch().reset(); return; } num_threads = std::min(num_threads, std::min(num_workers_, total_work)); size_t num_tasks = num_threads; @@ -527,14 +542,15 @@ namespace CactusThreading { const size_t per_task = total_work / num_tasks; const size_t remainder = total_work % num_tasks; + auto batch = std::make_shared(num_tasks); + current_batch() = batch; + { std::lock_guard lock(mutex); - pending_tasks.fetch_add(num_tasks, std::memory_order_relaxed); - for (size_t t = 0; t < num_tasks; ++t) { size_t start = t * per_task + std::min(t, remainder); size_t end = start + per_task + (t < remainder ? 1 : 0); - tasks.emplace_back([=]() { task_func(start, end); }); + tasks.emplace_back([=]() { task_func(start, end); finish_batch(batch); }); } } work_available.notify_all(); diff --git a/cactus-kernels/src/wav.h b/cactus-kernels/src/wav.h index 576ef43c1..d42ec0f7a 100644 --- a/cactus-kernels/src/wav.h +++ b/cactus-kernels/src/wav.h @@ -20,6 +20,10 @@ inline AudioFP32 load_wav_fp32(const std::string& path) { if (!file) throw std::runtime_error("Could not open WAV file: " + path); + file.seekg(0, std::ios::end); + const std::streamoff file_size = file.tellg(); + file.seekg(0, std::ios::beg); + char riff[4]; file.read(riff, 4); if (std::string(riff, 4) != "RIFF") throw std::runtime_error("Not RIFF"); @@ -66,12 +70,25 @@ inline AudioFP32 load_wav_fp32(const std::string& path) { if (std::string(data_id, 4) == "data") break; + const std::streamoff after_header = file.tellg(); + if (data_size > (uint64_t)(file_size - after_header)) + throw std::runtime_error("Malformed WAV: chunk exceeds file"); file.seekg(data_size, std::ios::cur); } + const std::streamoff data_offset = file.tellg(); + if (data_size > (uint64_t)(file_size - data_offset)) + throw std::runtime_error("Malformed WAV: data chunk exceeds file"); + + const size_t bytes_per_frame = (size_t)num_channels * 2; + if (num_channels == 0 || data_size % bytes_per_frame != 0) + throw std::runtime_error("Malformed WAV: data size not frame-aligned"); + size_t num_samples = data_size / 2; std::vector raw(num_samples); file.read(reinterpret_cast(raw.data()), data_size); + if ((uint64_t)file.gcount() != data_size) + throw std::runtime_error("Malformed WAV: truncated data chunk"); std::vector tmp(num_samples); constexpr float scale = 1.0f / 32768.0f; @@ -85,7 +102,7 @@ inline AudioFP32 load_wav_fp32(const std::string& path) { mono = std::move(tmp); } else if (num_channels == 2) { mono.reserve(num_samples / 2); - for (size_t i = 0; i < num_samples; i += 2) + for (size_t i = 0; i + 1 < num_samples; i += 2) mono.push_back(0.5f * (tmp[i] + tmp[i + 1])); } else { throw std::runtime_error("Unsupported channel count");