From ad725ffc71a2ca5cbd45d6bbdcc29276f0d4a0af Mon Sep 17 00:00:00 2001 From: Roman Lut <11955117+RomanLut@users.noreply.github.com> Date: Sat, 24 May 2025 23:45:10 +0200 Subject: [PATCH] fixed race conditions --- firmware-arduino/src/Audio.cpp | 133 +++++++++++++++++++------------- firmware-arduino/src/Audio.h | 4 +- firmware-arduino/src/Config.cpp | 2 +- firmware-arduino/src/Config.h | 5 +- firmware-arduino/src/main.cpp | 22 ++++-- 5 files changed, 104 insertions(+), 62 deletions(-) diff --git a/firmware-arduino/src/Audio.cpp b/firmware-arduino/src/Audio.cpp index d74aa33..d90d87d 100644 --- a/firmware-arduino/src/Audio.cpp +++ b/firmware-arduino/src/Audio.cpp @@ -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& 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 audioBuffer(AUDIO_BUFFER_SIZE, AUDIO_CHUNK_SIZE); -I2SStream i2s; -VolumeStream volume(i2s); -QueueStream queue(audioBuffer); +OpusAudioDecoder opusDecoder; //access guarded by wsmutex +BufferRTOS 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 queue(audioBuffer); //access from audioStreamTask only StreamCopy copier(volume, queue); AudioInfo info(SAMPLE_RATE, CHANNELS, BITS_PER_SAMPLE); +volatile bool outputFlushScheduled = 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; + outputFlushScheduled = 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 ( outputFlushScheduled) { + outputFlushScheduled = 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) @@ -276,7 +288,7 @@ void webSocketEvent(WStype_t type, uint8_t *payload, size_t length) // Check if volume_control is included in the message if (doc.containsKey("volume_control")) { int newVolume = doc["volume_control"].as(); - volume.setVolume(newVolume / 100.0f); + volume.setVolume(newVolume / 100.0f * 1.4f); } scheduleListeningRestart = true; @@ -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); } } - diff --git a/firmware-arduino/src/Audio.h b/firmware-arduino/src/Audio.h index ea7f8f8..7e43df0 100644 --- a/firmware-arduino/src/Audio.h +++ b/firmware-arduino/src/Audio.h @@ -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 queue; extern StreamCopy copier; extern AudioInfo info; +extern volatile bool outputFlushScheduled; // AUDIO INPUT extern I2SStream i2sInput; extern StreamCopy micToWsCopier; +extern volatile bool i2sInputFlushScheduled; // WEBSOCKET void webSocketEvent(WStype_t type, uint8_t *payload, size_t length); diff --git a/firmware-arduino/src/Config.cpp b/firmware-arduino/src/Config.cpp index dbb2f24..e5b3bcf 100644 --- a/firmware-arduino/src/Config.cpp +++ b/firmware-arduino/src/Config.cpp @@ -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; diff --git a/firmware-arduino/src/Config.h b/firmware-arduino/src/Config.h index 803f299..df66cbf 100644 --- a/firmware-arduino/src/Config.h +++ b/firmware-arduino/src/Config.h @@ -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; diff --git a/firmware-arduino/src/main.cpp b/firmware-arduino/src/main.cpp index a6fa52f..1a64049 100644 --- a/firmware-arduino/src/main.cpp +++ b/firmware-arduino/src/main.cpp @@ -20,27 +20,38 @@ 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; + outputFlushScheduled = 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(); - + outputFlushScheduled = true; + // 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); @@ -58,6 +69,7 @@ void enterSleep() #endif esp_deep_sleep_start(); + delay(1000); } void printOutESP32Error(esp_err_t err)