fixed race conditions

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<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;
+OpusAudioDecoder opusDecoder;  //access guarded by wsmutex
-BufferRTOS<uint8_t> audioBuffer(AUDIO_BUFFER_SIZE, AUDIO_CHUNK_SIZE);
+BufferRTOS<uint8_t> audioBuffer(AUDIO_BUFFER_SIZE, AUDIO_CHUNK_SIZE);  //producer: networkTask, consumer: audioStreamTask. Thread safe in single producer->single consumer scenario.
-I2SStream i2s; 
+I2SStream i2s; //access from audioStreamTask only
-VolumeStream volume(i2s);
+VolumeStream volume(i2s); //access from audioStreamTask only
-QueueStream<uint8_t> queue(audioBuffer);
+QueueStream<uint8_t> 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;
-        deviceState = SPEAKING;
+    digitalWrite(I2S_SD_OUT, HIGH);
-        digitalWrite(I2S_SD_OUT, HIGH);
+    speakingStartTime = millis();
-        speakingStartTime = millis();
+    
-        
+    webSocket.enableHeartbeat(30000, 15000, 3);
-        webSocket.enableHeartbeat(30000, 15000, 3);
-        xSemaphoreGive(wsMutex);
-    }
     
     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
+    i2sInputFlushScheduled = true;
-    i2s.flush();
+    outputFlushScheduled = true;
-    volume.flush();
-    queue.flush();
-    i2sInput.flush();
-    audioBuffer.reset();    
 
     Serial.println("Transitioned to listening mode");
 
-    if (xSemaphoreTake(wsMutex, pdMS_TO_TICKS(100)) == pdTRUE) {
+    deviceState = LISTENING;
-        deviceState = LISTENING;
+    digitalWrite(I2S_SD_OUT, LOW);
-        digitalWrite(I2S_SD_OUT, LOW);
+    webSocket.disableHeartbeat();
-        webSocket.disableHeartbeat();
-        xSemaphoreGive(wsMutex);
-    }
 }
 
+// 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) {
+        xSemaphoreTake(wsMutex, portMAX_DELAY);
-            webSocket.sendBIN(&b, 1);
+        webSocket.sendBIN(&b, 1);
-            xSemaphoreGive(wsMutex);
+        xSemaphoreGive(wsMutex);
-            return 1;
-        }
-        
         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;
         }
         
-        
+        xSemaphoreTake(wsMutex, portMAX_DELAY);
-        if (xSemaphoreTake(wsMutex, pdMS_TO_TICKS(100)) == pdTRUE) {
+        webSocket.sendBIN(buffer, size);
-            webSocket.sendBIN(buffer, size);
+        xSemaphoreGive(wsMutex);
-            xSemaphoreGive(wsMutex);
-            return size;
-        }
-
         return size;
     }
 };
 
-WebsocketStream wsStream;
+WebsocketStream wsStream; //guard with wsMutex
-I2SStream i2sInput;
+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 ( i2sInputFlushScheduled ) {
-        if (scheduleListeningRestart && millis() >= scheduledTime) {
+            i2sInputFlushScheduled = false;
-            transitionToListening();
+            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<int>();
-                    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);
     }
 }
-

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<uint8_t> 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);

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;

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;

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)