diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 9609e26..0c35a9c 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -85,7 +85,7 @@ jobs: # assembles hand-written ARM asm. These images carry a complete sysroot, gcc and # clang, and their older glibc keeps the published binaries widely usable. # - # Both images are amd64 — the aarch64 one bundles a cross toolchain — so the + # Both images are amd64 (the aarch64 one bundles a cross toolchain), so the # host-installed node_modules stay compatible inside the container. - host: ubuntu-latest target: x86_64-unknown-linux-gnu @@ -138,9 +138,9 @@ jobs: run: pnpm install # The build downloads the matching prebuilt native SDK for the target, so these steps # need network access. - # The image ships Rust 1.82, which predates edition 2024 — required by this crate and - # by `aic-sdk` and `aic-sdk-sys` themselves, so lowering our own edition would not - # help. The toolchain is updated in the container before building. + # The image ships Rust 1.82, which predates edition 2024. That edition is required + # by this crate and by `aic-sdk` and `aic-sdk-sys` themselves, so lowering our own + # would not help. The toolchain is updated in the container before building. # # The workspace is mounted at its own path rather than somewhere like /build, so the # absolute paths in pnpm's node_modules symlinks still resolve. diff --git a/.gitignore b/.gitignore index 0546ba4..778b085 100644 --- a/.gitignore +++ b/.gitignore @@ -6,10 +6,10 @@ debug/ target/ # Cargo.lock is deliberately committed, contrary to the template default below. This crate -# is a cdylib published as prebuilt binaries — an artifact, not a library other crates -# compile against — so its builds should be reproducible. Leaving it out meant every CI run -# re-resolved the whole Rust graph from crates.io, so an upstream release could break the -# build with no change to this repository. +# is a cdylib published as prebuilt binaries, an artifact rather than a library other +# crates compile against, so its builds should be reproducible. Leaving it out meant every +# CI run re-resolved the whole Rust graph from crates.io, so an upstream release could +# break the build with no change to this repository. # # Remove Cargo.lock from gitignore if creating an executable, leave it for libraries # More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html diff --git a/CHANGELOG.md b/CHANGELOG.md index 8374ad6..ab86635 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,18 @@ the binding and its type declarations are now generated from annotated Rust. at once. - `Model.download` is asynchronous and resolves to the model path, so a cold download no longer blocks the event loop. +- **`dispose()` on every class holding native resources** (`Model`, `Processor`, + `ProcessorAsync`, `Vad`, `VadAsync`, `Analyzer`), destroying the native object + immediately instead of waiting for garbage collection. Every later method throws and a + repeat `dispose()` does nothing; where work can still be in flight (the async classes, + or an `Analyzer` mid-`analyzeAsync`) it blocks until that work finishes. + `Model.dispose()` unmaps the model file while objects created from it keep working. + `ProcessorContext` and `VadContext` handles stay valid after their object is disposed; + calls on them just no longer reach a live one. +- **Native footprints are reported to V8's garbage collector** (`napi_adjust_external_memory`). + These classes hold large native allocations behind small JavaScript objects; without the + report the collector felt no pressure to reclaim dropped instances, so native memory + grew unbounded. ### Changed diff --git a/README.md b/README.md index 565fee7..20d2e0d 100644 --- a/README.md +++ b/README.md @@ -258,6 +258,43 @@ If your license key is a JWT, refresh it in place instead of rebuilding the obje context.updateBearerToken(renewedJwt) ``` +## Memory management + +`Model`, `Processor`, `ProcessorAsync`, `Vad`, `VadAsync` and `Analyzer` hold large native +allocations behind small JavaScript objects. The binding reports each object's native +footprint to V8 (`napi_adjust_external_memory`), so the garbage collector applies the right +amount of pressure and reclaims dropped instances promptly instead of letting native memory +grow unbounded. + +For deterministic cleanup, every one of these classes also exposes `dispose()`, which +destroys the native object immediately instead of waiting for garbage collection: + +```javascript +const processor = new Processor(model, licenseKey) +try { + processor.initialize(sampleRate, blockSize) + processor.process(block) +} finally { + processor.dispose() +} +``` + +After `dispose()`, every method on the object throws; calling `dispose()` again does +nothing. On the async classes it blocks until in-flight work on the libuv pool finishes. + +Two things to know about cleanup timing: + +- Native cleanup runs on the event loop when the object is finalized, not synchronously at + garbage collection. Finalizers run on event-loop turns, so batches that create many of + these objects back to back hold native memory until the loop turns. An `await` on an + already-resolved promise (a microtask) is not enough; real async boundaries such as I/O, + `setTimeout` or `setImmediate` are. A tight synchronous loop that creates thousands of + objects accumulates their native memory for the duration of the loop; create these + objects per unit of work behind real async boundaries, or reuse a single instance. +- RSS reflects the peak of simultaneously live (or not-yet-finalized) instances: the + allocator reuses freed native memory rather than returning it to the OS, so a burst of N + concurrent instances costs about N x their footprint even after they are dropped. + ## Development Requires a recent Rust toolchain and Node 18+. diff --git a/__test__/dispose.spec.ts b/__test__/dispose.spec.ts new file mode 100644 index 0000000..9b4cc06 --- /dev/null +++ b/__test__/dispose.spec.ts @@ -0,0 +1,228 @@ +import test from 'ava' + +import { Analyzer, Model, Processor, ProcessorAsync, Vad, VadAsync } from '../index.js' +import { licenseKey, modelPath } from './common.js' + +const enhancementModel = () => Model.fromFile(modelPath('enhancement')) +const vadModel = () => Model.fromFile(modelPath('vad')) +const analysisModel = () => Model.fromFile(modelPath('analysis')) + +/** An initialized analyzer with a span of audio already buffered. */ +function bufferedAnalyzer() { + const model = analysisModel() + const sampleRate = model.getOptimalSampleRate() + const blockSize = model.getOptimalBlockSize(sampleRate) + + const analyzer = new Analyzer(model, licenseKey()) + analyzer.initialize(sampleRate, blockSize) + + const audio = Float32Array.from({ length: blockSize }, (_, i) => Math.sin(i / 10) * 0.5) + for (let block = 0; block < 100; block += 1) { + analyzer.buffer(audio) + } + + return { analyzer, audio, blockSize } +} + +/** Milliseconds since a `process.hrtime.bigint()` reading. */ +function elapsedMs(since: bigint): number { + return Number(process.hrtime.bigint() - since) / 1e6 +} + +/** Holds the calling thread for `ms`, leaving worker threads free to run. */ +function spin(ms: number) { + const until = process.hrtime.bigint() + BigInt(Math.round(ms * 1e6)) + while (process.hrtime.bigint() < until) { + // Deliberately busy: yielding here would hand the event loop back and a timer would + // overshoot, either of which lets the analysis finish before dispose() is called. + } +} + +/** An initialized processor at the model's optimal settings, plus that block size. */ +function initializedProcessor() { + const model = enhancementModel() + const sampleRate = model.getOptimalSampleRate() + const blockSize = model.getOptimalBlockSize(sampleRate) + + const processor = new Processor(model, licenseKey()) + processor.initialize(sampleRate, blockSize) + + return { processor, sampleRate, blockSize, model } +} + +test('sync processor dispose releases memory and rejects later use', (t) => { + const { processor, blockSize, model } = initializedProcessor() + processor.dispose() + + t.throws(() => processor.process(new Float32Array(blockSize)), { message: /disposed/ }) + t.throws(() => processor.initialize(16000, blockSize), { message: /disposed/ }) + t.throws(() => processor.getContext(), { message: /disposed/ }) + t.throws(() => processor.terminateSession(), { message: /disposed/ }) + + // Dispose is idempotent: a second call does nothing and must not throw. + processor.dispose() + + // The model stays usable after one of its processors is disposed. + const second = new Processor(model, licenseKey()) + second.dispose() + t.pass() +}) + +test('sync vad dispose releases memory and rejects later use', (t) => { + const model = vadModel() + const vad = new Vad(model, licenseKey()) + vad.initialize(model.getOptimalSampleRate(), model.getOptimalBlockSize(model.getOptimalSampleRate())) + const context = vad.getContext() + vad.dispose() + + t.throws(() => vad.process(new Float32Array(160)), { message: /disposed/ }) + t.throws(() => vad.getContext(), { message: /disposed/ }) + vad.dispose() + + // The SDK guarantees a VAD context stays valid after its VAD is destroyed + // (the prediction just stops updating), so using it must not crash. + context.isSpeechDetected() + t.pass() +}) + +test('model dispose rejects later use but keeps created processors working', (t) => { + const model = enhancementModel() + const { processor, blockSize } = initializedProcessorOn(model) + model.dispose() + + t.throws(() => model.getId(), { message: /disposed/ }) + t.throws(() => model.getOptimalSampleRate(), { message: /disposed/ }) + + // The SDK reference-counts the weights, so the processor keeps working. + processor.process(new Float32Array(blockSize)) + t.pass() + + function initializedProcessorOn(model: Model) { + const sampleRate = model.getOptimalSampleRate() + const blockSize = model.getOptimalBlockSize(sampleRate) + const processor = new Processor(model, licenseKey()) + processor.initialize(sampleRate, blockSize) + return { processor, blockSize } + } +}) + +test('async processor dispose rejects later use and is idempotent', async (t) => { + const model = enhancementModel() + const sampleRate = model.getOptimalSampleRate() + const blockSize = model.getOptimalBlockSize(sampleRate) + + const processor = new ProcessorAsync(model, licenseKey()) + await processor.initialize(sampleRate, blockSize) + processor.dispose() + + await t.throwsAsync(() => processor.process(new Float32Array(blockSize)), { + message: /disposed/, + }) + await t.throwsAsync(() => processor.terminateSession(), { message: /disposed/ }) + processor.dispose() + t.pass() +}) + +test('async vad dispose rejects later use', async (t) => { + const model = vadModel() + const sampleRate = model.getOptimalSampleRate() + const blockSize = model.getOptimalBlockSize(sampleRate) + + const vad = new VadAsync(model, licenseKey()) + await vad.initialize(sampleRate, blockSize) + vad.dispose() + + await t.throwsAsync(() => vad.process(new Float32Array(blockSize)), { message: /disposed/ }) + t.pass() +}) + +test('analyzer dispose rejects later use and is idempotent', async (t) => { + const model = analysisModel() + const sampleRate = model.getOptimalSampleRate() + const blockSize = model.getOptimalBlockSize(sampleRate) + + const analyzer = new Analyzer(model, licenseKey()) + analyzer.initialize(sampleRate, blockSize) + analyzer.buffer(new Float32Array(blockSize)) + analyzer.dispose() + + t.throws(() => analyzer.buffer(new Float32Array(blockSize)), { message: /disposed/ }) + t.throws(() => analyzer.analyze(), { message: /disposed/ }) + await t.throwsAsync(() => analyzer.analyzeAsync(), { message: /disposed/ }) + + // Dispose is idempotent: a second call does nothing and must not throw. + analyzer.dispose() + t.pass() +}) + +test('analyzer dispose racing an analyzeAsync settles it without crashing', async (t) => { + const { analyzer } = bufferedAnalyzer() + + // Disposing in the same tick leaves who reaches the lock first genuinely undecided, so + // the promise may resolve or reject. Neither outcome is asserted here; that the race is + // resolved safely at all is the point, and the blocking path gets its own test below. + const inFlight = analyzer.analyzeAsync().catch(() => {}) + analyzer.dispose() + await inFlight + + t.throws(() => analyzer.analyze(), { message: /disposed/ }) + t.pass() +}) + +test('analyzer dispose waits for an in-flight analyzeAsync to finish', async (t) => { + const { analyzer } = bufferedAnalyzer() + + // Time a warm analysis so the thresholds below scale to this machine rather than to a + // guessed constant. The first run pays any lazy setup, so the second is the estimate. + await analyzer.analyzeAsync() + const calibrationStarted = process.hrtime.bigint() + await analyzer.analyzeAsync() + const analysisMs = elapsedMs(calibrationStarted) + + // Hand the task to libuv and let the worker take the analyzer lock. The delay is spun + // rather than timed: a timer's granularity could overshoot the whole analysis, which + // would leave dispose() with nothing to wait for and quietly void the test. + const inFlight = analyzer.analyzeAsync() + await new Promise((resolve) => setImmediate(resolve)) + spin(Math.min(analysisMs / 8, 5)) + + const disposeStarted = process.hrtime.bigint() + analyzer.dispose() + const blockedMs = elapsedMs(disposeStarted) + + // dispose() waited for the lock instead of pulling the analyzer out from under the + // worker, so the analysis ran to completion and produced a real result. + const result = await inFlight + t.is(typeof result.riskScore, 'number', 'the in-flight analysis must complete, not be cancelled') + + // And it was still running when dispose() was called, so that wait was the lock being + // held rather than a no-op on an already-finished analysis. + t.true( + blockedMs >= analysisMs / 4, + `dispose() blocked ${blockedMs.toFixed(1)}ms of a ~${analysisMs.toFixed(1)}ms analysis`, + ) + + t.throws(() => analyzer.analyze(), { message: /disposed/ }) +}) + +test('a withConfig handle outlives its original handle being collected', async (t) => { + const model = enhancementModel() + const sampleRate = model.getOptimalSampleRate() + const blockSize = model.getOptimalBlockSize(sampleRate) + + // The chaining form leaves the constructor's handle as garbage. Its finalizer must + // give back only its own claim, not destroy the shared native processor. + const processor = await new ProcessorAsync(model, licenseKey()).withConfig(sampleRate, blockSize) + + // Encourage a GC so the dropped handle's finalizer runs. + const churn: unknown[] = [] + for (let i = 0; i < 10_000; i++) churn.push({ i }) + churn.length = 0 + if (typeof global.gc === 'function') global.gc() + await new Promise((resolve) => setImmediate(resolve)) + + const block = new Float32Array(blockSize) + await processor.process(block) + processor.dispose() + t.pass() +}) diff --git a/__test__/index.spec.ts b/__test__/index.spec.ts index dc0089e..b7a7e9d 100644 --- a/__test__/index.spec.ts +++ b/__test__/index.spec.ts @@ -36,7 +36,7 @@ test('sdk reports a version', (t) => { }) test('sdk expects the model version the fixtures are published under', (t) => { - // Guards against the fixture URLs in models.ts drifting from the addon's expectation, + // Guards against the fixture IDs in models.ts drifting from the addon's expectation, // which would otherwise surface as a confusing "model version unsupported" much later. t.is(getCompatibleModelVersion(), TEST_MODEL_VERSION) }) diff --git a/__test__/models.ts b/__test__/models.ts index e56038f..b5208dd 100644 --- a/__test__/models.ts +++ b/__test__/models.ts @@ -17,8 +17,8 @@ export interface TestModel { * `scripts/fetch-test-models.mjs` resolves these through `Model.download()`, which * re-fetches the manifest and pulls the newest compatible model version. The model file * format version is tied to the SDK version, so `getCompatibleModelVersion()` reports - * the version the built addon expects, and `sdk exposes compatible model version` - * asserts the fixtures still match it. + * the version the built addon expects, and the `sdk expects the model version the + * fixtures are published under` test asserts the fixtures still match it. */ export const TEST_MODELS = { enhancement: { id: 'quail-vf-2.2-s-16khz' }, diff --git a/index.d.ts b/index.d.ts index 9724270..442e299 100644 --- a/index.d.ts +++ b/index.d.ts @@ -20,6 +20,14 @@ export declare class Analyzer { /** Creates an analyzer from an analysis model. Other model types are rejected. */ constructor(model: Model, licenseKey: string) + /** + * Destroys the native collector and analyzer immediately, releasing their memory + * without waiting for garbage collection. + * + * Every later method throws; calling `dispose()` again does nothing. Blocks until an + * in-flight `analyzeAsync` on a worker thread finishes. + */ + dispose(): void /** * Configures the analyzer for an audio format. Must be called before buffering. * @@ -89,6 +97,15 @@ export declare class Model { * {@link Model.download}. */ static fromFile(path: string): Model + /** + * Unmaps the model file immediately, releasing its footprint without waiting for + * garbage collection. + * + * Objects already created from the model keep working: the SDK keeps the weights + * alive through internal reference counting. Every later method on this handle + * throws; calling `dispose()` again does nothing. + */ + dispose(): void /** * Downloads a model from the ai-coustics artifact CDN and resolves to its path. * @@ -134,6 +151,13 @@ export declare class Processor { * instance. */ constructor(model: Model, licenseKey: string, otelConfig?: OtelConfig | undefined | null) + /** + * Destroys the native processor immediately, releasing its memory and telemetry + * session without waiting for garbage collection. + * + * Every later method throws; calling `dispose()` again does nothing. + */ + dispose(): void /** * Configures the processor for an audio format. Must be called before processing. * @@ -198,6 +222,14 @@ export declare class ProcessorAsync { * instance. */ constructor(model: Model, licenseKey: string, otelConfig?: OtelConfig | undefined | null) + /** + * Destroys the native processor immediately, releasing its memory and telemetry + * session without waiting for garbage collection. + * + * Every later method throws; calling `dispose()` again does nothing. Blocks until + * in-flight work on the libuv pool finishes. + */ + dispose(): void /** * Initializes the processor and resolves to a handle onto it, for chaining off the * constructor: @@ -253,8 +285,11 @@ export declare class ProcessorAsync { /** * Control handle for a {@link Processor}. * - * Every method may be called while audio is being processed. Releasing the handle does - * not destroy the processor it came from. + * Every method may be called while audio is being processed. The handle and the processor + * have independent lifetimes in both directions: releasing the handle does not destroy + * the processor it came from, and the handle stays valid after its processor is disposed + * or garbage-collected. Calls on it keep succeeding; they just no longer reach a live + * processor. */ export declare class ProcessorContext { /** Sets an enhancement parameter. Throws if the value is out of range. */ @@ -305,6 +340,13 @@ export declare class Vad { * instance. */ constructor(model: Model, licenseKey: string, otelConfig?: OtelConfig | undefined | null) + /** + * Destroys the native VAD immediately, releasing its memory and telemetry session + * without waiting for garbage collection. + * + * Every later method throws; calling `dispose()` again does nothing. + */ + dispose(): void /** * Configures the VAD for an audio format. Must be called before processing. * @@ -359,6 +401,14 @@ export declare class VadAsync { * instance. */ constructor(model: Model, licenseKey: string, otelConfig?: OtelConfig | undefined | null) + /** + * Destroys the native VAD immediately, releasing its memory and telemetry session + * without waiting for garbage collection. + * + * Every later method throws; calling `dispose()` again does nothing. Blocks until + * in-flight work on the libuv pool finishes. + */ + dispose(): void /** * Initializes the VAD and resolves to a handle onto it, for chaining off the * constructor: diff --git a/src/analyzer.rs b/src/analyzer.rs index 4e01c16..132ae38 100644 --- a/src/analyzer.rs +++ b/src/analyzer.rs @@ -1,13 +1,15 @@ use crate::{ claim_sdk_id, + disposable_slot::{DisposableSlot, lock}, error::{Result, map_err}, + mem, model::Model, processor::audio_config, - processor_async::{Shared, lock}, }; use napi::{ Env, Task, + bindgen_prelude::ObjectFinalize, bindgen_prelude::{AsyncTask, Float32Array}, }; use napi_derive::napi; @@ -66,7 +68,7 @@ impl From for AnalysisResult { /// as one object here, but the split still shows through: {@link Analyzer#buffer} drives /// the collector on the calling thread, while {@link Analyzer#analyzeAsync} moves the /// analyzer half onto a worker. The SDK guarantees the two are safe to use concurrently. -#[napi] +#[napi(custom_finalize)] pub struct Analyzer { // Neither half borrows the other, nor the model: `'static` here is the model weights' // lifetime, which `Model.fromFile` satisfies by memory-mapping the file. @@ -74,24 +76,67 @@ pub struct Analyzer { // Only the analyzer half is shared. The collector is owned outright, so `buffer`, the // one call on the audio path, takes no lock and cannot contend with an analysis running // on a worker thread. - collector: aic_sdk::Collector, - analyzer: Shared>, + collector: DisposableSlot, + analyzer: Arc>>>, +} + +impl ObjectFinalize for Analyzer { + fn finalize(mut self, env: Env) -> Result<()> { + // Each half is released on its own, and each release is a no-op when `dispose()` + // got there first. + // + // An in-flight `AnalyzeTask` holds the analyzer `Arc`, so that half is dropped + // only after the worker finishes, and its footprint is left for whoever holds the + // last handle. The collector drops here, possibly while the worker analyzes. That + // is safe per the C API, which destroys the paired halves independently, in any + // order (`aic_collector_destroy`). + self.collector.release(env); + if Arc::strong_count(&self.analyzer) == 1 { + lock(&self.analyzer).release(env); + } + Ok(()) + } } #[napi] impl Analyzer { /// Creates an analyzer from an analysis model. Other model types are rejected. #[napi(constructor)] - pub fn new(model: &Model, license_key: String) -> Result { + pub fn new(env: Env, model: &Model, license_key: String) -> Result { + let model_inner = model.inner()?; claim_sdk_id(); - let (collector, analyzer) = map_err(aic_sdk::analyzer_pair(&model.inner, &license_key))?; + let (collector, analyzer) = map_err(aic_sdk::analyzer_pair(model_inner, &license_key))?; Ok(Self { - collector, - analyzer: Arc::new(Mutex::new(analyzer)), + collector: DisposableSlot::new(env, collector, "Analyzer", mem::COLLECTOR_BYTES), + analyzer: Arc::new(Mutex::new(DisposableSlot::new( + env, + analyzer, + "Analyzer", + mem::ANALYZER_BYTES, + ))), }) } + /// Destroys the native collector and analyzer immediately, releasing their memory + /// without waiting for garbage collection. + /// + /// Every later method throws; calling `dispose()` again does nothing. Blocks until an + /// in-flight `analyzeAsync` on a worker thread finishes. + #[napi] + pub fn dispose(&mut self, env: Env) { + // The collector is released before the analyzer lock is taken, so it can be + // destroyed while an `analyzeAsync` is in flight on a worker. That is safe per the + // C API (`aic_collector_destroy`): the paired halves are destroyed independently, in + // any order, and the collector handle itself is only ever used on this thread. The + // analyzer half is destroyed under the lock, which is what blocks until the + // in-flight analysis finishes. + // + // Both releases are idempotent, so a second `dispose()` does nothing. + self.collector.release(env); + lock(&self.analyzer).release(env); + } + /// Configures the analyzer for an audio format. Must be called before buffering. /// /// The model's optimal sample rate and block size avoid internal resampling and @@ -103,17 +148,17 @@ impl Analyzer { block_size: u32, variable_block_size: Option, ) -> Result<()> { - map_err( - self - .collector - .initialize(&audio_config(sample_rate, block_size, variable_block_size)), - ) + map_err(self.collector.get_mut()?.initialize(&audio_config( + sample_rate, + block_size, + variable_block_size, + ))) } /// Buffers a mono audio block for later analysis, leaving the audio unmodified. #[napi] pub fn buffer(&mut self, audio: Float32Array) -> Result<()> { - map_err(self.collector.buffer(&audio)) + map_err(self.collector.get_mut()?.buffer(&audio)) } /// Runs the analysis model over the buffered audio, on the calling thread. @@ -127,7 +172,7 @@ impl Analyzer { /// nothing else is waiting on the event loop. #[napi] pub fn analyze(&self) -> Result { - map_err(lock(&self.analyzer).analyze_buffered()).map(AnalysisResult::from) + map_err(lock(&self.analyzer).get_mut()?.analyze_buffered()).map(AnalysisResult::from) } /// Runs the analysis model over the buffered audio on a worker thread. @@ -153,7 +198,7 @@ impl Analyzer { /// Clears buffered audio and internal state, keeping the configured audio settings. #[napi] pub fn reset(&self) -> Result<()> { - map_err(lock(&self.analyzer).reset()) + map_err(lock(&self.analyzer).get_mut()?.reset()) } /// Swaps in a renewed JWT without tearing down the analyzer. @@ -162,13 +207,13 @@ impl Analyzer { /// call is a no-op and the previous token stays active. #[napi] pub fn update_bearer_token(&self, token: String) -> Result<()> { - map_err(lock(&self.analyzer).update_bearer_token(&token)) + map_err(lock(&self.analyzer).get_mut()?.update_bearer_token(&token)) } /// Ends this analyzer's telemetry session, after which it can no longer analyze audio. #[napi] pub fn terminate_session(&self) -> Result<()> { - map_err(lock(&self.analyzer).terminate_session()) + map_err(lock(&self.analyzer).get_mut()?.terminate_session()) } } @@ -177,7 +222,7 @@ impl Analyzer { /// Holds only the analyzer half, so the collector stays on the JS thread where `buffer` can /// keep reaching it while this runs. pub struct AnalyzeTask { - analyzer: Shared>, + analyzer: Arc>>>, } impl Task for AnalyzeTask { @@ -185,7 +230,7 @@ impl Task for AnalyzeTask { type JsValue = AnalysisResult; fn compute(&mut self) -> Result { - map_err(lock(&self.analyzer).analyze_buffered()) + map_err(lock(&self.analyzer).get_mut()?.analyze_buffered()) } fn resolve(&mut self, _env: Env, result: aic_sdk::AnalysisResult) -> Result { diff --git a/src/disposable_slot.rs b/src/disposable_slot.rs new file mode 100644 index 0000000..b98853c --- /dev/null +++ b/src/disposable_slot.rs @@ -0,0 +1,88 @@ +//! One native SDK object, destroyed exactly once, with its footprint reported to V8 for +//! as long as it lives. +//! +//! Every binding class holds its SDK object in a slot, so the disposed error and the +//! footprint accounting are written once here rather than per class. The slot itself is +//! a plain owner with no interior mutability; the classes that share their object with +//! tasks on the libuv pool wrap it in an `Arc>` and reach it through [`lock`]. + +use std::sync::{Mutex, MutexGuard}; + +use napi::Env; + +use crate::{ + error::{Result, disposed_error}, + mem::adjust, +}; + +/// A slot holding a native SDK object until it is disposed, after which every access +/// fails with the disposed error. +/// +/// The slot owns both halves of the object's footprint report: [`new`](Self::new) reports +/// it to V8 and [`release`](Self::release) gives it back. `Option::take` makes the +/// give-back idempotent, so `dispose()` and a class finalizer cannot double-count it. +/// +/// A slot dropped without `release` having run still destroys the object, but those +/// bytes stay reported: giving them back takes an `Env`, which a `Drop` impl does not +/// have. That happens when the last JS handle onto a shared object is finalized while a +/// task still holds a clone; over-reporting only makes V8 a little more eager for the +/// rest of the process. +pub(crate) struct DisposableSlot { + inner: Option, + /// The JS class name, for the disposed error message. + class: &'static str, + /// The footprint reported to V8 while `inner` is live. + bytes: i64, +} + +impl DisposableSlot { + /// Takes ownership of `inner` and reports its `bytes` of native footprint to V8. + pub(crate) fn new(env: Env, inner: T, class: &'static str, bytes: i64) -> Self { + adjust(env, bytes); + + Self { + inner: Some(inner), + class, + bytes, + } + } + + /// The native object, or the disposed error once it is gone. + pub(crate) fn get(&self) -> Result<&T> { + self + .inner + .as_ref() + .ok_or_else(|| disposed_error(self.class)) + } + + /// The same, for a `&mut` call. + pub(crate) fn get_mut(&mut self) -> Result<&mut T> { + self + .inner + .as_mut() + .ok_or_else(|| disposed_error(self.class)) + } + + /// Destroys the native object, if it is still live, and gives its footprint back to + /// V8. Idempotent. + /// + /// Called by `dispose()` and by the class finalizer, in whichever order they happen. + pub(crate) fn release(&mut self, env: Env) { + if self.inner.take().is_some() { + adjust(env, -self.bytes); + } + } +} + +/// Locks a shared slot, recovering the guard if the lock is poisoned. +/// +/// A `Mutex` rather than an async lock: `compute` runs on a libuv worker, where blocking +/// is exactly what that thread is for. +/// +/// Poisoning would mean an earlier call panicked while holding the guard, which the SDK +/// does not do. Recovering keeps one hypothetical failure from turning every later call +/// into a panic, disposal included: a panic would happen inside an SDK call, leaving the +/// slot's own `Option` intact. +pub(crate) fn lock(slot: &Mutex) -> MutexGuard<'_, T> { + slot.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) +} diff --git a/src/error.rs b/src/error.rs index 2c6f409..a2d3050 100644 --- a/src/error.rs +++ b/src/error.rs @@ -32,3 +32,11 @@ pub type Result = napi::Result; pub fn map_err(result: std::result::Result) -> Result { result.map_err(|error| JsAicError(error).into()) } + +/// The error thrown by any method called after `dispose()`. +pub fn disposed_error(class: &str) -> napi::Error { + napi::Error::new( + napi::Status::GenericFailure, + format!("{class} has been disposed"), + ) +} diff --git a/src/lib.rs b/src/lib.rs index 8c43479..1e32358 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,7 +8,9 @@ use napi_derive::napi; mod analyzer; +mod disposable_slot; mod error; +mod mem; mod model; mod processor; mod processor_async; diff --git a/src/mem.rs b/src/mem.rs new file mode 100644 index 0000000..dc13c27 --- /dev/null +++ b/src/mem.rs @@ -0,0 +1,79 @@ +//! Reports the native footprint of SDK objects to V8's garbage collector. +//! +//! Each binding class is a small JS object of a few dozen bytes in front of a much larger +//! native allocation, from ~200 KiB for a processor up to the size of the model weights. +//! V8's heuristics only see the JS side: `heapUsed` and `external` barely move, so the +//! collector feels no pressure to reclaim dropped instances, and a workload that creates +//! processors per unit of work ratchets RSS up until the process is OOM-killed. +//! +//! `Env::adjust_external_memory` (`napi_adjust_external_memory`) is the Node-API mechanism +//! for reporting that hidden cost. Both halves of the ledger live in +//! [`DisposableSlot`](crate::disposable_slot::DisposableSlot): it reports its object's +//! footprint when constructed and reports the negation when released, by `dispose()` or by +//! the class finalizer, whichever gets there first. Either way the ledger balances. +//! +//! Footprints are per-class constants because the SDK exposes no per-instance memory +//! query. They are estimates keyed to measurement and deliberately err high: over-reporting +//! only makes V8 collect a little more eagerly, while under-reporting leaves the growth +//! above unchecked. + +use std::path::Path; + +use napi::Env; + +const MIB: i64 = 1024 * 1024; +const KIB: i64 = 1024; + +/// A `Processor` or `Vad` instance. Measured as the RSS delta per instance, at initialize +/// plus one `process` call: +/// +/// - `quail-vf-2.2-s` (5 MiB model): ~190 KiB +/// - `quail-vf-2.2-l` (20 MiB model): ~462 KiB +/// - `vad-2.1-xxs` (0.6 MiB model): ~197 KiB +/// +/// The workspace barely scales with the weights (those stay file-backed under `Model`), +/// so 512 KiB covers the largest measured enhancement model with headroom while staying +/// within ~3x of the smallest. +pub(crate) const PROCESSOR_BYTES: i64 = 512 * KIB; + +/// The analyzer half of an `Analyzer`, which holds the model workspace. Measured ~8.2 MiB +/// for `tyto-1.1-l` at construction, so 14 MiB leaves ~1.7x headroom for larger analysis +/// models. +/// +/// Reported separately from [`COLLECTOR_BYTES`] because the two halves are destroyed +/// independently: the collector can go while a worker thread still analyzes, so a single +/// report for the pair would be given back too early. +pub(crate) const ANALYZER_BYTES: i64 = 14 * MIB; + +/// The collector half of an `Analyzer`, which holds the buffered audio. Measured as the +/// ~0.7 MiB that `tyto-1.1-l` grows by once the collector is initialized and holding its +/// 5 s span, so 2 MiB leaves ~3x headroom. +pub(crate) const COLLECTOR_BYTES: i64 = 2 * MIB; + +/// Fallback footprint for a `Model` when its file cannot be stat'd. Deliberately +/// conservative: the loaded model is memory-mapped, so its resident share approaches the +/// file size as pages are touched. +const MODEL_FALLBACK_BYTES: i64 = 64 * MIB; + +/// Tells V8 that `delta_bytes` of external (native) memory changed. +/// +/// Positive when an object is created, negative (the same value) when it is finalized. +/// The result is a GC hint only: a failed adjustment is never worth failing the API call +/// over, so errors are ignored. +pub(crate) fn adjust(env: Env, delta_bytes: i64) { + if delta_bytes == 0 { + return; + } + + // A non-Ok status only means the hint was not applied; correctness does not depend on + // it, and there is nothing useful to do about a failed hint anyway. + let _ = env.adjust_external_memory(delta_bytes); +} + +/// The resident footprint to report for a model loaded from `path`: the file size, since +/// the weights are memory-mapped, falling back to a conservative estimate. +pub(crate) fn model_bytes(path: &Path) -> i64 { + std::fs::metadata(path) + .map(|meta| meta.len() as i64) + .unwrap_or(MODEL_FALLBACK_BYTES) +} diff --git a/src/model.rs b/src/model.rs index 7cafddd..10a262b 100644 --- a/src/model.rs +++ b/src/model.rs @@ -1,6 +1,10 @@ -use crate::error::{JsAicError, Result, map_err}; +use crate::{ + disposable_slot::DisposableSlot, + error::{JsAicError, Result, map_err}, + mem, +}; -use napi::{Env, Task, bindgen_prelude::AsyncTask}; +use napi::{Env, Task, bindgen_prelude::AsyncTask, bindgen_prelude::ObjectFinalize}; use napi_derive::napi; /// A loaded ai-coustics model. @@ -8,11 +12,29 @@ use napi_derive::napi; /// One model can back multiple processors, VADs and analyzers, according to its type. /// The underlying model data is kept alive by every object created from it through /// internal reference counting, so this handle may be released first. -#[napi] +#[napi(custom_finalize)] pub struct Model { // `from_file` memory-maps the file rather than borrowing a caller-owned buffer, so // the SDK model is `'static` and needs no lifetime plumbing here. - pub(crate) inner: aic_sdk::Model<'static>, + // + // The slot's footprint is this instance's alone: the mmap'd weights. It is per-instance + // rather than a per-class constant, since it is the model file's size. + slot: DisposableSlot>, +} + +impl ObjectFinalize for Model { + fn finalize(mut self, env: Env) -> Result<()> { + // A no-op when `dispose()` already gave the footprint back. + self.slot.release(env); + Ok(()) + } +} + +impl Model { + /// The inner SDK model, or the disposed error once `dispose()` ran. + pub(crate) fn inner(&self) -> Result<&aic_sdk::Model<'static>> { + self.slot.get() + } } #[napi] @@ -26,12 +48,26 @@ impl Model { /// Browse available models at , or fetch one with /// {@link Model.download}. #[napi(factory)] - pub fn from_file(path: String) -> Result { + pub fn from_file(env: Env, path: String) -> Result { + let inner = map_err(aic_sdk::Model::from_file(&path))?; + let bytes = mem::model_bytes(std::path::Path::new(&path)); + Ok(Self { - inner: map_err(aic_sdk::Model::from_file(path))?, + slot: DisposableSlot::new(env, inner, "Model", bytes), }) } + /// Unmaps the model file immediately, releasing its footprint without waiting for + /// garbage collection. + /// + /// Objects already created from the model keep working: the SDK keeps the weights + /// alive through internal reference counting. Every later method on this handle + /// throws; calling `dispose()` again does nothing. + #[napi] + pub fn dispose(&mut self, env: Env) { + self.slot.release(env); + } + /// Downloads a model from the ai-coustics artifact CDN and resolves to its path. /// /// The manifest is re-fetched on every call so the newest compatible model version @@ -51,8 +87,8 @@ impl Model { /// The model identifier, e.g. `quail-vf-2.2-s-16khz`. #[napi] - pub fn get_id(&self) -> String { - self.inner.id().to_owned() + pub fn get_id(&self) -> Result { + Ok(self.inner()?.id().to_owned()) } /// The sample rate in Hz the model was trained for. @@ -60,8 +96,8 @@ impl Model { /// Audio at any rate can be processed, but a model only enhances frequencies up to /// its own Nyquist limit, so matching this rate gives the best quality. #[napi] - pub fn get_optimal_sample_rate(&self) -> u32 { - self.inner.optimal_sample_rate() + pub fn get_optimal_sample_rate(&self) -> Result { + Ok(self.inner()?.optimal_sample_rate()) } /// The block size that avoids internal buffering at `sampleRate`. @@ -70,12 +106,12 @@ impl Model { /// The value changes with the sample rate, because the model works on a fixed time /// window: a 10 ms window is 480 samples at 48 kHz but 160 at 16 kHz. #[napi] - pub fn get_optimal_block_size(&self, sample_rate: u32) -> u32 { + pub fn get_optimal_block_size(&self, sample_rate: u32) -> Result { // The SDK reports sizes as `usize`, which napi would marshal as a JS BigInt. // A BigInt block size would throw on `new Float32Array(n)` and on arithmetic // against plain numbers, so it crosses the boundary as u32. Block sizes are a // few thousand samples at most. - self.inner.optimal_block_size(sample_rate) as u32 + Ok(self.inner()?.optimal_block_size(sample_rate) as u32) } } diff --git a/src/processor.rs b/src/processor.rs index 9c7c487..cbf92b2 100644 --- a/src/processor.rs +++ b/src/processor.rs @@ -1,10 +1,12 @@ use crate::{ claim_sdk_id, + disposable_slot::DisposableSlot, error::{Result, map_err}, + mem, model::Model, }; -use napi::bindgen_prelude::Float32Array; +use napi::{Env, bindgen_prelude::Float32Array, bindgen_prelude::ObjectFinalize}; use napi_derive::napi; /// Enhancement parameters, all changeable while audio is being processed. @@ -78,9 +80,19 @@ pub(crate) fn audio_config( /// and {@link Analyzer} for analysis models; passing the wrong kind throws. /// /// Create several processors to handle multiple streams or to switch models at runtime. -#[napi] +#[napi(custom_finalize)] pub struct Processor { - inner: aic_sdk::Processor<'static>, + // Owned outright, with no lock: every method here runs on the JS thread. The async + // class shares the same slot with its tasks instead. + slot: DisposableSlot>, +} + +impl ObjectFinalize for Processor { + fn finalize(mut self, env: Env) -> Result<()> { + // A no-op when `dispose()` already gave the footprint back. + self.slot.release(env); + Ok(()) + } } #[napi] @@ -90,19 +102,36 @@ impl Processor { /// Telemetry follows the runtime environment; pass `otelConfig` to override it for this /// instance. #[napi(constructor)] - pub fn new(model: &Model, license_key: String, otel_config: Option) -> Result { + pub fn new( + env: Env, + model: &Model, + license_key: String, + otel_config: Option, + ) -> Result { + let model_inner = model.inner()?; claim_sdk_id(); let inner = match otel_config { Some(config) => { - aic_sdk::Processor::with_otel_config(&model.inner, &license_key, &config.into()) + aic_sdk::Processor::with_otel_config(model_inner, &license_key, &config.into()) } - None => aic_sdk::Processor::new(&model.inner, &license_key), + None => aic_sdk::Processor::new(model_inner, &license_key), }; + let inner = map_err(inner)?; + Ok(Self { - inner: map_err(inner)?, + slot: DisposableSlot::new(env, inner, "Processor", mem::PROCESSOR_BYTES), }) } + /// Destroys the native processor immediately, releasing its memory and telemetry + /// session without waiting for garbage collection. + /// + /// Every later method throws; calling `dispose()` again does nothing. + #[napi] + pub fn dispose(&mut self, env: Env) { + self.slot.release(env); + } + /// Configures the processor for an audio format. Must be called before processing. /// /// For the lowest delay use {@link Model#getOptimalSampleRate} and @@ -117,11 +146,11 @@ impl Processor { block_size: u32, variable_block_size: Option, ) -> Result<()> { - map_err( - self - .inner - .initialize(&audio_config(sample_rate, block_size, variable_block_size)), - ) + map_err(self.slot.get_mut()?.initialize(&audio_config( + sample_rate, + block_size, + variable_block_size, + ))) } /// Enhances a mono audio block in place. @@ -140,17 +169,17 @@ impl Processor { // break that assumption, which is inherent to processing JS-owned buffers in place. let samples = unsafe { audio.as_mut() }; - map_err(self.inner.process(samples)) + map_err(self.slot.get_mut()?.process(samples)) } /// Creates a handle for reading and writing this processor's parameters and state. /// /// Each call returns an independent handle onto the same processor. #[napi] - pub fn get_context(&self) -> ProcessorContext { - ProcessorContext { - inner: self.inner.context(), - } + pub fn get_context(&self) -> Result { + Ok(ProcessorContext { + inner: self.slot.get()?.context(), + }) } /// Ends this processor's telemetry session, after which it can no longer process audio. @@ -159,14 +188,17 @@ impl Processor { /// is collected, but GC timing is not guaranteed. May block, so keep it off the audio path. #[napi] pub fn terminate_session(&mut self) -> Result<()> { - map_err(self.inner.terminate_session()) + map_err(self.slot.get_mut()?.terminate_session()) } } /// Control handle for a {@link Processor}. /// -/// Every method may be called while audio is being processed. Releasing the handle does -/// not destroy the processor it came from. +/// Every method may be called while audio is being processed. The handle and the processor +/// have independent lifetimes in both directions: releasing the handle does not destroy +/// the processor it came from, and the handle stays valid after its processor is disposed +/// or garbage-collected. Calls on it keep succeeding; they just no longer reach a live +/// processor. #[napi] pub struct ProcessorContext { pub(crate) inner: aic_sdk::ProcessorContext, diff --git a/src/processor_async.rs b/src/processor_async.rs index 7c82d10..064e2fa 100644 --- a/src/processor_async.rs +++ b/src/processor_async.rs @@ -1,32 +1,18 @@ use crate::{ claim_sdk_id, + disposable_slot::{DisposableSlot, lock}, error::{Result, map_err}, + mem, model::Model, processor::{OtelConfig, ProcessorContext, audio_config}, }; use napi::{ Env, Task, - bindgen_prelude::{AsyncTask, Float32Array}, + bindgen_prelude::{AsyncTask, Float32Array, ObjectFinalize}, }; use napi_derive::napi; -use std::sync::{Arc, Mutex, MutexGuard}; - -/// The SDK object shared between a binding class and the tasks it spawns. -/// -/// A `Mutex` rather than an async lock: `compute` runs on a libuv worker, where blocking -/// is exactly what that thread is for. -pub(crate) type Shared = Arc>; - -/// Locks a shared SDK object, recovering the guard if the lock is poisoned. -/// -/// Poisoning would mean an earlier call panicked mid-process, which the SDK does not do. -/// Recovering keeps one hypothetical failure from turning every later call into a panic. -pub(crate) fn lock(shared: &Mutex) -> MutexGuard<'_, T> { - shared - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) -} +use std::sync::{Arc, Mutex}; /// Speech enhancement processor that keeps its work off the main thread. /// @@ -47,9 +33,21 @@ pub(crate) fn lock(shared: &Mutex) -> MutexGuard<'_, T> { /// Raise `UV_THREADPOOL_SIZE` before Node starts to run more streams in parallel. The /// SDK's own `AIC_NUM_THREADS` does not apply here: that variable sizes a rayon pool this /// binding deliberately does not use. -#[napi] +#[napi(custom_finalize)] pub struct ProcessorAsync { - inner: Shared>, + slot: Arc>>>, +} + +impl ObjectFinalize for ProcessorAsync { + fn finalize(self, env: Env) -> Result<()> { + // Only the last handle onto the native object destroys it; with other handles or + // in-flight tasks holding an `Arc`, this leaves the object (and its footprint + // report) for them. Idempotent against `dispose()`. + if Arc::strong_count(&self.slot) == 1 { + lock(&self.slot).release(env); + } + Ok(()) + } } #[napi] @@ -62,19 +60,42 @@ impl ProcessorAsync { /// Telemetry follows the runtime environment; pass `otelConfig` to override it for this /// instance. #[napi(constructor)] - pub fn new(model: &Model, license_key: String, otel_config: Option) -> Result { + pub fn new( + env: Env, + model: &Model, + license_key: String, + otel_config: Option, + ) -> Result { + let model_inner = model.inner()?; claim_sdk_id(); let inner = match otel_config { Some(config) => { - aic_sdk::Processor::with_otel_config(&model.inner, &license_key, &config.into()) + aic_sdk::Processor::with_otel_config(model_inner, &license_key, &config.into()) } - None => aic_sdk::Processor::new(&model.inner, &license_key), + None => aic_sdk::Processor::new(model_inner, &license_key), }; + let inner = map_err(inner)?; + Ok(Self { - inner: Arc::new(Mutex::new(map_err(inner)?)), + slot: Arc::new(Mutex::new(DisposableSlot::new( + env, + inner, + "ProcessorAsync", + mem::PROCESSOR_BYTES, + ))), }) } + /// Destroys the native processor immediately, releasing its memory and telemetry + /// session without waiting for garbage collection. + /// + /// Every later method throws; calling `dispose()` again does nothing. Blocks until + /// in-flight work on the libuv pool finishes. + #[napi] + pub fn dispose(&self, env: Env) { + lock(&self.slot).release(env); + } + /// Initializes the processor and resolves to a handle onto it, for chaining off the /// constructor: /// @@ -92,7 +113,7 @@ impl ProcessorAsync { variable_block_size: Option, ) -> AsyncTask { AsyncTask::new(ProcessorWithConfigTask { - inner: self.inner.clone(), + slot: self.slot.clone(), config: audio_config(sample_rate, block_size, variable_block_size), }) } @@ -108,7 +129,7 @@ impl ProcessorAsync { variable_block_size: Option, ) -> AsyncTask { AsyncTask::new(ProcessorInitializeTask { - inner: self.inner.clone(), + slot: self.slot.clone(), config: audio_config(sample_rate, block_size, variable_block_size), }) } @@ -134,7 +155,7 @@ impl ProcessorAsync { #[napi(ts_return_type = "Promise>")] pub fn process(&self, audio: Float32Array) -> AsyncTask { AsyncTask::new(ProcessorProcessTask { - inner: self.inner.clone(), + slot: self.slot.clone(), // Copied on the JS thread so the worker owns its samples outright. A block is a // couple of kilobytes, far below the cost of running the model over it, and it // removes any chance of JS mutating the buffer mid-process. @@ -151,7 +172,7 @@ impl ProcessorAsync { #[napi(ts_return_type = "Promise")] pub fn get_context(&self) -> AsyncTask { AsyncTask::new(ProcessorContextTask { - inner: self.inner.clone(), + slot: self.slot.clone(), }) } @@ -161,14 +182,14 @@ impl ProcessorAsync { #[napi(ts_return_type = "Promise")] pub fn terminate_session(&self) -> AsyncTask { AsyncTask::new(ProcessorTerminateTask { - inner: self.inner.clone(), + slot: self.slot.clone(), }) } } /// Backs {@link ProcessorAsync#withConfig}. pub struct ProcessorWithConfigTask { - inner: Shared>, + slot: Arc>>>, config: aic_sdk::ProcessorConfig, } @@ -177,19 +198,22 @@ impl Task for ProcessorWithConfigTask { type JsValue = ProcessorAsync; fn compute(&mut self) -> Result<()> { - map_err(lock(&self.inner).initialize(&self.config)) + map_err(lock(&self.slot).get_mut()?.initialize(&self.config)) } fn resolve(&mut self, _env: Env, _: ()) -> Result { + // A second JS handle onto the same native processor. The footprint is reported once + // per object at construction, so there is nothing to report here; the last handle's + // finalizer gives it back. Born disposed when the processor was disposed mid-flight. Ok(ProcessorAsync { - inner: self.inner.clone(), + slot: self.slot.clone(), }) } } /// Backs {@link ProcessorAsync#initialize}. pub struct ProcessorInitializeTask { - inner: Shared>, + slot: Arc>>>, config: aic_sdk::ProcessorConfig, } @@ -198,7 +222,7 @@ impl Task for ProcessorInitializeTask { type JsValue = (); fn compute(&mut self) -> Result<()> { - map_err(lock(&self.inner).initialize(&self.config)) + map_err(lock(&self.slot).get_mut()?.initialize(&self.config)) } fn resolve(&mut self, _env: Env, _: ()) -> Result<()> { @@ -208,7 +232,7 @@ impl Task for ProcessorInitializeTask { /// Backs {@link ProcessorAsync#process}. pub struct ProcessorProcessTask { - inner: Shared>, + slot: Arc>>>, audio: Vec, } @@ -220,7 +244,7 @@ impl Task for ProcessorProcessTask { // Moved out rather than borrowed so the buffer can be handed to V8 in `resolve` // without another copy. The task is used once, so leaving an empty Vec behind is fine. let mut audio = std::mem::take(&mut self.audio); - map_err(lock(&self.inner).process(&mut audio))?; + map_err(lock(&self.slot).get_mut()?.process(&mut audio))?; Ok(audio) } @@ -234,7 +258,7 @@ impl Task for ProcessorProcessTask { /// Backs {@link ProcessorAsync#getContext}. pub struct ProcessorContextTask { - inner: Shared>, + slot: Arc>>>, } impl Task for ProcessorContextTask { @@ -242,7 +266,7 @@ impl Task for ProcessorContextTask { type JsValue = ProcessorContext; fn compute(&mut self) -> Result { - Ok(lock(&self.inner).context()) + Ok(lock(&self.slot).get()?.context()) } fn resolve(&mut self, _env: Env, context: aic_sdk::ProcessorContext) -> Result { @@ -252,7 +276,7 @@ impl Task for ProcessorContextTask { /// Backs {@link ProcessorAsync#terminateSession}. pub struct ProcessorTerminateTask { - inner: Shared>, + slot: Arc>>>, } impl Task for ProcessorTerminateTask { @@ -260,7 +284,7 @@ impl Task for ProcessorTerminateTask { type JsValue = (); fn compute(&mut self) -> Result<()> { - map_err(lock(&self.inner).terminate_session()) + map_err(lock(&self.slot).get_mut()?.terminate_session()) } fn resolve(&mut self, _env: Env, _: ()) -> Result<()> { diff --git a/src/vad.rs b/src/vad.rs index 7b29d5f..cb3e01f 100644 --- a/src/vad.rs +++ b/src/vad.rs @@ -1,11 +1,13 @@ use crate::{ claim_sdk_id, + disposable_slot::DisposableSlot, error::{Result, map_err}, + mem, model::Model, processor::{OtelConfig, audio_config}, }; -use napi::bindgen_prelude::Float32Array; +use napi::{Env, bindgen_prelude::Float32Array, bindgen_prelude::ObjectFinalize}; use napi_derive::napi; /// Voice activity detection parameters, all changeable while audio is being processed. @@ -53,9 +55,19 @@ impl From for aic_sdk::VadParameter { /// processor's output: enhancement changes the signal the VAD model expects, and stacks /// the processor's delay onto the prediction. Since `process` leaves its input untouched, /// calling it on the same block before `Processor#process` is enough. -#[napi] +#[napi(custom_finalize)] pub struct Vad { - inner: aic_sdk::Vad<'static>, + // Owned outright, with no lock: every method here runs on the JS thread. The async + // class shares the same slot with its tasks instead. + slot: DisposableSlot>, +} + +impl ObjectFinalize for Vad { + fn finalize(mut self, env: Env) -> Result<()> { + // A no-op when `dispose()` already gave the footprint back. + self.slot.release(env); + Ok(()) + } } #[napi] @@ -65,17 +77,34 @@ impl Vad { /// Telemetry follows the runtime environment; pass `otelConfig` to override it for this /// instance. #[napi(constructor)] - pub fn new(model: &Model, license_key: String, otel_config: Option) -> Result { + pub fn new( + env: Env, + model: &Model, + license_key: String, + otel_config: Option, + ) -> Result { + let model_inner = model.inner()?; claim_sdk_id(); let inner = match otel_config { - Some(config) => aic_sdk::Vad::with_otel_config(&model.inner, &license_key, &config.into()), - None => aic_sdk::Vad::new(&model.inner, &license_key), + Some(config) => aic_sdk::Vad::with_otel_config(model_inner, &license_key, &config.into()), + None => aic_sdk::Vad::new(model_inner, &license_key), }; + let inner = map_err(inner)?; + Ok(Self { - inner: map_err(inner)?, + slot: DisposableSlot::new(env, inner, "Vad", mem::PROCESSOR_BYTES), }) } + /// Destroys the native VAD immediately, releasing its memory and telemetry session + /// without waiting for garbage collection. + /// + /// Every later method throws; calling `dispose()` again does nothing. + #[napi] + pub fn dispose(&mut self, env: Env) { + self.slot.release(env); + } + /// Configures the VAD for an audio format. Must be called before processing. /// /// The model's optimal sample rate and block size give the most frequent prediction @@ -87,11 +116,11 @@ impl Vad { block_size: u32, variable_block_size: Option, ) -> Result<()> { - map_err( - self - .inner - .initialize(&audio_config(sample_rate, block_size, variable_block_size)), - ) + map_err(self.slot.get_mut()?.initialize(&audio_config( + sample_rate, + block_size, + variable_block_size, + ))) } /// Examines a mono audio block and updates the prediction, leaving the audio unmodified. @@ -99,23 +128,23 @@ impl Vad { pub fn process(&mut self, audio: Float32Array) -> Result<()> { // Read-only, so the safe `Deref` to `&[f32]` is enough here. Taking the view by // value does not copy the caller's samples. - map_err(self.inner.process(&audio)) + map_err(self.slot.get_mut()?.process(&audio)) } /// Creates a handle for reading predictions and controlling this VAD. /// /// Each call returns an independent handle onto the same VAD. #[napi] - pub fn get_context(&self) -> VadContext { - VadContext { - inner: self.inner.context(), - } + pub fn get_context(&self) -> Result { + Ok(VadContext { + inner: self.slot.get()?.context(), + }) } /// Ends this VAD's telemetry session, after which it can no longer process audio. #[napi] pub fn terminate_session(&mut self) -> Result<()> { - map_err(self.inner.terminate_session()) + map_err(self.slot.get_mut()?.terminate_session()) } } diff --git a/src/vad_async.rs b/src/vad_async.rs index c289e7b..36b347f 100644 --- a/src/vad_async.rs +++ b/src/vad_async.rs @@ -1,15 +1,16 @@ use crate::{ claim_sdk_id, + disposable_slot::{DisposableSlot, lock}, error::{Result, map_err}, + mem, model::Model, processor::{OtelConfig, audio_config}, - processor_async::{Shared, lock}, vad::VadContext, }; use napi::{ Env, Task, - bindgen_prelude::{AsyncTask, Float32Array}, + bindgen_prelude::{AsyncTask, Float32Array, ObjectFinalize}, }; use napi_derive::napi; use std::sync::{Arc, Mutex}; @@ -36,9 +37,21 @@ use std::sync::{Arc, Mutex}; /// Raise `UV_THREADPOOL_SIZE` before Node starts to run more streams in parallel. The /// SDK's own `AIC_NUM_THREADS` does not apply here: that variable sizes a rayon pool this /// binding deliberately does not use. -#[napi] +#[napi(custom_finalize)] pub struct VadAsync { - inner: Shared>, + slot: Arc>>>, +} + +impl ObjectFinalize for VadAsync { + fn finalize(self, env: Env) -> Result<()> { + // Only the last handle onto the native object destroys it; with other handles or + // in-flight tasks holding an `Arc`, this leaves the object (and its footprint + // report) for them. Idempotent against `dispose()`. + if Arc::strong_count(&self.slot) == 1 { + lock(&self.slot).release(env); + } + Ok(()) + } } #[napi] @@ -51,17 +64,40 @@ impl VadAsync { /// Telemetry follows the runtime environment; pass `otelConfig` to override it for this /// instance. #[napi(constructor)] - pub fn new(model: &Model, license_key: String, otel_config: Option) -> Result { + pub fn new( + env: Env, + model: &Model, + license_key: String, + otel_config: Option, + ) -> Result { + let model_inner = model.inner()?; claim_sdk_id(); let inner = match otel_config { - Some(config) => aic_sdk::Vad::with_otel_config(&model.inner, &license_key, &config.into()), - None => aic_sdk::Vad::new(&model.inner, &license_key), + Some(config) => aic_sdk::Vad::with_otel_config(model_inner, &license_key, &config.into()), + None => aic_sdk::Vad::new(model_inner, &license_key), }; + let inner = map_err(inner)?; + Ok(Self { - inner: Arc::new(Mutex::new(map_err(inner)?)), + slot: Arc::new(Mutex::new(DisposableSlot::new( + env, + inner, + "VadAsync", + mem::PROCESSOR_BYTES, + ))), }) } + /// Destroys the native VAD immediately, releasing its memory and telemetry session + /// without waiting for garbage collection. + /// + /// Every later method throws; calling `dispose()` again does nothing. Blocks until + /// in-flight work on the libuv pool finishes. + #[napi] + pub fn dispose(&self, env: Env) { + lock(&self.slot).release(env); + } + /// Initializes the VAD and resolves to a handle onto it, for chaining off the /// constructor: /// @@ -79,7 +115,7 @@ impl VadAsync { variable_block_size: Option, ) -> AsyncTask { AsyncTask::new(VadWithConfigTask { - inner: self.inner.clone(), + slot: self.slot.clone(), config: audio_config(sample_rate, block_size, variable_block_size), }) } @@ -95,7 +131,7 @@ impl VadAsync { variable_block_size: Option, ) -> AsyncTask { AsyncTask::new(VadInitializeTask { - inner: self.inner.clone(), + slot: self.slot.clone(), config: audio_config(sample_rate, block_size, variable_block_size), }) } @@ -119,7 +155,7 @@ impl VadAsync { #[napi(ts_return_type = "Promise>")] pub fn process(&self, audio: Float32Array) -> AsyncTask { AsyncTask::new(VadProcessTask { - inner: self.inner.clone(), + slot: self.slot.clone(), // Copied on the JS thread so the worker owns its samples outright. A block is a // couple of kilobytes, far below the cost of running the model over it, and it // removes any chance of JS mutating the buffer mid-process. @@ -136,7 +172,7 @@ impl VadAsync { #[napi(ts_return_type = "Promise")] pub fn get_context(&self) -> AsyncTask { AsyncTask::new(VadContextTask { - inner: self.inner.clone(), + slot: self.slot.clone(), }) } @@ -146,14 +182,14 @@ impl VadAsync { #[napi(ts_return_type = "Promise")] pub fn terminate_session(&self) -> AsyncTask { AsyncTask::new(VadTerminateTask { - inner: self.inner.clone(), + slot: self.slot.clone(), }) } } /// Backs {@link VadAsync#withConfig}. pub struct VadWithConfigTask { - inner: Shared>, + slot: Arc>>>, config: aic_sdk::ProcessorConfig, } @@ -162,19 +198,22 @@ impl Task for VadWithConfigTask { type JsValue = VadAsync; fn compute(&mut self) -> Result<()> { - map_err(lock(&self.inner).initialize(&self.config)) + map_err(lock(&self.slot).get_mut()?.initialize(&self.config)) } fn resolve(&mut self, _env: Env, _: ()) -> Result { + // A second JS handle onto the same native VAD. The footprint is reported once per + // object at construction, so there is nothing to report here; the last handle's + // finalizer gives it back. Born disposed when the VAD was disposed mid-flight. Ok(VadAsync { - inner: self.inner.clone(), + slot: self.slot.clone(), }) } } /// Backs {@link VadAsync#initialize}. pub struct VadInitializeTask { - inner: Shared>, + slot: Arc>>>, config: aic_sdk::ProcessorConfig, } @@ -183,7 +222,7 @@ impl Task for VadInitializeTask { type JsValue = (); fn compute(&mut self) -> Result<()> { - map_err(lock(&self.inner).initialize(&self.config)) + map_err(lock(&self.slot).get_mut()?.initialize(&self.config)) } fn resolve(&mut self, _env: Env, _: ()) -> Result<()> { @@ -193,7 +232,7 @@ impl Task for VadInitializeTask { /// Backs {@link VadAsync#process}. pub struct VadProcessTask { - inner: Shared>, + slot: Arc>>>, audio: Vec, } @@ -205,7 +244,7 @@ impl Task for VadProcessTask { // Moved out rather than borrowed so the buffer can be handed to V8 in `resolve` // without another copy. The task is used once, so leaving an empty Vec behind is fine. let audio = std::mem::take(&mut self.audio); - map_err(lock(&self.inner).process(&audio))?; + map_err(lock(&self.slot).get_mut()?.process(&audio))?; Ok(audio) } @@ -219,7 +258,7 @@ impl Task for VadProcessTask { /// Backs {@link VadAsync#getContext}. pub struct VadContextTask { - inner: Shared>, + slot: Arc>>>, } impl Task for VadContextTask { @@ -227,7 +266,7 @@ impl Task for VadContextTask { type JsValue = VadContext; fn compute(&mut self) -> Result { - Ok(lock(&self.inner).context()) + Ok(lock(&self.slot).get()?.context()) } fn resolve(&mut self, _env: Env, context: aic_sdk::VadContext) -> Result { @@ -237,7 +276,7 @@ impl Task for VadContextTask { /// Backs {@link VadAsync#terminateSession}. pub struct VadTerminateTask { - inner: Shared>, + slot: Arc>>>, } impl Task for VadTerminateTask { @@ -245,7 +284,7 @@ impl Task for VadTerminateTask { type JsValue = (); fn compute(&mut self) -> Result<()> { - map_err(lock(&self.inner).terminate_session()) + map_err(lock(&self.slot).get_mut()?.terminate_session()) } fn resolve(&mut self, _env: Env, _: ()) -> Result<()> {