From 885dd0002d02df54d3f811e350cf3d8e67f8fc1e Mon Sep 17 00:00:00 2001 From: harmon25 Date: Thu, 16 Jul 2026 13:29:00 -0400 Subject: [PATCH 1/2] Stream iodata in bounded chunks --- AGENTS.md | 4 ++ README.md | 8 ++- src/tcp_server.erl | 118 +++++++++++++++++++++++++++++---------- test/tcp_server_test.exs | 32 +++++++++++ 4 files changed, 133 insertions(+), 29 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index e2c39c5..fdb85a9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -113,6 +113,10 @@ handle_api_request(_M, _P, _R, _A) -> given up after the bounded budget. - Responses are sent in `chunk_size` slices (default 4096; configurable per server). lwIP accepts at most `TCP_MSS` (~1440 B) per `socket:send`, so larger chunks just loop internally. +- Never flatten a complete response with `iolist_to_binary/1`. `tcp_server` walks nested + iodata into chunks bounded by both byte size and entry count, and partial sends retry only + the unsent part of the current chunk. This is required to avoid response-sized allocations + and heap fragmentation on memory-constrained devices. - No `priv/` in this repo; `httpd_file_handler` serves from a *consumer* app's `priv`. ## Testing diff --git a/README.md b/README.md index 922d73a..8a16eda 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,11 @@ WiFi. ### Chunked sending with backpressure retry Responses are sent in `chunk_size` slices (default 4096 bytes) via -`tcp_server:do_send/3`, which runs in the worker so blocking is fine. On ESP32 the +`tcp_server:do_send/3`, which runs in the worker so blocking is fine. Nested +iodata is consumed incrementally rather than flattened into one response-sized +binary. The temporary chunk is bounded by both `chunk_size` and an internal +iodata-entry limit, keeping peak send memory independent of total response size. +On ESP32 the lwIP TCP **send buffer is small** (a few KB — roughly `4 × TCP_MSS`), so a single large response is many times the buffer size and must be paced: @@ -114,6 +118,8 @@ large response is many times the buffer size and must be paced: - `do_send` therefore **retries a failed chunk** with a short backoff (bounded — see `MAX_SEND_RETRIES`), only abandoning the response once retries are exhausted. A genuinely-dead connection keeps failing and is given up cleanly. +- Partial sends retry only the unsent portion of the current bounded chunk; they + do not concatenate it with the rest of the response. This retry is the difference between large responses (e.g. 64 KB) truncating a large fraction of the time under concurrency versus completing reliably. Tune diff --git a/src/tcp_server.erl b/src/tcp_server.erl index 22e9a11..5b5cb5a 100644 --- a/src/tcp_server.erl +++ b/src/tcp_server.erl @@ -39,6 +39,10 @@ -export([start/3, start/4, start_link/3, start_link/4, stop/1, send/3]). +-ifdef(TEST). +-export([next_chunk/2]). +-endif. + -behaviour(gen_server). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). @@ -90,6 +94,10 @@ %% Default send-chunk size. Matches gen_tcp_server: a comfortable fit within %% the ESP32 lwIP send buffer. -define(DEFAULT_SEND_CHUNK, 4096). +%% Bound the temporary list used to coalesce small iodata leaves. A byte limit +%% alone is not enough: a JSON response can contain thousands of one-byte +%% punctuation elements, each of which occupies a cons cell. +-define(MAX_CHUNK_ENTRIES, 64). %% Application-level keys (chunk_size, max_connections, recv_timeout) ride in %% the SocketOptions map but must never reach socket:setopt/3. They are %% stripped with nested maps:remove/2 in init/1 — AtomVM's maps module does not @@ -392,47 +400,101 @@ call_handle_info(Protocol, Msg, State) -> -define(MAX_SEND_RETRIES, 50). do_send(Socket, Packet, ChunkSize) -> - do_send(Socket, Packet, ChunkSize, 0). + do_send_iodata(Socket, Packet, ChunkSize). + +%% Walk nested iodata incrementally instead of flattening the complete response. +%% next_chunk/2 bounds both bytes and list entries, so peak temporary memory is +%% independent of the total response size. +do_send_iodata(Socket, Packet, ChunkSize) -> + case next_chunk(Packet, ChunkSize) of + done -> + ok; + {Chunk, Rest} -> + ToSend = chunk_to_binary(Chunk), + case send_chunk(Socket, ToSend, 0) of + ok -> + case Rest of + [] -> ok; + _ -> receive after 0 -> ok end + end, + do_send_iodata(Socket, Rest, ChunkSize); + {error, _Reason} = Error -> + Error + end + end. -do_send(_Socket, <<>>, _ChunkSize, _Retries) -> +%% @private +next_chunk(Iodata, ChunkSize) when is_integer(ChunkSize) andalso ChunkSize > 0 -> + take_chunk([Iodata], ChunkSize, ?MAX_CHUNK_ENTRIES, []); +next_chunk(_Iodata, _ChunkSize) -> + erlang:error(badarg). + +take_chunk([], _BytesLeft, _EntriesLeft, []) -> + done; +take_chunk([], _BytesLeft, _EntriesLeft, Acc) -> + {lists:reverse(Acc), []}; +take_chunk(Pending, 0, _EntriesLeft, Acc) -> + {lists:reverse(Acc), Pending}; +take_chunk(Pending, _BytesLeft, 0, Acc) -> + {lists:reverse(Acc), Pending}; +take_chunk([[] | Rest], BytesLeft, EntriesLeft, Acc) -> + take_chunk(Rest, BytesLeft, EntriesLeft, Acc); +take_chunk([Byte | Rest], BytesLeft, EntriesLeft, Acc) + when is_integer(Byte) andalso Byte >= 0 andalso Byte =< 255 -> + take_chunk(Rest, BytesLeft - 1, EntriesLeft - 1, [Byte | Acc]); +take_chunk([Bin | Rest], BytesLeft, EntriesLeft, Acc) when is_binary(Bin) -> + Size = byte_size(Bin), + case Size of + 0 -> + take_chunk(Rest, BytesLeft, EntriesLeft, Acc); + _ when Size =< BytesLeft -> + take_chunk(Rest, BytesLeft - Size, EntriesLeft - 1, [Bin | Acc]); + _ -> + <> = Bin, + {lists:reverse([Part | Acc]), [Remaining | Rest]} + end; +take_chunk([[Head | Tail] | Rest], BytesLeft, EntriesLeft, Acc) -> + take_chunk([Head, Tail | Rest], BytesLeft, EntriesLeft, Acc); +take_chunk(_Pending, _BytesLeft, _EntriesLeft, _Acc) -> + erlang:error(badarg). + +%% Avoid copying when a chunk is already represented by one binary leaf. +chunk_to_binary([Bin]) when is_binary(Bin) -> + Bin; +chunk_to_binary(Chunk) -> + erlang:iolist_to_binary(Chunk). + +send_chunk(_Socket, <<>>, _Retries) -> ok; -do_send(Socket, Packet, ChunkSize, Retries) when is_list(Packet) -> - do_send(Socket, erlang:iolist_to_binary(Packet), ChunkSize, Retries); -do_send(Socket, Packet, ChunkSize, Retries) when is_binary(Packet) -> - TotalSize = byte_size(Packet), - Chunk = erlang:min(TotalSize, ChunkSize), - <> = Packet, - case socket:send(Socket, ToSend) of +send_chunk(Socket, Chunk, Retries) -> + case socket:send(Socket, Chunk) of ok -> - case byte_size(Rest) > 0 of - true -> receive after 0 -> ok end; - false -> ok - end, - do_send(Socket, Rest, ChunkSize, 0); + ok; + {ok, <<>>} -> + ok; {ok, Unsent} -> - %% Partial send: lwIP send buffer full / TCP_MSS cap. Yield to let - %% the stack drain, then retry the remainder. - ?TRACE("Partial send: unsent=~p remaining=~p", [byte_size(Unsent), TotalSize]), + %% Retry only the unsent part of this bounded chunk. The remaining + %% response stays in its iodata traversal stack and is never copied + %% together with Unsent. + ?TRACE("Partial send: unsent=~p chunk=~p", [byte_size(Unsent), byte_size(Chunk)]), receive after 10 -> ok end, - do_send(Socket, <>, ChunkSize, 0); + send_chunk(Socket, Unsent, 0); {error, Reason} -> %% Send buffer backpressure on ESP32/lwIP is reported as a transient %% error (commonly {error, closed}) even though the connection is - %% still alive. Retry with a short backoff; only abandon the - %% response once retries are exhausted (a truly-dead connection keeps - %% failing and we give up after the bounded budget). - retry_send(Socket, Packet, ChunkSize, Retries, Reason, TotalSize) + %% still alive. Retry with a short backoff; only abandon the chunk + %% once retries are exhausted. + retry_send_chunk(Socket, Chunk, Retries, Reason) end. %% @private -retry_send(_Socket, _Packet, _ChunkSize, Retries, Reason, _TotalSize) - when Retries >= ?MAX_SEND_RETRIES -> - ?TRACE("Send giving up after ~p retries: ~p (remaining ~p)", [Retries, Reason, _TotalSize]), +retry_send_chunk(_Socket, _Chunk, Retries, Reason) when Retries >= ?MAX_SEND_RETRIES -> + ?TRACE("Send giving up after ~p retries: ~p (chunk ~p)", [Retries, Reason, byte_size(_Chunk)]), {error, Reason}; -retry_send(Socket, Packet, ChunkSize, Retries, _Reason, _TotalSize) -> - ?TRACE("Send error ~p; retry ~p (remaining ~p)", [_Reason, Retries, _TotalSize]), +retry_send_chunk(Socket, Chunk, Retries, _Reason) -> + ?TRACE("Send error ~p; retry ~p (chunk ~p)", [_Reason, Retries, byte_size(Chunk)]), receive after 10 -> ok end, - do_send(Socket, Packet, ChunkSize, Retries + 1). + send_chunk(Socket, Chunk, Retries + 1). %% @private try_close(Socket) -> diff --git a/test/tcp_server_test.exs b/test/tcp_server_test.exs index 3ca1a2f..ba6fff1 100644 --- a/test/tcp_server_test.exs +++ b/test/tcp_server_test.exs @@ -217,12 +217,44 @@ defmodule TcpServerTest do end end + describe "bounded iodata chunking" do + test "preserves deeply nested mixed iodata across byte boundaries" do + data = [<<"ab">>, [99, [<<"defgh">>, []]], [105 | <<"jkl">>]] + + chunks = collect_chunks(data, 4) + + assert Enum.all?(chunks, fn chunk -> :erlang.iolist_size(chunk) <= 4 end) + assert :erlang.iolist_to_binary(chunks) == "abcdefghijkl" + end + + test "bounds temporary entries for many one-byte elements" do + data = List.duplicate(?x, 1_000) + [first | _] = chunks = collect_chunks(data, 4_096) + + assert length(first) == 64 + assert Enum.all?(chunks, fn chunk -> length(chunk) <= 64 end) + assert :erlang.iolist_to_binary(chunks) == :binary.copy("x", 1_000) + end + + test "rejects invalid iodata and chunk sizes" do + assert catch_error(:tcp_server.next_chunk([256], 32)) == :badarg + assert catch_error(:tcp_server.next_chunk("ok", 0)) == :badarg + end + end + ## helpers defp connect(port) do :gen_tcp.connect(~c"localhost", port, [:binary, active: false, packet: :raw]) end + defp collect_chunks(iodata, chunk_size, acc \\ []) do + case :tcp_server.next_chunk(iodata, chunk_size) do + :done -> Enum.reverse(acc) + {chunk, rest} -> collect_chunks(rest, chunk_size, [chunk | acc]) + end + end + defp find_free_tcp_port do {:ok, socket} = :gen_tcp.listen(0, [:binary, active: false, packet: :raw, reuseaddr: true]) {:ok, port} = :inet.port(socket) From 8398a819420062c039b13acc375843af326ca141 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 17 Jul 2026 02:54:40 +0000 Subject: [PATCH 2/2] Fix do_send_iodata recursion on empty Rest and fix README reference --- README.md | 2 +- src/tcp_server.erl | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 8a16eda..7528d27 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,7 @@ WiFi. ### Chunked sending with backpressure retry Responses are sent in `chunk_size` slices (default 4096 bytes) via -`tcp_server:do_send/3`, which runs in the worker so blocking is fine. Nested +`tcp_server:send/3`, which runs in the worker so blocking is fine. Nested iodata is consumed incrementally rather than flattened into one response-sized binary. The temporary chunk is bounded by both `chunk_size` and an internal iodata-entry limit, keeping peak send memory independent of total response size. diff --git a/src/tcp_server.erl b/src/tcp_server.erl index 5b5cb5a..1fc147f 100644 --- a/src/tcp_server.erl +++ b/src/tcp_server.erl @@ -415,9 +415,10 @@ do_send_iodata(Socket, Packet, ChunkSize) -> ok -> case Rest of [] -> ok; - _ -> receive after 0 -> ok end - end, - do_send_iodata(Socket, Rest, ChunkSize); + _ -> + receive after 0 -> ok end, + do_send_iodata(Socket, Rest, ChunkSize) + end; {error, _Reason} = Error -> Error end