okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petopenai_api: fast paths for chat tool fragments and claude deltas, requests wait for a draining connection, parallel and tool-call benchmark6b24755

main
branchesmainall refs →
4.9 KiB114 linesraw
1-module(albedo_openai_rate_bench).
2-export([env/2, token/0, measure/2, trust/1]).
3
4%% Helpers for manual/stream_rate_benchmark: streams against
5%% test/manual/mock_openai_server.py, each in its own process, measured
6%% together.
7
8env(Name, Default) ->
9    case os:getenv(binary_to_list(Name)) of
10        false -> Default;
11        Value -> unicode:characters_to_binary(Value)
12    end.
13
14%% Trusts the CA in File for this run, for a mock serving https.
15trust(<<>>) -> nil;
16trust(File) ->
17    ok = public_key:cacerts_load(File),
18    nil.
19
20%% Called from the stream callback for every text or argument delta.
21token() ->
22    put(arrivals, [os:system_time(nanosecond) | get(arrivals)]),
23    nil.
24
25%% Runs every {StreamId, Run} at once, each in a fresh process; returns
26%% {Results, Metrics} where Metrics is a list of {Name, Number} pairs, per
27%% token across all streams. VM CPU is microstate accounting over every
28%% scheduler and thread, so the poller and the pool count too.
29measure(Origin, Runs) ->
30    Owner = self(),
31    erlang:garbage_collect(),
32    {Reductions0, _} = statistics(reductions),
33    {Runtime0, _} = statistics(runtime),
34    msacc:start(),
35    msacc:reset(),
36    Start = erlang:monotonic_time(microsecond),
37    Workers = [{Id, spawn_monitor(fun() ->
38        put(arrivals, []),
39        Result = Run(),
40        {reductions, Reductions} = process_info(self(), reductions),
41        {memory, Memory} = process_info(self(), memory),
42        Owner ! {self(), done, Result, Reductions, Memory, get(arrivals)}
43    end)} || {Id, Run} <- Runs],
44    Done = [receive
45        {Pid, done, Result, Reductions, Memory, Arrivals} ->
46            demonitor(Ref, [flush]),
47            {Id, Result, Reductions, Memory, lists:reverse(Arrivals)};
48        {'DOWN', Ref, process, Pid, Reason} -> error({benchmark_worker_crashed, Reason})
49    end || {Id, {Pid, Ref}} <- Workers],
50    Elapsed = erlang:monotonic_time(microsecond) - Start,
51    Stats = msacc:stats(),
52    Runtime = msacc:stats(system_runtime, Stats),
53    msacc:stop(),
54    {Runtime1, _} = statistics(runtime),
55    case os:getenv("ALBEDO_BENCH_MSACC") of false -> ok; _ -> msacc:print(Stats) end,
56    {Reductions1, _} = statistics(reductions),
57    Received = lists:sum([length(Arrivals) || {_, _, _, _, Arrivals} <- Done]),
58    Latency = latency([{timings(Origin, Id), Arrivals} || {Id, _, _, _, Arrivals} <- Done]),
59    Per = fun(X) -> X / max(1, Received) end,
60    Metrics = [
61        {<<"tokens">>, Received},
62        {<<"wall_ms">>, Elapsed / 1000},
63        {<<"tok_per_s">>, Received / (Elapsed / 1000000)},
64        {<<"vm_cpu_us_per_tok">>, Per(Runtime)},
65        {<<"vm_cpu_pct">>, 100 * Runtime / Elapsed},
66        {<<"os_cpu_us_per_tok">>, Per(1000 * (Runtime1 - Runtime0))},
67        {<<"worker_reds_per_tok">>, Per(lists:sum([R || {_, _, R, _, _} <- Done]))},
68        {<<"vm_reds_per_tok">>, Per(Reductions1 - Reductions0)},
69        {<<"worker_memory_kb">>, lists:max([M || {_, _, _, M, _} <- Done]) / 1024}
70        | Latency
71    ],
72    {[Result || {_, Result, _, _, _} <- Done], [{Name, float(Value)} || {Name, Value} <- Metrics]}.
73
74%% Send-to-callback latency per token, from each stream's send times.
75latency(Streams) ->
76    case lists:all(fun({Sent, Arrivals}) -> length(Sent) =:= length(Arrivals) end, Streams) of
77        false -> [{<<"latency_mismatch">>, 1}];
78        true ->
79            case lists:sort([(A - S) / 1000 || {Sent, Arrivals} <- Streams,
80                                                {S, A} <- lists:zip(Sent, Arrivals)]) of
81                [] -> [];
82                Sorted ->
83                    [{<<"lat_p50_us">>, pct(Sorted, 0.50)}, {<<"lat_p99_us">>, pct(Sorted, 0.99)},
84                     {<<"lat_max_us">>, lists:last(Sorted)}]
85            end
86    end.
87
88pct(Sorted, P) -> lists:nth(max(1, round(P * length(Sorted))), Sorted).
89
90timings(Origin, Id) ->
91    #{scheme := Scheme, host := Host, port := Port} = uri_string:parse(Origin),
92    Options = [binary, {active, false}],
93    {Module, {ok, Socket}} = case Scheme of
94        <<"https">> -> {ssl, ssl:connect(binary_to_list(Host), Port, [{verify, verify_none} | Options])};
95        <<"http">> -> {gen_tcp, gen_tcp:connect(binary_to_list(Host), Port, Options)}
96    end,
97    ok = Module:send(Socket, [<<"GET /timings/">>, Id, <<" HTTP/1.1\r\nhost: mock\r\n\r\n">>]),
98    Body = response_body(Module, Socket, <<>>),
99    Module:close(Socket),
100    json:decode(Body).
101
102response_body(Module, Socket, Read) ->
103    {ok, Bytes} = Module:recv(Socket, 0, 5000),
104    Response = <<Read/binary, Bytes/binary>>,
105    case binary:split(Response, <<"\r\n\r\n">>) of
106        [Head, Body] ->
107            {match, [Length]} = re:run(Head, <<"content-length: (\\d+)">>,
108                                       [caseless, {capture, all_but_first, binary}]),
109            case byte_size(Body) >= binary_to_integer(Length) of
110                true -> Body;
111                false -> response_body(Module, Socket, Response)
112            end;
113        [_] -> response_body(Module, Socket, Response)
114    end.