ssl: Internal active n must back off when user does not read data

TLS connections should not buffer too much application data if they
want to benefit from TCP flow control. Certain applications may want to
customize the value of internal_active_n as there is a tradeoff between
buffering memory  and throughput.

1 files changed, 55 insertions, 35 deletions

diff --git a/lib/ssl/src/tls_connection.erl b/lib/ssl/src/tls_connection.erl
index b848f82575..b88c7292a7 100644
--- a/lib/ssl/src/tls_connection.erl
+++ b/lib/ssl/src/tls_connection.erl
@@ -142,32 +142,60 @@ pids(#state{protocol_specific = #{sender := Sender}}) ->
 %%====================================================================
 %% State transition handling
 %%====================================================================
-next_record(#state{handshake_env = 
+next_record(_, #state{handshake_env = 
                        #handshake_env{unprocessed_handshake_events = N} = HsEnv} 
             = State) when N > 0 ->
     {no_record, State#state{handshake_env = 
                                 HsEnv#handshake_env{unprocessed_handshake_events = N-1}}};
-next_record(#state{protocol_buffers =
-                       #protocol_buffers{tls_cipher_texts = [_|_] = CipherTexts},
-                   connection_states = ConnectionStates,
-                   ssl_options = #ssl_options{padding_check = Check}} = State) ->
+next_record(_, #state{protocol_buffers =
+                          #protocol_buffers{tls_cipher_texts = [_|_] = CipherTexts},
+                      connection_states = ConnectionStates,
+                      ssl_options = #ssl_options{padding_check = Check}} = State) ->
     next_record(State, CipherTexts, ConnectionStates, Check);
-next_record(#state{protocol_buffers = #protocol_buffers{tls_cipher_texts = []},
-                   protocol_specific = #{active_n_toggle := true, active_n := N} = ProtocolSpec,
-                   static_env = #static_env{socket = Socket,
-                                            close_tag = CloseTag,
-                                            transport_cb = Transport}  
-                  } = State) ->
-    case tls_socket:setopts(Transport, Socket, [{active, N}]) of
- 	ok ->
-             {no_record, State#state{protocol_specific = ProtocolSpec#{active_n_toggle => false}}}; 
- 	_ ->
-             self() ! {CloseTag, Socket},
-             {no_record, State}
-     end;
-next_record(State) ->
+next_record(connection, #state{protocol_buffers = #protocol_buffers{tls_cipher_texts = []},
+                               protocol_specific = #{active_n_toggle := true}
+                              } = State) ->
+    %% If ssl application user is not reading data wait to activate socket
+    flow_ctrl(State); 
+  
+next_record(_, #state{protocol_buffers = #protocol_buffers{tls_cipher_texts = []},
+                      protocol_specific = #{active_n_toggle := true}
+                     } = State) ->
+    activate_socket(State);
+next_record(_, State) ->
     {no_record, State}.
 
+
+flow_ctrl(#state{user_data_buffer = {_,Size,_},
+                 socket_options = #socket_options{active = false},
+                 bytes_to_read = undefined} = State)  when Size =/= 0 ->
+    {no_record, State};
+flow_ctrl(#state{user_data_buffer = {_,Size,_},
+                 socket_options = #socket_options{active = false},
+                 bytes_to_read = 0} = State)  when Size =/= 0 ->
+    {no_record, State};
+flow_ctrl(#state{user_data_buffer = {_,Size,_}, 
+                 socket_options = #socket_options{active = false},
+                 bytes_to_read = BytesToRead} = State) when (Size >= BytesToRead) andalso
+                                                            (BytesToRead > 0) ->
+    {no_record, State};
+flow_ctrl(State) ->
+    activate_socket(State).
+
+
+activate_socket(#state{protocol_specific = #{active_n_toggle := true, active_n := N} = ProtocolSpec,
+                       static_env = #static_env{socket = Socket,
+                                                close_tag = CloseTag,
+                                                transport_cb = Transport}  
+                      } = State) ->                                                                                                            
+    case tls_socket:setopts(Transport, Socket, [{active, N}]) of
+        ok ->
+            {no_record, State#state{protocol_specific = ProtocolSpec#{active_n_toggle => false}}}; 
+        _ ->
+            self() ! {CloseTag, Socket},
+            {no_record, State}
+    end.
+
 %% Decipher next record and concatenate consecutive ?APPLICATION_DATA records into one
 %%
 next_record(State, CipherTexts, ConnectionStates, Check) ->
@@ -198,28 +226,20 @@ next_record_done(#state{protocol_buffers = Buffers} = State, CipherTexts, Connec
      State#state{protocol_buffers = Buffers#protocol_buffers{tls_cipher_texts = CipherTexts},
                  connection_states = ConnectionStates}}.
 
-
 next_event(StateName, Record, State) ->
     next_event(StateName, Record, State, []).
 %%
 next_event(StateName, no_record, State0, Actions) ->
-    case next_record(State0) of
+    case next_record(StateName, State0) of
  	{no_record, State} ->
             {next_state, StateName, State, Actions};
- 	{#ssl_tls{} = Record, State} ->
- 	    {next_state, StateName, State, [{next_event, internal, {protocol_record, Record}} | Actions]};
-	#alert{} = Alert ->
-	    {next_state, StateName, State0, [{next_event, internal, Alert} | Actions]}
+        {Record, State} ->
+            next_event(StateName, Record, State, Actions)
     end;
-next_event(StateName, Record, State, Actions) ->
-    case Record of
-	no_record ->
-	    {next_state, StateName, State, Actions};
-	#ssl_tls{} = Record ->
-	    {next_state, StateName, State, [{next_event, internal, {protocol_record, Record}} | Actions]};
-	#alert{} = Alert ->
-	    {next_state, StateName, State, [{next_event, internal, Alert} | Actions]}
-    end.
+next_event(StateName,  #ssl_tls{} = Record, State, Actions) ->
+    {next_state, StateName, State, [{next_event, internal, {protocol_record, Record}} | Actions]};
+next_event(StateName,  #alert{} = Alert, State, Actions) ->
+    {next_state, StateName, State, [{next_event, internal, Alert} | Actions]}.
 
 
 %%% TLS record protocol level application data messages 
@@ -869,7 +889,7 @@ next_tls_record(Data, StateName,
     case tls_record:get_tls_records(Data, Versions, Buf0) of
 	{Records, Buf1} ->
 	    CT1 = CT0 ++ Records,
-	    next_record(State0#state{protocol_buffers =
+	    next_record(StateName, State0#state{protocol_buffers =
 					 Buffers#protocol_buffers{tls_record_buffer = Buf1,
 								  tls_cipher_texts = CT1}});
 	#alert{} = Alert ->