diff --git a/test/datasets_contract_test.exs b/test/datasets_contract_test.exs index 3cdc7d8b..f5012289 100644 --- a/test/datasets_contract_test.exs +++ b/test/datasets_contract_test.exs @@ -206,11 +206,17 @@ defmodule DatasetsContractTest do File.write!(path, ["q,a\n" | rows] ++ ["bad,row,extra\n"]) caller = spawn(fn -> Datasets.csv(path, [:q]) end) - walk = wait_for_monitored(caller) + walk = wait_for_walk(caller) ref = Process.monitor(walk) + # Suspended, the walk cannot run to its end and exit on its own, so it + # ends only when something kills it. The suspension is lifted if the test + # process exits. + :erlang.suspend_process(walk) Process.exit(caller, :kill) - assert_receive {:DOWN, ^ref, :process, ^walk, :killed}, 1_000 + receive do + {:DOWN, ^ref, :process, ^walk, reason} -> assert reason == :killed + end after cleanup_tmp("killed-caller.csv") end @@ -434,13 +440,15 @@ defmodule DatasetsContractTest do cleanup_tmp("typed-math.jsonl") end - # The process `pid` monitors, once it monitors one. - # The caller also monitors other processes for a moment (the file server - # while it reads), so the walk is the monitored process running the CSV code. - defp wait_for_monitored(pid, tries \\ 500) do - walk = - case Process.info(pid, :monitors) do - {:monitors, monitors} -> + # The walk `caller` starts, once it has started it. The caller also monitors + # other processes for a moment (the file server while it reads), so the walk + # is the monitored process running the CSV code. The caller parses the whole + # file before it starts the walk, which takes as long as the machine is slow, + # so this waits for as long as the caller is alive. + defp wait_for_walk(caller) do + case Process.info(caller, :monitors) do + {:monitors, monitors} -> + walk = Enum.find_value(monitors, fn {:process, monitored} when is_pid(monitored) -> if csv_walk?(monitored), do: monitored @@ -449,15 +457,15 @@ defmodule DatasetsContractTest do nil end) - nil -> - nil - end + if walk do + walk + else + Process.sleep(5) + wait_for_walk(caller) + end - if walk || tries == 0 do - walk - else - Process.sleep(5) - wait_for_monitored(pid, tries - 1) + nil -> + flunk("the CSV caller ended before it started a walk") end end diff --git a/test/parity_sidecar_test.exs b/test/parity_sidecar_test.exs index 1935d9d9..d462feba 100644 --- a/test/parity_sidecar_test.exs +++ b/test/parity_sidecar_test.exs @@ -76,14 +76,19 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do refute process_group_alive?(group_pid) end + test "the timeout option ends a command that does not finish" do + assert {:error, :timeout, %Output{}} = + ParitySidecar.run(@python, ["-c", "import time; time.sleep(60)"], timeout: 150) + end + test "timeout returns captured output and terminates the process tree" do pid_path = tmp_path("timeout") - assert {:error, :timeout, %Output{} = output} = - ParitySidecar.run(@python, process_tree_args(pid_path), timeout: 150) + {result, [parent_pid, child_pid, group_pid]} = + time_out_ready_tree(process_tree_args(pid_path), pid_path) + assert {:error, :timeout, %Output{} = output} = result assert output.text =~ "tree ready" - [parent_pid, child_pid, group_pid] = await_pids(pid_path) refute await_alive?(parent_pid) refute await_alive?(child_pid) refute process_group_alive?(group_pid) @@ -92,14 +97,13 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do test "bounded cleanup kills a fully TERM-resistant process tree" do pid_path = tmp_path("term-resistant") - assert {:error, :timeout, %Output{}} = - ParitySidecar.run( - @python, - process_tree_args(pid_path, leader_ignore_term: true, child_ignore_term: true), - timeout: 150 - ) + {result, [parent_pid, child_pid, group_pid]} = + time_out_ready_tree( + process_tree_args(pid_path, leader_ignore_term: true, child_ignore_term: true), + pid_path + ) - [parent_pid, child_pid, group_pid] = await_pids(pid_path) + assert {:error, :timeout, %Output{}} = result refute await_alive?(parent_pid) refute await_alive?(child_pid) refute process_group_alive?(group_pid) @@ -108,18 +112,14 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do test "Port exit does not hide a TERM-resistant child with redirected stdio" do pid_path = tmp_path("mixed-tree") - assert {:error, :timeout, %Output{text: text}} = - ParitySidecar.run( - @python, - process_tree_args(pid_path, - child_ignore_term: true, - child_redirect_stdio: true - ), - timeout: 150 - ) + {result, [parent_pid, child_pid, group_pid]} = + time_out_ready_tree( + process_tree_args(pid_path, child_ignore_term: true, child_redirect_stdio: true), + pid_path + ) + assert {:error, :timeout, %Output{text: text}} = result assert text =~ "tree ready" - [parent_pid, child_pid, group_pid] = await_pids(pid_path) refute await_alive?(parent_pid) refute await_alive?(child_pid) refute process_group_alive?(group_pid) @@ -203,11 +203,11 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do #{if leader_ignore_term, do: "signal.signal(signal.SIGTERM, signal.SIG_IGN)", else: ""} child_code = #{inspect(child_code(child_ignore_term))} child = subprocess.Popen([sys.executable, "-c", child_code]#{child_stdio}) + print("tree ready", flush=True) with open(#{inspect(pid_path)}, "w", encoding="utf-8") as handle: handle.write(f"{os.getpid()}\\n{child.pid}\\n{os.getpgrp()}\\n") handle.flush() os.fsync(handle.fileno()) - print("tree ready", flush=True) time.sleep(60) """ @@ -220,25 +220,47 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do defp child_code(false), do: "import time; time.sleep(60)" - defp await_pids(path, attempts \\ 100) - defp await_pids(_path, 0), do: flunk("sidecar did not publish process ids") + # A timeout is a timer in the process that owns the sidecar's port, counted + # from the start. A timer short enough for a fast test fires, on a loaded + # machine, before Python has started the tree, and a long one makes every + # run wait it out. So these tests run the tree with no timer, wait until it + # is ready, and send its owner the message the timer sends. + defp time_out_ready_tree(args, pid_path) do + task = Task.async(fn -> ParitySidecar.run(@python, args) end) + [parent_pid | _rest] = pids = await_pids(pid_path) + send(port_owner(parent_pid), :command_timeout) + {Task.await(task, :infinity), pids} + end + + defp port_owner(os_pid) do + Enum.find_value(Port.list(), fn port -> + with {:os_pid, ^os_pid} <- Port.info(port, :os_pid), + {:connected, owner} <- Port.info(port, :connected) do + owner + else + _other -> nil + end + end) || flunk("no port runs OS process #{os_pid}") + end - defp await_pids(path, attempts) do + # The sidecar writes its process ids once the tree is running, which takes + # as long as the machine is slow. + defp await_pids(path) do case File.read(path) do {:ok, contents} -> case contents |> String.split("\n", trim: true) |> Enum.map(&Integer.parse/1) do [{parent, ""}, {child, ""}, {group, ""}] -> [parent, child, group] - _other -> retry_pids(path, attempts) + _other -> retry_pids(path) end _other -> - retry_pids(path, attempts) + retry_pids(path) end end - defp retry_pids(path, attempts) do + defp retry_pids(path) do Process.sleep(20) - await_pids(path, attempts - 1) + await_pids(path) end defp await_alive?(pid, attempts \\ 100) diff --git a/test/test_helper.exs b/test/test_helper.exs index 4e459308..9b0044e2 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -78,5 +78,14 @@ external_excludes = ExUnit.configure(exclude: external_excludes) +# ReqLLM reads its model catalog from LLMDB, which loads it on first use under +# a :global.trans lock. A process that finds the lock taken sleeps a random +# backoff that doubles up to 8 s before it asks again, so when the async tests +# at the start of the suite all make their first model lookup at once, some +# wait several times as long as the load itself (a 1.4 s load, waits of up to +# 4.6 s among 16 first lookups on an idle machine), and on a loaded runner past +# the 5 s of a Task.await. Loading it here, before any test runs, leaves nothing to contend for. +{:ok, _catalog} = LLMDB.load() + Imp.Test.OwnLog.install() ExUnit.start() diff --git a/test/timed_out_connection_test.exs b/test/timed_out_connection_test.exs index d84a4bf8..965fa9be 100644 --- a/test/timed_out_connection_test.exs +++ b/test/timed_out_connection_test.exs @@ -23,21 +23,24 @@ defmodule Imp.TimedOutConnectionTest do } test "the call after a timed-out call is answered on a new connection" do + test_pid = self() {:ok, seen} = Agent.start_link(fn -> [] end) # Bandit serves each HTTP/1 connection from one process, so the handler's - # pid names the connection a request came in on. The first request is held - # past the client's timeout; every later one is answered at once. + # pid names the connection a request came in on. The first request is not + # answered while the test runs, so the first call times out however slow + # the machine is, and its connection stays busy: a request written behind + # it would not be answered either. Every later request is answered at once. base_url = Imp.Test.LocalHTTP.start(fn _request -> connection = self() count = Agent.get_and_update(seen, &{length(&1) + 1, &1 ++ [connection]}) if count == 1 do + test_ref = Process.monitor(test_pid) + receive do - :release -> :ok - after - 2_000 -> :ok + {:DOWN, ^test_ref, :process, _pid, _reason} -> :ok end end @@ -61,13 +64,16 @@ defmodule Imp.TimedOutConnectionTest do # connection only when it draws the same shard. Choosing the first shard # for every request makes that meeting certain. finch = [name: ReqLLM.Application.finch_name(), pool_strategy: &hd/1] - opts = [timeout: 200, max_retries: 0, req_http_options: [finch: finch]] + opts = [max_retries: 0, req_http_options: [finch: finch]] messages = [%{role: :user, content: "ping"}] - assert {:error, %Imp.LMError{retryable: true}} = Imp.LM.generate(lm, messages, opts) - # Only the first call has to time out. The second gets room to be answered - # on a loaded machine; landing on the stuck connection still fails below. - assert {:ok, _reply} = Imp.LM.generate(lm, messages, Keyword.put(opts, :timeout, 5_000)) + assert {:error, %Imp.LMError{retryable: true}} = + Imp.LM.generate(lm, messages, Keyword.put(opts, :timeout, 200)) + + # The second call keeps ReqLLM's own receive timeout. On a new connection + # it is answered at once; on the stuck one it is not answered, and fails + # when that timeout runs out. + assert {:ok, _reply} = Imp.LM.generate(lm, messages, opts) assert [first, second] = Agent.get(seen, & &1) refute first == second