Put file ref_counting into it's own module, to make the speed of opening an already open database faster (by not waiting on file pending operations, particularly fsync).

git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@745076 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 142 insertions, 108 deletions

diff --git a/src/couchdb/Makefile.am b/src/couchdb/Makefile.am
index 1ad5d14a..8dd21256 100644
--- a/src/couchdb/Makefile.am
+++ b/src/couchdb/Makefile.am
@@ -62,6 +62,7 @@ source_files = \
     couch_log.erl \
     couch_os_process.erl \
     couch_query_servers.erl \
+    couch_ref_counter.erl \
     couch_rep.erl \
     couch_server.erl \
     couch_server_sup.erl \
@@ -99,6 +100,7 @@ compiled_files = \
     couch_log.beam \
     couch_os_process.beam \
     couch_query_servers.beam \
+    couch_ref_counter.beam \
     couch_rep.beam \
     couch_server.beam \
     couch_server_sup.beam \
diff --git a/src/couchdb/couch_db.erl b/src/couchdb/couch_db.erl
index ebc11bed..d9caa36f 100644
--- a/src/couchdb/couch_db.erl
+++ b/src/couchdb/couch_db.erl
@@ -14,7 +14,7 @@
 -behaviour(gen_server).
 
 -export([open/2,close/1,create/2,start_compact/1,get_db_info/1]).
--export([open_ref_counted/2,num_refs/1,monitor/1,count_changes_since/2]).
+-export([open_ref_counted/2,is_idle/1,monitor/1,count_changes_since/2]).
 -export([update_doc/3,update_docs/4,update_docs/2,update_docs/3,delete_doc/3]).
 -export([get_doc_info/2,open_doc/2,open_doc/3,open_doc_revs/4]).
 -export([get_missing_revs/2,name/1,doc_to_tree/1,get_update_seq/1,get_committed_update_seq/1]).
@@ -67,15 +67,15 @@ ensure_full_commit(#db{update_pid=UpdatePid,instance_start_time=StartTime}) ->
     ok = gen_server:call(UpdatePid, full_commit, infinity),
     {ok, StartTime}.
 
-close(#db{fd=Fd}) ->
-    couch_file:drop_ref(Fd).
+close(#db{fd_ref_counter=RefCntr}) ->
+    couch_ref_counter:drop(RefCntr).
 
 open_ref_counted(MainPid, UserCtx) ->
-    {ok, Db} = gen_server:call(MainPid, {open_ref_counted_instance, self()}),
+    {ok, Db} = gen_server:call(MainPid, {open_ref_count, self()}),
     {ok, Db#db{user_ctx=UserCtx}}.
 
-num_refs(MainPid) ->
-    gen_server:call(MainPid, num_refs).
+is_idle(MainPid) ->
+    gen_server:call(MainPid, is_idle).
 
 monitor(#db{main_pid=MainPid}) ->
     erlang:monitor(process, MainPid).
@@ -530,23 +530,28 @@ enum_docs(Db, StartId, InFun, Ctx) ->
 
 init({DbName, Filepath, Fd, Options}) ->
     {ok, UpdaterPid} = gen_server:start_link(couch_db_updater, {self(), DbName, Filepath, Fd, Options}, []),
-    ok = couch_file:add_ref(Fd),
-    gen_server:call(UpdaterPid, get_db).
+    {ok, #db{fd_ref_counter=RefCntr}=Db} = gen_server:call(UpdaterPid, get_db),
+    couch_ref_counter:add(RefCntr),
+    {ok, Db}.
 
 terminate(_Reason, _Db) ->
     ok.
     
-handle_call({open_ref_counted_instance, OpenerPid}, _From, #db{fd=Fd}=Db) ->
-    ok = couch_file:add_ref(Fd, OpenerPid),
+handle_call({open_ref_count, OpenerPid}, _, #db{fd_ref_counter=RefCntr}=Db) ->
+    ok = couch_ref_counter:add(RefCntr, OpenerPid),
     {reply, {ok, Db}, Db};
-handle_call(num_refs, _From, #db{fd=Fd}=Db) ->
-    {reply, couch_file:num_refs(Fd) - 1, Db};
-handle_call({db_updated, #db{fd=NewFd}=NewDb}, _From, #db{fd=OldFd}) ->
-    case NewFd == OldFd of
+handle_call(is_idle, _From,
+        #db{fd_ref_counter=RefCntr, compactor_pid=Compact}=Db) ->
+    % Idle means no referrers. Unless in the middle of a compaction file switch, 
+    % there are always at least 2 referrers, couch_db_updater and us.
+    {reply, (Compact == nil) and (couch_ref_counter:count(RefCntr) == 2), Db};
+handle_call({db_updated, #db{fd_ref_counter=NewRefCntr}=NewDb}, _From,
+        #db{fd_ref_counter=OldRefCntr}) ->
+    case NewRefCntr == OldRefCntr of
     true -> ok;
     false ->
-        couch_file:add_ref(NewFd),
-        couch_file:drop_ref(OldFd)
+        couch_ref_counter:add(NewRefCntr),
+        couch_ref_counter:drop(OldRefCntr)
     end,
     {reply, ok, NewDb}.
 
diff --git a/src/couchdb/couch_db.hrl b/src/couchdb/couch_db.hrl
index bfa16a3d..443ac73f 100644
--- a/src/couchdb/couch_db.hrl
+++ b/src/couchdb/couch_db.hrl
@@ -122,6 +122,7 @@
     compactor_pid=nil,
     instance_start_time, % number of microsecs since jan 1 1970 as a binary string
     fd,
+    fd_ref_counter,
     header = #db_header{},
     summary_stream,
     fulldocinfo_by_id_btree,
diff --git a/src/couchdb/couch_db_updater.erl b/src/couchdb/couch_db_updater.erl
index 800730d8..7790d7a4 100644
--- a/src/couchdb/couch_db_updater.erl
+++ b/src/couchdb/couch_db_updater.erl
@@ -22,7 +22,6 @@
 -define(HEADER_SIG, <<$g, $m, $k, 0>>).
 
 init({MainPid, DbName, Filepath, Fd, Options}) ->
-    link(Fd),
     case lists:member(create, Options) of
     true ->
         % create a new header and writes it to the file
@@ -38,8 +37,8 @@ init({MainPid, DbName, Filepath, Fd, Options}) ->
     Db2 = refresh_validate_doc_funs(Db),
     {ok, Db2#db{main_pid=MainPid}}.
 
-terminate(_Reason, Db) ->
-    close_db(Db).
+terminate(_Reason, _Db) ->
+    ok.
 
 handle_call(get_db, _From, Db) ->
     {reply, {ok, Db}, Db};
@@ -175,7 +174,7 @@ handle_cast({compact_done, CompactFilepath}, #db{filepath=Filepath}=Db) ->
         ok = file:rename(CompactFilepath, Filepath),
         
         couch_stream:close(Db#db.summary_stream),
-        couch_file:close_maybe(Db#db.fd),
+        couch_ref_counter:drop(Db#db.fd_ref_counter),
             
         ok = gen_server:call(Db#db.main_pid, {db_updated, NewDb2}),
         ?LOG_INFO("Compaction for db \"~s\" completed.", [Db#db.name]),
@@ -284,10 +283,11 @@ init_db(DbName, Filepath, Fd, Header0) ->
     {MegaSecs, Secs, MicroSecs} = now(),
     StartTime = ?l2b(io_lib:format("~p",
             [(MegaSecs*1000000*1000000) + (Secs*1000000) + MicroSecs])),
-    
+    {ok, RefCntr} = couch_ref_counter:start([Fd]),
     #db{
         update_pid=self(),
         fd=Fd,
+        fd_ref_counter = RefCntr,
         header=Header,
         summary_stream = SummaryStream,
         fulldocinfo_by_id_btree = IdBtree,
@@ -302,10 +302,6 @@ init_db(DbName, Filepath, Fd, Header0) ->
         }.
 
 
-close_db(#db{fd=Fd,summary_stream=Ss}) ->
-    couch_file:close(Fd),
-    couch_stream:close(Ss).
-
 refresh_validate_doc_funs(Db) ->
     {ok, DesignDocs} = get_design_docs(Db),
     ProcessDocFuns = lists:flatmap(
@@ -641,8 +637,7 @@ start_copy_compact(#db{name=Name,filepath=Filepath}=Db) ->
         ok = couch_file:write_header(Fd, ?HEADER_SIG, Header=#db_header{})
     end,
     NewDb = init_db(Name, CompactFile, Fd, Header),
-    NewDb2 = copy_compact(Db, NewDb, Retry),
-    close_db(NewDb2),
+    _NewDb2 = copy_compact(Db, NewDb, Retry),
     
     gen_server:cast(Db#db.update_pid, {compact_done, CompactFile}).
     
diff --git a/src/couchdb/couch_file.erl b/src/couchdb/couch_file.erl
index b29f45d2..d1103030 100644
--- a/src/couchdb/couch_file.erl
+++ b/src/couchdb/couch_file.erl
@@ -20,7 +20,6 @@
 -export([open/1, open/2, close/1, pread/3, pwrite/3, expand/2, bytes/1, sync/1]).
 -export([append_term/2, pread_term/2,write_header/3, read_header/2, truncate/2]).
 -export([init/1, terminate/2, handle_call/3, handle_cast/2, code_change/3, handle_info/2]).
--export([close_maybe/1,drop_ref/1,drop_ref/2,add_ref/1,add_ref/2,num_refs/1]).
 
 %%----------------------------------------------------------------------
 %% Args:   Valid Options are [create] and [create,overwrite].
@@ -168,26 +167,6 @@ close(Fd) ->
     Result = gen_server:cast(Fd, close),
     catch unlink(Fd),
     Result.
-    
-close_maybe(Fd) ->
-    catch unlink(Fd),
-    catch gen_server:cast(Fd, close_maybe).
-
-drop_ref(Fd) ->
-    drop_ref(Fd, self()).
-    
-drop_ref(Fd, Pid) ->
-    gen_server:cast(Fd, {drop_ref, Pid}).
-
-
-add_ref(Fd) ->
-    add_ref(Fd, self()).
-
-add_ref(Fd, Pid) ->
-    gen_server:call(Fd, {add_ref, Pid}).
-
-num_refs(Fd) ->
-    gen_server:call(Fd, num_refs).
 
 
 write_header(Fd, Prefix, Data) ->
@@ -291,7 +270,6 @@ init_status_error(ReturnPid, Ref, Error) ->
 % server functions
 
 init({Filepath, Options, ReturnPid, Ref}) ->
-    process_flag(trap_exit, true),
     case lists:member(create, Options) of
     true ->
         filelib:ensure_dir(Filepath),
@@ -357,66 +335,15 @@ handle_call({append_bin, Bin}, _From, Fd) ->
 handle_call({pread_bin, Pos}, _From, Fd) ->
     {ok, <<TermLen:32>>} = file:pread(Fd, Pos, 4),
     {ok, Bin} = file:pread(Fd, Pos + 4, TermLen),
-    {reply, {ok, Bin}, Fd};
-handle_call({add_ref, Pid},_From, Fd) ->
-    case get(Pid) of
-    undefined ->
-        put(Pid, {erlang:monitor(process, Pid), 1});
-    {MonRef, RefCnt} ->
-        put(Pid, {MonRef, RefCnt + 1})
-    end,
-    {reply, ok, Fd};
-handle_call(num_refs, _From, Fd) ->
-    {monitors, Monitors} =  process_info(self(), monitors),
-    {reply, length(Monitors), Fd}.
+    {reply, {ok, Bin}, Fd}.
 
 
 handle_cast(close, Fd) ->
-    {stop,normal,Fd};
-handle_cast(close_maybe, Fd) ->
-    maybe_close_async(Fd);
-handle_cast({drop_ref, Pid}, Fd) ->
-    case get(Pid) of
-    {MonRef, 1} ->
-        erase(Pid),
-        % don't check return of demonitor. The process could haved crashed causing
-        % the {'DOWN', ...} message to be sent and the process unmonitored.
-        erlang:demonitor(MonRef, [flush]);
-    {MonRef, Num} ->
-        put(Pid, {MonRef, Num-1})
-    end,
-    maybe_close_async(Fd).
+    {stop,normal,Fd}.
 
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-handle_info({'EXIT', _Pid, Reason}, Fd) ->
-    {stop, Reason, Fd};
-handle_info({'DOWN', MonitorRef, _Type, Pid, _Info}, Fd) ->
-    {MonitorRef, _RefCount} = erase(Pid),
-    maybe_close_async(Fd).
-
-
-
-should_close(_Fd) ->
-    case process_info(self(), links) of
-    {links, [_]} ->
-        % no linkers left (except our fd port). What about monitors?
-        case process_info(self(), monitors) of
-        {monitors, []} ->
-            true;
-        _ ->
-            false
-        end;
-    {links,  [_|_]} ->
-        false
-    end.
-
-maybe_close_async(Fd) ->
-    case should_close(Fd) of
-    true ->
-        {stop,normal,Fd};
-    false ->
-        {noreply,Fd}
-    end.
+handle_info(foo, Fd) ->
+    {stop, foo, Fd}.
diff --git a/src/couchdb/couch_httpd_show.erl b/src/couchdb/couch_httpd_show.erl
index 39c6cc1e..cc7d4d08 100644
--- a/src/couchdb/couch_httpd_show.erl
+++ b/src/couchdb/couch_httpd_show.erl
@@ -113,7 +113,7 @@ output_map_list(#httpd{mochi_req=MReq}=Req, Lang, ListSrc, View, Group, Db, Quer
         % pass it into the view fold with closures
         {ok, QueryServer} = couch_query_servers:start_view_list(Lang, ListSrc),
 
-        StartListRespFun = fun(Req2, Etag, TotalViewCount, Offset) ->
+        StartListRespFun = fun(Req2, _Etag, TotalViewCount, Offset) ->
             ExternalResp = couch_query_servers:render_list_head(QueryServer, 
                 Req2, Db, TotalViewCount, Offset),
             JsonResp = apply_etag(ExternalResp, CurrentEtag),
diff --git a/src/couchdb/couch_ref_counter.erl b/src/couchdb/couch_ref_counter.erl
new file mode 100644
index 00000000..95093f72
--- /dev/null
+++ b/src/couchdb/couch_ref_counter.erl
@@ -0,0 +1,104 @@
+% 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.
+
+-module(couch_ref_counter).
+-behaviour(gen_server).
+
+-export([start/1, init/1, terminate/2, handle_call/3, handle_cast/2, code_change/3, handle_info/2]).
+-export([drop/1,drop/2,add/1,add/2,count/1]).
+
+start(ChildProcs) ->
+    gen_server:start(couch_ref_counter, {self(), ChildProcs}, []).
+    
+    
+drop(RefCounterPid) ->
+    drop(RefCounterPid, self()).
+    
+drop(RefCounterPid, Pid) ->
+    gen_server:cast(RefCounterPid, {drop, Pid}).
+
+
+add(RefCounterPid) ->
+    add(RefCounterPid, self()).
+
+add(RefCounterPid, Pid) ->
+    gen_server:call(RefCounterPid, {add, Pid}).
+
+count(RefCounterPid) ->
+    gen_server:call(RefCounterPid, count).
+
+% server functions
+
+-record(srv,
+    {
+    referrers=dict:new() % a dict of each ref counting proc.
+    }).
+    
+init({Pid, ChildProcs}) ->
+    [link(ChildProc) || ChildProc <- ChildProcs],
+    Referrers = dict:from_list([{Pid, {erlang:monitor(process, Pid), 1}}]),
+    {ok, #srv{referrers=Referrers}}.
+
+
+terminate(_Reason, _Srv) ->
+    ok.
+
+
+handle_call({add, Pid},_From, #srv{referrers=Referrers}=Srv) ->
+    Referrers2 =
+    case dict:find(Pid, Referrers) of
+    error ->
+        dict:store(Pid, {erlang:monitor(process, Pid), 1}, Referrers);
+    {ok, {MonRef, RefCnt}} ->
+        dict:store(Pid, {MonRef, RefCnt + 1}, Referrers)
+    end,
+    {reply, ok, Srv#srv{referrers=Referrers2}};
+handle_call(count, _From, Srv) ->
+    {monitors, Monitors} =  process_info(self(), monitors),
+    {reply, length(Monitors), Srv}.
+
+
+handle_cast({drop, Pid}, #srv{referrers=Referrers}=Srv) ->
+    Referrers2 =
+    case dict:find(Pid, Referrers) of
+    {ok, {MonRef, 1}} ->
+        erlang:demonitor(MonRef, [flush]),
+        dict:erase(Pid, Referrers);
+    {ok, {MonRef, Num}} ->
+        dict:store(Pid, {MonRef, Num-1}, Referrers)
+    end,
+    maybe_close_async(Srv#srv{referrers=Referrers2}).
+
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+handle_info({'DOWN', MonRef, _, Pid, _}, #srv{referrers=Referrers}=Srv) ->
+    {ok, {MonRef, _RefCount}} = dict:find(Pid, Referrers),
+    maybe_close_async(Srv#srv{referrers=dict:erase(Pid, Referrers)}).
+
+
+should_close() ->
+    case process_info(self(), monitors) of
+    {monitors, []} ->
+        true;
+    _ ->
+        false
+    end.
+
+maybe_close_async(Srv) ->
+    case should_close() of
+    true ->
+        {stop,normal,Srv};
+    false ->
+        {noreply,Srv}
+    end.
diff --git a/src/couchdb/couch_server.erl b/src/couchdb/couch_server.erl
index 69cfa36c..39b33c63 100644
--- a/src/couchdb/couch_server.erl
+++ b/src/couchdb/couch_server.erl
@@ -195,15 +195,15 @@ try_close_lru(StartTime) ->
     true ->
         [{_, DbName}] = ets:lookup(couch_dbs_by_lru, LruTime),
         [{_, {MainPid, LruTime}}] = ets:lookup(couch_dbs_by_name, DbName),
-        case couch_db:num_refs(MainPid) of
-        0 ->
+        case couch_db:is_idle(MainPid) of
+        true ->
             exit(MainPid, kill),
             receive {'EXIT', MainPid, _Reason} -> ok end,
             true = ets:delete(couch_dbs_by_lru, LruTime),
             true = ets:delete(couch_dbs_by_name, DbName),
             true = ets:delete(couch_dbs_by_pid, MainPid),
             ok;
-        _NumRefs ->
+        false ->
             % this still has referrers. Go ahead and give it a current lru time
             % and try the next one in the table.
             NewLruTime = now(),