From 3aae06577a0594ef875256ff47895139f99a7b71 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Thu, 23 Apr 2026 17:38:17 +0800 Subject: [PATCH] fix docker_client --- src/docker/docker_client.erl | 202 ++++++++++------------------ src/docker/docker_commands.erl | 2 +- src/docker/docker_deployer.erl | 9 +- src/tests/docker_commands_tests.erl | 7 +- 4 files changed, 74 insertions(+), 146 deletions(-) diff --git a/src/docker/docker_client.erl b/src/docker/docker_client.erl index c6ffddd..6681faf 100644 --- a/src/docker/docker_client.erl +++ b/src/docker/docker_client.erl @@ -3,37 +3,20 @@ %%% Short-lived Docker API client process. %%% %%% Each request owns one process and one gun Unix socket connection. -%%% This keeps long-running stream calls isolated while preserving the -%%% existing open/request/close lifecycle. +%%% The process exits after the Docker API response finishes. %%% @end %%%------------------------------------------------------------------- -module(docker_client). --behaviour(gen_server). - %% API -export([request/4, start_stream/5]). -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). - -define(DOCKER_SOCKET, "/var/run/docker.sock"). -define(RESPONSE_TIMEOUT, 5000). -define(BODY_TIMEOUT, 10000). -define(STREAM_BODY_TIMEOUT, 30000). -define(CLIENT_TIMEOUT, 60000). --record(state, { - mode :: request | stream, - owner :: pid(), - owner_ref :: reference() | undefined, - ref :: reference(), - method :: string(), - path :: string(), - body :: binary(), - headers :: list() -}). - %%%=================================================================== %%% API %%%=================================================================== @@ -41,98 +24,52 @@ -spec request(Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) -> {ok, StatusCode :: integer(), RespHeaders :: proplists:proplist(), RespBody :: binary()} | {error, any()}. request(Method, Path, Body, Headers) when is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) -> - Ref = make_ref(), Owner = self(), - case gen_server:start(?MODULE, {request, Owner, Ref, Method, Path, Body, Headers}, []) of - {ok, Pid} -> - MRef = erlang:monitor(process, Pid), - receive - {docker_client, Ref, {result, Result}} -> - erlang:demonitor(MRef, [flush]), - Result; - {'DOWN', MRef, process, Pid, Reason} -> - {error, {client_down, Reason}} - after ?CLIENT_TIMEOUT -> - exit(Pid, shutdown), - erlang:demonitor(MRef, [flush]), - {error, timeout} - end; - {error, Reason} -> - {error, Reason} + Ref = make_ref(), + {Pid, MRef} = spawn_monitor(fun() -> request_worker(Owner, Ref, Method, Path, Body, Headers) end), + receive + {docker_client, Ref, {result, Result}} -> + erlang:demonitor(MRef, [flush]), + Result; + {'DOWN', MRef, process, Pid, Reason} -> + {error, {client_down, Reason}} + after ?CLIENT_TIMEOUT -> + exit(Pid, shutdown), + erlang:demonitor(MRef, [flush]), + {error, timeout} end. -spec start_stream(Owner :: pid(), Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) -> - {ok, Ref :: reference(), Pid :: pid()} | {error, any()}. + {ok, Ref :: reference(), Pid :: pid(), MRef :: reference()}. start_stream(Owner, Method, Path, Body, Headers) when is_pid(Owner), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) -> Ref = make_ref(), - case gen_server:start(?MODULE, {stream, Owner, Ref, Method, Path, Body, Headers}, []) of - {ok, Pid} -> - {ok, Ref, Pid}; - {error, Reason} -> - {error, Reason} - end. + {Pid, MRef} = spawn_monitor(fun() -> stream_worker(Owner, Ref, Method, Path, Body, Headers) end), + {ok, Ref, Pid, MRef}. %%%=================================================================== -%%% gen_server callbacks +%%% Worker functions %%%=================================================================== --spec init(term()) -> {ok, #state{}}. -init({Mode, Owner, Ref, Method, Path, Body, Headers}) - when (Mode =:= request orelse Mode =:= stream), - is_pid(Owner), is_reference(Ref), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) -> +-spec request_worker(pid(), reference(), string(), string(), binary(), list()) -> ok. +request_worker(Owner, Ref, Method, Path, Body, Headers) -> + Result = do_request(Method, Path, Body, Headers), + send_owner(Owner, Ref, {result, Result}). + +-spec stream_worker(pid(), reference(), string(), string(), binary(), list()) -> ok. +stream_worker(Owner, Ref, Method, Path, Body, Headers) -> OwnerRef = erlang:monitor(process, Owner), - erlang:send_after(0, self(), execute), - {ok, #state{ - mode = Mode, - owner = Owner, - owner_ref = OwnerRef, - ref = Ref, - method = Method, - path = Path, - body = Body, - headers = Headers - }}. - --spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, ok, #state{}}. -handle_call(_Request, _From, State) -> - {reply, ok, State}. - --spec handle_cast(term(), #state{}) -> {noreply, #state{}}. -handle_cast(_Request, State) -> - {noreply, State}. - --spec handle_info(term(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. -handle_info(execute, State = #state{mode = request}) -> - Result = do_request(State), - send_owner(State, {result, Result}), - {stop, normal, State}; -handle_info(execute, State = #state{mode = stream}) -> - _Result = do_stream(State), - {stop, normal, State}; -handle_info({'DOWN', OwnerRef, process, _Pid, Reason}, State = #state{owner_ref = OwnerRef}) -> - {stop, {owner_down, Reason}, State}; -handle_info(_Info, State) -> - {noreply, State}. - --spec terminate(term(), #state{}) -> ok. -terminate(_Reason, #state{owner_ref = undefined}) -> - ok; -terminate(_Reason, #state{owner_ref = OwnerRef}) -> + _Result = do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers), erlang:demonitor(OwnerRef, [flush]), ok. --spec code_change(term(), #state{}, term()) -> {ok, #state{}}. -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - %%%=================================================================== %%% Internal functions %%%=================================================================== --spec do_request(#state{}) -> +-spec do_request(string(), string(), binary(), list()) -> {ok, integer(), proplists:proplist(), binary()} | {error, any()}. -do_request(#state{method = Method, path = Path, body = Body, headers = Headers}) -> +do_request(Method, Path, Body, Headers) -> case open_connection() of {ok, ConnPid} -> try @@ -145,19 +82,20 @@ do_request(#state{method = Method, path = Path, body = Body, headers = Headers}) {error, Reason} end. --spec do_stream(#state{}) -> ok | {error, binary()}. -do_stream(State = #state{method = Method, path = Path, body = Body, headers = Headers}) -> +-spec do_stream(pid(), reference(), reference(), string(), string(), binary(), list()) -> + ok | {error, binary()}. +do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers) -> case open_connection() of {ok, ConnPid} -> try StreamRef = gun:request(ConnPid, Method, Path, Headers, Body), - receive_stream_response(State, ConnPid, StreamRef) + receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef) after close_connection(ConnPid) end; {error, Reason0} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason} end. @@ -198,97 +136,97 @@ receive_body(ConnPid, StreamRef, Status, Headers, Acc) -> {error, timeout} end. --spec receive_stream_response(#state{}, pid(), reference()) -> ok | {error, binary()}. -receive_stream_response(State, ConnPid, StreamRef) -> +-spec receive_stream_response(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}. +receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef) -> receive {gun_response, ConnPid, StreamRef, nofin, Status, Headers} when Status >= 200, Status < 300 -> - send_owner(State, {response, Status, Headers}), - receive_stream_body(State, ConnPid, StreamRef); + send_owner(Owner, Ref, {response, Status, Headers}), + receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef); {gun_response, ConnPid, StreamRef, fin, Status, Headers} when Status >= 200, Status < 300 -> - send_owner(State, {response, Status, Headers}), - send_owner(State, done), + send_owner(Owner, Ref, {response, Status, Headers}), + send_owner(Owner, Ref, done), ok; {gun_response, ConnPid, StreamRef, nofin, Status, _Headers} -> - receive_stream_error_body(State, ConnPid, StreamRef, Status, <<>>); + receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, <<>>); {gun_response, ConnPid, StreamRef, fin, Status, _Headers} -> Reason = http_status_error(Status, <<>>), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_error, ConnPid, StreamRef, Reason0} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_error, ConnPid, Reason0} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_down, ConnPid, _, Reason0, _} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; - {'DOWN', OwnerRef, process, _Pid, Reason0} when OwnerRef =:= State#state.owner_ref -> + {'DOWN', OwnerRef, process, _Pid, Reason0} -> {error, format_error({owner_down, Reason0})} after ?RESPONSE_TIMEOUT -> - send_owner(State, {error, <<"处理超时"/utf8>>}), + send_owner(Owner, Ref, {error, <<"处理超时"/utf8>>}), {error, <<"timeout">>} end. --spec receive_stream_body(#state{}, pid(), reference()) -> ok | {error, binary()}. -receive_stream_body(State, ConnPid, StreamRef) -> +-spec receive_stream_body(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}. +receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef) -> receive {gun_data, ConnPid, StreamRef, fin, Data} -> - maybe_send_stream_data(State, Data), - send_owner(State, done), + maybe_send_stream_data(Owner, Ref, Data), + send_owner(Owner, Ref, done), ok; {gun_data, ConnPid, StreamRef, nofin, Data} -> - maybe_send_stream_data(State, Data), - receive_stream_body(State, ConnPid, StreamRef); + maybe_send_stream_data(Owner, Ref, Data), + receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef); {gun_error, ConnPid, StreamRef, Reason0} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_error, ConnPid, Reason0} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_down, ConnPid, _, Reason0, _} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; - {'DOWN', OwnerRef, process, _Pid, Reason0} when OwnerRef =:= State#state.owner_ref -> + {'DOWN', OwnerRef, process, _Pid, Reason0} -> {error, format_error({owner_down, Reason0})} after ?STREAM_BODY_TIMEOUT -> - send_owner(State, {error, <<"timeout">>}), + send_owner(Owner, Ref, {error, <<"timeout">>}), {error, <<"timeout">>} end. --spec receive_stream_error_body(#state{}, pid(), reference(), integer(), iodata()) -> +-spec receive_stream_error_body(pid(), reference(), reference(), pid(), reference(), integer(), iodata()) -> {error, binary()}. -receive_stream_error_body(State, ConnPid, StreamRef, Status, Acc) -> +receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, Acc) -> receive {gun_data, ConnPid, StreamRef, fin, Data} -> Reason = http_status_error(Status, iolist_to_binary([Acc, Data])), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_data, ConnPid, StreamRef, nofin, Data} -> - receive_stream_error_body(State, ConnPid, StreamRef, Status, [Acc, Data]); + receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, [Acc, Data]); {gun_error, ConnPid, StreamRef, Reason0} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_error, ConnPid, Reason0} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; {gun_down, ConnPid, _, Reason0, _} -> Reason = format_error(Reason0), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason}; - {'DOWN', OwnerRef, process, _Pid, Reason0} when OwnerRef =:= State#state.owner_ref -> + {'DOWN', OwnerRef, process, _Pid, Reason0} -> {error, format_error({owner_down, Reason0})} after ?BODY_TIMEOUT -> Reason = http_status_error(Status, iolist_to_binary(Acc)), - send_owner(State, {error, Reason}), + send_owner(Owner, Ref, {error, Reason}), {error, Reason} end. @@ -312,18 +250,18 @@ close_connection(ConnPid) -> catch gun:close(ConnPid), ok. --spec send_owner(#state{}, term()) -> ok. -send_owner(#state{owner = Owner, ref = Ref}, Event) -> +-spec send_owner(pid(), reference(), term()) -> ok. +send_owner(Owner, Ref, Event) -> Owner ! {docker_client, Ref, Event}, ok. --spec maybe_send_stream_data(#state{}, iodata()) -> ok. -maybe_send_stream_data(State, Data) -> +-spec maybe_send_stream_data(pid(), reference(), iodata()) -> ok. +maybe_send_stream_data(Owner, Ref, Data) -> case iolist_to_binary(Data) of <<>> -> ok; DataBin -> - send_owner(State, {data, DataBin}) + send_owner(Owner, Ref, {data, DataBin}) end. -spec http_status_error(integer(), binary()) -> binary(). diff --git a/src/docker/docker_commands.erl b/src/docker/docker_commands.erl index 0660ce0..9bddcda 100644 --- a/src/docker/docker_commands.erl +++ b/src/docker/docker_commands.erl @@ -16,7 +16,7 @@ start_container/1, stop_container/1, stop_container/2, remove_container/1, remove_container/3, kill_container/1, kill_container/2, get_containers/0]). --spec pull_image(Image :: binary()) -> {ok, Ref :: reference(), Pid :: pid()} | {error, Reason :: any()}. +-spec pull_image(Image :: binary()) -> {ok, Ref :: reference(), Pid :: pid(), MRef :: reference()} | {error, Reason :: any()}. pull_image(Image) when is_binary(Image) -> Url = lists:flatten(io_lib:format("/images/create?fromImage=~s", [binary_to_list(Image)])), docker_client:start_stream(self(), "POST", Url, <<>>, []). diff --git a/src/docker/docker_deployer.erl b/src/docker/docker_deployer.erl index 4ddefcb..7d53633 100644 --- a/src/docker/docker_deployer.erl +++ b/src/docker/docker_deployer.erl @@ -126,8 +126,8 @@ ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) -> report_task_event(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>), report_task_event(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>), case docker_commands:pull_image(Image) of - {ok, Ref, Pid} -> - await_pull_image(TaskId, Ref, Pid), + {ok, Ref, Pid, MRef} -> + await_pull_image(TaskId, Ref, Pid, MRef), {ok, Image}; {error, Reason} when is_binary(Reason) -> throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>}); @@ -136,11 +136,6 @@ ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) -> throw({deploy_error, Error}) end. --spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid()) -> ok. -await_pull_image(TaskId, Ref, Pid) when is_integer(TaskId), is_reference(Ref), is_pid(Pid) -> - MRef = erlang:monitor(process, Pid), - await_pull_image(TaskId, Ref, Pid, MRef). - -spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok. await_pull_image(TaskId, Ref, Pid, MRef) -> receive diff --git a/src/tests/docker_commands_tests.erl b/src/tests/docker_commands_tests.erl index dff3787..a308759 100644 --- a/src/tests/docker_commands_tests.erl +++ b/src/tests/docker_commands_tests.erl @@ -67,12 +67,7 @@ test_all() -> -spec test_pull() -> ok. test_pull() -> - {ok, Ref, Pid} = docker_commands:pull_image(?TEST_IMAGE), - await_pull(Ref, Pid). - --spec await_pull(reference(), pid()) -> ok. -await_pull(Ref, Pid) when is_reference(Ref), is_pid(Pid) -> - MRef = erlang:monitor(process, Pid), + {ok, Ref, Pid, MRef} = docker_commands:pull_image(?TEST_IMAGE), await_pull(Ref, Pid, MRef). -spec await_pull(reference(), pid(), reference()) -> ok.