From 4e0f94ae35e7f8541367534800cce89798173aca Mon Sep 17 00:00:00 2001 From: Atomic Deploy Date: Mon, 19 Oct 2026 02:59:55 +0330 Subject: [PATCH] Preserve final response bytes through TCP backpressure --- src/WebResponseImpl.h | 2 + src/WebResponses.cpp | 47 ++++++----- tests/response_backpressure.cpp | 126 ++++++++++++++++++++++++++++ tests/run-response-backpressure.mjs | 37 ++++++++ 4 files changed, 193 insertions(+), 19 deletions(-) create mode 100644 tests/response_backpressure.cpp create mode 100644 tests/run-response-backpressure.mjs diff --git a/src/WebResponseImpl.h b/src/WebResponseImpl.h index 7f0d25ac6..a0da1ec79 100644 --- a/src/WebResponseImpl.h +++ b/src/WebResponseImpl.h @@ -82,6 +82,8 @@ class AsyncAbstractResponse : public AsyncWebServerResponse { std::unique_ptr > _send_buffer; // buffer data size specifiers size_t _send_buffer_offset{0}, _send_buffer_len{0}; + // Source EOF does not mean that the final buffer has reached the TCP queue. + bool _source_ended{false}; size_t _readDataFromCacheOrContent(uint8_t *data, const size_t len); size_t _fillBufferAndProcessTemplates(uint8_t *buf, size_t maxLen); diff --git a/src/WebResponses.cpp b/src/WebResponses.cpp index 83d3b7db8..772c56eff 100644 --- a/src/WebResponses.cpp +++ b/src/WebResponses.cpp @@ -351,6 +351,14 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s return 0; } + _ackedLength += len; + if (_state == RESPONSE_WAIT_ACK) { + if (_ackedLength >= _writtenLength) { + _state = RESPONSE_END; + } + return 0; + } + #if ASYNCWEBSERVER_USE_CHUNK_INFLIGHT /* for response payloads with unknown length or length larger than TCP_WND we need to control AsyncTCP's queue and in-flight fragmentation. @@ -378,9 +386,6 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s } #endif - // this is not functionally needed in AsyncAbstractResponse itself, but kept for compatibility if some of the derived classes are rely on it somehow - _ackedLength += len; - size_t payloadlen{0}; // amount of data to be written to tcp sockbuff during this call, used as return value of this method // send http headers first @@ -392,11 +397,14 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s if (_writtenHeadersLength < _assembled_headers.length()) { // we were not able to fit all headers in current buff, send this part here and return later for the rest #if ASYNCWEBSERVER_USE_CHUNK_INFLIGHT - _in_flight += pcb_written; - --_in_flight_credit; // take a credit + if (pcb_written) { + _in_flight += pcb_written; + --_in_flight_credit; // take a credit + } #endif - if (!request->client()->send()) { + if (pcb_written && !request->client()->send()) { // something is wrong, what should we do here? + _state = RESPONSE_FAILED; request->client()->close(); return 0; } @@ -415,6 +423,8 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s // data is pending in buffer from a previous call or previous iteration size_t const added_len = request->client()->add(reinterpret_cast(_send_buffer->data() + _send_buffer_offset), _send_buffer_len - _send_buffer_offset); + payloadlen += added_len; + _writtenLength += added_len; if (added_len != _send_buffer_len - _send_buffer_offset) { // we were not able to add entire buffer's content to tcp buffs, leave it for later // (this should not happen normally unless connection's TCP window suddenly changed from remote or mem pressure) @@ -423,11 +433,10 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s } else { _send_buffer_len = _send_buffer_offset = 0; // consider buffer empty } - payloadlen += added_len; } auto tcp_win = request->client()->space(); - if (tcp_win == 0 || _state == RESPONSE_END) { + if (tcp_win == 0 || _source_ended) { break; // no room left or no more data } @@ -467,7 +476,7 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s _sentLength += readLen; // data is not sent yet, but we won't get a chance to count this later properly for chunked data if (!readLen) { // last chunk? - _state = RESPONSE_END; + _source_ended = true; } } } else { @@ -483,21 +492,24 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s if (readLen == 0) { // no more data to send - _state = RESPONSE_END; + _source_ended = true; } else if (readLen != RESPONSE_TRY_AGAIN) { _send_buffer_len += readLen; // set buffers's size to match added data _sentLength += readLen; // data is not sent yet, but we need it to understand that it would be last block if (_sendContentLength && (_sentLength == _contentLength)) { // it was last piece of content - _state = RESPONSE_END; + _source_ended = true; } } } } while (_send_buffer_len); // go on till we have something in buffer pending to send // execute sending whatever we have in sock buffs now - request->client()->send(); - _writtenLength += payloadlen; + if (payloadlen && !request->client()->send()) { + _state = RESPONSE_FAILED; + request->client()->close(); + return payloadlen; + } #if ASYNCWEBSERVER_USE_CHUNK_INFLIGHT // Take a credit only for data actually put in flight. When the source // had nothing yet (RESPONSE_TRY_AGAIN), nothing will be acked to return @@ -511,16 +523,13 @@ size_t AsyncAbstractResponse::write_send_buffs(AsyncWebServerRequest *request, s if (_send_buffer_len == 0) { // buffer empty, we can release mem, otherwise need to keep it till next run (should not happen under normal conditions) _send_buffer.reset(); + if (_source_ended) { + _state = _ackedLength >= _writtenLength ? RESPONSE_END : RESPONSE_WAIT_ACK; + } } return payloadlen; } // (_state == RESPONSE_CONTENT) - // implicit check - if (_state == RESPONSE_WAIT_ACK) { - // we do not need to wait for any acks actually if we won't send any more data, - // connection would be closed gracefully with last piece of data (in AsyncWebServerRequest::_onAck) - _state = RESPONSE_END; - } return 0; } diff --git a/tests/response_backpressure.cpp b/tests/response_backpressure.cpp new file mode 100644 index 000000000..31d99cf0f --- /dev/null +++ b/tests/response_backpressure.cpp @@ -0,0 +1,126 @@ +// SPDX-License-Identifier: LGPL-3.0-or-later +// Host regression harness. The runner inserts the ACTUAL upstream/patched +// AsyncAbstractResponse::write_send_buffs below, not a rewritten response model. +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +using String=std::string; +constexpr size_t ASYNC_RESPONCE_BUFF_SIZE=16,CONFIG_LWIP_TCP_WND_DEFAULT=64,CONFIG_LWIP_TCP_MSS=32; +constexpr size_t RESPONSE_TRY_AGAIN=static_cast(-1); +enum {RESPONSE_SETUP,RESPONSE_HEADERS,RESPONSE_CONTENT,RESPONSE_WAIT_ACK,RESPONSE_END,RESPONSE_FAILED}; +void check(bool value,const char *message){if(!value)throw std::runtime_error(message);} +struct Client { + size_t room=64,sends=0;bool alive=true; + String wire,pending;std::deque limits;std::deque sendResults; + size_t space(){return room;} + size_t add(const char *data,size_t count){ + if(!alive)return 0; + size_t accepted=std::min(count,room); + if(!limits.empty()){accepted=std::min(accepted,limits.front());limits.pop_front();} + pending.append(data,accepted);room-=accepted;return accepted; + } + bool send(){ + ++sends;bool sent=true; + if(!sendResults.empty()){sent=sendResults.front();sendResults.pop_front();} + if(sent){wire+=pending;pending.clear();}return sent; + } + void close(){alive=false;} +}; +struct AsyncWebServerRequest {Client transport;Client *client(){return &transport;}}; +class AsyncAbstractResponse { +public: + String _assembled_headers="HEAD\n",source; + size_t _writtenHeadersLength=0,_contentLength=0,_sentLength=0,_writtenLength=0,_ackedLength=0; + size_t _send_buffer_offset=0,_send_buffer_len=0,_in_flight=0,_in_flight_credit=2; + bool _source_ended=false,_chunked=false,_sendContentLength=true,valid=true; + int _state=RESPONSE_HEADERS; + size_t position=0,fills=0,notReady=0; + std::unique_ptr> _send_buffer; + bool _sourceValid()const{return valid;} + size_t _fillBufferAndProcessTemplates(uint8_t *data,size_t count){ + ++fills;if(notReady){--notReady;return RESPONSE_TRY_AGAIN;} + const auto n=std::min(count,source.size()-position); + std::memcpy(data,source.data()+position,n);position+=n;return n; + } + size_t write_send_buffs(AsyncWebServerRequest*,size_t,uint32_t); +}; +// ACTUAL_IMPLEMENTATION + +String unchunk(const String &wire){ + size_t at=5;String body; + for(;;){check(at+6<=wire.size(),"truncated chunk header"); + check(wire.substr(at+4,2)=="\r\n","chunk header framing"); + const auto count=std::stoul(wire.substr(at,4),nullptr,16);at+=6; + check(at+count+2<=wire.size(),"truncated chunk body");body+=wire.substr(at,count);at+=count; + check(wire.substr(at,2)=="\r\n","chunk tail framing");at+=2; + if(!count){check(at==wire.size(),"duplicate chunk terminator or trailing bytes");return body;} + } +} +void finish(AsyncAbstractResponse &r,AsyncWebServerRequest &q){ + for(unsigned budget=0;budget<200000;++budget){ + check(q.transport.alive,"response closed while healthy"); + check(r._state!=RESPONSE_FAILED,"healthy response failed"); + if(r._state==RESPONSE_END){ + check(q.transport.pending.empty(),"finished with unsent TCP data"); + check(r._ackedLength>=q.transport.wire.size(),"finished before actual admitted bytes ACK");return; + } + const auto outstanding=q.transport.wire.size()-r._ackedLength; + // Real ACK callbacks can be smaller than the last admitted tail. + const auto ack=std::min(outstanding,3);q.transport.room=64; + r.write_send_buffs(&q,ack,0); + } + throw std::runtime_error("response failed to progress under ACK/poll"); +} +void content(const String &body,bool chunked,bool known){ + std::cout<<"case bytes="<source.slice(start,end)); +} +for(const baseline of options['--baseline']?[true,false]:[false])for(const credits of [0,1]){ + const path=baseline?options['--baseline']:resolve(root,'src/WebResponses.cpp'); + const compiled=spawnSync(process.env.CXX||'g++',['-std=c++17','-Wall','-Wextra','-Werror','-O0','-g0', + '-DASYNCWEBSERVER_USE_CHUNK_INFLIGHT='+credits,'-x','c++','-','-o',out], + {input:method(path,baseline),encoding:'utf8',windowsHide:true,timeout:30000}); + if(compiled.error||compiled.status!==0)throw Error('Fixture compile failed: '+compiled.stderr); + const result=spawnSync(out,[],{encoding:'utf8',windowsHide:true,timeout:20000}); + console.log((baseline?'BASELINE':'PATCHED')+' credits='+credits+' exit='+result.status+' '+result.stdout+result.stderr); + if(result.error||result.status!==(baseline?1:0))throw Error('Unexpected baseline/patch fixture outcome'); +}