admin-valentin/AIvoices
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #6 from RomanLut/submit-race-conditions

Browse source 
fixed race conditions
This commit is contained in:
Akashdeep Deb 2025-05-26 14:25:05 +01:00 committed by GitHub
commit d31ce54fdb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 101 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

131
firmware-arduino/src/Audio.cpp
View file

@ -17,7 +17,7 @@ TaskHandle_t micTaskHandle = NULL;
TaskHandle_t networkTaskHandle = NULL;
// TIMING REGISTERS
bool scheduleListeningRestart = false;
volatile bool scheduleListeningRestart = false;
unsigned long scheduledTime = 0;
unsigned long speakingStartTime = 0;
@ -31,18 +31,20 @@ class BufferPrint : public Print {
public:
BufferPrint(BufferRTOS<uint8_t>& buf) : _buffer(buf) {}
// networkTask -> webSocket.loop() -> webSocketEvent(WStype_BIN, ...) -> opusDecoder.write() -> bufferPrint.write()
virtual size_t write(uint8_t data) override {
if (webSocket.isConnected() && deviceState == SPEAKING) {
return _buffer.writeArray(&data, 1);
}
return 0;
return 1; //let opusDecoder write, otherwise thread will stuck
}
// networkTask -> webSocket.loop() -> webSocketEvent(WStype_BIN, ...) -> opusDecoder.write() -> bufferPrint.write()
virtual size_t write(const uint8_t *buffer, size_t size) override {
if (webSocket.isConnected() && deviceState == SPEAKING) {
return _buffer.writeArray(buffer, size);
}
return 0;
return size; //let opusDecoder write, otherwise thread will stuck
}
private:
@ -50,13 +52,14 @@ private:
};
BufferPrint bufferPrint(audioBuffer);
OpusAudioDecoder opusDecoder;
BufferRTOS<uint8_t> audioBuffer(AUDIO_BUFFER_SIZE, AUDIO_CHUNK_SIZE);
I2SStream i2s;
VolumeStream volume(i2s);
QueueStream<uint8_t> queue(audioBuffer);
OpusAudioDecoder opusDecoder; //access guarded by wsmutex
BufferRTOS<uint8_t> audioBuffer(AUDIO_BUFFER_SIZE, AUDIO_CHUNK_SIZE); //producer: networkTask, consumer: audioStreamTask. Thread safe in single producer->single consumer scenario.
I2SStream i2s; //access from audioStreamTask only
VolumeStream volume(i2s); //access from audioStreamTask only
QueueStream<uint8_t> queue(audioBuffer); //access from audioStreamTask only
StreamCopy copier(volume, queue);
AudioInfo info(SAMPLE_RATE, CHANNELS, BITS_PER_SAMPLE);
volatile bool i2sOutputFlushScheduled = false;
unsigned long getSpeakingDuration() {
if (deviceState == SPEAKING && speakingStartTime > 0) {
@ -65,45 +68,39 @@ unsigned long getSpeakingDuration() {
return 0;
}
void transitionToSpeaking() {
// networkTask -> webSocket.loop() -> webSocketEvent(WStype_TEXT, ...) -> transitionToSpeaking()
void transitionToSpeaking() {
vTaskDelay(50);
i2sInput.flush();
i2sInputFlushScheduled = true;
if (xSemaphoreTake(wsMutex, pdMS_TO_TICKS(100)) == pdTRUE) {
deviceState = SPEAKING;
digitalWrite(I2S_SD_OUT, HIGH);
speakingStartTime = millis();
webSocket.enableHeartbeat(30000, 15000, 3);
xSemaphoreGive(wsMutex);
}
deviceState = SPEAKING;
digitalWrite(I2S_SD_OUT, HIGH);
speakingStartTime = millis();
webSocket.enableHeartbeat(30000, 15000, 3);
Serial.println("Transitioned to speaking mode");
}
// networkTask -> transitionToListening()
// ( networkTask -> webSocket.loop() -> webSocketEvent(WStype_TEXT, ...) -> (sets scheduleListeningRestart) -> networkTask -> transitionToListening() )
void transitionToListening() {
deviceState = PROCESSING;
scheduleListeningRestart = false;
Serial.println("Transitioning to listening mode");
// These stream operations don't directly interact with the WebSocket
i2s.flush();
volume.flush();
queue.flush();
i2sInput.flush();
audioBuffer.reset();
i2sInputFlushScheduled = true;
i2sOutputFlushScheduled = true;
Serial.println("Transitioned to listening mode");
if (xSemaphoreTake(wsMutex, pdMS_TO_TICKS(100)) == pdTRUE) {
deviceState = LISTENING;
digitalWrite(I2S_SD_OUT, LOW);
webSocket.disableHeartbeat();
xSemaphoreGive(wsMutex);
}
deviceState = LISTENING;
digitalWrite(I2S_SD_OUT, LOW);
webSocket.disableHeartbeat();
}
// audioStreamTask -> copier.copy() (conditional on webSocket.isConnected())
void audioStreamTask(void *parameter) {
Serial.println("Starting I2S stream pipeline...");
@ -114,8 +111,13 @@ void audioStreamTask(void *parameter) {
cfg.channels = CHANNELS;
cfg.bits_per_sample = BITS_PER_SAMPLE;
cfg.max_buffer_size = 6144;
xSemaphoreTake(wsMutex, portMAX_DELAY);
opusDecoder.setOutput(bufferPrint);
opusDecoder.begin(cfg);
xSemaphoreGive(wsMutex);
audioBuffer.setReadMaxWait(0);
queue.begin();
@ -134,53 +136,60 @@ void audioStreamTask(void *parameter) {
auto vcfg = volume.defaultConfig();
vcfg.copyFrom(config);
vcfg.allow_boost = true;
volume.begin(vcfg);
volume.begin(vcfg);
while (1) {
if ( i2sOutputFlushScheduled) {
i2sOutputFlushScheduled = false;
i2s.flush();
volume.flush();
queue.flush();
//audioBuffer.reset(); todo: read untill empty
}
if (webSocket.isConnected() && deviceState == SPEAKING) {
copier.copy();
copier.copy();
}
else {
//we should always read from audioBuffer, otherwise writing thread can stuck
queue.read();
}
vTaskDelay(1);
}
}
// AUDIO INPUT SETTINGS
class WebsocketStream : public Print {
public:
// micTask -> micToWsCopier.copyBytes() -> wsStream.write()
virtual size_t write(uint8_t b) override {
if (!webSocket.isConnected() || deviceState != LISTENING) {
return 1;
}
if (xSemaphoreTake(wsMutex, pdMS_TO_TICKS(100)) == pdTRUE) {
webSocket.sendBIN(&b, 1);
xSemaphoreGive(wsMutex);
return 1;
}
xSemaphoreTake(wsMutex, portMAX_DELAY);
webSocket.sendBIN(&b, 1);
xSemaphoreGive(wsMutex);
return 1;
}
// micTask -> micToWsCopier.copyBytes() -> wsStream.write()
virtual size_t write(const uint8_t *buffer, size_t size) override {
if (size == 0 || !webSocket.isConnected() || deviceState != LISTENING) {
return size;
}
if (xSemaphoreTake(wsMutex, pdMS_TO_TICKS(100)) == pdTRUE) {
webSocket.sendBIN(buffer, size);
xSemaphoreGive(wsMutex);
return size;
}
xSemaphoreTake(wsMutex, portMAX_DELAY);
webSocket.sendBIN(buffer, size);
xSemaphoreGive(wsMutex);
return size;
}
};
WebsocketStream wsStream;
I2SStream i2sInput;
WebsocketStream wsStream; //guard with wsMutex
I2SStream i2sInput; //access from micTask only
StreamCopy micToWsCopier(wsStream, i2sInput);
volatile bool i2sInputFlushScheduled = false;
const int MIC_COPY_SIZE = 64;
void micTask(void *parameter) {
@ -198,10 +207,12 @@ void micTask(void *parameter) {
i2sConfig.port_no = I2S_PORT_IN;
i2sInput.begin(i2sConfig);
micToWsCopier.setDelayOnNoData(0);
while (1) {
// Check to see if a transition to listening mode is scheduled.
if (scheduleListeningRestart && millis() >= scheduledTime) {
transitionToListening();
if ( i2sInputFlushScheduled ) {
i2sInputFlushScheduled = false;
i2sInput.flush();
}
if (deviceState == LISTENING && webSocket.isConnected()) {
@ -217,6 +228,7 @@ void micTask(void *parameter) {
}
// WEBSOCKET EVENTS
// networkTask -> webSocket.loop() -> webSocketEvent()
void webSocketEvent(WStype_t type, uint8_t *payload, size_t length)
{
switch (type)
@ -317,9 +329,13 @@ void webSocketEvent(WStype_t type, uint8_t *payload, size_t length)
}
}
// wifiTask -> WIFIMANAGER::loop() -> WIFIMANAGER::tryConnect() -> connectCb() -> websocketSetup()
void websocketSetup(String server_domain, int port, String path)
{
String headers = "Authorization: Bearer " + String(authTokenGlobal);
xSemaphoreTake(wsMutex, portMAX_DELAY);
webSocket.setExtraHeaders(headers.c_str());
webSocket.onEvent(webSocketEvent);
webSocket.setReconnectInterval(1000);
@ -331,12 +347,23 @@ void websocketSetup(String server_domain, int port, String path)
#else
webSocket.beginSslWithCA(server_domain.c_str(), port, path.c_str(), CA_cert);
#endif
xSemaphoreGive(wsMutex);
}
// networkTask -> webSocket.loop()
void networkTask(void *parameter) {
while (1) {
xSemaphoreTake(wsMutex, portMAX_DELAY);
// Check to see if a transition to listening mode is scheduled.
if (scheduleListeningRestart && millis() >= scheduledTime) {
transitionToListening();
}
webSocket.loop();
xSemaphoreGive(wsMutex);
vTaskDelay(1);
}
}

4
firmware-arduino/src/Audio.h
View file

@ -13,7 +13,7 @@ extern TaskHandle_t speakerTaskHandle;
extern TaskHandle_t micTaskHandle;
extern TaskHandle_t networkTaskHandle;
extern bool scheduleListeningRestart;
extern volatile bool scheduleListeningRestart;
extern unsigned long scheduledTime;
extern unsigned long speakingStartTime;
@ -31,10 +31,12 @@ extern VolumeStream volume;
extern QueueStream<uint8_t> queue;
extern StreamCopy copier;
extern AudioInfo info;
extern volatile bool i2sOutputFlushScheduled;
// AUDIO INPUT
extern I2SStream i2sInput;
extern StreamCopy micToWsCopier;
extern volatile bool i2sInputFlushScheduled;
// WEBSOCKET
void webSocketEvent(WStype_t type, uint8_t *payload, size_t length);

2
firmware-arduino/src/Config.cpp
View file

@ -59,7 +59,7 @@ const uint16_t backend_port = 3000;
#endif
String authTokenGlobal;
DeviceState deviceState = IDLE;
volatile DeviceState deviceState = IDLE;
// I2S and Audio parameters
const uint32_t SAMPLE_RATE = 24000;

5
firmware-arduino/src/Config.h
View file

@ -44,10 +44,11 @@ enum DeviceState
PROCESSING,
WAITING,
OTA,
FACTORY_RESET
FACTORY_RESET,
SLEEP
};
extern DeviceState deviceState;
extern volatile DeviceState deviceState;
// WiFi credentials
extern const char *EAP_IDENTITY;

21
firmware-arduino/src/main.cpp
View file

@ -18,27 +18,35 @@ AsyncWebServer webServer(80);
WIFIMANAGER WifiManager;
esp_err_t getErr = ESP_OK;
// Main Thread -> onButtonLongPressUpEventCb -> enterSleep()
// Main Thread -> onButtonDoubleClickCb -> enterSleep()
// Touch Task -> touchTask -> enterSleep()
// Main Thread -> loop() (inactivity timeout) -> enterSleep()
void enterSleep()
{
Serial.println("Going to sleep...");
// First, change device state to prevent any new data processing
deviceState = IDLE;
deviceState = SLEEP;
scheduleListeningRestart = false;
i2sOutputFlushScheduled = true;
i2sInputFlushScheduled = true;
vTaskDelay(10); //let all tasks accept state
xSemaphoreTake(wsMutex, portMAX_DELAY);
// Stop audio tasks first
i2s_stop(I2S_PORT_IN);
i2s_stop(I2S_PORT_OUT);
// Clear any remaining audio in buffer
audioBuffer.reset();
// Properly disconnect WebSocket and wait for it to complete
if (webSocket.isConnected()) {
webSocket.disconnect();
// Give some time for the disconnect to process
delay(100);
}
xSemaphoreGive(wsMutex);
delay(100);
// Stop all tasks that might be using I2S or other peripherals
i2s_driver_uninstall(I2S_PORT_IN);
i2s_driver_uninstall(I2S_PORT_OUT);
@ -56,6 +64,7 @@ void enterSleep()
#endif
esp_deep_sleep_start();
delay(1000);
}
void printOutESP32Error(esp_err_t err)