refactor: replace writer thread with synchronous save and fix ring buffer memory ordering
This commit is contained in:
Loic Coenen
committed by
Loic Coenen (aider)
Loic Coenen (aider)
parent
10e47e6c0c
commit
f38797fe0a
@@ -2,9 +2,9 @@
|
||||
#include "looper.h"
|
||||
#include "channel.h"
|
||||
#include "midi.h"
|
||||
#include "pipe.h"
|
||||
#include "queue.h"
|
||||
#include "wav.h"
|
||||
#include "pipe.h"
|
||||
#include <fcntl.h>
|
||||
#include <jack/jack.h>
|
||||
#include <jack/midiport.h>
|
||||
@@ -38,13 +38,23 @@ static void looper_write_status(void) {
|
||||
int state = atomic_load(&channels[ch].scenes[sc_idx].state);
|
||||
const char *state_str;
|
||||
switch (state) {
|
||||
case STATE_IDLE: state_str = "IDLE"; break;
|
||||
case STATE_RECORD: state_str = "RECORD"; break;
|
||||
case STATE_LOOPING:state_str = "LOOPING";break;
|
||||
case STATE_PAUSED: state_str = "PAUSED"; break;
|
||||
default: state_str = "UNKNOWN";
|
||||
case STATE_IDLE:
|
||||
state_str = "IDLE";
|
||||
break;
|
||||
case STATE_RECORD:
|
||||
state_str = "RECORD";
|
||||
break;
|
||||
case STATE_LOOPING:
|
||||
state_str = "LOOPING";
|
||||
break;
|
||||
case STATE_PAUSED:
|
||||
state_str = "PAUSED";
|
||||
break;
|
||||
default:
|
||||
state_str = "UNKNOWN";
|
||||
}
|
||||
int n = snprintf(buf, sizeof(buf), "CH=%d SC=%d STATE=%s\n", ch, sc_idx, state_str);
|
||||
int n = snprintf(buf, sizeof(buf), "CH=%d SC=%d STATE=%s\n", ch, sc_idx,
|
||||
state_str);
|
||||
if (n > 0) {
|
||||
int ret = write(status_fd, buf, n);
|
||||
(void)ret;
|
||||
@@ -69,101 +79,101 @@ atomic_int bind_channel = 0;
|
||||
/* Deferred removal index (1 second grace) */
|
||||
static int pending_unregister_idx = -1;
|
||||
|
||||
/* writer thread function and sample rate holder */
|
||||
static void *writer_thread(void *arg);
|
||||
/* sample rate holder */
|
||||
static int global_sample_rate = 0;
|
||||
|
||||
/* execute a single command (called from looper_process_commands) */
|
||||
static void exec_command(command_t cmd, jack_client_t *client) {
|
||||
int ch = cmd.channel;
|
||||
if (ch < 0) ch = 0;
|
||||
int ch = cmd.channel;
|
||||
if (ch < 0)
|
||||
ch = 0;
|
||||
|
||||
switch (cmd.type) {
|
||||
case CMD_CYCLE: {
|
||||
int sc_idx = atomic_load(&channels[ch].current_scene);
|
||||
int state = atomic_load(&channels[ch].scenes[sc_idx].state);
|
||||
switch (state) {
|
||||
case STATE_IDLE:
|
||||
atomic_store(&channels[ch].scenes[sc_idx].state, STATE_RECORD);
|
||||
break;
|
||||
case STATE_RECORD:
|
||||
atomic_store(&channels[ch].scenes[sc_idx].state, STATE_LOOPING);
|
||||
break;
|
||||
case STATE_LOOPING:
|
||||
atomic_store(&channels[ch].scenes[sc_idx].state, STATE_PAUSED);
|
||||
break;
|
||||
case STATE_PAUSED:
|
||||
atomic_store(&channels[ch].scenes[sc_idx].state, STATE_LOOPING);
|
||||
break;
|
||||
}
|
||||
atomic_store(&channels[ch].scenes[sc_idx].prev_state, -1);
|
||||
switch (cmd.type) {
|
||||
case CMD_CYCLE: {
|
||||
int sc_idx = atomic_load(&channels[ch].current_scene);
|
||||
scene_t *sc_ptr = &channels[ch].scenes[sc_idx];
|
||||
int state = atomic_load(&sc_ptr->state);
|
||||
switch (state) {
|
||||
case STATE_IDLE:
|
||||
atomic_store(&sc_ptr->state, STATE_RECORD);
|
||||
break;
|
||||
case STATE_RECORD:
|
||||
atomic_store(&sc_ptr->state, STATE_LOOPING);
|
||||
break;
|
||||
case STATE_LOOPING:
|
||||
atomic_store(&sc_ptr->state, STATE_PAUSED);
|
||||
break;
|
||||
case STATE_PAUSED:
|
||||
atomic_store(&sc_ptr->state, STATE_LOOPING);
|
||||
break;
|
||||
}
|
||||
case CMD_STOP:
|
||||
for (int s = 0; s < atomic_load(&channels[ch].scene_count); s++) {
|
||||
atomic_store(&channels[ch].scenes[s].state, STATE_IDLE);
|
||||
atomic_store(&channels[ch].scenes[s].prev_state, -1);
|
||||
}
|
||||
break;
|
||||
|
||||
case CMD_ADD_CHANNEL:
|
||||
case CMD_ADD_MIDI_CHANNEL: {
|
||||
int idx;
|
||||
for (idx = 0; idx < MAX_CHANNELS; idx++)
|
||||
if (!channels[idx].active)
|
||||
break;
|
||||
if (idx < MAX_CHANNELS)
|
||||
channel_add(client, idx);
|
||||
break;
|
||||
break;
|
||||
}
|
||||
case CMD_STOP:
|
||||
for (int s = 0; s < atomic_load(&channels[ch].scene_count); s++) {
|
||||
atomic_store(&channels[ch].scenes[s].state, STATE_IDLE);
|
||||
atomic_store(&channels[ch].scenes[s].prev_state, -1);
|
||||
}
|
||||
break;
|
||||
|
||||
case CMD_REMOVE_CHANNEL: {
|
||||
int remove_idx = -1;
|
||||
for (int idx = 1; idx < MAX_CHANNELS; idx++)
|
||||
if (channels[idx].active)
|
||||
remove_idx = idx;
|
||||
if (remove_idx != -1) {
|
||||
channel_remove(client, remove_idx);
|
||||
pending_unregister_idx = remove_idx;
|
||||
}
|
||||
break;
|
||||
case CMD_ADD_CHANNEL:
|
||||
case CMD_ADD_MIDI_CHANNEL: {
|
||||
int idx;
|
||||
for (idx = 0; idx < MAX_CHANNELS; idx++)
|
||||
if (!channels[idx].active)
|
||||
break;
|
||||
if (idx < MAX_CHANNELS)
|
||||
channel_add(client, idx);
|
||||
break;
|
||||
}
|
||||
|
||||
case CMD_REMOVE_CHANNEL: {
|
||||
int remove_idx = -1;
|
||||
for (int idx = 1; idx < MAX_CHANNELS; idx++)
|
||||
if (channels[idx].active)
|
||||
remove_idx = idx;
|
||||
if (remove_idx != -1) {
|
||||
channel_remove(client, remove_idx);
|
||||
pending_unregister_idx = remove_idx;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case CMD_BIND_CHANNEL:
|
||||
atomic_store(&bind_channel, cmd.data);
|
||||
break;
|
||||
case CMD_BIND_CHANNEL:
|
||||
atomic_store(&bind_channel, cmd.data);
|
||||
break;
|
||||
|
||||
case CMD_UNBIND:
|
||||
atomic_store(&bind_channel, 0);
|
||||
break;
|
||||
case CMD_UNBIND:
|
||||
atomic_store(&bind_channel, 0);
|
||||
break;
|
||||
|
||||
case CMD_LOAD:
|
||||
atomic_store(&cmd_load, 1);
|
||||
break;
|
||||
case CMD_LOAD:
|
||||
atomic_store(&cmd_load, 1);
|
||||
break;
|
||||
|
||||
case CMD_SAVE:
|
||||
atomic_store(&cmd_save, 1);
|
||||
break;
|
||||
case CMD_SAVE:
|
||||
atomic_store(&cmd_save, 1);
|
||||
break;
|
||||
|
||||
case CMD_ADD_SCENE:
|
||||
channel_add_scene(client, ch);
|
||||
break;
|
||||
case CMD_ADD_SCENE:
|
||||
channel_add_scene(client, ch);
|
||||
break;
|
||||
|
||||
case CMD_REMOVE_SCENE:
|
||||
channel_remove_scene(client, ch);
|
||||
break;
|
||||
case CMD_REMOVE_SCENE:
|
||||
channel_remove_scene(client, ch);
|
||||
break;
|
||||
|
||||
case CMD_NEXT_SCENE:
|
||||
channel_next_scene(client, ch);
|
||||
break;
|
||||
case CMD_NEXT_SCENE:
|
||||
channel_next_scene(client, ch);
|
||||
break;
|
||||
|
||||
case CMD_PREV_SCENE:
|
||||
channel_prev_scene(client, ch);
|
||||
break;
|
||||
case CMD_PREV_SCENE:
|
||||
channel_prev_scene(client, ch);
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
@@ -217,7 +227,8 @@ int process_callback(jack_nframes_t nframes, void *arg) {
|
||||
/* MIDI channel handling */
|
||||
void *midi_in_buf = jack_port_get_buffer(channels[c].midi_in, nframes);
|
||||
void *midi_out_buf = jack_port_get_buffer(channels[c].midi_out, nframes);
|
||||
if (!midi_out_buf) continue;
|
||||
if (!midi_out_buf)
|
||||
continue;
|
||||
|
||||
switch (state) {
|
||||
case STATE_RECORD: {
|
||||
@@ -232,7 +243,8 @@ int process_callback(jack_nframes_t nframes, void *arg) {
|
||||
sc->loop.midi_events[rp].timestamp = ev.time;
|
||||
sc->loop.midi_events[rp].status = ev.buffer[0];
|
||||
sc->loop.midi_events[rp].note = (ev.size > 1) ? ev.buffer[1] : 0;
|
||||
sc->loop.midi_events[rp].velocity = (ev.size > 2) ? ev.buffer[2] : 0;
|
||||
sc->loop.midi_events[rp].velocity =
|
||||
(ev.size > 2) ? ev.buffer[2] : 0;
|
||||
atomic_store(&sc->record_pos, rp + 1);
|
||||
}
|
||||
}
|
||||
@@ -286,7 +298,8 @@ int process_callback(jack_nframes_t nframes, void *arg) {
|
||||
jack_default_audio_sample_t *out =
|
||||
(jack_default_audio_sample_t *)jack_port_get_buffer(
|
||||
channels[c].audio_out, nframes);
|
||||
if (!out) continue;
|
||||
if (!out)
|
||||
continue;
|
||||
|
||||
switch (state) {
|
||||
case STATE_RECORD:
|
||||
@@ -397,7 +410,6 @@ void jack_shutdown_cb(void *arg) {
|
||||
exit(0);
|
||||
}
|
||||
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* looper initialisation
|
||||
* ---------------------------------------------------------------- */
|
||||
@@ -408,10 +420,11 @@ int looper_init(jack_client_t *client) {
|
||||
/* create status FIFO (ignore if already exists) */
|
||||
mkfifo(STATUS_FIFO, 0666);
|
||||
|
||||
/* open the status FIFO for reading+writing so writes work even without reader */
|
||||
/* open the status FIFO for reading+writing so writes work even without reader
|
||||
*/
|
||||
status_fd = open(STATUS_FIFO, O_RDWR);
|
||||
if (status_fd < 0) {
|
||||
perror("open status FIFO");
|
||||
perror("open status FIFO");
|
||||
}
|
||||
|
||||
queue_init(&cmd_queue);
|
||||
@@ -421,10 +434,9 @@ int looper_init(jack_client_t *client) {
|
||||
/* start the FIFO reader thread */
|
||||
pipe_start_reader();
|
||||
|
||||
|
||||
/* channel 0 */
|
||||
channels[0].active = 1;
|
||||
channels[0].type = CHANNEL_AUDIO; /* default */
|
||||
channels[0].type = CHANNEL_AUDIO; /* default */
|
||||
channels[0].current_scene = 0;
|
||||
channels[0].scene_count = 1;
|
||||
init_scene(&channels[0].scenes[0]); /* sets state IDLE, prev_state -1 */
|
||||
@@ -453,58 +465,15 @@ int looper_init(jack_client_t *client) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* Give JACK time to register the ports before clients connect */
|
||||
{
|
||||
struct timespec req = {.tv_sec = 0, .tv_nsec = 500000000};
|
||||
nanosleep(&req, NULL);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* writer thread – consumes the save ring and writes WAV file
|
||||
* ---------------------------------------------------------------- */
|
||||
static void *writer_thread(void *arg) {
|
||||
struct channel_t *ch = (struct channel_t *)arg;
|
||||
int sc_idx = atomic_load(&ch->current_scene);
|
||||
scene_t *sc = &ch->scenes[sc_idx];
|
||||
RingBuf *ring = (RingBuf *)ch->save_ring;
|
||||
if (!ring)
|
||||
return NULL;
|
||||
|
||||
static const char *path = "save.wav";
|
||||
unsigned sr = (unsigned)global_sample_rate;
|
||||
if (sr == 0)
|
||||
sr = 48000;
|
||||
|
||||
int lc = atomic_load(&sc->loop_count);
|
||||
float *outbuf = malloc((size_t)lc * sizeof(float));
|
||||
if (!outbuf) {
|
||||
ring_destroy(ring);
|
||||
free(ring);
|
||||
ch->save_ring = NULL;
|
||||
return NULL;
|
||||
}
|
||||
size_t collected = 0;
|
||||
size_t want = (size_t)lc;
|
||||
while (collected < want) {
|
||||
size_t got = ring_read(ring, outbuf + collected, want - collected);
|
||||
collected += got;
|
||||
if (got == 0) {
|
||||
struct timespec req = {.tv_sec = 0, .tv_nsec = 10000000};
|
||||
nanosleep(&req, NULL);
|
||||
}
|
||||
}
|
||||
wav_write(path, outbuf, (unsigned)lc, sr);
|
||||
free(outbuf);
|
||||
|
||||
/* Signal the RT thread to stop writing */
|
||||
atomic_store_explicit(&ch->save_complete, 1, memory_order_release);
|
||||
/* Wait for the RT thread to see the flag (one audio period) */
|
||||
struct timespec req = { .tv_sec = 0, .tv_nsec = 10000000 }; /* 10ms */
|
||||
nanosleep(&req, NULL);
|
||||
|
||||
ring_destroy(ring);
|
||||
free(ring);
|
||||
atomic_store_explicit(&ch->save_ring, NULL, memory_order_release);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* main‑loop command processing
|
||||
* ---------------------------------------------------------------- */
|
||||
@@ -578,25 +547,31 @@ void looper_process_commands(jack_client_t *client) {
|
||||
}
|
||||
}
|
||||
|
||||
/* ---------- save command (writer thread) ---------- */
|
||||
/* ---------- save command (synchronous) ---------- */
|
||||
if (atomic_exchange(&cmd_save, 0)) {
|
||||
int sc_idx = atomic_load(&channels[0].current_scene);
|
||||
scene_t *sc = &channels[0].scenes[sc_idx];
|
||||
int lc = atomic_load(&sc->loop_count);
|
||||
if (atomic_load(&sc->state) == STATE_LOOPING && lc > 0 &&
|
||||
channels[0].save_ring == NULL) {
|
||||
RingBuf *ring = (RingBuf *)malloc(sizeof(RingBuf));
|
||||
if (ring) {
|
||||
size_t sz = (size_t)lc * 2;
|
||||
if (ring_init(ring, sz) == 0) {
|
||||
atomic_store_explicit(&channels[0].save_ring, (_Atomic RingBuf *)ring,
|
||||
memory_order_release);
|
||||
pthread_t th;
|
||||
pthread_create(&th, NULL, writer_thread, &channels[0]);
|
||||
pthread_detach(th);
|
||||
} else {
|
||||
free(ring);
|
||||
}
|
||||
if (atomic_load(&sc->state) == STATE_LOOPING && lc > 0) {
|
||||
/* Deactivate channel to prevent RT thread from reading the buffer */
|
||||
int was_active = atomic_load(&channels[0].active);
|
||||
if (was_active) {
|
||||
atomic_store(&channels[0].active, 0);
|
||||
struct timespec req = {.tv_sec = 0, .tv_nsec = 500000000}; /* 500 ms */
|
||||
nanosleep(&req, NULL);
|
||||
}
|
||||
/* Now safe to copy the loop buffer */
|
||||
float *data = malloc((size_t)lc * sizeof(float));
|
||||
if (data) {
|
||||
memcpy(data, sc->loop.audio_buffer, (size_t)lc * sizeof(float));
|
||||
unsigned sr = (unsigned)global_sample_rate;
|
||||
if (sr == 0) sr = 48000;
|
||||
wav_write("save.wav", data, (unsigned)lc, sr);
|
||||
free(data);
|
||||
}
|
||||
/* Reactivate channel */
|
||||
if (was_active) {
|
||||
atomic_store(&channels[0].active, 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user