Unable to stop ICYStream in a multithreaded context

[vladkorotnev]

I have a setup where I'm using 3 tasks for an internet radio:

1. Net task: downloads using ICYStream into a buffer in PS-RAM
2. Codec task: decodes using libhelix from the PS-RAM net buffer into a PCM buffer in DRAM
3. Sound task: provides bits of the PCM buffer into the I2S output

This allows a highly resilient buffering pipeline which can survive several multi-second network dropouts before it stutters. However I am failing to stop the pipeline due to a crash in the ICYStream.end() method.

Here is how I'm starting and stopping the stream:

void start(const std::string url, std::function<void(bool)> loadingCallback) {
            running = true;
            loadingCallback(true);
            ESP_LOGI(LOG_TAG, "Streamer is starting");
            queueNetData.begin();
            queuePcmData.begin();
            netTask.begin([this, url, loadingCallback]() { 
                    TickType_t last_shart_time = xTaskGetTickCount();
                    urlStream.setMetadataCallback(_update_meta_global);
                    urlStream.httpRequest().setTimeout(NET_CLIENT_TIMEOUT);
                    urlStream.httpRequest().header().setProtocol("HTTP/1.0"); // <- important, because chunked transfer of some servers seems to cause trouble with buffering!
                    urlStream.begin(url.c_str());
                    ESP_LOGI(LOG_TAG, "Streamer did begin URL");

                    const char * _srv_mime = urlStream.httpRequest().reply().get("Content-Type");
                    if(_srv_mime != nullptr) {
                        const std::string srv_mime = _srv_mime;
                        ESP_LOGI(LOG_TAG, "Server reports content-type: %s", srv_mime.c_str());
                        if(srv_mime == "audio/mpeg" || srv_mime == "audio/mp3") {
                            activeCodec = new MP3DecoderHelix();
                        }
                        else if(srv_mime == "audio/aac") {
                            activeCodec = new AACDecoderHelix();
                        }
                        else {
                            ESP_LOGE(LOG_TAG, "Content-type is not supported, TODO show error");
                            loadingCallback(false);
                            vTaskDelay(portMAX_DELAY);
                        }
                        decoder.setDecoder(activeCodec);
                        activeCodec->setNotifyActive(true);
                        activeCodec->addNotifyAudioChange(*outPort);
                        decoder.setOutput(queuePcmData);
                    } else {
                        ESP_LOGE(LOG_TAG, "No content-type reported, TODO show error");
                        loadingCallback(false);
                        vTaskDelay(portMAX_DELAY);
                    }
                    decoder.setNotifyActive(true);
                    decoder.begin();
                    ESP_LOGI(LOG_TAG, "Streamer did begin decoder");
    
                    TickType_t last_successful_copy = xTaskGetTickCount();
                    TickType_t last_stats = xTaskGetTickCount();

                    codecTask.begin([this]() { 
                        while(bufferNetData.levelPercent() < NET_DATA_ENOUGH_MARK_PCT && running) { 
                            delay(125);
                        }
                        
                        if(!running) {
                            ESP_LOGI(LOG_TAG, "Finishing up codec task early");
                            decoder.end();
                            xSemaphoreGive(semaCodec);
                            vTaskDelay(portMAX_DELAY);
                        }
    
                        while(copierDecoding.copy() > 0 && !bufferNetData.isEmpty() && running) {
                            delay(10);
                        }
                        while(bufferNetData.levelPercent() < NET_DATA_ENOUGH_MARK_PCT && running) delay(250);

    
                        if(!running) {
                            ESP_LOGI(LOG_TAG, "Finishing up codec task");
                            decoder.end();
                            xSemaphoreGive(semaCodec);
                            vTaskDelay(portMAX_DELAY);
                        }

                        delay(5);
                        xSemaphoreGive(semaCodec);
                    });

                    sndTask.begin([this, loadingCallback]() { 
                        xSemaphoreTake(semaSnd, portMAX_DELAY);
                        if(!copierPlaying.copy() && running) {
                            ESP_LOGD(LOG_TAG, "copierPlaying underrun!? PCM buffer %.00f%%, net buffer %.00f%%", bufferPcmData.levelPercent(), bufferNetData.levelPercent()); 
                            loadingCallback(true);
                            while(bufferPcmData.isEmpty() && running) { 
                                xSemaphoreGive(semaSnd);
                                delay(100);
                                xSemaphoreTake(semaSnd, portMAX_DELAY);
                            }
                            loadingCallback(false);
                        }
                        xSemaphoreGive(semaSnd);
                        delay(2); 
                        if(!running) { 
                            ESP_LOGI(LOG_TAG, "Finishing up sound task");
                            vTaskDelay(portMAX_DELAY);
                        }
                    });

                    while(running) {
                        xSemaphoreTake(semaNet, portMAX_DELAY);
                        TickType_t now = xTaskGetTickCount();
                        int copied = copierDownloading.copy();
                        if(copied > 0) {
                            last_successful_copy = now;
                            delay(2);
                        } else if(now - last_successful_copy >= pdTICKS_TO_MS(NET_NO_DATA_TIMEOUT)  && (bufferNetData.levelPercent() < 20.0 || bufferPcmData.isEmpty())) {
                            ESP_LOGE(LOG_TAG, "streamer is stalled!? time since last shart = [ %i ms ]", now - last_shart_time);
                            last_shart_time = now;
                            loadingCallback(true);
                            urlStream.end();
                            int retryDelay = 100;
                            do {
                                if(bufferPcmData.isEmpty()) bufferNetData.clear(); // prevent click artifact on restart playback
                                delay(retryDelay);
                                retryDelay += 1000;
                                if(retryDelay > NET_CLIENT_TIMEOUT) retryDelay = NET_CLIENT_TIMEOUT;
                                ESP_LOGE(LOG_TAG, "streamer is trying to begin URL again");
                            } while(!urlStream.begin(url.c_str()));
                            last_successful_copy = now;
                        } else if(copied < 0) {
                            ESP_LOGE(LOG_TAG, "copied = %i ???", copied);
                        } else if(bufferNetData.isFull()) {
                            ESP_LOGW(LOG_TAG, "Net buffer full, no space to receive more data!!");
                            delay(10);
                        } else {
                            // copied nothing, maybe yield to another task
                            delay(10);
                        }
                        if(now - last_stats >= pdTICKS_TO_MS(LOG_STATS_INTERVAL)) {
                            ESP_LOGI(LOG_TAG, "Stats: PCM buffer %.00f%%, net buffer %.00f%%", bufferPcmData.levelPercent(), bufferNetData.levelPercent()); 
                            last_stats = now;
                        }

                        xSemaphoreGive(semaNet);
                    }
                    ESP_LOGI(LOG_TAG, "Finishing up net task");
                    vTaskDelay(portMAX_DELAY);
                }
            );
        }
    
        void stop() {
            NullStream tmp;
            decoder.setOutput(tmp);
            if(activeCodec != nullptr) {
                activeCodec->setOutput(tmp);
            }

            running = false;
            xSemaphoreTake(semaNet, portMAX_DELAY);
            ESP_LOGI(LOG_TAG, "Streamer finalizing net task");
            netTask.end();
    
            ESP_LOGI(LOG_TAG, "Streamer finalizing sound task");
            xSemaphoreTake(semaSnd, portMAX_DELAY);
            sndTask.end();
    
            // Small shitshow because the codec thread sometimes gets stuck and never exits and leaves the semaphore
            ESP_LOGI(LOG_TAG, "Streamer finalizing codec task");
            bool codecNotStuck = xSemaphoreTake(semaCodec, pdMS_TO_TICKS(1000));
            codecTask.end();

            if(!codecNotStuck) { 
                delay(1000); // give it time to unshit itself to prevent explosions 
            }
            if(activeCodec != nullptr) {
                activeCodec->setOutput(tmp);
                delete activeCodec;
                activeCodec = nullptr;
            }
        }

As you can see, everything is locked by mutexes, and nothing is released until the respective task's mutex has been taken.

This method is then called in the destructor for the pipeline, to prevent any object being deleted while the task is still running:

        ~StreamingPipeline() {
            if(running) stop();
    
            vSemaphoreDelete(semaSnd);
            vSemaphoreDelete(semaNet);
            vSemaphoreDelete(semaCodec);
        }

However, it ultimately leaves me with a nullpointerexception:

Guru Meditation Error: Core  0 panic'ed (InstrFetchProhibited). Exception was unhandled.

Core  0 register dump:
PC      : 0x00000000  PS      : 0x00060130  A0      : 0x800ecd60  A1      : 0x3ffda420
A2      : 0x00000000  A3      : 0x3f817798  A4      : 0x40277c08  A5      : 0x3f81a418  
A6      : 0x007bf168  A7      : 0x003fffff  A8      : 0x80277c68  A9      : 0x3ffda400
A10     : 0x3ffb1d9c  A11     : 0x00000000  A12     : 0x00000000  A13     : 0x0000ff00  
A14     : 0x00ff0000  A15     : 0xff000000  SAR     : 0x0000000a  EXCCAUSE: 0x00000014
EXCVADDR: 0x00000000  LBEG    : 0x40001609  LEND    : 0x4000160d  LCOUNT  : 0x00000000  


Backtrace: 0xfffffffd:0x3ffda420 0x400ecd5d:0x3ffda440 0x400f2d2c:0x3ffda460 0x400f3156:0x3ffda480 0x400f32fa:0x3ffda4a0 0x400f7158:0x3ffda4c0 0x400f752e:0x3ffda5b0 0x400de332:0x3ffda5d0 0x40189650:0x3ffda7c0
  #0  0x400ecd5d in audio_tools::HttpRequest::end() at .pio/libdeps/OTA/ArduinoAudioTools/src/AudioTools/CoreAudio/AudioHttp/HttpRequest.h:63
  #1  0x400f2d2c in audio_tools::HttpRequest::~HttpRequest() at .pio/libdeps/OTA/ArduinoAudioTools/src/AudioTools/CoreAudio/AudioHttp/HttpRequest.h:32
      (inlined by) audio_tools::URLStream::~URLStream() at .pio/libdeps/OTA/ArduinoAudioTools/src/AudioTools/CoreAudio/AudioHttp/URLStream.h:49
  #2  0x400f3156 in audio_tools::ICYStream::~ICYStream() at .pio/libdeps/OTA/ArduinoAudioTools/src/AudioTools/CoreAudio/AudioHttp/ICYStream.h:22
      (inlined by) InternetRadioMode::StreamingPipeline::~StreamingPipeline() at src/mode/netradio/pipeline.h:77
  #3  0x400f32fa in InternetRadioMode::stop() at src/mode/netradio_mode.cpp:253 (discriminator 1)
  #4  0x400f7158 in InternetRadioMode::select_station(int) at src/mode/netradio_mode.cpp:196
  #5  0x400f752e in InternetRadioMode::on_remote_key_pressed(VirtualKey) at src/mode/netradio_mode.cpp:299
  #6  0x400de332 in ModeHost::loop() at include/mode_host.h:138
      (inlined by) loop() at src/main.cpp:155
  #7  0x40189650 in loopTask(void*) at C:/Users/akasaka/.platformio/packages/framework-arduinoespressif32/cores/esp32/main.cpp:50

It appears as if the HttpRequest is being ended twice or something. If I remove the call to end() from HttpRequest's destructor, this crash doesn't happen anymore — and, frankly, I do not observe any memory leaks either.

Does HttpRequest create some sort of object that cannot be accessed cross-thread even if locked behind a mutex? Or am I missing something simple?

[pschatzmann]

Not sure what is going wrong: the following test sketch is working perfectly

#include "AudioTools.h"

void setup(){
  Serial.begin(115200);
  AudioToolsLogger.begin(Serial, AudioToolsLogLevel::Info);  
}

void loop() {
  Serial.begin(115200);
  Serial.println("---> loop:");
  ICYStream *p_url = new ICYStream("ssid","pwd");  // or replace with ICYStream to get metadata
  if (!p_url->begin("https://stream.srg-ssr.ch/m/rsj/mp3_128","audio/mp3")){
    stop();
  }
  p_url->end();
  delete p_url;

}

I have the following comments though:

• on microcontrollers avoid to allocate objects dynamically: the best is to allocate them once in the setup or better as global data. You are running the risk of heap fragmentation which leads to issues which are impossible to find.
• instead of calling xSemaphoreTake and xSemaphoreGive use my LockGuard and let RAII do the management.

[vladkorotnev]

If I stop it before the net buffer has been filled at least partially, it doesn't explode, so it might be something related to the buffers down the line. Just in case, am I correct that end()'ing a single item in the stream chain doesn't implicitly end() anything connected to it?

I'll look into the LockGuard and how it works, and whether I can reuse an existing pipeline. Either way I need to be able to deallocate it when the user exits the Radio app to switch to another source (keeping them all in memory at once is somewhat tough).

P.S. Watch out for the WiFi password when copying code...