feat(socket): abandon socket through linger=0 in fast_close/1

.github/workflows/run_test_case.yaml

@@ -20,7 +20,7 @@ jobs:
             image: ${{ matrix.otp }}
 
         steps:
-        - uses: actions/checkout@v2
+        - uses: actions/checkout@v4
         - name: Code dialyzer
           run: |
             make xref
@@ -31,3 +31,19 @@ jobs:
             make ct
             make cover
 
+    run_compose_tests:
+
+        runs-on: ubuntu-latest
+
+        strategy:
+          fail-fast: false
+          matrix:
+            otp:
+              - erlang:27
+              - erlang:26
+              - erlang:25
+
+        steps:
+        - uses: actions/checkout@v4
+        - name: Run compose-based tests
+          run: make ESOCKD_IMAGE_ERLANG=${{ matrix.otp }} ct-compose

Makefile

@@ -25,6 +25,18 @@ eunit: compile
 ct: compile
 	$(REBAR) as test ct -v
 
+COMPOSE := docker compose -f test/docker-compose.yml
+COMPOSE_CT_SUITES := test/esockd_socket_SUITE.erl
+
+.PHONY: ct-compose
+ct-compose:
+	$(COMPOSE) up --build --quiet-pull -d
+	sleep 5
+	$(COMPOSE) exec esockd rebar3 ct -v --suite $(COMPOSE_CT_SUITES) || $(COMPOSE) logs
+	$(COMPOSE) logs tcptest
+	$(COMPOSE) logs tc
+	$(COMPOSE) down -t1
+
 cover:
 	$(REBAR) cover
 

src/esockd_socket.erl

@@ -113,9 +113,9 @@ upgrade(Sock, [{Fun, Args} | More]) ->
 
 -spec fast_close(socket()) -> ok.
 fast_close(Sock) ->
-    %% TODO: Research better alternatives.
-    _Pid = erlang:spawn(socket, close, [Sock]),
-    ok.
+    %% NOTE: Unexpected to fail on active socket.
+    _ = socket:setopt(Sock, {socket, linger}, #{onoff => true, linger => 0}),
+    socket:close(Sock).
 
 %% @doc Sockname
 %% Returns original destination address and port if proxy protocol is enabled.

test/Dockerfile.tcptest

@@ -0,0 +1,11 @@
+ARG ESOCKD_IMAGE_ERLANG=erlang:27
+FROM ${ESOCKD_IMAGE_ERLANG}
+
+COPY . /esockd
+WORKDIR /esockd
+RUN rebar3 compile
+RUN rebar3 as test compile
+
+CMD ["sh", "-c", "erl -pa _build/default/lib/*/ebin _build/test/lib/esockd/test -s esockd_tcptest start 8080"]
+
+EXPOSE 8080

test/docker-compose.yml

@@ -0,0 +1,53 @@
+services:
+  tcptest:
+    build:
+      context: ..
+      dockerfile: test/Dockerfile.tcptest
+      args:
+        ESOCKD_IMAGE_ERLANG: ${ESOCKD_IMAGE_ERLANG:-erlang:27}
+    ports:
+      - 8080
+    tty: true
+    networks:
+      - esockd-bridge
+
+  tc:
+    image: docker.io/lukaszlach/docker-tc
+    cap_add:
+      - NET_ADMIN
+    ports:
+      - 4080
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+      - /var/docker-tc:/var/docker-tc
+    environment:
+      HTTP_BIND: "${HTTP_BIND:-0.0.0.0}"
+      HTTP_PORT: "${HTTP_PORT:-4080}"
+    # network_mode: host
+    networks:
+      - esockd-bridge
+
+  esockd:
+    image: ${ESOCKD_IMAGE_ERLANG:-erlang:27}
+    volumes:
+      - ${PWD}:/esockd
+    working_dir: /esockd
+    environment:
+      ESOCKD_CT_TCPTEST_HOST: tcptest
+      ESOCKD_CT_TCPTEST_PORT: 8080
+      ESOCKD_CT_TC_ID_TCPTEST: tcptest-1
+      ESOCKD_CT_TC_URL: http://tc:4080
+    command:
+      - epmd
+    networks:
+      - esockd-bridge
+
+networks:
+  esockd-bridge:
+    driver: bridge
+    name: esockd-bridge
+    ipam:
+      driver: default
+      config:
+        - subnet: 172.10.100.0/24
+          gateway: 172.10.100.1

test/esockd_socket_SUITE.erl

@@ -0,0 +1,142 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2025 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(esockd_socket_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(FAST_ENOUGH_MS, 25).
+
+all() -> esockd_ct:all(?MODULE).
+
+init_per_suite(Config) ->
+    Host = os:getenv("ESOCKD_CT_TCPTEST_HOST"),
+    Port = os:getenv("ESOCKD_CT_TCPTEST_PORT"),
+    TCUrl = os:getenv("ESOCKD_CT_TC_URL"),
+    TCContainer = os:getenv("ESOCKD_CT_TC_ID_TCPTEST"),
+    case Host =/= false andalso Port =/= false of
+        true ->
+            [{tcptest_host, Host},
+             {tcptest_port, list_to_integer(Port)},
+             {tc_url, TCUrl},
+             {tc_container, TCContainer}
+            | Config];
+        false ->
+            {skip, "Remote TCP test server not running"}
+    end.
+
+end_per_suite(_Config) ->
+    ok.
+
+init_per_testcase(t_fast_close_lost_network, Config) ->
+    case proplists:get_value(tc_url, Config) of
+        false ->
+            {skip, "Docker Traffic Control not running"};
+        _ ->
+            Config
+    end;
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(t_fast_close_lost_network, Config) ->
+    traffic_control_stop(Config);
+end_per_testcase(_TestCase, _Config) ->
+    ok.
+
+%%
+
+t_fast_close(Config) ->
+    Sock = connect_tcptest(Config),
+    T0 = erlang:monotonic_time(millisecond),
+    ok = esockd_socket:fast_close(Sock),
+    TD = erlang:monotonic_time(millisecond) - T0,
+    ?assert(TD < ?FAST_ENOUGH_MS, TD).
+
+t_fast_close_send_buffers(Config) ->
+    Sock = connect_tcptest(Config),
+    ok = socket:send(Sock, esockd_tcptest:command({sleep, 10_000})),
+    _Select = send_until_select(Sock, 1),
+    T0 = erlang:monotonic_time(millisecond),
+    ok = esockd_socket:fast_close(Sock),
+    TD = erlang:monotonic_time(millisecond) - T0,
+    ?assert(TD < ?FAST_ENOUGH_MS, TD).
+
+t_fast_close_lost_network(Config) ->
+    Sock = connect_tcptest(Config),
+    ok = socket:send(Sock, esockd_tcptest:command(echo)),
+    ok = socket:send(Sock, <<"Hello!">>),
+    ?assertEqual({ok, <<"Hello!">>}, socket:recv(Sock, 0, 5_000)),
+    %% Turn on complete 100% packet loss:
+    ok = traffic_control_set([{"loss", "100%"}], Config),
+    %% Stuff the send buffer full:
+    _Select = send_until_select(Sock, 1),
+    %% Close is still expected to be fast:
+    T0 = erlang:monotonic_time(millisecond),
+    ok = esockd_socket:fast_close(Sock),
+    TD = erlang:monotonic_time(millisecond) - T0,
+    ?assert(TD < ?FAST_ENOUGH_MS, TD).
+
+%%
+
+connect_tcptest(Config) ->
+    Host = proplists:get_value(tcptest_host, Config),
+    Port = proplists:get_value(tcptest_port, Config),
+    Addr = case inet:parse_address(Host) of
+               {ok, X} -> X;
+               {error, einval} ->
+                   {ok, X} = inet:getaddr(Host, inet),
+                   X
+           end,
+    {ok, Sock} = socket:open(inet, stream, tcp),
+    ok = esockd_socket:setopts(Sock, [{{otp, rcvbuf}, 2048},
+                                      {{socket, rcvbuf}, 2048},
+                                      {{socket, sndbuf}, 2048}]),
+    ok = socket:connect(Sock, #{family => inet,
+                                addr => Addr,
+                                port => Port}),
+    Sock.
+
+send_until_select(Sock, N) ->
+    case socket:send(Sock, binary:copy(<<N:32>>, 200), nowait) of
+        ok ->
+            send_until_select(Sock, N + 1);
+        {select, _} = Select ->
+            Select;
+        {error, closed} ->
+            ct:fail("Socket closed unexpectedly")
+    end.
+
+traffic_control_set(Policy, Config) ->
+    TCContainer = proplists:get_value(tc_container, Config),
+    TCUrl = proplists:get_value(tc_url, Config),
+    Request = {TCUrl ++ "/" ++ TCContainer,
+               [],
+               _ContentType = "application/x-www-form-urlencoded",
+               uri_string:compose_query(Policy)},
+    {ok, {{_HTTP, 200, _OK}, _Headers, ""}} =
+        httpc:request(post, Request, [], []),
+    ok.
+
+traffic_control_stop(Config) ->
+    TCContainer = proplists:get_value(tc_container, Config),
+    TCUrl = proplists:get_value(tc_url, Config),
+    Request = {TCUrl ++ "/" ++ TCContainer, []},
+    {ok, {{_HTTP, 200, _OK}, _Headers, ""}} =
+        httpc:request(delete, Request, [], []),
+    ok.

test/esockd_tcptest.erl

@@ -0,0 +1,126 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(esockd_tcptest).
+
+-export([start/1]).
+-export([command/1]).
+
+-export([start_connection/2]).
+-export([init_connection/2]).
+
+-define(COMMAND_SIZE, 64).
+
+-define(RECV_TIMEOUT_MS, 5_000).
+-define(SLEEP_TIME_MS, 15_000).
+
+%%
+
+start(Args) ->
+    {ok, _} = application:ensure_all_started(esockd),
+    Port = list_to_integer(argv(1, Args)),
+    ok = logger:set_primary_config(level, info),
+    {ok, Pid} = esockd:open(tcp, {"0.0.0.0", Port}, [
+        {acceptors, 4},
+        {backlog, 8},
+        {buffer, 8192},
+        {send_timeout, 5000},
+        {tcp_options, [binary, {show_econnreset, true}]},
+        {connection_mfargs, {?MODULE, start_connection}}
+    ]),
+    logger:info("Started esockd ~p on port ~p", [Pid, Port]),
+    ok.
+
+argv(N, Args) ->
+    to_string(lists:nth(N, Args)).
+
+to_string(Atom) when is_atom(Atom) ->
+    atom_to_list(Atom);
+to_string(Binary) when is_binary(Binary) ->
+    binary_to_list(Binary);
+to_string(String) when is_list(String) ->
+    String.
+
+%%
+
+command({sleep, Time} = Command) when is_integer(Time) ->
+    encode_command(Command);
+command(echo) ->
+    encode_command(echo);
+command(close) ->
+    encode_command(close).
+
+encode_command(Command) ->
+    Serial = term_to_binary(Command),
+    <<Serial/bytes, 0:(?COMMAND_SIZE - byte_size(Serial))/integer-unit:8>>.
+
+decode_command(Command) ->
+    binary_to_term(Command).
+
+%%
+
+start_connection(Transport, SockIn) ->
+    {ok, spawn_link(?MODULE, init_connection, [Transport, SockIn])}.
+
+init_connection(Transport, SockIn) ->
+    case Transport:wait(SockIn) of
+        {ok, Sock} ->
+            before_loop(Transport, Sock);
+        {error, Reason} ->
+            {error, Reason}
+    end.
+
+before_loop(Transport, Sock) ->
+    logger:info("[~s] Waiting for command", [format_peername(Transport, Sock)]),
+    case Transport:recv(Sock, ?COMMAND_SIZE, ?RECV_TIMEOUT_MS) of
+        {ok, Packet} ->
+            Command = decode_command(Packet),
+            logger:info("[~s] Incoming command: ~0p", [format_peername(Transport, Sock), Command]),
+            case Command of
+                {sleep, Time} ->
+                    ok = timer:sleep(Time),
+                    echo_loop(Transport, Sock);
+                echo ->
+                    echo_loop(Transport, Sock);
+                close ->
+                    Transport:close(Sock)
+            end;
+        {error, timeout} ->
+            logger:info("[~s] Command timeout", [format_peername(Transport, Sock)]),
+            Transport:close(Sock);
+        {error, Reason} ->
+            logger:info("[~s] Connection error: ~0p", [format_peername(Transport, Sock), Reason]),
+            ok
+    end.
+
+echo_loop(Transport, Sock) ->
+    case Transport:recv(Sock, 0, infinity) of
+        {ok, Data} ->
+            logger:info("[~s] Echoing back: ~0P", [format_peername(Transport, Sock), Data, 100]),
+            Transport:send(Sock, Data),
+            echo_loop(Transport, Sock);
+        {error, Reason} ->
+            logger:info("[~s] Connection error: ~0p", [format_peername(Transport, Sock), Reason]),
+            ok
+    end.
+
+format_peername(Transport, Sock) ->
+    case Transport:peername(Sock) of
+        {ok, Peername} ->
+            esockd:format(Peername);
+        {error, Reason} ->
+            io_lib:format("~0p", [Reason])
+    end.