Make sure disk_log does not write too much data

While disk_log eagerly collects logged data for better performance,
collecting too much data may choke the system and cause huge binaries
to be written. (Reported by Klarna.)

(Logging data faster than the speed disk_log can write to disk can
still cause problems.)

1 files changed, 28 insertions, 23 deletions

diff --git a/lib/kernel/src/disk_log.erl b/lib/kernel/src/disk_log.erl
index d6bc23be6d..6fb5b6e2ad 100644
--- a/lib/kernel/src/disk_log.erl
+++ b/lib/kernel/src/disk_log.erl
@@ -64,7 +64,7 @@
 %%-define(PROFILE(C), C).
 -define(PROFILE(C), void).
 
--compile({inline,[{log_loop,4},{log_end_sync,2},{replies,2},{rflat,1}]}).
+-compile({inline,[{log_loop,5},{log_end_sync,2},{replies,2},{rflat,1}]}).
 
 %%%----------------------------------------------------------------------
 %%% Contract type specifications
@@ -685,7 +685,7 @@ handle({From, {log, B}}, S) ->
 	L when L#log.mode =:= read_only ->
 	    reply(From, {error, {read_only_mode, L#log.name}}, S);
 	L when L#log.status =:= ok, L#log.format =:= internal ->
-	    log_loop(S, From, [B], []);
+	    log_loop(S, From, [B], [], iolist_size(B));
 	L when L#log.status =:= ok, L#log.format =:= external ->
 	    reply(From, {error, {format_external, L#log.name}}, S);
 	L when L#log.status =:= {blocked, false} ->
@@ -700,7 +700,7 @@ handle({From, {blog, B}}, S) ->
 	L when L#log.mode =:= read_only ->
 	    reply(From, {error, {read_only_mode, L#log.name}}, S);
 	L when L#log.status =:= ok ->
-	    log_loop(S, From, [B], []);
+	    log_loop(S, From, [B], [], iolist_size(B));
 	L when L#log.status =:= {blocked, false} ->
 	    reply(From, {error, {blocked_log, L#log.name}}, S);
 	L when L#log.blocked_by =:= From ->
@@ -714,7 +714,7 @@ handle({alog, B}, S) ->
 	    notify_owners({read_only,B}),
 	    loop(S);
 	L when L#log.status =:= ok, L#log.format =:= internal ->
-	    log_loop(S, [], [B], []);
+	    log_loop(S, [], [B], [], iolist_size(B));
 	L when L#log.status =:= ok ->
 	    notify_owners({format_external, B}),
 	    loop(S);
@@ -730,7 +730,7 @@ handle({balog, B}, S) ->
 	    notify_owners({read_only,B}),
 	    loop(S);
 	L when L#log.status =:= ok ->
-	    log_loop(S, [], [B], []);
+	    log_loop(S, [], [B], [], iolist_size(B));
 	L when L#log.status =:= {blocked, false} ->
 	    notify_owners({blocked_log, B}),
 	    loop(S);
@@ -1029,38 +1029,43 @@ handle(_, S) ->
     loop(S).
 
 sync_loop(From, S) ->
-    log_loop(S, [], [], From).
+    log_loop(S, [], [], From, 0).
+
+-define(MAX_LOOK_AHEAD, 64*1024).
 
 %% Inlined.
-log_loop(S, Pids, _Bins, _Sync) when S#state.cache_error =/= ok ->
+log_loop(S, Pids, _Bins, _Sync, _Sz) when S#state.cache_error =/= ok ->
     loop(cache_error(S, Pids));
-log_loop(S, Pids, Bins, Sync) when S#state.messages =:= [] ->
+log_loop(#state{messages = []}=S, Pids, Bins, Sync, Sz)
+            when Sz > ?MAX_LOOK_AHEAD ->
+erlang:display({rad,12}),
+    loop(log_end(S, Pids, Bins, Sync));
+log_loop(#state{messages = []}=S, Pids, Bins, Sync, Sz) ->
     receive 
 	Message ->
-	    log_loop(Message, Pids, Bins, Sync, S, get(log))
+            log_loop(Message, Pids, Bins, Sync, Sz, S, get(log))
     after 0 ->
 	    loop(log_end(S, Pids, Bins, Sync))
     end;
-log_loop(S, Pids, Bins, Sync) ->
+log_loop(S, Pids, Bins, Sync, Sz) ->
     [M | Ms] = S#state.messages,
     S1 = S#state{messages = Ms},
-    log_loop(M, Pids, Bins, Sync, S1, get(log)).
+    log_loop(M, Pids, Bins, Sync, Sz, S1, get(log)).
 
 %% Items logged after the last sync request found are sync:ed as well.
-log_loop({alog,B}, Pids, Bins, Sync, S, L) when L#log.format =:= internal ->
+log_loop({alog,B}, Pids, Bins, Sync, Sz, S, #log{format = internal}) ->
     %% {alog, _} allowed for the internal format only.
-    log_loop(S, Pids, [B | Bins], Sync);
-log_loop({balog, B}, Pids, Bins, Sync, S, _L) ->
-    log_loop(S, Pids, [B | Bins], Sync);
-log_loop({From, {log, B}}, Pids, Bins, Sync, S, L) 
-                                           when L#log.format =:= internal ->
+    log_loop(S, Pids, [B | Bins], Sync, Sz+iolist_size(B));
+log_loop({balog, B}, Pids, Bins, Sync, Sz, S, _L) ->
+    log_loop(S, Pids, [B | Bins], Sync, Sz+iolist_size(B));
+log_loop({From, {log, B}}, Pids, Bins, Sync, Sz, S, #log{format = internal}) ->
     %% {log, _} allowed for the internal format only.
-    log_loop(S, [From | Pids], [B | Bins], Sync);
-log_loop({From, {blog, B}}, Pids, Bins, Sync, S, _L) ->
-    log_loop(S, [From | Pids], [B | Bins], Sync);
-log_loop({From, sync}, Pids, Bins, Sync, S, _L) ->
-    log_loop(S, Pids, Bins, [From | Sync]);
-log_loop(Message, Pids, Bins, Sync, S, _L) ->
+    log_loop(S, [From | Pids], [B | Bins], Sync, Sz+iolist_size(B));
+log_loop({From, {blog, B}}, Pids, Bins, Sync, Sz, S, _L) ->
+    log_loop(S, [From | Pids], [B | Bins], Sync, Sz+iolist_size(B));
+log_loop({From, sync}, Pids, Bins, Sync, Sz, S, _L) ->
+    log_loop(S, Pids, Bins, [From | Sync], Sz);
+log_loop(Message, Pids, Bins, Sync, _Sz, S, _L) ->
     NS = log_end(S, Pids, Bins, Sync),
     handle(Message, NS).