From ead8d3a41d224c6a2e9a4a4b8005e18159218de9 Mon Sep 17 00:00:00 2001
From: Kresten Krab Thorup <krab@trifork.com>
Date: Thu, 19 Jan 2012 14:19:16 +0100
Subject: [PATCH] Make lsm_btree:close/1 stop more processes

Closing a tree did not stop ongoing merge
processes beyond the current top level.
Now close synchronously calls down through all
levels and closes each one.
---
 src/lsm_btree.erl       | 20 ++++++++++++++++----
 src/lsm_btree_level.erl | 15 ++++++++++++++-
 2 files changed, 30 insertions(+), 5 deletions(-)

diff --git a/src/lsm_btree.erl b/src/lsm_btree.erl
index 2c32290..a7f98d7 100644
--- a/src/lsm_btree.erl
+++ b/src/lsm_btree.erl
@@ -19,7 +19,12 @@ open(Dir) ->
     gen_server:start(?MODULE, [Dir], []).
 
 close(Ref) ->
-    gen_server:call(Ref, close).
+    try
+        gen_server:call(Ref, close)
+    catch
+        exit:{noproc,_} -> ok
+    end.
+
 
 lookup(Ref,Key) when is_binary(Key) ->
     gen_server:call(Ref, {lookup, Key}).
@@ -147,9 +152,16 @@ handle_call({lookup, Key}, _From, State=#state{ top=Top, nursery=Nursery } ) whe
             {reply, Reply, State}
     end;
 
-handle_call(close, _From, State) ->
-    {ok, State2} = flush_nursery(State),
-    {stop, normal, ok, State2}.
+handle_call(close, _From, State=#state{top=Top}) ->
+    try
+        {ok, State2} = flush_nursery(State),
+        ok = lsm_btree_level:close(Top),
+        {stop, normal, ok, State2}
+    catch
+        E:R ->
+            error_logger:info_msg("exception from close ~p:~p~n", [E,R]),
+            {stop, normal, ok, State}
+    end.
 
 do_put(Key, Value, State=#state{ nursery=Nursery, top=Top }) ->
     {ok, Nursery2} = lsm_btree_nursery:add_maybe_flush(Key, Value, Nursery, Top),
diff --git a/src/lsm_btree_level.erl b/src/lsm_btree_level.erl
index c93ff45..21fffb0 100644
--- a/src/lsm_btree_level.erl
+++ b/src/lsm_btree_level.erl
@@ -41,7 +41,12 @@ inject(Ref, FileName) ->
     Result.
 
 close(Ref) ->
-    call(Ref, close).
+    try
+        call(Ref, close)
+    catch
+        exit:{noproc,_} -> ok
+    end.
+
 
 range_fold(Ref, SendTo, From, To) ->
     {ok, FoldWorkerPID} = lsm_btree_fold_worker:start(SendTo),
@@ -170,6 +175,14 @@ main_loop(State = #state{ next=Next }) ->
             close_if_defined(State#state.b),
             stop_if_defined(State#state.merge_pid),
             reply(From, ok),
+
+            %% this is synchronous all the way down, because our
+            %% caller is monitoring *this* proces, and thus the
+            %% rpc would fail when we fall off the cliff
+            if Next == undefined -> ok;
+               true ->
+                    lsm_btree_level:close(Next)
+            end,
             ok;
 
         ?REQ(From, {init_range_fold, WorkerPID, FromKey, ToKey, List}) when State#state.folding == 0 ->