%%% -*- erlang -*- %%% %%% QUIC Socket Abstraction with UDP Packet Batching %%% Supports GSO/GRO on Linux (OTP 27+), gen_udp fallback elsewhere %%% %%% Copyright (c) 2024-2026 Benoit Chesneau %%% Apache License 2.0 %%% %%% @doc UDP socket abstraction with packet batching support. %%% %%% This module provides a unified socket interface that: %%% - Uses the OTP 27+ `socket' module with GSO/GRO on Linux %%% - Falls back to gen_udp on macOS/Windows/other platforms %%% - Batches outgoing packets for improved throughput %%% - Handles coalesced packets on receive (GRO) %%% %%% == Architecture == %%% ``` %%% quic_connection/quic_listener %%% | %%% quic_socket (this module) %%% | %%% socket module (OTP 27+) with GSO/GRO on Linux %%% or %%% gen_udp fallback on macOS/Windows %%% ''' %%% %%% == Configuration == %%% ``` %%% quic:start_server(Name, Port, #{ %%% batching => #{ %%% enabled => true, %%% max_packets => 64 %%% } %%% }). %%% ''' -module(quic_socket). -export([ open/2, open_for_send/2, open_server_send/2, wrap/2, close/1, send/4, flush/1, recv/2, sockname/1, controlling_process/2, setopts/2, detect_capabilities/0, get_fd/1, get_socket/1, gso_supported/1 ]). -include("quic.hrl"). -include_lib("kernel/include/logger.hrl"). %% GSO/GRO socket option constants for Linux %% UDP_SEGMENT = 103 (for GSO) %% UDP_GRO = 104 (for GRO) -define(UDP_SEGMENT, 103). -define(UDP_GRO, 104). -record(socket_state, { %% The underlying socket socket :: socket:socket() | gen_udp:socket(), %% Which backend we're using backend :: socket | gen_udp, %% Whether this socket_state owns the socket (true = close on cleanup) %% When wrapping an existing socket, this is false to avoid closing caller's socket owns_socket = true :: boolean(), %% GSO support detected and enabled gso_supported = false :: boolean(), %% GSO segment size for batching gso_size = ?DEFAULT_GSO_SEGMENT_SIZE :: non_neg_integer(), %% GRO enabled for receive gro_enabled = false :: boolean(), %% Batching enabled batching_enabled = true :: boolean(), %% Batched packets waiting to be sent (stored as packet_view()) batch_buffer = [] :: [packet_view()], %% Number of packets in batch (avoids O(n) length/1) batch_count = 0 :: non_neg_integer(), %% Current batch destination address batch_addr :: {inet:ip_address(), inet:port_number()} | undefined, %% Maximum packets per batch max_batch_packets = ?DEFAULT_MAX_BATCH_PACKETS :: pos_integer() }). %% Packet can be: %% - binary() - already flat %% - {iov, [binary()]} - explicit iov parts (preferred for zero-copy) %% - iodata() - any iolist (for backwards compatibility) -type packet_view() :: binary() | {iov, [binary()]} | iodata(). -opaque socket_state() :: #socket_state{}. -export_type([socket_state/0, packet_view/0]). %%==================================================================== %% API %%==================================================================== %% @doc Open a UDP socket with batching support. %% Options: %% - All standard gen_udp options %% - batching => #{enabled => true, max_packets => 64, flush_timeout_ms => 1} -spec open(inet:port_number(), map()) -> {ok, socket_state()} | {error, term()}. open(Port, Opts) -> Capabilities = detect_capabilities(), Backend = maps:get(backend, Capabilities, gen_udp), GSOSupported = maps:get(gso, Capabilities, false), GROSupported = maps:get(gro, Capabilities, false), BatchOpts = maps:get(batching, Opts, #{}), BatchingEnabled = maps:get(enabled, BatchOpts, true), MaxBatch = maps:get(max_packets, BatchOpts, ?DEFAULT_MAX_BATCH_PACKETS), GSOSize = maps:get(gso_size, BatchOpts, ?DEFAULT_GSO_SEGMENT_SIZE), case Backend of socket -> open_socket_backend(Port, Opts, #{ gso_supported => GSOSupported, gro_supported => GROSupported, batching_enabled => BatchingEnabled, max_batch => MaxBatch, gso_size => GSOSize }); gen_udp -> open_genudp_backend(Port, Opts, #{ batching_enabled => BatchingEnabled, max_batch => MaxBatch, gso_size => GSOSize }) end. %% @doc Open a UDP socket optimized for sending (client connections). %% Detects platform capabilities and enables GSO if available. %% Unlike open/2, this is optimized for a single destination. -spec open_for_send(inet:ip_address(), map()) -> {ok, socket_state()} | {error, term()}. open_for_send(RemoteIP, Opts) -> Capabilities = detect_capabilities(), Backend = maps:get(backend, Capabilities, gen_udp), GSOSupported = maps:get(gso, Capabilities, false), BatchOpts = maps:get(batching, Opts, #{}), BatchingEnabled = maps:get(enabled, BatchOpts, true), MaxBatch = maps:get(max_packets, BatchOpts, ?DEFAULT_MAX_BATCH_PACKETS), GSOSize = maps:get(gso_size, BatchOpts, ?DEFAULT_GSO_SEGMENT_SIZE), case Backend of socket -> open_send_socket_backend(RemoteIP, Opts, #{ gso_supported => GSOSupported, batching_enabled => BatchingEnabled, max_batch => MaxBatch, gso_size => GSOSize }); gen_udp -> open_send_genudp_backend(RemoteIP, Opts, #{ batching_enabled => BatchingEnabled, max_batch => MaxBatch, gso_size => GSOSize }) end. %% @doc Open a server send socket bound to a local address with reuseport. %% Uses OTP socket backend with GSO support on Linux for high throughput. %% This is for server connections that need to send from a specific local port. -spec open_server_send({inet:ip_address(), inet:port_number()}, map()) -> {ok, socket_state()} | {error, term()}. open_server_send({LocalIP, LocalPort}, Opts) -> Capabilities = detect_capabilities(), Backend = maps:get(backend, Capabilities, gen_udp), GSOSupported = maps:get(gso, Capabilities, false), BatchOpts = maps:get(batching, Opts, #{}), BatchingEnabled = maps:get(enabled, BatchOpts, true), MaxBatch = maps:get(max_packets, BatchOpts, ?DEFAULT_MAX_BATCH_PACKETS), GSOSize = maps:get(gso_size, BatchOpts, ?DEFAULT_GSO_SEGMENT_SIZE), BatchConfig = #{ gso_supported => GSOSupported, batching_enabled => BatchingEnabled, max_batch => MaxBatch, gso_size => GSOSize }, case Backend of socket -> open_server_send_socket(LocalIP, LocalPort, Opts, BatchConfig); gen_udp -> open_server_send_genudp(LocalIP, LocalPort, Opts, BatchConfig) end. %% Open OTP socket for server send with reuseport binding open_server_send_socket(LocalIP, LocalPort, Opts, BatchConfig) -> Family = family(LocalIP), case socket:open(Family, dgram, udp) of {ok, Socket} -> configure_server_send_socket(Socket, LocalIP, LocalPort, Family, Opts, BatchConfig); {error, _} = Error -> Error end. configure_server_send_socket(Socket, LocalIP, LocalPort, Family, Opts, BatchConfig) -> %% Set reuseaddr for binding to same port as listener. %% NOTE: Do NOT use reuseport here! With reuseport, the kernel distributes %% incoming packets between all sockets bound to the port. Since this socket %% is only for sending (nobody reads from it), any packets directed here would %% be dropped, causing the listener to miss incoming data. ok = socket:setopt(Socket, {socket, reuseaddr}, true), set_socket_buffer_sizes(Socket, Opts), %% Bind to local address SockAddr = #{family => Family, addr => LocalIP, port => LocalPort}, case socket:bind(Socket, SockAddr) of ok -> GSOEnabled = maybe_enable_gso(Socket, BatchConfig), State = #socket_state{ socket = Socket, backend = socket, owns_socket = true, gso_supported = GSOEnabled, gso_size = maps:get(gso_size, BatchConfig), gro_enabled = false, batching_enabled = maps:get(batching_enabled, BatchConfig), max_batch_packets = maps:get(max_batch, BatchConfig) }, {ok, State}; {error, _} = Error -> socket:close(Socket), Error end. %% Fallback to gen_udp for server send (non-Linux platforms) %% NOTE: Do NOT use reuseport here! With reuseport, the kernel distributes %% incoming packets between all sockets bound to the port. Since this socket %% is only for sending, any packets directed here would be dropped. open_server_send_genudp(LocalIP, LocalPort, Opts, BatchConfig) -> RecBuf = maps:get(recbuf, Opts, ?DEFAULT_UDP_RECBUF), SndBuf = maps:get(sndbuf, Opts, ?DEFAULT_UDP_SNDBUF), SocketOpts = [ binary, {ip, LocalIP}, {active, false}, {reuseaddr, true}, {recbuf, RecBuf}, {sndbuf, SndBuf} ], case gen_udp:open(LocalPort, SocketOpts) of {ok, Socket} -> {ok, build_genudp_state(Socket, BatchConfig)}; {error, _} = Error -> Error end. %% @doc Get the underlying socket from a socket_state. -spec get_socket(socket_state()) -> socket:socket() | gen_udp:socket(). get_socket(#socket_state{socket = Socket}) -> Socket. %% @doc Check if GSO is supported for this socket_state. -spec gso_supported(socket_state()) -> boolean(). gso_supported(#socket_state{gso_supported = Supported}) -> Supported. %% @doc Wrap an existing gen_udp socket with batching support. %% This allows adding batching to connections that already have a socket. %% Note: GSO/GRO are not available when wrapping existing gen_udp sockets. %% The wrapped socket is NOT owned by this state - close/1 will not close it. -spec wrap(gen_udp:socket(), map()) -> {ok, socket_state()}. wrap(Socket, Opts) -> BatchOpts = maps:get(batching, Opts, #{}), BatchingEnabled = maps:get(enabled, BatchOpts, true), MaxBatch = maps:get(max_packets, BatchOpts, ?DEFAULT_MAX_BATCH_PACKETS), GSOSize = maps:get(gso_size, BatchOpts, ?DEFAULT_GSO_SEGMENT_SIZE), State = #socket_state{ socket = Socket, backend = gen_udp, owns_socket = false, gso_supported = false, gso_size = GSOSize, gro_enabled = false, batching_enabled = BatchingEnabled, max_batch_packets = MaxBatch }, {ok, State}. %% @doc Close the socket and flush any pending packets. %% Only closes the socket if owns_socket is true (i.e., socket was created by us). -spec close(socket_state()) -> ok. close(#socket_state{owns_socket = false}) -> %% Don't close wrapped sockets - caller owns them ok; close(#socket_state{socket = Socket, backend = socket}) -> _ = socket:close(Socket), ok; close(#socket_state{socket = Socket, backend = gen_udp}) -> _ = gen_udp:close(Socket), ok. %% @doc Send a packet, buffering for batch send if enabled. %% Packet can be: %% - binary() - already flat packet %% - {iov, [binary()]} - packet parts for scatter/gather (avoids copy) %% %% Returns updated state. Auto-flushes when: %% - Batch is full (max_batch_packets reached) %% - Destination address changes -spec send(socket_state(), inet:ip_address(), inet:port_number(), packet_view()) -> {ok, socket_state()} | {error, term()}. send(#socket_state{batching_enabled = false} = State, IP, Port, Packet) -> %% Batching disabled - send immediately do_send_immediate(State, IP, Port, Packet); send(#socket_state{batch_addr = undefined} = State, IP, Port, Packet) -> %% First packet in batch add_to_batch(State#socket_state{batch_addr = {IP, Port}}, Packet); send(#socket_state{batch_addr = {IP, Port}} = State, IP, Port, Packet) -> %% Same destination - add to batch add_to_batch(State, Packet); send(#socket_state{} = State, IP, Port, Packet) -> %% Different destination - flush current batch first case flush(State) of {ok, State1} -> add_to_batch(State1#socket_state{batch_addr = {IP, Port}}, Packet); {error, _} = Error -> Error end. %% @doc Flush all buffered packets. -spec flush(socket_state()) -> {ok, socket_state()} | {error, term()}. flush(#socket_state{batch_count = 0} = State) -> %% Nothing to flush {ok, State}; flush(#socket_state{batch_count = Count, batch_addr = undefined} = State) -> %% No address set but have data - shouldn't happen, but clear buffer ?LOG_WARNING(#{what => flush_no_addr, buffer_size => Count}), {ok, clear_batch(State)}; flush(#socket_state{gso_supported = true} = State) -> %% GSO path - send all packets in one syscall flush_gso(State); flush(#socket_state{} = State) -> %% Fallback path - send packets individually flush_individual(State). %% @doc Receive packets from the socket. %% On Linux with GRO, may return multiple coalesced packets. -spec recv(socket_state(), timeout()) -> {ok, {inet:ip_address(), inet:port_number()}, [binary()]} | {error, term()}. recv(#socket_state{socket = Socket, backend = socket, gro_enabled = true}, Timeout) -> %% GRO path - receive potentially coalesced packets recv_gro(Socket, Timeout); recv(#socket_state{socket = Socket, backend = socket}, Timeout) -> %% Socket backend without GRO case socket:recvfrom(Socket, 0, [], Timeout) of {ok, {#{addr := IP, port := Port}, Data}} -> {ok, {IP, Port}, [Data]}; {error, _} = Error -> Error end; recv(#socket_state{socket = Socket, backend = gen_udp}, Timeout) -> %% gen_udp backend receive {udp, Socket, IP, Port, Data} -> {ok, {IP, Port}, [Data]} after Timeout -> {error, timeout} end. %% @doc Get the local address and port. -spec sockname(socket_state()) -> {ok, {inet:ip_address(), inet:port_number()}} | {error, term()}. sockname(#socket_state{socket = Socket, backend = socket}) -> case socket:sockname(Socket) of {ok, #{addr := IP, port := Port}} -> {ok, {IP, Port}}; {error, _} = Error -> Error end; sockname(#socket_state{socket = Socket, backend = gen_udp}) -> inet:sockname(Socket). %% @doc Set the controlling process. -spec controlling_process(socket_state(), pid()) -> ok | {error, term()}. controlling_process(#socket_state{socket = Socket, backend = gen_udp}, Pid) -> gen_udp:controlling_process(Socket, Pid); controlling_process(#socket_state{backend = socket}, _Pid) -> %% socket module doesn't have controlling_process concept ok. %% @doc Set socket options. -spec setopts(socket_state(), list()) -> ok | {error, term()}. setopts(#socket_state{socket = Socket, backend = gen_udp}, Opts) -> inet:setopts(Socket, Opts); setopts(#socket_state{socket = Socket, backend = socket}, Opts) -> %% Convert gen_udp style opts to socket module set_socket_opts(Socket, Opts). %% @doc Get the underlying file descriptor. -spec get_fd(socket_state()) -> {ok, integer()} | {error, term()}. get_fd(#socket_state{socket = Socket, backend = gen_udp}) -> case inet:getfd(Socket) of {ok, Fd} -> {ok, Fd}; Error -> Error end; get_fd(#socket_state{socket = Socket, backend = socket}) -> %% socket module - get native fd try Fd = socket:info(Socket), case maps:get(fd, Fd, undefined) of undefined -> {error, no_fd}; FdVal -> {ok, FdVal} end catch _:_ -> {error, not_supported} end. %% @doc Detect platform capabilities for GSO/GRO. -spec detect_capabilities() -> map(). detect_capabilities() -> case os:type() of {unix, linux} -> detect_linux_capabilities(); _ -> #{gso => false, gro => false, backend => gen_udp} end. %%==================================================================== %% Internal Functions - Socket Backend %%==================================================================== open_socket_backend(Port, Opts, BatchConfig) -> case socket:open(inet, dgram, udp) of {ok, Socket} -> configure_and_bind_socket(Socket, Port, Opts, BatchConfig); {error, _} = Error -> Error end. configure_and_bind_socket(Socket, Port, Opts, BatchConfig) -> ok = socket:setopt(Socket, {socket, reuseaddr}, true), set_socket_buffer_sizes(Socket, Opts), maybe_set_reuseport(Socket, Opts), bind_and_finalize_socket(Socket, Port, BatchConfig). set_socket_buffer_sizes(Socket, Opts) -> RecBuf = maps:get(recbuf, Opts, ?DEFAULT_UDP_RECBUF), SndBuf = maps:get(sndbuf, Opts, ?DEFAULT_UDP_SNDBUF), _ = socket:setopt(Socket, {socket, rcvbuf}, RecBuf), _ = socket:setopt(Socket, {socket, sndbuf}, SndBuf), ok. maybe_set_reuseport(Socket, Opts) -> case maps:get(reuseport, Opts, false) of true -> _ = socket:setopt(Socket, {socket, reuseport}, true); false -> ok end. bind_and_finalize_socket(Socket, Port, BatchConfig) -> Addr = #{family => inet, addr => any, port => Port}, case socket:bind(Socket, Addr) of ok -> build_socket_state(Socket, BatchConfig); {error, _} = Error -> socket:close(Socket), Error end. build_socket_state(Socket, BatchConfig) -> GSOEnabled = maybe_enable_gso(Socket, BatchConfig), GROEnabled = maybe_enable_gro(Socket, BatchConfig), State = #socket_state{ socket = Socket, backend = socket, gso_supported = GSOEnabled, gso_size = maps:get(gso_size, BatchConfig), gro_enabled = GROEnabled, batching_enabled = maps:get(batching_enabled, BatchConfig), max_batch_packets = maps:get(max_batch, BatchConfig) }, {ok, State}. maybe_enable_gso(Socket, #{gso_supported := true, gso_size := Size}) -> %% Try to set GSO segment size %% UDP_SEGMENT = 103 case socket:setopt_native(Socket, {udp, ?UDP_SEGMENT}, <>) of ok -> true; {error, _} -> false end; maybe_enable_gso(_, _) -> false. maybe_enable_gro(Socket, #{gro_supported := true}) -> %% Try to enable GRO %% UDP_GRO = 104 case socket:setopt_native(Socket, {udp, ?UDP_GRO}, <<1:32/native>>) of ok -> true; {error, _} -> false end; maybe_enable_gro(_, _) -> false. %%==================================================================== %% Internal Functions - Send-optimized Socket (for client connections) %%==================================================================== %% Open socket backend optimized for sending (no binding to specific port) open_send_socket_backend(RemoteIP, Opts, BatchConfig) -> Family = family(RemoteIP), case socket:open(Family, dgram, udp) of {ok, Socket} -> configure_send_socket(Socket, Opts, BatchConfig); {error, _} = Error -> Error end. configure_send_socket(Socket, Opts, BatchConfig) -> ok = socket:setopt(Socket, {socket, reuseaddr}, true), set_socket_buffer_sizes(Socket, Opts), GSOEnabled = maybe_enable_gso(Socket, BatchConfig), State = #socket_state{ socket = Socket, backend = socket, owns_socket = true, gso_supported = GSOEnabled, gso_size = maps:get(gso_size, BatchConfig), gro_enabled = false, batching_enabled = maps:get(batching_enabled, BatchConfig), max_batch_packets = maps:get(max_batch, BatchConfig) }, {ok, State}. %% Open gen_udp backend optimized for sending (ephemeral port) open_send_genudp_backend(_RemoteIP, Opts, BatchConfig) -> SocketOpts = build_send_genudp_opts(Opts), case gen_udp:open(0, SocketOpts) of {ok, Socket} -> {ok, build_genudp_state(Socket, BatchConfig)}; {error, _} = Error -> Error end. build_send_genudp_opts(Opts) -> ActiveN = maps:get(active_n, Opts, 100), ExtraFlags = maps:get(extra_socket_opts, Opts, []), RecBuf = maps:get(recbuf, Opts, ?DEFAULT_UDP_RECBUF), SndBuf = maps:get(sndbuf, Opts, ?DEFAULT_UDP_SNDBUF), [ binary, inet, {active, ActiveN}, {reuseaddr, true}, {recbuf, RecBuf}, {sndbuf, SndBuf} ] ++ ExtraFlags. %%==================================================================== %% Internal Functions - gen_udp Backend %%==================================================================== open_genudp_backend(Port, Opts, BatchConfig) -> SocketOpts = build_genudp_opts(Opts), case gen_udp:open(Port, SocketOpts) of {ok, Socket} -> {ok, build_genudp_state(Socket, BatchConfig)}; {error, _} = Error -> Error end. build_genudp_opts(Opts) -> ActiveN = maps:get(active_n, Opts, 100), ReusePort = maps:get(reuseport, Opts, false), ExtraFlags = maps:get(extra_socket_opts, Opts, []), RecBuf = maps:get(recbuf, Opts, ?DEFAULT_UDP_RECBUF), SndBuf = maps:get(sndbuf, Opts, ?DEFAULT_UDP_SNDBUF), BaseOpts = [ binary, inet, {active, ActiveN}, {reuseaddr, true}, {recbuf, RecBuf}, {sndbuf, SndBuf} ], ReuseOpts = case ReusePort of true -> [{reuseport, true}, {reuseport_lb, true}]; false -> [] end, BaseOpts ++ ReuseOpts ++ ExtraFlags. build_genudp_state(Socket, BatchConfig) -> #socket_state{ socket = Socket, backend = gen_udp, gso_supported = false, gso_size = maps:get(gso_size, BatchConfig), gro_enabled = false, batching_enabled = maps:get(batching_enabled, BatchConfig), max_batch_packets = maps:get(max_batch, BatchConfig) }. %%==================================================================== %% Internal Functions - Batching %%==================================================================== add_to_batch( #socket_state{batch_buffer = Buffer, batch_count = Count, max_batch_packets = Max} = State, Packet ) -> %% Store packet as-is (binary or {iov, Parts}) - no flattening yet NewBuffer = [Packet | Buffer], NewCount = Count + 1, State1 = State#socket_state{batch_buffer = NewBuffer, batch_count = NewCount}, case NewCount >= Max of true -> %% Batch full - flush now flush(State1); false -> %% Just accumulate - caller flushes at send cycle boundaries {ok, State1} end. flush_gso( #socket_state{ socket = Socket, batch_buffer = Buffer, batch_addr = {IP, Port}, gso_size = SegmentSize } = State ) -> %% Combine all packets into a single super-datagram %% Packets are in reverse order, so reverse them first %% Normalize packet_view to binary for GSO (kernel needs contiguous buffer) Packets = lists:reverse(Buffer), PacketBins = [normalize_packet(P) || P <- Packets], CombinedData = iolist_to_binary(PacketBins), %% Build sendmsg with GSO control message Msg = #{ addr => #{family => family(IP), addr => IP, port => Port}, iov => [CombinedData], ctrl => [#{level => udp, type => ?UDP_SEGMENT, data => <>}] }, case socket:sendmsg(Socket, Msg) of ok -> {ok, clear_batch(State)}; {ok, RestData} -> %% Partial send - GSO didn't send all data ?LOG_WARNING(#{ what => gso_partial_send, sent => byte_size(CombinedData) - iolist_size(RestData), remaining => iolist_size(RestData) }), %% Disable GSO and retry remaining data individually State1 = clear_batch(State#socket_state{gso_supported = false}), send_remaining_individually(State1, IP, Port, RestData); {error, _} = Error -> Error end. flush_individual(#socket_state{backend = socket} = State) -> flush_individual_socket(State); flush_individual(#socket_state{backend = gen_udp} = State) -> flush_individual_genudp(State). flush_individual_socket( #socket_state{ socket = Socket, batch_buffer = Buffer, batch_addr = {IP, Port} } = State ) -> %% Send each packet using sendmsg with iov (no flattening) Packets = lists:reverse(Buffer), Dest = #{family => family(IP), addr => IP, port => Port}, case send_packets_socket(Socket, Dest, Packets, 0) of {ok, _Sent} -> {ok, clear_batch(State)}; {error, Reason, _Sent} -> {error, Reason} end. flush_individual_genudp( #socket_state{ socket = Socket, batch_buffer = Buffer, batch_addr = {IP, Port} } = State ) -> %% Send each packet using gen_udp (must flatten) Packets = lists:reverse(Buffer), case send_packets_genudp(Socket, IP, Port, Packets, 0) of {ok, _Sent} -> {ok, clear_batch(State)}; {error, Reason, _Sent} -> {error, Reason} end. %% Send packets using socket:sendmsg with iov (no flattening for socket backend) send_packets_socket(_Socket, _Dest, [], Sent) -> {ok, Sent}; send_packets_socket(Socket, Dest, [Packet | Rest], Sent) -> Msg = #{ addr => Dest, iov => packet_iov(Packet) }, case socket:sendmsg(Socket, Msg) of ok -> send_packets_socket(Socket, Dest, Rest, Sent + 1); {ok, _Remaining} -> {error, partial_send, Sent}; {error, Reason} -> {error, Reason, Sent} end. %% Send packets using gen_udp (must flatten to binary) send_packets_genudp(_Socket, _IP, _Port, [], Sent) -> {ok, Sent}; send_packets_genudp(Socket, IP, Port, [Packet | Rest], Sent) -> Bin = normalize_packet(Packet), case gen_udp:send(Socket, IP, Port, Bin) of ok -> send_packets_genudp(Socket, IP, Port, Rest, Sent + 1); {error, Reason} -> {error, Reason, Sent} end. %% Send remaining data after GSO partial write (RestData is iolist) %% Split into segment-sized packets to preserve QUIC datagram boundaries send_remaining_individually( #socket_state{socket = Socket, gso_size = SegmentSize} = State, IP, Port, RestData ) -> Dest = #{family => family(IP), addr => IP, port => Port}, Data = iolist_to_binary(RestData), Packets = split_into_segments(Data, SegmentSize), send_segments(Socket, Dest, Packets, State). %% Split binary data into segment-sized chunks split_into_segments(Data, SegmentSize) -> split_into_segments(Data, SegmentSize, []). split_into_segments(<<>>, _SegmentSize, Acc) -> lists:reverse(Acc); split_into_segments(Data, SegmentSize, Acc) when byte_size(Data) =< SegmentSize -> lists:reverse([Data | Acc]); split_into_segments(Data, SegmentSize, Acc) -> <> = Data, split_into_segments(Rest, SegmentSize, [Segment | Acc]). %% Send each segment as a separate datagram send_segments(_Socket, _Dest, [], State) -> {ok, State}; send_segments(Socket, Dest, [Packet | Rest], State) -> case socket:sendto(Socket, Packet, Dest) of ok -> send_segments(Socket, Dest, Rest, State); {error, _} = Error -> Error end. do_send_immediate(#socket_state{socket = Socket, backend = socket} = State, IP, Port, Packet) -> %% Use sendmsg with iov - no flattening needed Msg = #{ addr => #{family => family(IP), addr => IP, port => Port}, iov => packet_iov(Packet) }, case socket:sendmsg(Socket, Msg) of ok -> {ok, State}; {ok, _Remaining} -> {error, partial_send}; {error, Reason} -> {error, Reason} end; do_send_immediate(#socket_state{socket = Socket, backend = gen_udp} = State, IP, Port, Packet) -> %% gen_udp needs flat binary Bin = normalize_packet(Packet), case gen_udp:send(Socket, IP, Port, Bin) of ok -> {ok, State}; {error, _} = Error -> Error end. %%==================================================================== %% Internal Functions - Packet View Helpers %%==================================================================== %% Convert packet_view to iov list for sendmsg %% {iov, Parts} uses parts directly (zero-copy path) %% binary is wrapped in list %% iodata is flattened to list of binaries (preserves binary boundaries) packet_iov(Bin) when is_binary(Bin) -> [Bin]; packet_iov({iov, Parts}) -> Parts; packet_iov(IoData) when is_list(IoData) -> %% Flatten iolist to list of binaries (preserves binary boundaries) flatten_iodata(IoData, []). %% Flatten iodata to list of binaries (for iov) %% Preserves binary boundaries, concatenates byte sequences flatten_iodata([], Acc) -> lists:reverse(Acc); flatten_iodata([H | T], Acc) when is_binary(H) -> flatten_iodata(T, [H | Acc]); flatten_iodata([H | T], Acc) when is_list(H) -> %% Nested list - recurse flatten_iodata(T, flatten_iodata(H, Acc)); flatten_iodata([H | T], Acc) when is_integer(H), H >= 0, H =< 255 -> %% Byte value - collect consecutive bytes into a binary {Bytes, Rest} = collect_bytes([H | T], []), flatten_iodata(Rest, [list_to_binary(Bytes) | Acc]). %% Collect consecutive byte integers collect_bytes([H | T], Acc) when is_integer(H), H >= 0, H =< 255 -> collect_bytes(T, [H | Acc]); collect_bytes(Rest, Acc) -> {lists:reverse(Acc), Rest}. %% Flatten packet_view to binary (for GSO and gen_udp) normalize_packet(Bin) when is_binary(Bin) -> Bin; normalize_packet({iov, Parts}) -> iolist_to_binary(Parts); normalize_packet(IoData) when is_list(IoData) -> iolist_to_binary(IoData). %% Clear batch state clear_batch(State) -> State#socket_state{ batch_buffer = [], batch_count = 0, batch_addr = undefined }. %% Get address family family({_, _, _, _}) -> inet; family({_, _, _, _, _, _, _, _}) -> inet6. %%==================================================================== %% Internal Functions - GRO Receive %%==================================================================== recv_gro(Socket, Timeout) -> %% Receive with GRO - may get coalesced packets case socket:recvmsg(Socket, 0, 128, [], Timeout) of {ok, #{addr := #{addr := IP, port := Port}, iov := [Data], ctrl := Ctrl}} -> %% Check for GRO segment size in control messages case extract_gro_segment_size(Ctrl) of undefined -> %% No GRO - single packet {ok, {IP, Port}, [Data]}; SegmentSize -> %% Split coalesced data into individual packets Packets = split_gro_packets(Data, SegmentSize), {ok, {IP, Port}, Packets} end; {ok, #{addr := #{addr := IP, port := Port}, iov := [Data]}} -> %% No control messages {ok, {IP, Port}, [Data]}; {error, _} = Error -> Error end. extract_gro_segment_size([]) -> undefined; extract_gro_segment_size([#{level := udp, type := ?UDP_GRO, data := <>} | _]) -> Size; extract_gro_segment_size([_ | Rest]) -> extract_gro_segment_size(Rest). split_gro_packets(Data, SegmentSize) -> split_gro_packets(Data, SegmentSize, []). split_gro_packets(<<>>, _SegmentSize, Acc) -> lists:reverse(Acc); split_gro_packets(Data, SegmentSize, Acc) when byte_size(Data) =< SegmentSize -> lists:reverse([Data | Acc]); split_gro_packets(Data, SegmentSize, Acc) -> <> = Data, split_gro_packets(Rest, SegmentSize, [Packet | Acc]). %%==================================================================== %% Internal Functions - Linux Capability Detection %%==================================================================== detect_linux_capabilities() -> %% Check OTP version - need 27+ for socket module features case otp_version_check() of false -> #{gso => false, gro => false, backend => gen_udp}; true -> %% Try to create a test socket and check GSO/GRO test_linux_socket_capabilities() end. otp_version_check() -> %% Check if OTP version is 27 or higher try OtpRelease = erlang:system_info(otp_release), Version = list_to_integer(OtpRelease), Version >= 27 catch _:_ -> false end. test_linux_socket_capabilities() -> case socket:open(inet, dgram, udp) of {ok, Socket} -> %% Test GSO GSOSupported = test_gso(Socket), %% Test GRO GROSupported = test_gro(Socket), socket:close(Socket), #{ gso => GSOSupported, gro => GROSupported, backend => socket }; {error, _} -> #{gso => false, gro => false, backend => gen_udp} end. test_gso(Socket) -> %% Try to set UDP_SEGMENT option case socket:setopt_native(Socket, {udp, ?UDP_SEGMENT}, <<1200:16/native>>) of ok -> true; {error, _} -> false end. test_gro(Socket) -> %% Try to set UDP_GRO option case socket:setopt_native(Socket, {udp, ?UDP_GRO}, <<1:32/native>>) of ok -> true; {error, _} -> false end. %%==================================================================== %% Internal Functions - Socket Option Conversion %%==================================================================== set_socket_opts(_Socket, []) -> ok; set_socket_opts(Socket, [{active, N} | Rest]) when is_integer(N) -> %% socket module doesn't have active mode - skip set_socket_opts(Socket, Rest); set_socket_opts(Socket, [{active, _} | Rest]) -> %% Skip active mode set_socket_opts(Socket, Rest); set_socket_opts(Socket, [{recbuf, Size} | Rest]) -> _ = socket:setopt(Socket, {socket, rcvbuf}, Size), set_socket_opts(Socket, Rest); set_socket_opts(Socket, [{sndbuf, Size} | Rest]) -> _ = socket:setopt(Socket, {socket, sndbuf}, Size), set_socket_opts(Socket, Rest); set_socket_opts(Socket, [_ | Rest]) -> %% Skip unknown options set_socket_opts(Socket, Rest).