How does ackLater() actually work?

[the-programmer]

In my current project I need to stream data from an FTP server to the AsyncWebserver with a chunked response (since I don't know the size in advance). So, my ESP32 is working as a client to the FTP server and as a server to the browser. (The requirement here is that the user should be able to download files from the FTP server wirelessly and there is no other network available).
The download from the FTP server (in PASV mode) (to ram/serial console) works as intended.

Since the file is bigger than the RAM of the ESP32, i'm looking for a way to stream it.

So, now I'm writing the data into a buffer (in the onData callback) and reading/writing it to the webserver (in the beginChunkedResponse callback).

For some reason, the readout of the webserver is slower than the writing of the FTP server. So, my buffer fills up to the point of overflow. I try to mitigate this by calling client->ackLater(); once the buffer is full. However even after calling that, the data is still being pushed in. (The buffer is a std::vector<uint8_t> _buffer; and I have some space extra, so nothing get's lost).

After that, the webserver does read the data (although not all the time) and the data get's pulled out slowly.

In the onPoll callback I'm checking if ackLater() was set, and if it was, if I have space in my buffer. If I do have space in the buffer, I make the following call client->ack(freeSpace); (but this is a pure guess).

I would like to add that I have the feeling ackLater() also influences the response of the webserver. However I don't have proof for this.

---

Small note to add: With wireshark I can see that the data is still being 'acked' even if I call ackLater() on the 1st onData callback.

[mathieucarbou]

@me-no-dev @vortigont : if you can help ? thanks!

[mathieucarbou]

ackLater() has to be called from onData callback and is a way to skip the ack that AsyncTCP will send after the onData callback is called.

    } else {
      if (_recv_cb) {
        _recv_cb(_recv_cb_arg, this, b->payload, b->len);
      }
      if (!_ack_pcb) {
        _rx_ack_len += b->len;
      } else if (_pcb) {
        _tcp_recved(_pcb, _closed_slot, b->len);
      }
    }

if ackLater() is called within onData callback, _ack_pcb is false then _tcp_recved is skipped.

[mathieucarbou]

Would you be able to show us a MRE ? IT is not clear to me what ESPAsyncWebServer (http) does in the context of commmunication with a FTP server from AsyncTCP client...

[the-programmer]

I'll see what I can do.
Do you only need the FTP Client or the entire setup?

[mathieucarbou]

Just the AsyncTCP client I guess: what do you do in the onData ? Just accumulating on a buffer ? or streaming it to websocket, SSE or something like that ?

Basically, would be nice to see what you do in the onData.

[the-programmer]

Ok, here is just the onData callback on the PASV socket. In the current code it is wrapped into another callback. But for ease of reading, I just flattened everything.

//In the constructor
    _dataClient.onData([this](void* obj, AsyncClient* client, void* data, size_t len) { 
        log_v("Got data onData");
        onDataData(client, (uint8_t*)data, len); 
    });

//The (flat) code.
void ESP32AsyncFtpClient::onDataData(AsyncClient* client, const uint8_t* data, size_t len)
{
    if (len == 0)
        return;

    bool bufferFull = false;

    std::lock_guard<std::mutex> lock(_bufferMutex);
            
    // Add data to the end of our buffer
    _buffer.insert(_buffer.end(), data, data + length);

    // did we exeed the buffer limit
    if (_buffer.size() > _maxBufferSize) 
    {
        // Signal to pause data flow
        _flowControlPaused = true;
        bufferFull = true;
    }

    if(bufferFull)
    {
        log_d("Buffer full, ack later");
        client->ackLater();
    }
}

But even if I change my OnDataData to this

void ESP32AsyncFtpClient::onDataData(AsyncClient* client, const uint8_t* data, size_t len)
{
    if (len == 0)
        return;

    client->ackLater();
}

Wireshark still shows the following.

A quick rundown of the packets:
38 - 40 Create secondary (PASV/data) connection
40 Request file
42 - 43 Acknowledge the Request (and confirm that it will come via the data connection
44 - 45 Data blocks 1 and 2
46 Acknowledgement of the first 2 blocks (note that ackLater() should already have been called)
47 - 48 Data blocks 3 and 4
49 Another ACK, but now no more data

The file can also be found here Packet dump.pcapng.gz (note that the numbers can be a bit off due to filtering when exporting)

[mathieucarbou]

Thanks for these details!
I am wondering if ackLater() correctly works or is something to use in your case....
I don't know why it was added and me-no-dev does not remember either...

If I understand your use case, you are trying to "slow down" the reading in order to let time for an asynchronous task on the background to process the queue when it is full, otherwise keep reading would fill the MCU RAM and crash ?

[mathieucarbou]

yeah that's your vector that is not able to reallocate.

you might want to

1. reserve a fixed size and not go further to avoid reallocation
2. maybe switch to a ring buffer like cbuf ?

[the-programmer]

That isn't the issue.

The problem is that my FTP RX thread is pushing in data, and the webserver isn't pulling out data fast enough.

From the logfile. This is the normal behavior:

[ 19476][D][ESP32AsyncFtpClient.cpp:223] handleResponse(): FTP <<< 125 Data connection already open
[ 19487][D][ESP32AsyncFtpClient.cpp:412] setState(): FTP state: WAIT_COMMAND_RESPONSE -> RECEIVING_DATA
[ 19499][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 19508][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 19518][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[...]
[ 19887][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd03e9 with max len 5752 and index 0, wrote 5176
[...]
[ 19925][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 19938][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd095e with max len 5752 and index 5176, wrote 5176
[...]
[ 19977][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 19990][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd037a with max len 5752 and index 10352, wrote 5176
[...]

[...]
[ 20862][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5752 and index 109296, wrote 5176
[ 20874][D][LocalWebserver.cpp:47] operator()(): Got data of len 501 and event 0
[ 20882][W][ESP32AsyncFtpClient.cpp:519] onDataDisconnect(): Data connection closed
[ 20892][D][ESP32AsyncFtpClient.cpp:412] setState(): FTP state: RECEIVING_DATA -> WAIT_TRANSFER_COMPLETE
[ 20903][V][ESP32AsyncFtpClient.cpp:464] onControlData(): Got control data with len 23 and content '226 Transfer complete'
[ 20917][D][ESP32AsyncFtpClient.cpp:223] handleResponse(): FTP <<< 226 Transfer complete
[ 20926][D][LocalWebserver.cpp:47] operator()(): Got data of len 0 and event 1
[ 20938][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5752 and index 114472, wrote 5176
[ 20953][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 3440 and index 119648, wrote 3096
[ 20967][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5520 and index 122744, wrote 4968
[ 20980][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5084 and index 127712, wrote 4575
[ 20994][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd1c1a with max len 5477 and index 132287, wrote 4929
[ 21007][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0826 with max len 5123 and index 137216, wrote 4610
[ 21020][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5442 and index 141826, wrote 4897
[ 21033][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5155 and index 146723, wrote 4639
[ 21045][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd1c62 with max len 5413 and index 151362, wrote 4871
[ 21057][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5181 and index 156233, wrote 792
[ 21070][V][LocalWebserver.cpp:68] operator()(): Can write to 0x3ffd0216 with max len 5752 and index 157025, wrote 0

Line 47 is writing to the buffer and Line 68 is reading from the buffer.
However on the 3rd attempt you have this:

[ 39402][D][ESP32AsyncFtpClient.cpp:223] handleResponse(): FTP <<< 125 Data connection already open
[ 39413][D][ESP32AsyncFtpClient.cpp:412] setState(): FTP state: WAIT_COMMAND_RESPONSE -> RECEIVING_DATA
[ 39424][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 39434][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 39443][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 39453][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 39462][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[ 39471][D][LocalWebserver.cpp:47] operator()(): Got data of len 1436 and event 0
[...]
abort() was called at PC 0x40149b8f on core 1 [Allocator failed]

On this attempt only line 47 is logging data into the buffer until the RAM is full (resulting in a crash).

To prevent this a form "backpressure" is required (eg. tell the FTP server to pause data transmission) so the buffer can be read by the chunked response callback.

This is the reason for my original question. How to use ackLater() so the FTP server pauses transmission.

[mathieucarbou]

@the-programmer : ack are sent AFTER the onData callback is called.
My point was to do some sort of tunneling from within the onData cb: you receive the data, you send it, and wait for it to be sent (or buffered).
It would be far easier to do that using binary / text websocket though because the nature of a request (polling and not pushing) is not easy to adapt.

[the-programmer]

Since the file has to be downloaded on the client side, I think i'm going to keep the setup "as is" (with flash buffering).

However it might be an idea to remove/deprecate ackLater() since it doesn't seem to work as intended. Just do future users know what to expect.

Since the answer also seems to be "don't use ackLater(), it doesn't work as intended". I'm going to mark the start of this thread as the answer. But thanks for thinking along.

[mathieucarbou]

However it might be an idea to remove/deprecate ackLater() since it doesn't seem to work as intended. Just do future users know what to expect.

I agree. This thing is weird.

[mathieucarbou]

@the-programmer I dig a little to see the issues encountered with a tunnel:

So I did the following code:

// SPDX-License-Identifier: LGPL-3.0-or-later
// Copyright 2016-2025 Hristo Gochkov, Mathieu Carbou, Emil Muratov

//
// Shows how to trigger an async client request from a browser request and send the client response back to the browser
//

#include <Arduino.h>
#ifdef ESP32
#include <AsyncTCP.h>
#include <WiFi.h>
#elif defined(ESP8266)
#include <ESP8266WiFi.h>
#include <ESPAsyncTCP.h>
#elif defined(TARGET_RP2040) || defined(TARGET_RP2350) || defined(PICO_RP2040) || defined(PICO_RP2350)
#include <RPAsyncTCP.h>
#include <WiFi.h>
#endif

#include <ESPAsyncWebServer.h>
#include <cbuf.h>

#define WIFI_SSID     "IoT"
#define WIFI_PASSWORD ""

static AsyncWebServer server(80);

void setup() {
  Serial.begin(115200);

#ifndef CONFIG_IDF_TARGET_ESP32H2
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
  }
  Serial.println("Connected to WiFi!");
  Serial.println(WiFi.localIP());
#endif

  //
  // HOW TO RUN THIS EXAMPLE: curl -v http://192.168.125.127?host=www.google.com
  //
  server.on("/", HTTP_GET, [](AsyncWebServerRequest *request) {
    const AsyncWebParameter *param = request->getParam("host");

    if (!param || !param->value().length()) {
      request->send(400, "text/plain", "Missing host parameter");
      return;
    }

    AsyncClient *client = new AsyncClient();
    cbuf *buffer = new cbuf(1024);
    request->_tempObject = buffer;

    // when client disconnects, we clear the buffer which signals the end of the chunk request
    // note: onDisconnect also called on error
    client->onDisconnect([request](void *arg, AsyncClient *client) {
      Serial.printf("Tunnel disconnected!\n");
      cbuf *content = (cbuf *)request->_tempObject;
      delete content;
      request->_tempObject = nullptr;
    });

    // register a callback when an error occurs
    client->onError([request](void *arg, AsyncClient *client, int8_t error) {
      Serial.printf("Tunnel error: %s\n", client->errorToString(error));
    });

    // register a callback when data arrives, to accumulate it
    client->onData([request](void *arg, AsyncClient *client, void *data, size_t len) {
      Serial.printf("Tunnel received %u bytes...\n", len);
      size_t total = 0;
      while (total < len) {
        cbuf *content = (cbuf *)request->_tempObject;
        if (content) {
          if (content->full()) {
            Serial.printf("Buffer is full: waiting for more space...\n");
            // IMPORTANT NOTE: a normal request being a poll model, it is not suited with this async design.
            // To correctly implement such tunnel, WebSocket should be used and onData should send websocket frames.
            delay(100);
          } else {
            const size_t written = content->write((const char *)data, len);
            Serial.printf("Wrote %u bytes to buffer\n", written);
            total += written;
          }
        } else {
          Serial.printf("Buffer is null: ignoring data\n");
          break;
        }
      }
    });

    // register a callback when we are connected
    client->onConnect([request](void *arg, AsyncClient *client) {
      Serial.printf("HTTP Handshake...\n");

      // send request
      client->write("GET / HTTP/1.1\r\n");
      client->write("Host: ");
      client->write(request->getParam("host")->value().c_str());
      client->write("\r\n");
      client->write("User-Agent: ESP32\r\n");
      client->write("Connection: close\r\n");
      client->write("\r\n");
    });

    Serial.printf("Fetching: http://%s...\n", param->value().c_str());

    if (client->connect(param->value().c_str(), 80)) {
      Serial.printf("Tunnel connected!\n");

      AsyncWebServerResponse *response = request->beginChunkedResponse("text/plain", [request](uint8_t *buffer, size_t maxLen, size_t index) -> size_t {
        // are we done ?
        if (request->_tempObject == nullptr) {
          return 0;
        }

        cbuf *content = (cbuf *)request->_tempObject;
        const size_t avail = content->available();

        // no data ? wait...
        if (!avail) {
          return RESPONSE_TRY_AGAIN;
        }

        Serial.printf("Available in buffer: %u bytes\n", avail);

        const size_t len = maxLen < avail ? maxLen : avail;
        const size_t read = content->read((char *)buffer, len);

        Serial.printf("Read %u bytes from buffer\n", read);
        Serial.printf("%.*s\n", read, buffer);

        return read;
      });

      request->send(response);

    } else {
      Serial.printf("Failed to open tunnel!\n");
      cbuf *content = (cbuf *)request->_tempObject;
      delete client;
      delete content;
      request->_tempObject = nullptr;
      request->send(500, "text/plain", "Failed to open tunnel");
    }
  });

  server.begin();
}

void loop() {
  delay(100);
}

And actually, many issues prevent it to work:

1. A request is a poll model, not push, so it is hard to use it. If using websocket, the complete data received in onData() cb could be sent to a websocket frame. A little like you do with the FS.

2. onData executes in the async_tcp task which CANNOT be blocked (or TWDT triggers). So there is no way to wait. There is also no throttling in place for the onData. So if onData blocks, async_tcp task is not able to process anymore work.

3. If the buffer (in the case above) is lower (1024) than the incoming data, issue 2 happens, where we need to wait for the data to be sent before processing more and this is not possible within onData.

4. ack later expects ALL the data to be processed (_rx_ack_len += b->len;). There is no way to has onData callback called again with the non-processed data.

      if (!_ack_pcb) {
        _rx_ack_len += b->len;
      } else if (_pcb) {
        _tcp_recved(_pcb, _closed_slot, b->len);
      }

Then I crated another example which works fine with websocket:

// SPDX-License-Identifier: LGPL-3.0-or-later
// Copyright 2016-2025 Hristo Gochkov, Mathieu Carbou, Emil Muratov

//
// Shows how to trigger an async client request from a browser request and send the client response back to the browser through websocket
//

#include <Arduino.h>
#ifdef ESP32
#include <AsyncTCP.h>
#include <WiFi.h>
#elif defined(ESP8266)
#include <ESP8266WiFi.h>
#include <ESPAsyncTCP.h>
#elif defined(TARGET_RP2040) || defined(TARGET_RP2350) || defined(PICO_RP2040) || defined(PICO_RP2350)
#include <RPAsyncTCP.h>
#include <WiFi.h>
#endif

#include <ESPAsyncWebServer.h>

#define WIFI_SSID     "IoT"
#define WIFI_PASSWORD ""

static AsyncWebServer server(80);
static AsyncWebSocketMessageHandler wsHandler;
static AsyncWebSocket ws("/ws", wsHandler.eventHandler());

void setup() {
  Serial.begin(115200);

#ifndef CONFIG_IDF_TARGET_ESP32H2
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
  }
  Serial.println("Connected to WiFi!");
  Serial.println(WiFi.localIP());
#endif

  //
  // HOW TO RUN THIS EXAMPLE:
  // 1. websocat --binary ws://192.168.125.127/ws
  // 2. curl -v "http://192.168.125.127?host=www.google.com&port=80"
  //
  server.on("/", HTTP_GET, [](AsyncWebServerRequest *request) {
    if (!request->getParam("host") || !request->getParam("host")->value().length()) {
      request->send(400, "text/plain", "Missing host parameter");
      return;
    }

    if (!request->getParam("port") || !request->getParam("port")->value().length()) {
      request->send(400, "text/plain", "Missing port parameter");
      return;
    }

    const String host = request->getParam("host")->value();
    const String port = request->getParam("port")->value();
    AsyncClient *client = new AsyncClient();

    // when client disconnects, we clear the buffer which signals the end of the chunk request
    // note: onDisconnect also called on error
    client->onDisconnect([](void *arg, AsyncClient *client) {
      Serial.printf("Tunnel disconnected!\n");
      delete client;
    });

    // register a callback when an error occurs
    client->onError([](void *arg, AsyncClient *client, int8_t error) {
      Serial.printf("Tunnel error: %s\n", client->errorToString(error));
    });

    // register a callback when data arrives, to accumulate it
    client->onData([](void *arg, AsyncClient *client, void *data, size_t len) {
      Serial.printf("Tunnel received %u bytes...\n", len);
      ws.binaryAll(static_cast<const uint8_t *>(data), len);
    });

    client->onConnect([host, port](void *arg, AsyncClient *client) {
      Serial.printf("Tunnel connected!\n");

      client->write("GET / HTTP/1.1\r\n");
      client->write("Host: ");
      client->write(host.c_str());
      client->write(":");
      client->write(port.c_str());
      client->write("\r\n");
      client->write("User-Agent: ESP32\r\n");
      client->write("Accept: */*\r\n");
      client->write("Connection: close\r\n");
      client->write("\r\n");
    });

    Serial.printf("Fetching: http://%s:%s...\n", host.c_str(), port.c_str());

    if (!client->connect(host.c_str(), port.toInt())) {
      Serial.printf("Failed to open tunnel!\n");
      delete client;
      request->send(500, "text/plain", "Failed to open tunnel");
    } else {
      request->send(200, "text/plain", "Tunnel opened, data will be sent through websocket\n");
    }
  });

  wsHandler.onConnect([](AsyncWebSocket *server, AsyncWebSocketClient *client) {
    Serial.printf("Client %" PRIu32 " connected\n", client->id());
    server->textAll("New client: " + String(client->id()));
  });

  wsHandler.onDisconnect([](AsyncWebSocket *server, uint32_t clientId) {
    Serial.printf("Client %" PRIu32 " disconnected\n", clientId);
    server->textAll("Client " + String(clientId) + " disconnected");
  });

  server.addHandler(&ws);
  server.begin();
}

void loop() {
  delay(100);
}

[mathieucarbou]

if you don't want to revert the API change, we should add some clarifying documentation on how to use ackPacket safely.

@willmmiles this is really an advanced usage of the lib only a few people are able to correctly achieve, so I do not have nay string opinion on that myself regarding what is better.

I am not sure @TienHuyIoT though as deep as you on the implications of his PR....

@TienHuyIoT @me-no-dev @vortigont : do you have some opinion ?

I have created PR #59 in case we want to rollback.

[me-no-dev]

My opinion is that the feature is now broken. Just like @willmmiles is saying it. It should be fixed back, unless @TienHuyIoT gives a good reason

[mathieucarbou]

AsyncTCP v3.4.1 released with the fix

[mathieucarbou]

PR with the AsyncTunnel example updated (ESP32Async/ESPAsyncWebServer#171) and works fine without memory leak ;-)

@the-programmer : this is the concept you are looking for. AsyncTunnel example works with WebSocket but you can make it work with a normal Http request and chunk callback the same way I think.

[TienHuyIoT]

Hi all,
Sorry for my late feedback!
This above investigation was very interesting. It helped correct a misunderstanding I had during the PR review.

@willmmiles, Thank you for clearly explaining the motivation behind the use of the onPacket() callback.
@me-no-dev @mathieucarbou To be honest It very happy that the fantastic AsyncTCP library is continuously maintainable.
@the-programmer Thanks for the analysis and contribution.