-module(async_ports_SUITE).

-export([all/0, suite/0]).
-export([permanent_busy_test/1]).
-export([run_loop/5]).

-include_lib("common_test/include/ct.hrl").

-define(PACKET_SIZE, (10 * 1024 * 8)).
-define(CPORT_DELAY, 100).
-define(TEST_LOOPS_COUNT, 100000).
-define(SLEEP_BEFORE_CHECK, 1000).
-define(TEST_PROCS_COUNT, 2).
-define(TC_TIMETRAP_SECONDS, 10).

suite() ->
    [{ct_hooks,[ts_install_cth]},
     {timetrap, {seconds, ?TC_TIMETRAP_SECONDS}}].

all() ->
    [permanent_busy_test].

permanent_busy_test(Config) ->
    ExePath = filename:join(proplists:get_value(data_dir, Config), "cport"),
    Self = self(),
    spawn_link(
      fun() ->
              Block = <<0:?PACKET_SIZE>>,

              Port = open_port(ExePath),
              
              Testers = lists:map(
                          fun(_) ->
                                  spawn_link(?MODULE, run_loop,
                                             [Self,
                                              Port,
                                              Block,
                                              ?TEST_LOOPS_COUNT,
                                              0])
                          end,
                          lists:seq(1, ?TEST_PROCS_COUNT)),
              Self ! {test_info, Port, Testers},
              endless_flush(Port)
      end),

    receive
        {test_info, Port, Testers} ->
            MaxWaitTime = round(0.7 * ?TC_TIMETRAP_SECONDS * 1000),
            ct:log("wait testers, maximum ~w mcsec~n", [MaxWaitTime]),
            ok = wait_testers(MaxWaitTime, Testers),
            timer:sleep(?SLEEP_BEFORE_CHECK),
            case erlang:port_command(Port, <<"test">>, [nosuspend]) of
                false ->
                    exit(port_dead);
                true ->
                    ok
            end
    end.

wait_testers(Timeout, Testers) ->
    lists:foldl(
      fun(Pid, AccIn) ->
              StartWait = os:timestamp(),
              receive
                  {Pid, port_dead} ->
                      recalc_timeout(AccIn, StartWait)
              after AccIn ->
                      Pid ! stop,
                      recalc_timeout(AccIn, StartWait)
              end
      end, Timeout, Testers),
    ok.

recalc_timeout(TimeoutIn, WaitStart) ->
    erlang:max(0, TimeoutIn - round(timer:now_diff(os:timestamp(), WaitStart)) div 1000).

open_port(ExePath) ->
    erlang:open_port({spawn, ExePath ++ " 100"}, [{packet, 4}, eof, exit_status, use_stdio, binary]).

run_loop(RootProc, Port, Block, CheckLimit, BusyCnt) ->
    receive
        stop ->
            ok
    after 0 ->
            case erlang:port_command(Port, Block, [nosuspend]) of
                true ->
                    run_loop(RootProc, Port, Block, CheckLimit, 0);
                false ->
                    if
                        BusyCnt + 1 > CheckLimit ->
                            check_dead(RootProc, Port, Block, CheckLimit);
                        true ->
                            run_loop(RootProc, Port, Block, CheckLimit, BusyCnt + 1)
                    end
            end
    end.

check_dead(RootProc, Port, Block, CheckLimit) ->
    ct:log("~p: check port dead~n", [self()]),
    timer:sleep(?SLEEP_BEFORE_CHECK),
    case erlang:port_command(Port, Block, [nosuspend]) of
        true ->
            ct:log("not dead~n"),
            run_loop(RootProc, Port, Block, CheckLimit, 0);
        false ->
            ct:log("port dead: ~p~n", [Port]),
            RootProc ! {self(), port_dead},
            ok
    end.

endless_flush(Port) ->
    receive
        {Port, {data, _}} ->
            endless_flush(Port);
        {Port, SomethingWrong} ->
            erlang:error({someting_wrong, SomethingWrong})
    end.