w1n5t0n/memlnaut-nisps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(vcv): close the audio-thread race and remove JSON from process()

Browse source 
Phase 2 (L34, L35). Both findings' line citations were accurate this time.

- L34 data race: process() (Rack's audio thread) called
  imlShadow.get_example_features()/get_example_labels() directly on the WORKER
  thread's private engine — which the file's own THREADING INVARIANT comment
  says only the worker may touch — while workerLoop() concurrently
  clear/refills those same std::vector<std::vector<float>> members via
  load_examples(), train_(), randomise_weights and clear_dataset. Unsynchronised
  reader/writer on a non-atomic vector: undefined behaviour.
  Fix extends the staged handoff the file ALREADY uses for pendingWeights
  rather than adding a second mutex: the worker deep-copies features/labels
  into pendingFeatures/pendingLabels at the same instant it copies
  pendingWeights, immediately before weightsPending.store(true), and the flag is
  now released only after the whole batch is consumed — closing an early-release
  window the old code had. process() no longer references imlShadow at all
  (verified: the only surviving mention is a comment).
- L35: process() ran full jansson serialize on every weight-swap OSC push and
  full json_loads + dataFromJson on incoming OSC state — heap-heavy tree work
  on the audio thread. The plan said to move it to the worker. It is DELETED
  instead: reading the actual transport shows both directions talk to nobody —
  osc-client.ts only ever sends {params|input|feedback}, and bridge.ts has no
  state/weights case and explicitly drops other addresses. Relocating
  heap-heavy work to serve a confirmed-zero consumer is complexity without a
  requirement; removing the cause is the smaller coherent design.
  dataToJson/dataFromJson are UNTOUCHED — they remain the live consumers for
  Rack patch save/load and the .nisps preset menu, both off the audio thread.

Neither is empirically reproduced: a real race needs a live Rack engine under
TSan, which is not available here. Justified by reading, and verified by
`cd vcv && make -j4` (clean) plus the host suite including
test_vcv_iml_parity.cpp, which pins iml.hpp bit-exactly against the core MLP —
iml.hpp was not modified, and parity holds.

Known remaining, pre-existing and out of scope: process() still takes a brief
lock_guard on feedbackMutex to copy a small staged struct, and several config
fields (slewMs, oscPort, output/input range flags) are written by the UI thread
without atomics.
This commit is contained in:
monkey-w1n5t0n 2026-07-21 13:23:11 +02:00
parent 75e0e58067
commit 1a78ed9597
2 changed files with 26 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
vcv/src/MEMLNaut.cpp
View file

@ -78,12 +78,21 @@ struct MEMLNaut : Module {
// ── ML Engine (double-buffered) ─────────────────────────────────
// THREADING INVARIANT: only the audio thread touches `iml`; the worker
// thread operates exclusively on `imlShadow`. Hand-off is through atomic-
// flagged staging buffers (see startOsc + workerLoop).
// flagged staging buffers (see stageForWorker + workerLoop) — the audio
// thread must never read imlShadow's members directly (see pendingWeights).
nisps::IML<float> iml{NUM_ML_INPUTS, NUM_ML_OUTPUTS, {16, 24, 16}};
nisps::IML<float> imlShadow{NUM_ML_INPUTS, NUM_ML_OUTPUTS, {16, 24, 16}};
// Worker → Audio: staged weights ready for swap (core-exact flat layout)
// Worker → Audio: staged weights + example set ready for swap (core-exact
// flat weight layout). The worker deep-copies imlShadow's examples here
// BEFORE flipping weightsPending, so the audio thread never touches
// imlShadow directly — it only ever reads this completed private copy
// (same round-trip handshake as pendingWeights: the worker only writes
// these once weightsPending has been observed false, i.e. the audio
// thread has fully consumed the previous batch).
nisps::IML<float>::Weights pendingWeights;
std::vector<std::vector<float>> pendingFeatures;
std::vector<std::vector<float>> pendingLabels;
std::atomic<bool> weightsPending{false};
// Audio → Worker: staged weight snapshot for the worker to start from
@ -120,9 +129,7 @@ struct MEMLNaut : Module {
float bridgedInputs[MAX_ML_INPUTS] = {};
std::mutex bridgedInputMutex;
// OSC → Audio: staged JSON (state/weights) + staged feedback op
std::string oscStagedJson;
std::atomic<bool> oscJsonPending{false};
// OSC → Audio: staged feedback op
StagedFeedback stagedFeedback;
std::atomic<bool> feedbackPending{false};
std::mutex feedbackMutex;
@ -133,20 +140,11 @@ struct MEMLNaut : Module {
int oscPort = OSC_DEFAULT_PORT;
int oscSendCounter = 0;
static constexpr int OSC_SEND_INTERVAL_SAMPLES = 4410; // ~100ms at 44.1kHz
std::atomic<bool> stateDirty{false}; // set after a weight swap → push /nisps/state
void startOsc() {
if (oscServer && oscServer->isRunning()) return;
oscServer = std::make_unique<memlnaut::OscServer>();
// Full state / weights JSON → stage for the audio thread to apply.
oscServer->onState([this](const std::string& json) {
if (!oscJsonPending.load()) { oscStagedJson = json; oscJsonPending.store(true); }
});
oscServer->onWeights([this](const std::string& json) {
if (!oscJsonPending.load()) { oscStagedJson = json; oscJsonPending.store(true); }
});
// Live input vector from the browser → drive the model inputs.
oscServer->onInput([this](const std::vector<float>& values) {
{
@ -239,16 +237,6 @@ struct MEMLNaut : Module {
return fb;
}
// Build a compact JSON state snapshot (for module → browser sync).
std::string buildStateJson() {
json_t* root = dataToJson();
char* str = json_dumps(root, JSON_COMPACT);
json_decref(root);
std::string out = str ? str : "{}";
free(str);
return out;
}
// ── Triggers ──────────────────────────────────────────────────────
dsp::BooleanTrigger randTrigger;
dsp::BooleanTrigger thumbsUpTrigger;
@ -361,7 +349,14 @@ struct MEMLNaut : Module {
std::this_thread::sleep_for(std::chrono::microseconds(100));
if (shouldStop.load()) break;
// Deep-copy the example set alongside the weights BEFORE flipping
// the ready flag, so the audio thread's consumer below only ever
// reads a completed private copy — it must never read imlShadow's
// features_/labels_ directly, since the worker keeps mutating
// them (next job's load_examples, training, clear) concurrently.
pendingWeights = imlShadow.get_weights();
pendingFeatures = imlShadow.get_example_features();
pendingLabels = imlShadow.get_example_labels();
weightsPending.store(true);
if (imlShadow.get_example_count() > 0) {
@ -470,14 +465,6 @@ struct MEMLNaut : Module {
lights[LIGHT_LEARN].setBrightness(learn ? 1.f : 0.f);
lights[LIGHT_TRAINING].setBrightness(isTraining.load() ? 1.f : 0.f);
// Apply staged OSC state/weights JSON.
if (oscJsonPending.load()) {
json_error_t error;
json_t* root = json_loads(oscStagedJson.c_str(), 0, &error);
if (root) { dataFromJson(root); json_decref(root); }
oscJsonPending.store(false);
}
// Apply staged remote feedback (browser verdict over the bridge) — routes
// through the same paths as the panel buttons.
if (feedbackPending.load()) {
@ -507,16 +494,16 @@ struct MEMLNaut : Module {
}
}
// Apply new weights from the background thread.
// Apply new weights + example set from the background thread. Both
// were deep-copied into pendingWeights/pendingFeatures/pendingLabels
// by the worker before it set the flag — this thread never reads
// imlShadow directly (see workerLoop's staging comment).
if (weightsPending.load()) {
iml.set_weights(pendingWeights);
iml.load_examples(pendingFeatures, pendingLabels);
weightsPending.store(false);
auto newFeats = imlShadow.get_example_features();
auto newLabels = imlShadow.get_example_labels();
iml.load_examples(newFeats, newLabels);
for (int i = 0; i < NUM_ML_OUTPUTS; i++) prevOutputs[i] = cachedOutputs[i];
crossfadeProgress = 0.f;
stateDirty.store(true); // push fresh /nisps/state to the browser
}
// RAND button → enqueue Randomize.
@ -607,11 +594,8 @@ struct MEMLNaut : Module {
derivedDelta = std::sqrt(delta);
}
// OSC send (throttled to ~100ms), plus an immediate state push when dirty.
// OSC send (throttled to ~100ms).
if (oscServer && oscServer->isRunning()) {
if (stateDirty.exchange(false)) {
oscServer->sendState(buildStateJson());
}
oscSendCounter++;
if (oscSendCounter >= OSC_SEND_INTERVAL_SAMPLES) {
oscSendCounter = 0;

32
vcv/src/osc_server.hpp
View file

@ -146,12 +146,8 @@ public:
OscServer& operator=(const OscServer&) = delete;
// Register handlers before starting.
// onState — full JSON state snapshot (/nisps/state <s>)
// onWeights — weights-only JSON (/nisps/weights <s>)
// onInput — live input vector (browser drives the model) (/nisps/input <f…f>)
// onFeedback— verdict op JSON (thumbs/place/rand/clear) (/nisps/feedback <s>)
void onState(StringCallback cb) { stateCallback_ = std::move(cb); }
void onWeights(StringCallback cb) { weightsCallback_ = std::move(cb); }
void onInput(FloatVecCallback cb) { inputCallback_ = std::move(cb); }
void onFeedback(StringCallback cb) { feedbackCallback_ = std::move(cb); }
@ -253,18 +249,6 @@ public:
sendPacket(msg);
}
// Send a full JSON state snapshot (module → browser).
void sendState(const std::string& json) {
auto msg = osc::messageString("/nisps/state", json);
sendPacket(msg);
}
// Send weights-only JSON (module → browser).
void sendWeights(const std::string& json) {
auto msg = osc::messageString("/nisps/weights", json);
sendPacket(msg);
}
private:
void recvLoop() {
uint8_t buf[65536];
@ -289,19 +273,7 @@ private:
if (tags.empty() || tags[0] != ',') return;
// Dispatch based on address
if (address == "/nisps/state") {
// Expect a single string argument
if (tags.size() >= 2 && tags[1] == 's') {
std::string payload = osc::readString(buf, len, offset);
if (stateCallback_) stateCallback_(payload);
}
} else if (address == "/nisps/weights") {
// Expect a single string argument (JSON)
if (tags.size() >= 2 && tags[1] == 's') {
std::string payload = osc::readString(buf, len, offset);
if (weightsCallback_) weightsCallback_(payload);
}
} else if (address == "/nisps/feedback") {
if (address == "/nisps/feedback") {
// Verdict op as a JSON string:
// {"op":"up|down|rand|clear","spread":f,"input":[…],"output":[…]}
if (tags.size() >= 2 && tags[1] == 's') {
@ -355,8 +327,6 @@ private:
std::atomic<bool> running_{false};
// Callbacks
StringCallback stateCallback_;
StringCallback weightsCallback_;
StringCallback feedbackCallback_;
FloatVecCallback inputCallback_;