From d92fff65f71af87d475ec6f1ee109c81e9386281 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 25 Sep 2026 07:19:08 +0000 Subject: [PATCH 1/3] Extract the edge port multiplexer into Network.Mux. Network.Sender still writes the socket, but partition ordering and 64KB coalescing are now a pure queue. CI runs the stress tests that show flat_size priority and unbounded buffering under two downloads. Co-authored-by: dominic.letz --- .github/workflows/ci.yml | 2 + lib/network/mux.ex | 112 ++++++++++++++++++++++++ lib/network/sender.ex | 103 +++++++--------------- test/network/mux_test.exs | 176 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 322 insertions(+), 71 deletions(-) create mode 100644 lib/network/mux.ex create mode 100644 test/network/mux_test.exs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c048c1d5..cf0f03ed 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -71,5 +71,7 @@ jobs: # Wiring all of that up is tracked separately. mix test.turn mix test.nat + # Port multiplexer unit tests (no anvil, no WireGuard). + DIODE_MINIMAL_TEST=1 mix test --no-start test/network/mux_test.exs # Snapcraft recipe contract (unpinned libc6 / USN rebuilds); no anvil. DIODE_MINIMAL_TEST=1 mix test --no-start test/snap/snapcraft_test.exs diff --git a/lib/network/mux.ex b/lib/network/mux.ex new file mode 100644 index 00000000..85873ffa --- /dev/null +++ b/lib/network/mux.ex @@ -0,0 +1,112 @@ +defmodule Network.Mux do + @moduledoc """ + Fair-queue of outbound frames for one edge socket. + + `Network.Sender` used to own this map inline. Frames are grouped by + partition (`{:port, ref}` for `portsend`, plus the RPC partition). + `pop/1` always takes the partition whose queued term has the smaller + `:erts_debug.flat_size/1`. `coalesce/2` concatenates pops up to 64KB + for one `:ssl.send/2`. + + The client scheduler is `DiodeClient.Mux`. Same shortest-queue idea, + different weight and backpressure. See that module for merge options. + Keeping a pure `enqueue` / `pop` / `coalesce` here is option 1: the + GenServer only does socket IO and ack replies. + + Known stress behavior (covered by `test/network/mux_test.exs`): + + * `flat_size` does not count refc binary bytes. One 15MB chunk looks + smaller than a handful of sub-64-byte frames, so the large transfer + is preferred and the small one stalls. + * Equal weights do not share. `Enum.min/3` picks one partition, + that pop makes it strictly lighter, and it is drained to + completion before the other port runs. Two simultaneous + transfers do not interleave. + * `enqueue/4` appends with `++`. A multi-megabyte download that + outruns the socket copies the whole queue on every chunk. + * There is no byte cap. Two simultaneous downloads buffer without + bound while the shorter queue is served exclusively. + """ + + @coalesce_limit 64_000 + + defstruct partitions: %{} + + def new, do: %__MODULE__{} + + def coalesce_limit, do: @coalesce_limit + + @doc "Queue `frame` on `partition`, remembering `meta` for the matching pop." + def enqueue(mux = %__MODULE__{partitions: partitions}, partition, frame, meta) + when is_binary(frame) do + partitions = + Map.update(partitions, partition, {[frame], [meta]}, fn {queue, metas} -> + {queue ++ [frame], metas ++ [meta]} + end) + + %{mux | partitions: partitions} + end + + @doc """ + Pop the lightest partition. + + Returns `{:empty, mux}` or `{:ok, mux, frame, meta}`. + """ + def pop(mux = %__MODULE__{partitions: partitions}) do + case pick(partitions) do + nil -> + {:empty, mux} + + {partition, {[frame | queue], [meta | metas]}} -> + partitions = store(partitions, partition, queue, metas) + {:ok, %{mux | partitions: partitions}, frame, meta} + end + end + + @doc """ + Pop and concatenate until `limit` bytes or the queue is empty. + + Returns `{mux, binary, metas}`. + """ + def coalesce(mux, limit \\ @coalesce_limit) do + do_coalesce(mux, limit, [], []) + end + + def queued_frames(%__MODULE__{partitions: partitions}, partition) do + case Map.get(partitions, partition) do + nil -> [] + {queue, _metas} -> queue + end + end + + def partition_count(%__MODULE__{partitions: partitions}), do: map_size(partitions) + + defp do_coalesce(mux, limit, frames, metas) do + size = IO.iodata_length(frames) + + if partition_count(mux) > 0 and size < limit do + {:ok, mux, frame, meta} = pop(mux) + do_coalesce(mux, limit, [frames, frame], [meta | metas]) + else + {mux, IO.iodata_to_binary(frames), Enum.reverse(metas)} + end + end + + defp store(partitions, partition, [], []) do + Map.delete(partitions, partition) + end + + defp store(partitions, partition, queue, metas) do + Map.put(partitions, partition, {queue, metas}) + end + + defp pick(partitions) do + Enum.min(partitions, &lighter?/2, fn -> nil end) + end + + defp lighter?({_, {frames_a, _}}, {_, {frames_b, _}}) do + weight(frames_a) < weight(frames_b) + end + + defp weight(frames), do: :erts_debug.flat_size(frames) +end diff --git a/lib/network/sender.ex b/lib/network/sender.ex index 5e816136..534f4201 100644 --- a/lib/network/sender.ex +++ b/lib/network/sender.ex @@ -1,12 +1,14 @@ defmodule Network.Sender do - @doc """ - Quality-Of-Service aware network sender. The idea is to wrap - around a real socket and send on multiple "partitions". Small partitions - with fewer data in the queue are always preferred over the large partitions + @moduledoc """ + Quality-Of-Service aware network sender. Wraps a socket and sends on + multiple partitions. Small partitions are preferred over large ones. + + Partition ordering lives in `Network.Mux`. """ use GenServer + alias Network.Mux alias Network.Sender - defstruct [:partitions, :waiting, :relay] + defstruct [:mux, :waiting, :relay] def new(socket) do {:ok, pid} = GenServer.start_link(__MODULE__, [socket], hibernate_after: 5_000) @@ -34,86 +36,37 @@ defmodule Network.Sender do end @impl true - def handle_call( - {:push_async, partition, data, trace}, - _from, - state = %Sender{partitions: partitions, waiting: nil} - ) do - ps = - Map.update(partitions, partition, {[data], [{:trace, trace}]}, fn {q, rest} -> - {q ++ [data], rest ++ [{:trace, trace}]} - end) - - {:reply, :ok, %Sender{state | partitions: ps}} + def handle_call({:push_async, partition, data, trace}, _from, state = %Sender{waiting: nil}) do + {:reply, :ok, enqueue(state, partition, data, {:trace, trace})} end @impl true - def handle_call( - {:push_async, _partition, data, trace}, - _from, - state = %Sender{waiting: from} - ) do + def handle_call({:push_async, _partition, data, trace}, _from, state = %Sender{waiting: from}) do GenServer.reply(from, data) Network.EdgeV2.trace(trace) {:reply, :ok, %Sender{state | waiting: nil}} end @impl true - def handle_call( - {:push, partition, data}, - from, - state = %Sender{partitions: partitions, waiting: nil} - ) do - ps = - Map.update(partitions, partition, {[data], [from]}, fn {q, rest} -> - {q ++ [data], rest ++ [from]} - end) - - {:noreply, %Sender{state | partitions: ps}} + def handle_call({:push, partition, data}, from, state = %Sender{waiting: nil}) do + {:noreply, enqueue(state, partition, data, from)} end @impl true - def handle_call( - {:push, _partition, data}, - _from, - state = %Sender{waiting: from} - ) do + def handle_call({:push, _partition, data}, _from, state = %Sender{waiting: from}) do GenServer.reply(from, data) {:reply, :ok, %Sender{state | waiting: nil}} end @impl true - def handle_call(:pop, _from, state = %Sender{partitions: partitions}) do - min = - Enum.min( - partitions, - fn {_ka, {data_a, _from_a}}, {_kb, {data_b, _from_b}} -> - :erts_debug.flat_size(data_a) < :erts_debug.flat_size(data_b) - end, - fn -> nil end - ) - - case min do - nil -> - {:reply, nil, state} - - {partition, {[data | q], [from | rest]}} -> - ps = - if q == [] do - Map.delete(partitions, partition) - else - Map.put(partitions, partition, {q, rest}) - end - - state = %Sender{state | partitions: ps} - - case from do - nil -> :nop - {:trace, trace} -> Network.EdgeV2.trace(trace) - from -> GenServer.reply(from, :ok) - end - - {:reply, data, state} + def handle_call(:pop, _from, state = %Sender{}) do + case Mux.pop(state.mux) do + {:empty, mux} -> + {:reply, nil, %Sender{state | mux: mux}} + + {:ok, mux, data, meta} -> + ack(meta) + {:reply, data, %Sender{state | mux: mux}} end end @@ -122,9 +75,17 @@ defmodule Network.Sender do do_await(from, state) end + defp enqueue(state = %Sender{}, partition, data, meta) do + %Sender{state | mux: Mux.enqueue(state.mux, partition, data, meta)} + end + + defp ack(nil), do: :ok + defp ack({:trace, trace}), do: Network.EdgeV2.trace(trace) + defp ack(from), do: GenServer.reply(from, :ok) + # Coalescing data frames into 64kb at least when available - defp do_await(data \\ "", from, state = %Sender{partitions: partitions, waiting: nil}) do - if map_size(partitions) > 0 and byte_size(data) < 64_000 do + defp do_await(data \\ "", from, state = %Sender{waiting: nil}) do + if Mux.partition_count(state.mux) > 0 and byte_size(data) < Mux.coalesce_limit() do {:reply, new_data, state} = handle_call(:pop, from, state) do_await(data <> new_data, from, state) else @@ -149,7 +110,7 @@ defmodule Network.Sender do def init([socket]) do q = self() relay = spawn_link(__MODULE__, :relayer_loop, [q, socket]) - {:ok, %Sender{partitions: %{}, waiting: nil, relay: relay}} + {:ok, %Sender{mux: Mux.new(), waiting: nil, relay: relay}} end def relayer_loop(q, socket) do diff --git a/test/network/mux_test.exs b/test/network/mux_test.exs new file mode 100644 index 00000000..d66c72e2 --- /dev/null +++ b/test/network/mux_test.exs @@ -0,0 +1,176 @@ +defmodule Network.MuxTest do + use ExUnit.Case, async: true + + alias Network.Mux + alias Network.Sender + + defp enqueue_all(mux, partition, frames) do + Enum.reduce(frames, mux, fn frame, mux -> + Mux.enqueue(mux, partition, frame, partition) + end) + end + + defp pop_metas(mux, n) do + Enum.map_reduce(1..n, mux, fn _, mux -> + {:ok, mux, _frame, meta} = Mux.pop(mux) + {meta, mux} + end) + end + + describe "partition priority" do + test "equal depth does not alternate; one partition is drained before the other" do + mux = + Mux.new() + |> enqueue_all(:port_b, repeated(3, 1000)) + |> enqueue_all(:port_a, repeated(3, 1000)) + + {metas, _} = pop_metas(mux, 6) + assert metas == [:port_a, :port_a, :port_a, :port_b, :port_b, :port_b] + end + + test "one refc bulk chunk outranks many small frames (flat_size ignores payload bytes)" do + # 15MB-style chunk. flat_size counts the list cell and the binary + # reference, not the 200KB. Twenty 1-byte frames weigh more, so the + # bulk port is sent first and the small port waits. + bulk = :binary.copy(<<1>>, 200_000) + small = repeated(20, 1) + + mux = + Mux.new() + |> Mux.enqueue(:bulk, bulk, :bulk) + |> enqueue_all(:small, small) + + {:ok, mux, frame, :bulk} = Mux.pop(mux) + assert frame == bulk + + {metas, _} = pop_metas(mux, 20) + assert metas == List.duplicate(:small, 20) + end + + test "the shorter queue is served exclusively until depths match" do + mux = + Mux.new() + |> enqueue_all(:deep, repeated(10, 8_000)) + |> enqueue_all(:shallow, repeated(2, 8_000)) + + {metas, mux} = pop_metas(mux, 2) + assert metas == [:shallow, :shallow] + {:ok, _mux, _frame, :deep} = Mux.pop(mux) + end + + test "a late small port preempts a port that already buffered a large transfer" do + mux = enqueue_all(Mux.new(), :download_a, repeated(50, 16_000)) + mux = Mux.enqueue(mux, :download_b, <<0::size(16_000 * 8)>>, :download_b) + + {:ok, _, _, :download_b} = Mux.pop(mux) + end + end + + describe "coalesce" do + test "packs frames from both ports into one 64KB socket write" do + mux = + Mux.new() + |> enqueue_all(:port_a, repeated(40, 1_000)) + |> enqueue_all(:port_b, repeated(40, 1_000)) + + {mux, blob, metas} = Mux.coalesce(mux) + assert byte_size(blob) >= Mux.coalesce_limit() + assert byte_size(blob) < Mux.coalesce_limit() + 1_000 + assert :port_a in metas and :port_b in metas + assert Mux.queued_frames(mux, :port_a) == [] + assert Mux.queued_frames(mux, :port_b) != [] + assert Mux.partition_count(mux) == 1 + end + + test "returns empty when nothing is queued" do + assert Mux.coalesce(Mux.new()) == {Mux.new(), "", []} + end + + test "preserves order inside one partition" do + frames = for n <- 1..5, do: <> + + mux = enqueue_all(Mux.new(), :only, frames) + {_, blob, metas} = Mux.coalesce(mux) + assert blob == IO.iodata_to_binary(frames) + assert metas == List.duplicate(:only, 5) + end + end + + describe "stress" do + test "two deep ports buffer without a byte cap" do + frames = repeated(100, 16_000) + + mux = + Mux.new() + |> enqueue_all(:download_a, frames) + |> enqueue_all(:download_b, frames) + + assert length(Mux.queued_frames(mux, :download_a)) == 100 + assert length(Mux.queued_frames(mux, :download_b)) == 100 + assert Mux.partition_count(mux) == 2 + end + + test "enqueue keeps order on a long single-port queue" do + frames = for n <- 1..300, do: <> + mux = enqueue_all(Mux.new(), :port, frames) + assert Mux.queued_frames(mux, :port) == frames + end + + test "pop deletes a partition once it is empty and leaves the other" do + mux = + Mux.new() + |> Mux.enqueue(:gone, <<1>>, :gone) + |> Mux.enqueue(:stay, <<2>>, :stay) + + {:ok, mux, <<2>>, :stay} = Mux.pop(mux) + assert Mux.queued_frames(mux, :stay) == [] + assert Mux.queued_frames(mux, :gone) == [<<1>>] + assert Mux.partition_count(mux) == 1 + end + end + + describe "sender idle bypass" do + test "a push while the writer is waiting skips the queue and the partition" do + tag = make_ref() + from = {self(), tag} + + state = %Sender{ + mux: Mux.new() |> Mux.enqueue(:queued, <<9>>, :queued), + waiting: from, + relay: self() + } + + assert {:reply, :ok, %{waiting: nil} = state} = + Sender.handle_call( + {:push_async, :racer, <<1, 2, 3>>, nil}, + {self(), make_ref()}, + state + ) + + assert_received {^tag, <<1, 2, 3>>} + assert Mux.queued_frames(state.mux, :queued) == [<<9>>] + assert Mux.queued_frames(state.mux, :racer) == [] + end + + test "await coalesces queued partitions and does not wait" do + tag_a = make_ref() + tag_b = make_ref() + + mux = + Mux.new() + |> Mux.enqueue(:port_a, <<0::size(1000 * 8)>>, {self(), tag_a}) + |> Mux.enqueue(:port_b, <<1::size(1000 * 8)>>, {self(), tag_b}) + + state = %Sender{mux: mux, waiting: nil, relay: self()} + + assert {:reply, blob, %{waiting: nil}} = + Sender.handle_call(:await, {self(), make_ref()}, state) + + assert byte_size(blob) == 2_000 + assert_receive {^tag_a, :ok} + assert_receive {^tag_b, :ok} + end + end + + defp repeated(n, size), do: for(_ <- 1..n, do: <<0::size(size * 8)>>) +end From 0424ccb6b287b2865e058d5493ebe5ba1ed36134 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 25 Sep 2026 07:25:18 +0000 Subject: [PATCH 2/3] Stop pinning which partition wins an equal-weight tie. Map iteration order differs between OTP 28.3 and 28.5. The stress tests now assert that one partition is drained before the other, without naming the winner. Co-authored-by: dominic.letz --- test/network/mux_test.exs | 23 +++++++++++++++++------ 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/test/network/mux_test.exs b/test/network/mux_test.exs index d66c72e2..09dd177b 100644 --- a/test/network/mux_test.exs +++ b/test/network/mux_test.exs @@ -25,7 +25,7 @@ defmodule Network.MuxTest do |> enqueue_all(:port_a, repeated(3, 1000)) {metas, _} = pop_metas(mux, 6) - assert metas == [:port_a, :port_a, :port_a, :port_b, :port_b, :port_b] + assert one_then_the_other?(metas, 3, 3) end test "one refc bulk chunk outranks many small frames (flat_size ignores payload bytes)" do @@ -77,8 +77,8 @@ defmodule Network.MuxTest do assert byte_size(blob) >= Mux.coalesce_limit() assert byte_size(blob) < Mux.coalesce_limit() + 1_000 assert :port_a in metas and :port_b in metas - assert Mux.queued_frames(mux, :port_a) == [] - assert Mux.queued_frames(mux, :port_b) != [] + emptied = Enum.filter([:port_a, :port_b], &(Mux.queued_frames(mux, &1) == [])) + assert length(emptied) == 1 assert Mux.partition_count(mux) == 1 end @@ -122,9 +122,10 @@ defmodule Network.MuxTest do |> Mux.enqueue(:gone, <<1>>, :gone) |> Mux.enqueue(:stay, <<2>>, :stay) - {:ok, mux, <<2>>, :stay} = Mux.pop(mux) - assert Mux.queued_frames(mux, :stay) == [] - assert Mux.queued_frames(mux, :gone) == [<<1>>] + {:ok, mux, frame, meta} = Mux.pop(mux) + assert frame in [<<1>>, <<2>>] + assert meta in [:gone, :stay] + assert Mux.queued_frames(mux, meta) == [] assert Mux.partition_count(mux) == 1 end end @@ -172,5 +173,15 @@ defmodule Network.MuxTest do end end + defp one_then_the_other?(ids, first_n, second_n) do + case Enum.uniq(ids) do + [first, second] -> + ids == List.duplicate(first, first_n) ++ List.duplicate(second, second_n) + + _ -> + false + end + end + defp repeated(n, size), do: for(_ <- 1..n, do: <<0::size(size * 8)>>) end From d4bc3c4b6932c91be18a7385fc9e8b74aa201438 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 25 Sep 2026 10:10:47 +0000 Subject: [PATCH 3/3] Share edge portsend bandwidth by a 64KB quantum. Partitions rotate once they have sent 64KB, weighted by payload bytes. A push no longer skips the queue while the writer is waiting. Tests cover interleaving and linear enqueue cost. Co-authored-by: dominic.letz --- lib/network/mux.ex | 148 ++++++++++++++++++++++---------------- lib/network/sender.ex | 29 +++++--- test/network/mux_test.exs | 130 ++++++++++++++++++++++----------- 3 files changed, 197 insertions(+), 110 deletions(-) diff --git a/lib/network/mux.ex b/lib/network/mux.ex index 85873ffa..c9245cdf 100644 --- a/lib/network/mux.ex +++ b/lib/network/mux.ex @@ -1,65 +1,60 @@ defmodule Network.Mux do @moduledoc """ - Fair-queue of outbound frames for one edge socket. - - `Network.Sender` used to own this map inline. Frames are grouped by - partition (`{:port, ref}` for `portsend`, plus the RPC partition). - `pop/1` always takes the partition whose queued term has the smaller - `:erts_debug.flat_size/1`. `coalesce/2` concatenates pops up to 64KB - for one `:ssl.send/2`. - - The client scheduler is `DiodeClient.Mux`. Same shortest-queue idea, - different weight and backpressure. See that module for merge options. - Keeping a pure `enqueue` / `pop` / `coalesce` here is option 1: the - GenServer only does socket IO and ack replies. - - Known stress behavior (covered by `test/network/mux_test.exs`): - - * `flat_size` does not count refc binary bytes. One 15MB chunk looks - smaller than a handful of sub-64-byte frames, so the large transfer - is preferred and the small one stalls. - * Equal weights do not share. `Enum.min/3` picks one partition, - that pop makes it strictly lighter, and it is drained to - completion before the other port runs. Two simultaneous - transfers do not interleave. - * `enqueue/4` appends with `++`. A multi-megabyte download that - outruns the socket copies the whole queue on every chunk. - * There is no byte cap. Two simultaneous downloads buffer without - bound while the shorter queue is served exclusively. + Round-robin queue of outbound frames for one edge socket. + + Frames are grouped by partition (`{:port, ref}` for `portsend`, plus + the RPC partition). `pop/1` serves the current partition until it has + sent `quantum/0` bytes (64KB), then rotates. Weight is the payload + byte size. `coalesce/2` concatenates pops up to 64KB for one + `:ssl.send/2`. + + The client scheduler is `DiodeClient.Mux`. """ @coalesce_limit 64_000 + @quantum 64_000 - defstruct partitions: %{} + defstruct partitions: %{}, ring: [], turn: 0 def new, do: %__MODULE__{} def coalesce_limit, do: @coalesce_limit + def quantum, do: @quantum + @doc "Queue `frame` on `partition`, remembering `meta` for the matching pop." - def enqueue(mux = %__MODULE__{partitions: partitions}, partition, frame, meta) - when is_binary(frame) do - partitions = - Map.update(partitions, partition, {[frame], [meta]}, fn {queue, metas} -> - {queue ++ [frame], metas ++ [meta]} - end) - - %{mux | partitions: partitions} + def enqueue(mux = %__MODULE__{}, partition, frame, meta) when is_binary(frame) do + partitions = mux.partitions + ring = if Map.has_key?(partitions, partition), do: mux.ring, else: mux.ring ++ [partition] + + slot = + case Map.get(partitions, partition) do + nil -> {:queue.new(), :queue.new()} + existing -> existing + end + + {frames, metas} = slot + slot = {:queue.in(frame, frames), :queue.in(meta, metas)} + %{mux | partitions: Map.put(partitions, partition, slot), ring: ring} end @doc """ - Pop the lightest partition. + Pop one frame from the partition whose turn it is. Returns `{:empty, mux}` or `{:ok, mux, frame, meta}`. """ - def pop(mux = %__MODULE__{partitions: partitions}) do - case pick(partitions) do - nil -> - {:empty, mux} - - {partition, {[frame | queue], [meta | metas]}} -> - partitions = store(partitions, partition, queue, metas) - {:ok, %{mux | partitions: partitions}, frame, meta} + def pop(mux = %__MODULE__{ring: []}), do: {:empty, %{mux | turn: 0}} + + def pop(mux = %__MODULE__{ring: [partition | rest], turn: turn}) do + cond do + empty_slot?(mux, partition) -> + pop(drop_partition(mux, partition)) + + turn >= @quantum -> + pop(%{mux | ring: rest ++ [partition], turn: 0}) + + true -> + take_frame(mux, partition, rest) end end @@ -75,38 +70,69 @@ defmodule Network.Mux do def queued_frames(%__MODULE__{partitions: partitions}, partition) do case Map.get(partitions, partition) do nil -> [] - {queue, _metas} -> queue + {frames, _metas} -> :queue.to_list(frames) end end def partition_count(%__MODULE__{partitions: partitions}), do: map_size(partitions) - defp do_coalesce(mux, limit, frames, metas) do - size = IO.iodata_length(frames) + defp take_frame(mux, partition, rest) do + {frames, metas} = Map.fetch!(mux.partitions, partition) + {:value, next} = :queue.peek(frames) - if partition_count(mux) > 0 and size < limit do - {:ok, mux, frame, meta} = pop(mux) - do_coalesce(mux, limit, [frames, frame], [meta | metas]) + if mux.turn > 0 and mux.turn + byte_size(next) > @quantum do + pop(%{mux | ring: rest ++ [partition], turn: 0}) else - {mux, IO.iodata_to_binary(frames), Enum.reverse(metas)} + emit(mux, partition, rest, frames, metas) end end - defp store(partitions, partition, [], []) do - Map.delete(partitions, partition) - end + defp emit(mux, partition, rest, frames, metas) do + {{:value, frame}, frames} = :queue.out(frames) + {{:value, meta}, metas} = :queue.out(metas) + partitions = Map.put(mux.partitions, partition, {frames, metas}) + turn = mux.turn + byte_size(frame) + mux = %{mux | partitions: partitions, turn: turn} + + mux = + cond do + empty_slot?(mux, partition) -> + drop_partition(%{mux | ring: [partition | rest]}, partition) - defp store(partitions, partition, queue, metas) do - Map.put(partitions, partition, {queue, metas}) + turn >= @quantum -> + %{mux | ring: rest ++ [partition], turn: 0} + + true -> + mux + end + + {:ok, mux, frame, meta} end - defp pick(partitions) do - Enum.min(partitions, &lighter?/2, fn -> nil end) + defp drop_partition(mux, partition) do + %{ + mux + | partitions: Map.delete(mux.partitions, partition), + ring: Enum.reject(mux.ring, &(&1 == partition)), + turn: 0 + } end - defp lighter?({_, {frames_a, _}}, {_, {frames_b, _}}) do - weight(frames_a) < weight(frames_b) + defp empty_slot?(mux, partition) do + case Map.get(mux.partitions, partition) do + nil -> true + {frames, _metas} -> :queue.is_empty(frames) + end end - defp weight(frames), do: :erts_debug.flat_size(frames) + defp do_coalesce(mux, limit, frames, metas) do + size = IO.iodata_length(frames) + + if partition_count(mux) > 0 and size < limit do + {:ok, mux, frame, meta} = pop(mux) + do_coalesce(mux, limit, [frames, frame], [meta | metas]) + else + {mux, IO.iodata_to_binary(frames), Enum.reverse(metas)} + end + end end diff --git a/lib/network/sender.ex b/lib/network/sender.ex index 534f4201..907e42c3 100644 --- a/lib/network/sender.ex +++ b/lib/network/sender.ex @@ -1,7 +1,7 @@ defmodule Network.Sender do @moduledoc """ Quality-Of-Service aware network sender. Wraps a socket and sends on - multiple partitions. Small partitions are preferred over large ones. + multiple partitions, rotating after each 64KB quantum. Partition ordering lives in `Network.Mux`. """ @@ -41,10 +41,10 @@ defmodule Network.Sender do end @impl true - def handle_call({:push_async, _partition, data, trace}, _from, state = %Sender{waiting: from}) do - GenServer.reply(from, data) - Network.EdgeV2.trace(trace) - {:reply, :ok, %Sender{state | waiting: nil}} + def handle_call({:push_async, partition, data, trace}, _from, state = %Sender{waiting: from}) + when from != nil do + state = enqueue(state, partition, data, {:trace, trace}) + deliver_waiting(state) end @impl true @@ -53,9 +53,10 @@ defmodule Network.Sender do end @impl true - def handle_call({:push, _partition, data}, _from, state = %Sender{waiting: from}) do - GenServer.reply(from, data) - {:reply, :ok, %Sender{state | waiting: nil}} + def handle_call({:push, partition, data}, from_push, state = %Sender{waiting: from}) + when from != nil do + state = enqueue(state, partition, data, from_push) + deliver_waiting(state) end @impl true @@ -79,6 +80,18 @@ defmodule Network.Sender do %Sender{state | mux: Mux.enqueue(state.mux, partition, data, meta)} end + defp deliver_waiting(state = %Sender{waiting: from}) do + case Mux.pop(state.mux) do + {:empty, mux} -> + {:reply, :ok, %Sender{state | mux: mux}} + + {:ok, mux, data, meta} -> + ack(meta) + GenServer.reply(from, data) + {:reply, :ok, %Sender{state | mux: mux, waiting: nil}} + end + end + defp ack(nil), do: :ok defp ack({:trace, trace}), do: Network.EdgeV2.trace(trace) defp ack(from), do: GenServer.reply(from, :ok) diff --git a/test/network/mux_test.exs b/test/network/mux_test.exs index 09dd177b..3d5e1854 100644 --- a/test/network/mux_test.exs +++ b/test/network/mux_test.exs @@ -18,56 +18,73 @@ defmodule Network.MuxTest do end describe "partition priority" do - test "equal depth does not alternate; one partition is drained before the other" do + test "equal quantum-sized frames alternate in insertion order" do + q = Mux.quantum() + mux = Mux.new() - |> enqueue_all(:port_b, repeated(3, 1000)) - |> enqueue_all(:port_a, repeated(3, 1000)) + |> enqueue_all(:port_b, repeated(3, q)) + |> enqueue_all(:port_a, repeated(3, q)) {metas, _} = pop_metas(mux, 6) - assert one_then_the_other?(metas, 3, 3) + assert metas == [:port_b, :port_a, :port_b, :port_a, :port_b, :port_a] end - test "one refc bulk chunk outranks many small frames (flat_size ignores payload bytes)" do - # 15MB-style chunk. flat_size counts the list cell and the binary - # reference, not the 200KB. Twenty 1-byte frames weigh more, so the - # bulk port is sent first and the small port waits. + test "a large refc chunk does not outrank frames already queued" do bulk = :binary.copy(<<1>>, 200_000) small = repeated(20, 1) mux = Mux.new() - |> Mux.enqueue(:bulk, bulk, :bulk) |> enqueue_all(:small, small) + |> Mux.enqueue(:bulk, bulk, :bulk) - {:ok, mux, frame, :bulk} = Mux.pop(mux) - assert frame == bulk - - {metas, _} = pop_metas(mux, 20) - assert metas == List.duplicate(:small, 20) + {metas, _} = pop_metas(mux, 21) + assert metas == List.duplicate(:small, 20) ++ [:bulk] end - test "the shorter queue is served exclusively until depths match" do + test "the shorter queue waits its turn instead of being served exclusively" do mux = Mux.new() |> enqueue_all(:deep, repeated(10, 8_000)) |> enqueue_all(:shallow, repeated(2, 8_000)) - {metas, mux} = pop_metas(mux, 2) - assert metas == [:shallow, :shallow] - {:ok, _mux, _frame, :deep} = Mux.pop(mux) + {metas, _} = pop_metas(mux, 10) + assert hd(metas) == :deep + assert :shallow in metas + assert Enum.take(metas, 2) != [:shallow, :shallow] end - test "a late small port preempts a port that already buffered a large transfer" do - mux = enqueue_all(Mux.new(), :download_a, repeated(50, 16_000)) + test "a late small port does not preempt a port already in the ring" do + mux = enqueue_all(Mux.new(), :download_a, repeated(8, 16_000)) mux = Mux.enqueue(mux, :download_b, <<0::size(16_000 * 8)>>, :download_b) - {:ok, _, _, :download_b} = Mux.pop(mux) + {metas, _} = pop_metas(mux, 5) + assert Enum.take(metas, 4) == List.duplicate(:download_a, 4) + assert Enum.at(metas, 4) == :download_b + end + + test "small frames from two ports share bandwidth within a quantum" do + mux = + Mux.new() + |> enqueue_all(:port_a, repeated(200, 1_000)) + |> enqueue_all(:port_b, repeated(200, 1_000)) + + {metas, _} = pop_metas(mux, 400) + + bursts = + metas + |> Enum.chunk_by(& &1) + |> Enum.map(&(length(&1) * 1_000)) + + assert Enum.max(bursts) <= Mux.quantum() + assert Enum.count(metas, &(&1 == :port_a)) == 200 + assert Enum.count(metas, &(&1 == :port_b)) == 200 end end describe "coalesce" do - test "packs frames from both ports into one 64KB socket write" do + test "packs a quantum from the first port, then the next, into one 64KB write" do mux = Mux.new() |> enqueue_all(:port_a, repeated(40, 1_000)) @@ -77,9 +94,8 @@ defmodule Network.MuxTest do assert byte_size(blob) >= Mux.coalesce_limit() assert byte_size(blob) < Mux.coalesce_limit() + 1_000 assert :port_a in metas and :port_b in metas - emptied = Enum.filter([:port_a, :port_b], &(Mux.queued_frames(mux, &1) == [])) - assert length(emptied) == 1 - assert Mux.partition_count(mux) == 1 + assert Mux.queued_frames(mux, :port_a) == [] + assert Mux.queued_frames(mux, :port_b) != [] end test "returns empty when nothing is queued" do @@ -88,7 +104,6 @@ defmodule Network.MuxTest do test "preserves order inside one partition" do frames = for n <- 1..5, do: <> - mux = enqueue_all(Mux.new(), :only, frames) {_, blob, metas} = Mux.coalesce(mux) assert blob == IO.iodata_to_binary(frames) @@ -122,21 +137,22 @@ defmodule Network.MuxTest do |> Mux.enqueue(:gone, <<1>>, :gone) |> Mux.enqueue(:stay, <<2>>, :stay) - {:ok, mux, frame, meta} = Mux.pop(mux) - assert frame in [<<1>>, <<2>>] - assert meta in [:gone, :stay] - assert Mux.queued_frames(mux, meta) == [] + {:ok, mux, <<1>>, :gone} = Mux.pop(mux) + assert Mux.queued_frames(mux, :gone) == [] + assert Mux.queued_frames(mux, :stay) == [<<2>>] assert Mux.partition_count(mux) == 1 end end describe "sender idle bypass" do - test "a push while the writer is waiting skips the queue and the partition" do + test "a push while the writer is waiting enters the queue behind what is already waiting" do tag = make_ref() from = {self(), tag} + queued = make_ref() + state = %Sender{ - mux: Mux.new() |> Mux.enqueue(:queued, <<9>>, :queued), + mux: Mux.new() |> Mux.enqueue(:queued, <<9>>, {self(), queued}), waiting: from, relay: self() } @@ -148,9 +164,10 @@ defmodule Network.MuxTest do state ) - assert_received {^tag, <<1, 2, 3>>} - assert Mux.queued_frames(state.mux, :queued) == [<<9>>] - assert Mux.queued_frames(state.mux, :racer) == [] + assert_received {^tag, <<9>>} + assert_received {^queued, :ok} + assert Mux.queued_frames(state.mux, :racer) == [<<1, 2, 3>>] + assert Mux.queued_frames(state.mux, :queued) == [] end test "await coalesces queued partitions and does not wait" do @@ -173,15 +190,46 @@ defmodule Network.MuxTest do end end - defp one_then_the_other?(ids, first_n, second_n) do - case Enum.uniq(ids) do - [first, second] -> - ids == List.duplicate(first, first_n) ++ List.duplicate(second, second_n) + describe "performance" do + test "enqueue and pop of many frames stays linear" do + small = time_pop(2_000) + large = time_pop(8_000) + assert large < small * 10 + assert large < 1_000_000 + end - _ -> - false + test "one hundred partitions each pop once quickly" do + mux = + Enum.reduce(1..100, Mux.new(), fn n, mux -> + Mux.enqueue(mux, n, <>, n) + end) + + {micro, metas} = + :timer.tc(fn -> + {metas, _} = pop_metas(mux, 100) + metas + end) + + assert metas == Enum.to_list(1..100) + assert micro < 100_000 end end + defp time_pop(n) do + frames = repeated(n, 64) + + {micro, _} = + :timer.tc(fn -> + mux = + Mux.new() + |> enqueue_all(:a, frames) + |> enqueue_all(:b, frames) + + pop_metas(mux, n * 2) + end) + + micro + end + defp repeated(n, size), do: for(_ <- 1..n, do: <<0::size(size * 8)>>) end