Add option to compress data when copying tables between Mnesia nodes

Optionally using data compression for copying Mnesia tables allows
the system to be tuned to eliminate network bottlenecks from
deployments where enough CPU is available to use zlib.

With this patch, running erl -mnesia send_compressed <level>
will enable compression for sending tables between nodes. The
compression level can be any integer in [0, 9], with 0 (the
default) meaning no compression (exactly the previous behaviour)
and 9 being the highest compression level.

To set any non-zero compression level at the sender, both nodes
must have the updated Mnesia modules (the receiver will work
according to the sender's configuration).

4 files changed, 53 insertions, 5 deletions

diff --git a/lib/mnesia/doc/src/mnesia.xml b/lib/mnesia/doc/src/mnesia.xml
index 3484cd104a..33387a88f4 100644
--- a/lib/mnesia/doc/src/mnesia.xml
+++ b/lib/mnesia/doc/src/mnesia.xml
@@ -3045,6 +3045,17 @@ raise(Name, Amount) ->
           </p>
       </item>
       <item>
+        <p><c>-mnesia send_compressed Level</c> specifies the level of
+          compression to be used when copying a table from the local node to
+          another one. The default level is 0.
+          </p>
+        <p><c>Level</c> must be an integer in the interval [0, 9], with 0
+          representing no compression and 9 representing maximum compression.
+          Before setting it to a non-zero value, make sure the remote nodes
+          understand this configuration.
+          </p>
+      </item>
+      <item>
         <p><c>-mnesia schema_location Loc</c> controls where
           Mnesia will look for its schema. The parameter
           <c>Loc</c> may be one of the following atoms: </p>
diff --git a/lib/mnesia/src/mnesia.erl b/lib/mnesia/src/mnesia.erl
index 9a630f18eb..c9599a1911 100644
--- a/lib/mnesia/src/mnesia.erl
+++ b/lib/mnesia/src/mnesia.erl
@@ -2200,6 +2200,7 @@ system_info2(transaction_log_writes) -> mnesia_dumper:get_log_writes();
 system_info2(core_dir) ->  mnesia_monitor:get_env(core_dir); 
 system_info2(no_table_loaders) ->  mnesia_monitor:get_env(no_table_loaders); 
 system_info2(dc_dump_limit) ->  mnesia_monitor:get_env(dc_dump_limit); 
+system_info2(send_compressed) -> mnesia_monitor:get_env(send_compressed);
 
 system_info2(Item) -> exit({badarg, Item}).
 
@@ -2244,6 +2245,7 @@ system_info_items(yes) ->
      core_dir,
      no_table_loaders,
      dc_dump_limit,
+     send_compressed,
      version
     ];
 system_info_items(no) ->
diff --git a/lib/mnesia/src/mnesia_loader.erl b/lib/mnesia/src/mnesia_loader.erl
index 77c317abc5..5657f6e299 100644
--- a/lib/mnesia/src/mnesia_loader.erl
+++ b/lib/mnesia/src/mnesia_loader.erl
@@ -438,6 +438,9 @@ make_table_fun(Pid, TabRec) ->
 
 get_data(Pid, TabRec) ->
     receive 
+	{Pid, {more_z, CompressedRecs}} when is_binary(CompressedRecs) ->
+	    Pid ! {TabRec, more},
+	    {zlib_uncompress(CompressedRecs), make_table_fun(Pid,TabRec)};
 	{Pid, {more, Recs}} ->
 	    Pid ! {TabRec, more},
 	    {Recs, make_table_fun(Pid,TabRec)};
@@ -769,6 +772,27 @@ dets_bchunk(Tab, Chunk) -> %% Arrg
 	Else -> Else
     end.	    
 
+zlib_compress(Data, Level) ->
+    BinData = term_to_binary(Data),
+    Z = zlib:open(),
+    zlib:deflateInit(Z, Level),
+    Bs = zlib:deflate(Z, BinData, finish),
+    zlib:deflateEnd(Z),
+    zlib:close(Z),
+    list_to_binary(Bs).
+
+zlib_uncompress(Data) when is_binary(Data) ->
+    binary_to_term(zlib:uncompress(Data)).
+
+compression_level() ->
+    NoCompression = 0,
+    case ?catch_val(send_compressed) of
+	{'EXIT', _} ->
+	    mnesia_lib:set(send_compressed, NoCompression),
+	    NoCompression;
+	Val -> Val
+    end.
+
 send_packet(N, Pid, _Chunk, '$end_of_table', OldNode) ->
     case OldNode of
 	true -> ignore; %% Old nodes can't handle the new no_more
@@ -779,8 +803,15 @@ send_packet(N, Pid, Chunk, {[], Cont}, OldNode) ->
     send_packet(N, Pid, Chunk, Chunk(Cont), OldNode);
 send_packet(N, Pid, Chunk, {Recs, Cont}, OldNode) when N < ?MAX_NOPACKETS ->
     case OldNode of
-	true -> Pid ! {self(), {more, [Recs]}}; %% Old need's wrapping list
-	false -> Pid ! {self(), {more, Recs}}
+        true ->
+            Pid ! {self(), {more, [Recs]}}; %% Old need's wrapping list
+        false ->
+            case compression_level() of
+                0 ->
+                    Pid ! {self(), {more, Recs}};
+                Level ->
+                    Pid ! {self(), {more_z, zlib_compress(Recs, Level)}}
+            end
     end,
     send_packet(N+1, Pid, Chunk, Chunk(Cont), OldNode);
 send_packet(_N, _Pid, _Chunk, DataState, _OldNode) ->
diff --git a/lib/mnesia/src/mnesia_monitor.erl b/lib/mnesia/src/mnesia_monitor.erl
index 05ae943e3b..b39ee92fa8 100644
--- a/lib/mnesia/src/mnesia_monitor.erl
+++ b/lib/mnesia/src/mnesia_monitor.erl
@@ -674,7 +674,8 @@ env() ->
      core_dir,
      pid_sort_order,
      no_table_loaders,
-     dc_dump_limit
+     dc_dump_limit,
+     send_compressed
     ].
 
 default_env(access_module) -> 
@@ -717,7 +718,9 @@ default_env(pid_sort_order) ->
 default_env(no_table_loaders) ->
     2;
 default_env(dc_dump_limit) ->
-    4.
+    4;
+default_env(send_compressed) ->
+    0.
 
 check_type(Env, Val) ->
     case catch do_check_type(Env, Val) of
@@ -763,7 +766,8 @@ do_check_type(pid_sort_order, standard) -> standard;
 do_check_type(pid_sort_order, "standard") -> standard;
 do_check_type(pid_sort_order, _) -> false;
 do_check_type(no_table_loaders, N) when is_integer(N), N > 0 -> N;
-do_check_type(dc_dump_limit,N) when is_number(N), N > 0 -> N.
+do_check_type(dc_dump_limit,N) when is_number(N), N > 0 -> N;
+do_check_type(send_compressed, L) when is_integer(L), L >= 0, L =< 9 -> L.
 
 bool(true) -> true;
 bool(false) -> false.