Add LRU for view groups

src/couch/include/couch_db.hrl

@@ -231,6 +231,12 @@
     atts = []
 }).
 
+-record(couch_lru, {
+    count=0,
+    updates,
+    counts,
+    close_fun
+}).
 
 -type doc() :: #doc{}.
 -type ddoc() :: #doc{}.

src/couch/src/couch_lru.erl

@@ -11,61 +11,56 @@
 % the License.
 
 -module(couch_lru).
--export([new/0, insert/2, update/2, close/1]).
+-export([new/1, insert/2, update/2, close/1]).
 
 -include_lib("couch/include/couch_db.hrl").
 
-new() ->
+new(CloseFun) ->
     Updates = ets:new(couch_lru_updates, [ordered_set]),
-    Dbs = ets:new(couch_lru_dbs, [set]),
-    {0, Updates, Dbs}.
+    Counts = ets:new(couch_lru_counts, [set]),
+    #couch_lru{updates=Updates, counts=Counts, close_fun=CloseFun}.
 
-insert(DbName, {Count, Updates, Dbs}) ->
-    update(DbName, {Count, Updates, Dbs}).
+insert(Name, Lru) ->
+    update(Name, Lru).
 
-update(DbName, {Count, Updates, Dbs}) ->
-    case ets:lookup(Dbs, DbName) of
+update(Name, Lru) ->
+    #couch_lru{counts=Counts, updates=Updates, count=Count} = Lru,
+    case ets:lookup(Counts, Name) of
         [] ->
-            true = ets:insert(Dbs, {DbName, Count});
-        [{DbName, OldCount}] ->
-            true = ets:update_element(Dbs, DbName, {2, Count}),
-            true = ets:delete(Updates, {OldCount, DbName})
+            true = ets:insert(Counts, {Name, Count});
+        [{Name, OldCount}] ->
+            true = ets:update_element(Counts, Name, {2, Count}),
+            true = ets:delete(Updates, {OldCount, Name})
     end,
-    true = ets:insert(Updates, {{Count, DbName}}),
-    {Count + 1, Updates, Dbs}.
+    true = ets:insert(Updates, {{Count, Name}}),
+    Lru#couch_lru{count=Count+1}.
 
 
-close({Count, Updates, Dbs}) ->
-    case close_int(ets:next(Updates, {-1, <<>>}), Updates, Dbs) of
+close(Lru) ->
+    #couch_lru{updates=Updates} = Lru,
+    case close_int(ets:next(Updates, {-1, <<>>}), Lru) of
         true ->
-            {true, {Count, Updates, Dbs}};
+            {true, Lru};
         false ->
             false
     end.
 
 
 %% internals
 
-close_int('$end_of_table', _Updates, _Dbs) ->
+close_int('$end_of_table', _Lru) ->
     false;
-close_int({_Count, DbName} = Key, Updates, Dbs) ->
-    case ets:update_element(couch_dbs, DbName, {#db.fd_monitor, locked}) of
-    true ->
-        [#db{main_pid = Pid} = Db] = ets:lookup(couch_dbs, DbName),
-        case couch_db:is_idle(Db) of true ->
-            true = ets:delete(couch_dbs, DbName),
-            true = ets:delete(couch_dbs_pid_to_name, Pid),
-            exit(Pid, kill),
+close_int({_Count, Name} = Key, Lru) ->
+    #couch_lru{updates=Updates, counts=Counts, close_fun=CloseFun} = Lru,
+    {Stop, Remove} = CloseFun(Name),
+    case Remove of
+        true ->
             true = ets:delete(Updates, Key),
-            true = ets:delete(Dbs, DbName),
-            true;
+            true = ets:delete(Counts, Name);
         false ->
-            true = ets:update_element(couch_dbs, DbName, {#db.fd_monitor, nil}),
-            couch_stats:increment_counter([couchdb, couch_server, lru_skip]),
-            close_int(ets:next(Updates, Key), Updates, Dbs)
-        end;
-    false ->
-        true = ets:delete(Updates, Key),
-        true = ets:delete(Dbs, DbName),
-        close_int(ets:next(Updates, Key), Updates, Dbs)
+            ok
+    end,
+    case Stop of
+        false -> close_int(ets:next(Updates, Key), Lru);
+        true -> true
     end.

src/couch/src/couch_server.erl

@@ -20,7 +20,7 @@
 -export([init/1, handle_call/3,sup_start_link/0]).
 -export([handle_cast/2,code_change/3,handle_info/2,terminate/2]).
 -export([dev_start/0,is_admin/2,has_admins/0,get_stats/0]).
--export([close_lru/0]).
+-export([close_lru/0, maybe_close_db/1]).
 
 % config_listener api
 -export([handle_config_change/5, handle_config_terminate/3]).
@@ -36,7 +36,7 @@
     dbs_open=0,
     start_time="",
     update_lru_on_read=true,
-    lru = couch_lru:new()
+    lru = couch_lru:new(fun maybe_close_db/1)
     }).
 
 dev_start() ->
@@ -97,6 +97,26 @@ update_lru(DbName) ->
 close_lru() ->
     gen_server:call(couch_server, close_lru).
 
+maybe_close_db(DbName) ->
+    % First element of return indicates if we should stop closing DBs from the
+    % LRU and the second element indicates if we closed a DB.
+    case ets:update_element(couch_dbs, DbName, {#db.fd_monitor, locked}) of
+    true ->
+        [#db{main_pid = Pid} = Db] = ets:lookup(couch_dbs, DbName),
+        case couch_db:is_idle(Db) of true ->
+            true = ets:delete(couch_dbs, DbName),
+            true = ets:delete(couch_dbs_pid_to_name, Pid),
+            exit(Pid, kill),
+            {true, true};
+        false ->
+            true = ets:update_element(couch_dbs, DbName, {#db.fd_monitor, nil}),
+            couch_stats:increment_counter([couchdb, couch_server, lru_skip]),
+            {false, false}
+        end;
+    false ->
+        {false, true}
+    end.
+
 create(DbName, Options0) ->
     Options = maybe_add_sys_db_callbacks(DbName, Options0),
     case gen_server:call(couch_server, {create, DbName, Options}, infinity) of

src/couch/test/couch_lru_tests.erl

@@ -20,7 +20,7 @@ setup() ->
     ok = meck:new(couch_db, [passthrough]),
     ets:new(couch_dbs, [set, public, named_table, {keypos, #db.name}]),
     ets:new(couch_dbs_pid_to_name, [set, public, named_table]),
-    couch_lru:new().
+    couch_lru:new(fun couch_server:maybe_close_db/1).
 
 teardown(_) ->
     ets:delete(couch_dbs),
@@ -29,18 +29,18 @@ teardown(_) ->
 
 new_test_() ->
     {setup,
-        fun() -> couch_lru:new() end,
+        fun() -> couch_lru:new(fun couch_server:maybe_close_db/1) end,
         fun(Lru) ->
-            ?_assertMatch({0, _, _}, Lru)
+            ?_assertEqual(0, Lru#couch_lru.count)
         end
     }.
 
 insert_test_() ->
     {setup,
-        fun() -> couch_lru:new() end,
+        fun() -> couch_lru:new(fun couch_server:maybe_close_db/1) end,
         fun(Lru) ->
             Key = <<"test">>,
-            {1, Updates, Dbs} = couch_lru:insert(Key, Lru),
+            #couch_lru{count=1, updates=Updates, counts=Dbs} = couch_lru:insert(Key, Lru),
             [
                 ?_assertEqual(1, ets_size(Dbs)),
                 ?_assert(ets:member(Dbs, Key)),
@@ -52,11 +52,11 @@ insert_test_() ->
 
 insert_same_test_() ->
     {setup,
-        fun() -> couch_lru:new() end,
+        fun() -> couch_lru:new(fun couch_server:maybe_close_db/1) end,
         fun(Lru) ->
             Key = <<"test">>,
-            Lru1 = {1, Updates, Dbs} = couch_lru:insert(Key, Lru),
-            {2, Updates, Dbs} = couch_lru:insert(Key, Lru1),
+            Lru1 = #couch_lru{count=1} = couch_lru:insert(Key, Lru),
+            #couch_lru{count=2, updates=Updates, counts=Dbs} = couch_lru:insert(Key, Lru1),
             [
                 ?_assertEqual(1, ets_size(Dbs)),
                 ?_assert(ets:member(Dbs, Key)),
@@ -68,11 +68,11 @@ insert_same_test_() ->
 
 update_test_() ->
     {setup,
-        fun() -> couch_lru:new() end,
+        fun() -> couch_lru:new(fun couch_server:maybe_close_db/1) end,
         fun(Lru) ->
             Key = <<"test">>,
-            Lru1 = {1, Updates, Dbs} = couch_lru:update(Key, Lru),
-            {2, Updates, Dbs} = couch_lru:update(Key, Lru1),
+            Lru1 = #couch_lru{count=1} = couch_lru:update(Key, Lru),
+            #couch_lru{count=2, updates=Updates, counts=Dbs} = couch_lru:update(Key, Lru1),
             [
                 ?_assertEqual(1, ets_size(Dbs)),
                 ?_assert(ets:member(Dbs, Key)),
@@ -90,7 +90,7 @@ close_test_() ->
             ok = meck:expect(couch_db, is_idle, 1, true),
             {ok, Lru1} = add_record(Lru, <<"test1">>, c:pid(0, 1001, 0)),
             {ok, Lru2} = add_record(Lru1, <<"test2">>, c:pid(0, 2001, 0)),
-            {true, {2, Updates, Dbs}} = couch_lru:close(Lru2),
+            {true, #couch_lru{count=2, updates=Updates, counts=Dbs}} = couch_lru:close(Lru2),
             [
                 ?_assertEqual(1, ets_size(Dbs)),
                 ?_assert(ets:member(Dbs, <<"test2">>)),

src/couch/test/couchdb_compaction_daemon_tests.erl

@@ -222,8 +222,9 @@ spawn_compaction_monitor(DbName) ->
         DbPid = couch_util:with_db(DbName, fun(Db) ->
             Db#db.main_pid
         end),
-        {ok, ViewPid} = couch_index_server:get_index(couch_mrview_index,
+        {ok, ViewPid, Mon} = couch_index_server:get_index(couch_mrview_index,
                 DbName, <<"_design/foo">>),
+        couch_index_server:close(Mon),
         TestPid ! {self(), started},
         receive
             {TestPid, go} -> ok

src/couch/test/couchdb_views_tests.erl

@@ -226,7 +226,7 @@ should_cleanup_all_index_files({DbName, {FooRev, BooRev}})->
 
 couchdb_1138(DbName) ->
     ?_test(begin
-        {ok, IndexerPid} = couch_index_server:get_index(
+        {ok, IndexerPid, Mon} = couch_index_server:get_index(
             couch_mrview_index, DbName, <<"_design/foo">>),
         ?assert(is_pid(IndexerPid)),
         ?assert(is_process_alive(IndexerPid)),
@@ -259,12 +259,14 @@ couchdb_1138(DbName) ->
         ?assertEqual(5, length(Rows2)),
         ?assertEqual(2, count_users(DbName)),
 
-        ?assert(is_process_alive(IndexerPid))
+        ?assert(is_process_alive(IndexerPid)),
+
+        couch_index_server:close(Mon)
     end).
 
 couchdb_1309(DbName) ->
     ?_test(begin
-        {ok, IndexerPid} = couch_index_server:get_index(
+        {ok, IndexerPid, Mon} = couch_index_server:get_index(
             couch_mrview_index, DbName, <<"_design/foo">>),
         ?assert(is_pid(IndexerPid)),
         ?assert(is_process_alive(IndexerPid)),
@@ -281,7 +283,7 @@ couchdb_1309(DbName) ->
         ?assert(is_process_alive(IndexerPid)),
 
         update_design_doc(DbName,  <<"_design/foo">>, <<"bar">>),
-        {ok, NewIndexerPid} = couch_index_server:get_index(
+        {ok, NewIndexerPid, NewMon} = couch_index_server:get_index(
             couch_mrview_index, DbName, <<"_design/foo">>),
         ?assert(is_pid(NewIndexerPid)),
         ?assert(is_process_alive(NewIndexerPid)),
@@ -301,12 +303,15 @@ couchdb_1309(DbName) ->
         ?assertEqual(4, length(Rows2)),
 
         ok = stop_indexer( %% FIXME we need to grab monitor earlier
-               fun() -> ok end,
+               fun() -> couch_index_server:close(Mon) end,
                IndexerPid, ?LINE,
                "old view group is not dead after ddoc update"),
 
         ok = stop_indexer(
-               fun() -> couch_server:delete(DbName, [?ADMIN_USER]) end,
+               fun() ->
+                   couch_index_server:close(NewMon),
+                   couch_server:delete(DbName, [?ADMIN_USER])
+               end,
                NewIndexerPid, ?LINE,
                "new view group did not die after DB deletion")
     end).
@@ -363,9 +368,10 @@ couchdb_1283() ->
         %% because we need have access to compaction Pid, not a Ref.
         %% {ok, MonRef} = couch_mrview:compact(MDb1#db.name, <<"_design/foo">>,
         %%                                     [monitor]),
-        {ok, Pid} = couch_index_server:get_index(
+        {ok, Pid, Mon} = couch_index_server:get_index(
             couch_mrview_index, MDb1#db.name, <<"_design/foo">>),
         {ok, CPid} = gen_server:call(Pid, compact),
+        couch_index_server:close(Mon),
         %% By suspending compaction process we ensure that compaction won't get
         %% finished too early to make get_writer_status assertion fail.
         erlang:suspend_process(CPid),

src/couch_index/priv/stats_descriptions.cfg

@@ -0,0 +1,20 @@
+%% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+%% use this file except in compliance with the License. You may obtain a copy of
+%% the License at
+%%
+%%   http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+%% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+%% License for the specific language governing permissions and limitations under
+%% the License.
+
+% Style guide for descriptions: Start with a lowercase letter & do not add
+% a trailing full-stop / period
+% Please keep this in alphabetical order
+
+{[couchdb, couch_index_server, lru_skip], [
+    {type, counter},
+    {desc, <<"number of couch_index_server LRU operations skipped">>}
+]}.

src/couch_index/src/couch_index.erl

@@ -240,8 +240,11 @@ handle_cast({new_state, NewIdxState}, State) ->
     couch_log:debug("Updated index for db: ~s idx: ~s seq: ~B", Args),
     Rest = send_replies(State#st.waiters, CurrSeq, NewIdxState),
     case State#st.committed of
-        true -> erlang:send_after(commit_delay(), self(), commit);
-        false -> ok
+        true ->
+            ok = couch_index_server:set_committing(self(), true),
+            erlang:send_after(commit_delay(), self(), commit);
+        false ->
+            ok
     end,
     {noreply, State#st{
         idx_state=NewIdxState,
@@ -297,6 +300,7 @@ handle_info(commit, State) ->
             % Commit the updates
             ok = Mod:commit(IdxState),
             couch_event:notify(DbName, {index_commit, IdxName}),
+            ok = couch_index_server:set_committing(self(), false),
             {noreply, State#st{committed=true}};
         _ ->
             % We can't commit the header because the database seq that's
@@ -305,6 +309,7 @@ handle_info(commit, State) ->
             % forever out of sync with the database. But a crash before we
             % commit these changes, no big deal, we only lose incremental
             % changes since last committal.
+            ok = couch_index_server:set_committing(self(), true),
             erlang:send_after(commit_delay(), self(), commit),
             {noreply, State}
     end;
@@ -385,8 +390,11 @@ commit_compacted(NewIdxState, State) ->
         false -> ok
     end,
     case State#st.committed of
-        true -> erlang:send_after(commit_delay(), self(), commit);
-        false -> ok
+        true ->
+            ok = couch_index_server:set_committing(self(), true),
+            erlang:send_after(commit_delay(), self(), commit);
+        false ->
+            ok
     end,
     State#st{
         idx_state=NewIdxState1,

src/couch_index/src/couch_index_compactor.erl

@@ -64,12 +64,14 @@ handle_call({compact, _}, _From, #st{pid=Pid}=State) when is_pid(Pid) ->
     {reply, {ok, Pid}, State};
 handle_call({compact, IdxState}, _From, #st{idx=Idx}=State) ->
     Pid = spawn_link(fun() -> compact(Idx, State#st.mod, IdxState) end),
+    ok = couch_index_server:set_compacting(Idx, true),
     {reply, {ok, Pid}, State#st{pid=Pid}};
 handle_call(cancel, _From, #st{pid=undefined}=State) ->
     {reply, ok, State};
 handle_call(cancel, _From, #st{pid=Pid}=State) ->
     unlink(Pid),
     exit(Pid, kill),
+    ok = couch_index_server:set_compacting(State#st.idx, false),
     {reply, ok, State#st{pid=undefined}};
 handle_call(get_compacting_pid, _From, #st{pid=Pid}=State) ->
     {reply, {ok, Pid}, State};
@@ -84,6 +86,7 @@ handle_cast(_Mesg, State) ->
 
 
 handle_info({'EXIT', Pid, normal}, #st{pid=Pid}=State) ->
+    ok = couch_index_server:set_compacting(State#st.idx, false),
     {noreply, State#st{pid=undefined}};
 handle_info({'EXIT', Pid, Reason}, #st{pid = Pid} = State) ->
     #st{idx = Idx, mod = Mod} = State,
@@ -92,11 +95,10 @@ handle_info({'EXIT', Pid, Reason}, #st{pid = Pid} = State) ->
     IdxName = Mod:get(idx_name, IdxState),
     Args = [DbName, IdxName, Reason],
     couch_log:error("Compaction failed for db: ~s idx: ~s reason: ~p", Args),
+    ok = couch_index_server:set_compacting(State#st.idx, false),
     {noreply, State#st{pid = undefined}};
 handle_info({'EXIT', _Pid, normal}, State) ->
     {noreply, State};
-handle_info({'EXIT', Pid, _Reason}, #st{idx=Pid}=State) ->
-    {stop, normal, State};
 handle_info(_Mesg, State) ->
     {stop, unknown_info, State}.