File 4476-Clean-up-output_handler.patch of Package erlang

Overview Repositories Revisions Requests Users Attributes Meta

File 4476-Clean-up-output_handler.patch of Package erlang (Revision 08848386a4ea82f272aa9c86a95948ae)

Currently displaying revision 08848386a4ea82f272aa9c86a95948ae , Show latest

From a626e320710c1a1572126e7aa5e2285a42cb747e Mon Sep 17 00:00:00 2001
From: Raimo Niskanen <raimo@erlang.org>
Date: Mon, 24 Oct 2022 17:11:18 +0200
Subject: [PATCH 26/27] Clean up output_handler

---
 lib/ssl/test/inet_crypto_dist.erl | 240 +++++++++++++++++-------------
 1 file changed, 139 insertions(+), 101 deletions(-)

diff --git a/lib/ssl/test/inet_crypto_dist.erl b/lib/ssl/test/inet_crypto_dist.erl
index b9a7e5ba1f..f28da210f7 100644
--- a/lib/ssl/test/inet_crypto_dist.erl
+++ b/lib/ssl/test/inet_crypto_dist.erl
@@ -1093,26 +1093,8 @@ handshake(
                             link(Controller),
                             receive
                                 DistHandle ->
-                                    ok =
-                                        inet:setopts(
-                                          Socket,
-                                          [{active, ?TCP_ACTIVE},
-                                           inet_tcp_dist:nodelay()]),
-                                    try
-                                        input_handler(
-                                          RecvParams#params{
-                                            dist_handle = DistHandle},
-                                          RecvSeq)
-                                    catch
-                                        Class : Reason : Stacktrace ->
-                                            error_logger:info_report(
-                                              [input_handler_exception,
-                                               {class, Class},
-                                               {reason, Reason},
-                                               {stacktrace, Stacktrace}]),
-                                            erlang:raise(
-                                              Class, Reason, Stacktrace)
-                                    end
+                                    input_handler(
+                                      RecvParams, RecvSeq, DistHandle)
                             end
                     end,
                     [link,
@@ -1137,29 +1119,12 @@ handshake(
                      {fullsweep_after, 0}])),
             _ = monitor(process, InputHandler), % For the benchmark test
             ok = gen_tcp:controlling_process(Socket, InputHandler),
+            false = erlang:dist_ctrl_set_opt(DistHandle, get_size, true),
             ok = erlang:dist_ctrl_input_handler(DistHandle, InputHandler),
             InputHandler ! DistHandle,
-            crypto:rand_seed_alg(crypto_cache),
             reply(From, ok),
             process_flag(priority, normal),
-            erlang:dist_ctrl_get_data_notification(DistHandle),
-            try
-                output_handler(
-                  SendParams#params{
-                    dist_handle = DistHandle,
-                    rekey_msg =
-                        start_rekey_timer(SendParams#params.rekey_time)},
-                  SendSeq)
-            catch
-                Class : Reason : Stacktrace ->
-                    error_logger:info_report(
-                      [output_handler_exception,
-                       {class, Class},
-                       {reason, Reason},
-                       {stacktrace, Stacktrace}]),
-                    erlang:raise(
-                      Class, Reason, Stacktrace)
-            end;
+            output_handler(SendParams, SendSeq, DistHandle);
         %%
         {?MODULE, From, {send, Data}} ->
             {SendParams_1, SendSeq_1, Result} =
@@ -1223,10 +1188,38 @@ recv_and_decrypt_chunk(#params{socket = Socket} = RecvParams, RecvSeq) ->
 %% -------------------------------------------------------------------------
 %% Output handler process
 %%
-%% The game here is to flush all dist_data and dist_tick messages,
-%% prioritize dist_data over dist_tick, and to not use selective receive
-%% because that would hurt performance during overload
+%% Await an event about what to do; fetch dist data from the VM,
+%% send a dist tick, or rekey outbound encryption parameters.
+%%
+%% In case we are overloaded and could get many accumulated
+%% dist_data or dist_tick messages; make sure to flush all of them
+%% before proceeding with what to do.  But, do not use selective
+%% receive since that does not perform well when there are
+%% many messages in the process mailbox.
+
+%% Entry function
+output_handler(Params, Seq, DistHandle) ->
+    try
+        _ = crypto:rand_seed_alg(crypto_cache),
+        erlang:dist_ctrl_get_data_notification(DistHandle),
+        output_handler(
+          Params#params{
+            dist_handle = DistHandle,
+            rekey_msg = start_rekey_timer(Params#params.rekey_time)},
+          Seq)
+    catch
+        Class : Reason : Stacktrace ->
+            error_logger:info_report(
+              [output_handler_exception,
+               {class, Class},
+               {reason, Reason},
+               {stacktrace, Stacktrace}]),
+            erlang:raise(Class, Reason, Stacktrace)
+    end.
 
+%% Loop top
+%%
+%% State: lurking until any interesting message
 output_handler(Params, Seq) ->
     receive
         Msg ->
@@ -1305,64 +1298,130 @@ output_handler_rekey(Params, Seq) ->
     end.
 
 
-
+%% Get outbound data from VM; encrypt and send,
+%% until the VM has no more
+%%
 output_handler_xfer(Params, Seq) ->
     output_handler_xfer(Params, Seq, [], 0, []).
 %%
-output_handler_xfer(Params, Seq, {Front, Size, Rear}) ->
-    output_handler_xfer(Params, Seq, Front, Size, Rear).
+%% Front,Size,Rear is an Okasaki queue of binaries with total byte Size
 %%
 output_handler_xfer(Params, Seq, Front, Size, Rear)
   when ?CHUNK_SIZE =< Size ->
-    {Data, Q} = deq_iovec(?CHUNK_SIZE, Front, Size, Rear),
-    {Params_1, Seq_1, Result} =
-        encrypt_and_send_chunk(
-          Params, Seq, [?DATA_CHUNK, Data], 1 + ?CHUNK_SIZE),
-    if
-        Result =:= ok ->
-            output_handler_xfer(Params_1, Seq_1, Q);
-        true ->
-            death_row({send_chunk, trace(Result)})
-    end;
+    %%
+    %% We have a full chunk or more
+    %% -> collect one chunk or less and send
+    output_handler_collect(Params, Seq, Front, Size, Rear);
 output_handler_xfer(Params, Seq, Front, Size, Rear) ->
+    %% when Size < ?CHUNK_SIZE ->
+    %%
+    %% We do not have a full chunk -> try to fetch more from VM
     case erlang:dist_ctrl_get_data(Params#params.dist_handle) of
         none ->
             if
                 Size =:= 0 ->
+                    %% No more data from VM, nothing buffered
+                    %% -> go back to lurking
                     {Params, Seq};
                 true ->
-                    Data = Front ++ lists:reverse(Rear),
-                    {Params_1, Seq_1, Result} =
-                        encrypt_and_send_chunk(
-                          Params, Seq, [?DATA_CHUNK, Data], 1 + Size),
-                    if
-                        Result =:= ok ->
-                            {Params_1, Seq_1};
-                        true ->
-                            death_row({send_chunk, trace(Result)})
-                    end
+                    %% The VM had no more -> send what we have
+                    output_handler_collect(Params, Seq, Front, Size, Rear)
             end;
-        Bin when is_binary(Bin) ->
-            Len = byte_size(Bin),
-            output_handler_xfer(
-              Params, Seq, Front,
-              Size + 4 + Len, [Bin, <<Len:32>>|Rear]);
-        [Bin1, Bin2] ->
-            Len = byte_size(Bin1) + byte_size(Bin2),
-            output_handler_xfer(
-              Params, Seq, Front,
-              Size + 4 + Len, [Bin2, Bin1, <<Len:32>>|Rear]);
-        Iovec ->
-            Len = iolist_size(Iovec),
-            output_handler_xfer(
-              Params, Seq, Front,
-              Size + 4 + Len, lists:reverse(Iovec, [<<Len:32>>|Rear]))
+        {Len,Iov} ->
+            output_handler_enq(
+              Params, Seq, Front, Size + 4 + Len, [<<Len:32>>|Rear], Iov)
+    end.
+
+%% Enqueue VM data while splitting large binaries into ?CHUNK_SIZE
+%%
+output_handler_enq(Params, Seq, Front, Size, Rear, []) ->
+    output_handler_xfer(Params, Seq, Front, Size, Rear);
+output_handler_enq(Params, Seq, Front, Size, Rear, [Bin|Iov]) ->
+    output_handler_enq(Params, Seq, Front, Size, Rear, Iov, Bin).
+%%
+output_handler_enq(Params, Seq, Front, Size, Rear, Iov, Bin) ->
+    BinSize = byte_size(Bin),
+    if
+        BinSize =< ?CHUNK_SIZE ->
+            output_handler_enq(
+              Params, Seq, Front, Size, [Bin|Rear], Iov);
+        true ->
+            <<Bin1:?CHUNK_SIZE/binary, Bin2/binary>> = Bin,
+            output_handler_enq(
+              Params, Seq, Front, Size, [Bin1|Rear], Iov, Bin2)
+    end.
+
+%% Collect small binaries into chunks of at most ?CHUNK_SIZE
+%%
+output_handler_collect(Params, Seq, [], Zero, []) ->
+    0 = Zero, % Assert
+    %% No more enqueued -> try to get more form VM
+    output_handler_xfer(Params, Seq);
+output_handler_collect(Params, Seq, Front, Size, Rear) ->
+    output_handler_collect(Params, Seq, Front, Size, Rear, [], 0).
+%%
+output_handler_collect(Params, Seq, [], Zero, [], Acc, DataSize) ->
+    0 = Zero, % Assert
+    output_handler_chunk(Params, Seq, [], Zero, [], Acc, DataSize);
+output_handler_collect(Params, Seq, [], Size, Rear, Acc, DataSize) ->
+    %% Okasaki queue transfer Rear -> Front
+    output_handler_collect(
+      Params, Seq, lists:reverse(Rear), Size, [], Acc, DataSize);
+output_handler_collect(
+  Params, Seq, [Bin|Iov] = Front, Size, Rear, Acc, DataSize) ->
+    BinSize = byte_size(Bin),
+    DataSize_1 = DataSize + BinSize,
+    if
+        ?CHUNK_SIZE < DataSize_1 ->
+            %% Bin does not fit in chunk -> send Acc
+            output_handler_chunk(
+              Params, Seq, Front, Size, Rear, Acc, DataSize);
+        DataSize_1 < ?CHUNK_SIZE ->
+            %% Chunk not full yet -> try to accumulate more
+            output_handler_collect(
+              Params, Seq, Iov, Size - BinSize, Rear, [Bin|Acc], DataSize_1);
+        true -> % DataSize_1 == ?CHUNK_SIZE ->
+            %% Optimize one iteration; Bin fits exactly -> accumulate and send
+            output_handler_chunk(
+              Params, Seq, Iov, Size - BinSize, Rear, [Bin|Acc], DataSize_1)
+    end.
+
+%% Encrypt and send a chunk
+%%
+output_handler_chunk(Params, Seq, Front, Size, Rear, Acc, DataSize) ->
+    Data = lists:reverse(Acc),
+    {Params_1, Seq_1, Result} =
+        encrypt_and_send_chunk(Params, Seq, [?DATA_CHUNK|Data], 1 + DataSize),
+    if
+        Result =:= ok ->
+            %% Try to collect another chunk
+            output_handler_collect(Params_1, Seq_1, Front, Size, Rear);
+        true ->
+            death_row({send_chunk, trace(Result)})
     end.
 
 %% -------------------------------------------------------------------------
 %% Input handler process
 %%
 
+%% Entry function
+input_handler(#params{socket = Socket} = Params, Seq, DistHandle) ->
+    try
+        ok = inet:setopts(Socket, [{active, ?TCP_ACTIVE}, nodelay()]),
+        input_handler(
+          Params#params{dist_handle = DistHandle},
+          Seq)
+    catch
+        Class : Reason : Stacktrace ->
+            error_logger:info_report(
+              [input_handler_exception,
+               {class, Class},
+               {reason, Reason},
+               {stacktrace, Stacktrace}]),
+            erlang:raise(Class, Reason, Stacktrace)
+    end.
+
+%% Loop top
 input_handler(Params, Seq) ->
     %% Shortcut into the loop
     {Params_1, Seq_1, Data} = input_data(Params, Seq),
@@ -1626,27 +1685,6 @@ block_decrypt(
             error
     end.
 
-%% -------------------------------------------------------------------------
-%% Queue of binaries i.e an iovec queue
-
-deq_iovec(GetSize, Front, Size, Rear) ->
-    deq_iovec(GetSize, Front, Size, Rear, []).
-%%
-deq_iovec(GetSize, [], Size, Rear, Acc) ->
-    deq_iovec(GetSize, lists:reverse(Rear), Size, [], Acc);
-deq_iovec(GetSize, [Bin|Front], Size, Rear, Acc) ->
-    BinSize = byte_size(Bin),
-    if
-        BinSize < GetSize ->
-            deq_iovec(
-              GetSize - BinSize, Front, Size - BinSize, Rear, [Bin|Acc]);
-        GetSize < BinSize ->
-            {Bin1,Bin2} = erlang:split_binary(Bin, GetSize),
-            {lists:reverse(Acc, [Bin1]), {[Bin2|Front], Size - GetSize, Rear}};
-        true ->
-            {lists:reverse(Acc, [Bin]), {Front, Size - BinSize, Rear}}
-    end.
-
 %% -------------------------------------------------------------------------
 
 death_row(Reason) ->
-- 
2.35.3

Places

File 4476-Clean-up-output_handler.patch of Package erlang (Revision 08848386a4ea82f272aa9c86a95948ae)

Places