couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dam...@apache.org
Subject svn commit: r682560 - in /incubator/couchdb/trunk/src/couchdb: couch_db.erl couch_db.hrl couch_db_updater.erl couch_file.erl couch_httpd.erl couch_rep.erl couch_server.erl couch_server_sup.erl couch_view.erl
Date Tue, 05 Aug 2008 01:43:41 GMT
Author: damien
Date: Mon Aug  4 18:43:40 2008
New Revision: 682560

URL: http://svn.apache.org/viewvc?rev=682560&view=rev
Log:
Added concurrent open db limit and a LRU cache for closing old databases when limit reached (configurable via MaxDbsOpen var in couch.ini). Refactored db update code in couch_db.erl into couch_db_updater.erl.

Added:
    incubator/couchdb/trunk/src/couchdb/couch_db_updater.erl
Modified:
    incubator/couchdb/trunk/src/couchdb/couch_db.erl
    incubator/couchdb/trunk/src/couchdb/couch_db.hrl
    incubator/couchdb/trunk/src/couchdb/couch_file.erl
    incubator/couchdb/trunk/src/couchdb/couch_httpd.erl
    incubator/couchdb/trunk/src/couchdb/couch_rep.erl
    incubator/couchdb/trunk/src/couchdb/couch_server.erl
    incubator/couchdb/trunk/src/couchdb/couch_server_sup.erl
    incubator/couchdb/trunk/src/couchdb/couch_view.erl

Modified: incubator/couchdb/trunk/src/couchdb/couch_db.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_db.erl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_db.erl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_db.erl Mon Aug  4 18:43:40 2008
@@ -13,63 +13,24 @@
 -module(couch_db).
 -behaviour(gen_server).
 
--export([open/2,create/2,create/3,get_doc_info/2,start_compact/1]).
--export([save_docs/2, save_docs/3, get_db_info/1, update_doc/3, update_docs/2, update_docs/3]).
--export([delete_doc/3,open_doc/2,open_doc/3,enum_docs_since/4,enum_docs_since/5]).
--export([enum_docs/4,enum_docs/5, open_doc_revs/4, get_missing_revs/2]).
+-export([open/2,close/1,create/2,start_compact/1,get_db_info/1]).
+-export([open_ref_counted/2,num_refs/1,monitor/1]).
+-export([save_docs/3,update_doc/3,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]).
+-export([enum_docs/4,enum_docs/5,enum_docs_since/4,enum_docs_since/5]).
 -export([enum_docs_since_reduce_to_count/1,enum_docs_reduce_to_count/1]).
 -export([increment_update_seq/1]).
--export([start_update_loop/2]).
+-export([start_link/3]).
 -export([init/1,terminate/2,handle_call/3,handle_cast/2,code_change/3,handle_info/2]).
--export([start_copy_compact_int/2]).
 
--export([btree_by_id_split/1,
-            btree_by_id_join/2,
-            btree_by_id_reduce/2,
-            btree_by_seq_split/1,
-            btree_by_seq_join/2,
-            btree_by_seq_reduce/2]).
 
 -include("couch_db.hrl").
 
--record(db_header,
-    {write_version = 0,
-     update_seq = 0,
-     summary_stream_state = nil,
-     fulldocinfo_by_id_btree_state = nil,
-     docinfo_by_seq_btree_state = nil,
-     local_docs_btree_state = nil,
-     doc_count=0,
-     doc_del_count=0
-    }).
-
--record(db,
-    {main_pid=nil,
-    update_pid=nil,
-    compactor_pid=nil,
-    fd,
-    header = #db_header{},
-    summary_stream,
-    fulldocinfo_by_id_btree,
-    docinfo_by_seq_btree,
-    local_docs_btree,
-    update_seq,
-    doc_count,
-    doc_del_count,
-    name,
-    filepath
-    }).
-
-% small value used in revision trees to indicate the revision isn't stored
--define(REV_MISSING, []).
-
--define(HEADER_SIG, <<$g, $m, $k, 0>>).
-
 start_link(DbName, Filepath, Options) ->
     catch start_link0(DbName, Filepath, Options).
         
 start_link0(DbName, Filepath, Options) ->
-     % first delete the old file previous compaction
     Fd = 
     case couch_file:open(Filepath, Options) of
     {ok, Fd0} ->
@@ -105,33 +66,38 @@
     end,
     StartResult.
 
-%%% Interface functions %%%
 
-create(Filepath, Options) ->
-    create(Filepath, Filepath, Options).
+create(DbName, Options) ->
+    couch_server:create(DbName, Options).
+
+open(DbName, Options) ->
+    couch_server:open(DbName, Options).
 
-create(DbName, Filepath, Options) when is_list(Options) ->
-    start_link(DbName, Filepath, [create | Options]).
+close(#db{fd=Fd}) ->
+    couch_file:drop_ref(Fd).
 
-open(DbName, Filepath) ->
-    start_link(DbName, Filepath, []).
+open_ref_counted(MainPid, OpeningPid) ->
+    gen_server:call(MainPid, {open_ref_counted_instance, OpeningPid}).
 
+num_refs(MainPid) ->
+    gen_server:call(MainPid, num_refs).
 
-% Compaction still needs work. Right now readers and writers can get an error 
-% file compaction changeover. This doesn't need to be the case.
-start_compact(MainPid) ->
-    gen_server:cast(MainPid, start_compact).
+monitor(#db{main_pid=MainPid}) ->
+    erlang:monitor(process, MainPid).
 
-delete_doc(MainPid, Id, Revisions) ->
+start_compact(#db{update_pid=Pid}) ->
+    gen_server:cast(Pid, start_compact).
+
+delete_doc(Db, Id, Revisions) ->
     DeletedDocs = [#doc{id=Id, revs=[Rev], deleted=true} || Rev <- Revisions],
-    {ok, [Result]} = update_docs(MainPid, DeletedDocs, []),
+    {ok, [Result]} = update_docs(Db, DeletedDocs, []),
     {ok, Result}.
 
-open_doc(MainPid, IdOrDocInfo) ->
-    open_doc(MainPid, IdOrDocInfo, []).
+open_doc(Db, IdOrDocInfo) ->
+    open_doc(Db, IdOrDocInfo, []).
 
-open_doc(MainPid, Id, Options) ->
-    case open_doc_int(get_db(MainPid), Id, Options) of
+open_doc(Db, Id, Options) ->
+    case open_doc_int(Db, Id, Options) of
     {ok, #doc{deleted=true}=Doc} ->
         case lists:member(deleted, Options) of
         true ->
@@ -143,13 +109,13 @@
         Else
     end.
 
-open_doc_revs(MainPid, Id, Revs, Options) ->
-    [Result] = open_doc_revs_int(get_db(MainPid), [{Id, Revs}], Options),
+open_doc_revs(Db, Id, Revs, Options) ->
+    [Result] = open_doc_revs_int(Db, [{Id, Revs}], Options),
     Result.
 
-get_missing_revs(MainPid, IdRevsList) ->
+get_missing_revs(Db, IdRevsList) ->
     Ids = [Id1 || {Id1, _Revs} <- IdRevsList],
-    FullDocInfoResults = get_full_doc_infos(MainPid, Ids),
+    FullDocInfoResults = get_full_doc_infos(Db, Ids),
     Results = lists:zipwith(
         fun({Id, Revs}, FullDocInfoResult) ->
             case FullDocInfoResult of
@@ -177,18 +143,12 @@
     [Result] = get_full_doc_infos(Db, [Id]),
     Result.
 
-
-get_full_doc_infos(MainPid, Ids) when is_pid(MainPid) ->
-    get_full_doc_infos(get_db(MainPid), Ids);
-get_full_doc_infos(#db{}=Db, Ids) ->
+get_full_doc_infos(Db, Ids) ->
     couch_btree:lookup(Db#db.fulldocinfo_by_id_btree, Ids).
 
-increment_update_seq(MainPid) ->
-    gen_server:call(MainPid, increment_update_seq).
-        
+increment_update_seq(#db{update_pid=UpdatePid}) ->
+    gen_server:call(UpdatePid, increment_update_seq).
         
-get_db_info(MainPid) when is_pid(MainPid) ->
-    get_db_info(get_db(MainPid));
 get_db_info(Db) ->
     #db{fd=Fd,
         compactor_pid=Compactor,
@@ -205,12 +165,12 @@
         ],
     {ok, InfoList}.
 
-update_doc(MainPid, Doc, Options) ->
-    {ok, [NewRev]} = update_docs(MainPid, [Doc], Options),
+update_doc(Db, Doc, Options) ->
+    {ok, [NewRev]} = update_docs(Db, [Doc], Options),
     {ok, NewRev}.
 
-update_docs(MainPid, Docs) ->
-    update_docs(MainPid, Docs, []).
+update_docs(Db, Docs) ->
+    update_docs(Db, Docs, []).
     
 % group_alike_docs groups the sorted documents into sublist buckets, by id.
 % ([DocA, DocA, DocB, DocC], []) -> [[DocA, DocA], [DocB], [DocC]]
@@ -263,7 +223,7 @@
         end
     end.
 
-update_docs(MainPid, Docs, Options) ->
+update_docs(#db{update_pid=UpdatePid}=Db, Docs, Options) ->
     % go ahead and generate the new revision ids for the documents.
     Docs2 = lists:map(
         fun(#doc{id=Id,revs=Revs}=Doc) ->
@@ -278,7 +238,6 @@
     NewRevs = [NewRev || #doc{revs=[NewRev|_]} <- Docs2],
     DocBuckets = group_alike_docs(Docs2),
     Ids = [Id || [#doc{id=Id}|_] <- DocBuckets],
-    Db = get_db(MainPid),
     
     % lookup the doc by id and get the most recent
     
@@ -298,13 +257,14 @@
     % flush unwritten binaries to disk.
     DocBuckets3 = [[doc_flush_binaries(Doc, Db#db.fd) || Doc <- Bucket] || Bucket <- DocBuckets2],
 
-    case gen_server:call(MainPid, {update_docs, DocBuckets3, [new_edits | Options]}, infinity) of
+    case gen_server:call(UpdatePid, {update_docs, DocBuckets3, [new_edits | Options]}, infinity) of
     ok -> {ok, NewRevs};
     retry ->
-        Db2 = get_db(MainPid),
+        Db2 = open_ref_counted(Db#db.main_pid, self()),
         DocBuckets4 = [[doc_flush_binaries(Doc, Db2#db.fd) || Doc <- Bucket] || Bucket <- DocBuckets3],
         % We only retry once
-        case gen_server:call(MainPid, {update_docs, DocBuckets4, [new_edits | Options]}, infinity) of
+        ok = close(Db2),
+        case gen_server:call(UpdatePid, {update_docs, DocBuckets4, [new_edits | Options]}, infinity) of
         ok -> {ok, NewRevs};
         Else -> throw(Else)
         end;
@@ -312,15 +272,11 @@
         throw(Else)
     end.
 
-save_docs(MainPid, Docs) ->
-    save_docs(MainPid, Docs, []).
-
-save_docs(MainPid, Docs, Options) ->
+save_docs(#db{update_pid=UpdatePid, fd=Fd}, Docs, Options) ->
     % flush unwritten binaries to disk.
-    Db = get_db(MainPid),
     DocBuckets = group_alike_docs(Docs),
-    DocBuckets2 = [[doc_flush_binaries(Doc, Db#db.fd) || Doc <- Bucket] || Bucket <- DocBuckets],
-    ok = gen_server:call(MainPid, {update_docs, DocBuckets2, Options}, infinity).
+    DocBuckets2 = [[doc_flush_binaries(Doc, Fd) || Doc <- Bucket] || Bucket <- DocBuckets],
+    ok = gen_server:call(UpdatePid, {update_docs, DocBuckets2, Options}, infinity).
 
 
 doc_flush_binaries(Doc, Fd) ->
@@ -379,125 +335,51 @@
     Doc#doc{attachments = NewBins}.
 
 enum_docs_since_reduce_to_count(Reds) ->
-    couch_btree:final_reduce(fun btree_by_seq_reduce/2, Reds).
+    couch_btree:final_reduce(fun couch_db_updater:btree_by_seq_reduce/2, Reds).
 
 enum_docs_reduce_to_count(Reds) ->
-    couch_btree:final_reduce(fun btree_by_id_reduce/2, Reds).
+    couch_btree:final_reduce(fun couch_db_updater:btree_by_id_reduce/2, Reds).
 
-enum_docs_since(MainPid, SinceSeq, Direction, InFun, Ctx) ->
-    Db = get_db(MainPid),
+enum_docs_since(Db, SinceSeq, Direction, InFun, Ctx) ->
     couch_btree:fold(Db#db.docinfo_by_seq_btree, SinceSeq + 1, Direction, InFun, Ctx).
 
-enum_docs_since(MainPid, SinceSeq, InFun, Acc) ->
-    enum_docs_since(MainPid, SinceSeq, fwd, InFun, Acc).
+enum_docs_since(Db, SinceSeq, InFun, Acc) ->
+    enum_docs_since(Db, SinceSeq, fwd, InFun, Acc).
 
-enum_docs(MainPid, StartId, Direction, InFun, InAcc) ->
-    Db = get_db(MainPid),
+enum_docs(Db, StartId, Direction, InFun, InAcc) ->
     couch_btree:fold(Db#db.fulldocinfo_by_id_btree, StartId, Direction, InFun, InAcc).
 
-enum_docs(MainPid, StartId, InFun, Ctx) ->
-    enum_docs(MainPid, StartId, fwd, InFun, Ctx).
+enum_docs(Db, StartId, InFun, Ctx) ->
+    enum_docs(Db, StartId, fwd, InFun, Ctx).
 
 % server functions
 
-init(InitArgs) ->
-    spawn_link(couch_db, start_update_loop, [self(), InitArgs]),
-    receive
-    {initialized, Db} ->
-        {ok, Db}
-    end.
-
-btree_by_seq_split(DocInfo) ->
-    #doc_info{
-        id = Id,
-        rev = Rev,
-        update_seq = Seq,
-        summary_pointer = Sp,
-        conflict_revs = Conflicts,
-        deleted_conflict_revs = DelConflicts,
-        deleted = Deleted} = DocInfo,
-    {Seq,{Id, Rev, Sp, Conflicts, DelConflicts, Deleted}}.
-    
-btree_by_seq_join(Seq,{Id, Rev, Sp, Conflicts, DelConflicts, Deleted}) ->
-    #doc_info{
-        id = Id,
-        rev = Rev,
-        update_seq = Seq,
-        summary_pointer = Sp,
-        conflict_revs = Conflicts,
-        deleted_conflict_revs = DelConflicts,
-        deleted = Deleted}.
-
-btree_by_id_split(#full_doc_info{id=Id, update_seq=Seq,
-        deleted=Deleted, rev_tree=Tree}) ->
-    {Id, {Seq, case Deleted of true -> 1; false-> 0 end, Tree}}.
+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).
 
-btree_by_id_join(Id, {Seq, Deleted, Tree}) ->
-    #full_doc_info{id=Id, update_seq=Seq, deleted=Deleted==1, rev_tree=Tree}.
-    
-
-
-btree_by_id_reduce(reduce, FullDocInfos) ->
-    % count the number of deleted documents
-    length([1 || #full_doc_info{deleted=false} <- FullDocInfos]);
-btree_by_id_reduce(rereduce, Reds) ->
-    lists:sum(Reds).
-            
-btree_by_seq_reduce(reduce, DocInfos) ->
-    % count the number of deleted documents
-    length(DocInfos);
-btree_by_seq_reduce(rereduce, Reds) ->
-    lists:sum(Reds).
-
-init_db(DbName, Filepath, Fd, Header) ->
-    {ok, SummaryStream} = couch_stream:open(Header#db_header.summary_stream_state, Fd),
-    ok = couch_stream:set_min_buffer(SummaryStream, 10000),
-    {ok, IdBtree} = couch_btree:open(Header#db_header.fulldocinfo_by_id_btree_state, Fd,
-        [{split, fun btree_by_id_split/1},
-        {join, fun btree_by_id_join/2},
-        {reduce, fun btree_by_id_reduce/2}]),
-    {ok, SeqBtree} = couch_btree:open(Header#db_header.docinfo_by_seq_btree_state, Fd,
-            [{split, fun btree_by_seq_split/1},
-            {join, fun btree_by_seq_join/2},
-            {reduce, fun btree_by_seq_reduce/2}]),
-    {ok, LocalDocsBtree} = couch_btree:open(Header#db_header.local_docs_btree_state, Fd),
-
-    #db{
-        update_pid=self(),
-        fd=Fd,
-        header=Header,
-        summary_stream = SummaryStream,
-        fulldocinfo_by_id_btree = IdBtree,
-        docinfo_by_seq_btree = SeqBtree,
-        local_docs_btree = LocalDocsBtree,
-        update_seq = Header#db_header.update_seq,
-        doc_count = Header#db_header.doc_count,
-        doc_del_count = Header#db_header.doc_del_count,
-        name = DbName,
-        filepath=Filepath }.
-
-close_db(#db{fd=Fd,summary_stream=Ss}) ->
-    couch_file:close(Fd),
-    couch_stream:close(Ss).
-    
 terminate(_Reason, Db) ->
     exit(Db#db.update_pid, kill).
     
-handle_call({update_docs, DocActions, Options}, From, #db{update_pid=Updater}=Db) ->
-    Updater ! {From, update_docs, DocActions, Options},
-    {noreply, Db};
-handle_call(increment_update_seq, From, #db{update_pid=Updater}=Db) ->
-    Updater ! {From, increment_update_seq},
-    {noreply, Db};
-handle_call(get_db, _From, Db) ->
+handle_call({open_ref_counted_instance, OpenerPid}, _From, #db{fd=Fd}=Db) ->
+    ok = couch_file:add_ref(Fd, OpenerPid),
     {reply, {ok, Db}, Db};
-handle_call({db_updated, NewDb}, _From, _OldDb) ->
+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
+    true -> ok;
+    false ->
+        couch_file:add_ref(NewFd),
+        couch_file:drop_ref(OldFd)
+    end,
     {reply, ok, NewDb}.
 
 
-handle_cast(start_compact, #db{update_pid=Updater}=Db) ->
-    Updater ! compact,
-    {noreply, Db}.
+handle_cast(Msg, Db) ->
+    ?LOG_ERROR("Bad cast message received for db ~s: ~p", [Db#db.name, Msg]),
+    exit({error, Msg}).
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
@@ -508,114 +390,6 @@
 
 
 %%% Internal function %%%
-
-start_update_loop(MainPid, {DbName, Filepath, Fd, Options}) ->
-    link(Fd),
-    
-    case lists:member(create, Options) of
-    true ->
-        % create a new header and writes it to the file
-        Header =  #db_header{},
-        ok = couch_file:write_header(Fd, ?HEADER_SIG, Header),
-        % delete any old compaction files that might be hanging around
-        file:delete(Filepath ++ ".compact"),
-        file:delete(Filepath ++ ".old");
-    false ->
-        {ok, Header} = couch_file:read_header(Fd, ?HEADER_SIG)
-    end,
-    
-    Db = init_db(DbName, Filepath, Fd, Header),
-    Db2 = Db#db{main_pid=MainPid},
-    MainPid ! {initialized, Db2},
-    update_loop(Db2).
-    
-update_loop(#db{fd=Fd,name=Name,
-            filepath=Filepath,
-            main_pid=MainPid,
-            update_seq=UpdateSeq}=Db) ->
-    receive
-    {OrigFrom, update_docs, DocActions, Options} ->
-        case (catch update_docs_int(Db, DocActions, Options)) of
-        {ok, Db2} ->
-            ok = gen_server:call(MainPid, {db_updated, Db2}),
-            gen_server:reply(OrigFrom, ok),
-            couch_db_update_notifier:notify({updated, Name}),
-            update_loop(Db2);
-        retry ->
-            gen_server:reply(OrigFrom, retry),
-            update_loop(Db);
-        conflict ->
-            gen_server:reply(OrigFrom, conflict),
-            update_loop(Db);
-        Error ->
-            exit(Error) % we crashed
-        end;
-    compact ->
-        case Db#db.compactor_pid of
-        nil ->
-            ?LOG_INFO("Starting compaction for db \"~s\"", [Name]),
-            Pid = spawn_link(couch_db, start_copy_compact_int, [Db, true]),
-            Db2 = Db#db{compactor_pid=Pid},
-            ok = gen_server:call(MainPid, {db_updated, Db2}),
-            update_loop(Db2);
-        _ ->
-            update_loop(Db) % already started
-        end;
-    {compact_done, CompactFilepath} ->
-        {ok, NewFd} = couch_file:open(CompactFilepath),
-        {ok, NewHeader} = couch_file:read_header(NewFd, ?HEADER_SIG),
-        #db{update_seq=NewSeq}= NewDb =
-                init_db(Name, CompactFilepath, NewFd, NewHeader),
-        case Db#db.update_seq == NewSeq of
-        true ->
-            NewDb2 = commit_data(
-                NewDb#db{
-                    main_pid = Db#db.main_pid,
-                    doc_count = Db#db.doc_count,
-                    doc_del_count = Db#db.doc_del_count,
-                    filepath = Filepath}),
-                
-            ?LOG_DEBUG("CouchDB swapping files ~s and ~s.", [Filepath, CompactFilepath]),
-            ok = file:rename(Filepath, Filepath ++ ".old"),
-            ok = file:rename(CompactFilepath, Filepath),
-            
-            couch_stream:close(Db#db.summary_stream),
-            % close file handle async.
-            % wait 5 secs before closing, allowing readers to finish
-            unlink(Fd),
-            spawn_link(fun() ->
-                receive after 5000 -> ok end,
-                couch_file:close(Fd),
-                file:delete(Filepath ++ ".old")
-                end),
-                
-            ok = gen_server:call(MainPid, {db_updated, NewDb2}),
-            ?LOG_INFO("Compaction for db ~p completed.", [Name]),
-            update_loop(NewDb2#db{compactor_pid=nil});
-        false ->
-            ?LOG_INFO("Compaction file still behind main file "
-                "(update seq=~p. compact update seq=~p). Retrying.",
-                [Db#db.update_seq, NewSeq]),
-            Pid = spawn_link(couch_db, start_copy_compact_int, [Db, false]),
-            Db2 = Db#db{compactor_pid=Pid},
-            couch_file:close(NewFd),
-            update_loop(Db2)
-        end;
-    {OrigFrom, increment_update_seq} ->
-        Db2 = commit_data(Db#db{update_seq=UpdateSeq+1}),
-        ok = gen_server:call(MainPid, {db_updated, Db2}),
-        gen_server:reply(OrigFrom, {ok, UpdateSeq+1}),
-        couch_db_update_notifier:notify({updated, Name}),
-        update_loop(Db2);
-    Else ->
-        ?LOG_ERROR("Unknown message received in db ~s:~p", [Db#db.name, Else]),
-        exit({error, Else})
-    end.
-
-get_db(MainPid) ->
-    {ok, Db} = gen_server:call(MainPid, get_db),
-    Db.
-
 open_doc_revs_int(Db, IdRevs, Options) ->
     Ids = [Id || {Id, _Revs} <- IdRevs],
     LookupResults = get_full_doc_infos(Db, Ids),
@@ -711,16 +485,6 @@
         end
     end.
 
-% rev tree functions
-
-doc_to_tree(Doc) ->
-    doc_to_tree(Doc, lists:reverse(Doc#doc.revs)).
-
-doc_to_tree(Doc, [RevId]) ->
-    [{RevId, Doc, []}];
-doc_to_tree(Doc, [RevId | Rest]) ->
-    [{RevId, ?REV_MISSING, doc_to_tree(Doc, Rest)}].
-
 make_doc(Db, Id, Deleted, SummaryPointer, RevisionPath) ->
     {BodyData, BinValues} =
     case SummaryPointer of
@@ -737,303 +501,6 @@
         attachments = BinValues,
         deleted = Deleted
         }.
-
-flush_trees(_Db, [], AccFlushedTrees) ->
-    {ok, lists:reverse(AccFlushedTrees)};
-flush_trees(#db{fd=Fd}=Db, [InfoUnflushed | RestUnflushed], AccFlushed) ->
-        #full_doc_info{rev_tree=Unflushed} = InfoUnflushed,
-        Flushed = couch_key_tree:map(
-        fun(_Rev, Value) ->
-            case Value of
-            #doc{attachments=Atts,deleted=IsDeleted}=Doc ->
-                % this node value is actually an unwritten document summary,
-                % write to disk.
-                % make sure the Fd in the written bins is the same Fd we are.
-                Bins =
-                case Atts of
-                [] -> [];
-                [{_BName, {_Type, {BinFd, _Sp, _Len}}} | _ ] when BinFd == Fd ->
-                    % convert bins, removing the FD.
-                    % All bins should have been flushed to disk already.
-                    [{BinName, {BinType, BinSp, BinLen}}
-                        || {BinName, {BinType, {_Fd, BinSp, BinLen}}}
-                        <- Atts];
-                _ ->
-                    % BinFd must not equal our Fd. This can happen when a database
-                    % is being updated during a compaction
-                    ?LOG_DEBUG("File where the attachments are written has changed. Possibly retrying.", []),
-                    throw(retry)
-                end,
-                {ok, NewSummaryPointer} = couch_stream:write_term(Db#db.summary_stream, {Doc#doc.body, Bins}),
-                {IsDeleted, NewSummaryPointer};
-            _ ->
-                Value
-            end
-        end, Unflushed),
-    flush_trees(Db, RestUnflushed, [InfoUnflushed#full_doc_info{rev_tree=Flushed} | AccFlushed]).
-
-merge_rev_trees(_NoConflicts, [], [], AccNewInfos, AccSeq) ->
-    {ok, lists:reverse(AccNewInfos), AccSeq};
-merge_rev_trees(NoConflicts, [NewDocs|RestDocsList],
-        [OldDocInfo|RestOldInfo], AccNewInfos, AccSeq) ->
-    #full_doc_info{id=Id,rev_tree=OldTree}=OldDocInfo,
-    UpdatesRevTree = lists:foldl(
-        fun(NewDoc, AccTree) ->
-            couch_key_tree:merge(AccTree, doc_to_tree(NewDoc))
-        end,
-        [], NewDocs),
-    NewRevTree = couch_key_tree:merge(OldTree, UpdatesRevTree),
-    if NewRevTree == OldTree ->
-        % nothing changed
-        merge_rev_trees(NoConflicts, RestDocsList, RestOldInfo, AccNewInfos, AccSeq);
-    true ->
-        if NoConflicts andalso OldTree /= [] ->
-            OldConflicts = couch_key_tree:count_leafs(OldTree),
-            NewConflicts = couch_key_tree:count_leafs(NewRevTree),
-            if NewConflicts > OldConflicts ->
-                throw(conflict);
-            true -> ok
-            end;
-        true -> ok
-        end,
-        NewInfo = #full_doc_info{id=Id,update_seq=AccSeq+1,rev_tree=NewRevTree},
-        merge_rev_trees(NoConflicts, RestDocsList,RestOldInfo, 
-                [NewInfo|AccNewInfos],AccSeq+1)
-    end.
-
-new_index_entries([], DocCount, DelCount, AccById, AccBySeq) ->
-    {ok, DocCount, DelCount, AccById, AccBySeq};
-new_index_entries([FullDocInfo|RestInfos], DocCount, DelCount, AccById, AccBySeq) ->
-    #doc_info{deleted=Deleted} = DocInfo = couch_doc:to_doc_info(FullDocInfo),
-    {DocCount2, DelCount2} =
-    if Deleted -> {DocCount, DelCount + 1};
-    true -> {DocCount + 1, DelCount} 
-    end,
-    new_index_entries(RestInfos, DocCount2, DelCount2, 
-        [FullDocInfo#full_doc_info{deleted=Deleted}|AccById],
-        [DocInfo|AccBySeq]).
-
-update_docs_int(Db, DocsList, Options) ->
-    #db{
-        fulldocinfo_by_id_btree = DocInfoByIdBTree,
-        docinfo_by_seq_btree = DocInfoBySeqBTree,
-        update_seq = LastSeq,
-        doc_count = FullDocCount,
-        doc_del_count = FullDelCount
-        } = Db,
-
-    % separate out the NonRep documents from the rest of the documents
-    {DocsList2, NonRepDocs} = lists:foldl(
-        fun([#doc{id=Id}=Doc | Rest]=Docs, {DocsListAcc, NonRepDocsAcc}) ->
-            case Id of
-            ?LOCAL_DOC_PREFIX ++ _ when Rest==[] ->
-                % when saving NR (non rep) documents, you can only save a single rev
-                {DocsListAcc, [Doc | NonRepDocsAcc]};
-            Id->
-                {[Docs | DocsListAcc], NonRepDocsAcc}
-            end
-        end, {[], []}, DocsList),
-    
-    Ids = [Id || [#doc{id=Id}|_] <- DocsList2], 
-    
-    % lookup up the existing documents, if they exist.
-    OldDocLookups = couch_btree:lookup(DocInfoByIdBTree, Ids),
-    OldDocInfos = lists:zipwith(
-        fun(_Id, {ok, FullDocInfo}) ->
-            FullDocInfo;
-        (Id, not_found) ->
-            #full_doc_info{id=Id}
-        end,
-        Ids, OldDocLookups),
-    
-    {OldCount, OldDelCount} = lists:foldl(
-        fun({ok, FullDocInfo}, {OldCountAcc, OldDelCountAcc}) ->
-            case couch_doc:to_doc_info(FullDocInfo) of
-            #doc_info{deleted=false} ->
-                {OldCountAcc + 1, OldDelCountAcc};
-            _ ->
-                {OldCountAcc, OldDelCountAcc + 1}
-            end;
-        (not_found, Acc) ->
-            Acc
-        end, {0, 0}, OldDocLookups),
-    
-    % Merge the new docs into the revision trees.
-    NoConflicts = lists:member(new_edits, Options),
-    {ok, NewDocInfos, NewSeq} = merge_rev_trees(NoConflicts, DocsList2, OldDocInfos, [], LastSeq),
-    
-    RemoveSeqs =
-        [ OldSeq || {ok, #full_doc_info{update_seq=OldSeq}} <- OldDocLookups],
-    
-    % All regular documents are now ready to write.
-    
-    % Try to write the local documents first, a conflict might be generated
-    {ok, Db2}  = update_local_docs(Db, NonRepDocs),
-    
-    % Write out the documents summaries (they are stored in the nodes of the rev trees)
-    {ok, FlushedDocInfos} = flush_trees(Db2, NewDocInfos, []),
-    
-    {ok, NewDocsCount, NewDelCount, InfoById, InfoBySeq} =
-        new_index_entries(FlushedDocInfos, 0, 0, [], []),
-
-    % and the indexes to the documents
-    {ok, DocInfoBySeqBTree2} = couch_btree:add_remove(DocInfoBySeqBTree, InfoBySeq, RemoveSeqs),
-    {ok, DocInfoByIdBTree2} = couch_btree:add_remove(DocInfoByIdBTree, InfoById, []),
-
-    Db3 = Db2#db{
-        fulldocinfo_by_id_btree = DocInfoByIdBTree2,
-        docinfo_by_seq_btree = DocInfoBySeqBTree2,
-        update_seq = NewSeq,
-        doc_count = FullDocCount + NewDocsCount - OldCount,
-        doc_del_count = FullDelCount + NewDelCount - OldDelCount},
-
-    case lists:member(delay_commit, Options) of
-    true ->
-        {ok, Db3};
-    false ->
-        {ok, commit_data(Db3)}
-    end.
-
-update_local_docs(#db{local_docs_btree=Btree}=Db, Docs) ->
-    Ids = [Id || #doc{id=Id} <- Docs],
-    OldDocLookups = couch_btree:lookup(Btree, Ids),
-    BtreeEntries = lists:zipwith(
-        fun(#doc{id=Id,deleted=Delete,revs=Revs,body=Body}, OldDocLookup) ->
-            NewRev =
-            case Revs of
-                [] -> 0;
-                [RevStr|_] -> list_to_integer(RevStr)
-            end,
-            OldRev =
-            case OldDocLookup of
-                {ok, {_, {OldRev0, _}}} -> OldRev0;
-                not_found -> 0
-            end,
-            case OldRev + 1 == NewRev of
-            true ->
-                case Delete of
-                    false -> {update, {Id, {NewRev, Body}}};
-                    true  -> {remove, Id}
-                end;
-            false ->
-                throw(conflict)
-            end
-            
-        end, Docs, OldDocLookups),
-
-    BtreeIdsRemove = [Id || {remove, Id} <- BtreeEntries],
-    BtreeIdsUpdate = [ByIdDocInfo || {update, ByIdDocInfo} <- BtreeEntries],
-
-    {ok, Btree2} =
-        couch_btree:add_remove(Btree, BtreeIdsUpdate, BtreeIdsRemove),
-
-    {ok, Db#db{local_docs_btree = Btree2}}.
-
-
-
-commit_data(#db{fd=Fd, header=Header} = Db) ->
-    Header2 = Header#db_header{
-        update_seq = Db#db.update_seq,
-        summary_stream_state = couch_stream:get_state(Db#db.summary_stream),
-        docinfo_by_seq_btree_state = couch_btree:get_state(Db#db.docinfo_by_seq_btree),
-        fulldocinfo_by_id_btree_state = couch_btree:get_state(Db#db.fulldocinfo_by_id_btree),
-        local_docs_btree_state = couch_btree:get_state(Db#db.local_docs_btree),
-        doc_count = Db#db.doc_count,
-        doc_del_count = Db#db.doc_del_count
-        },
-    if Header == Header2 ->
-        Db; % unchanged. nothing to do
-    true ->
-        ok = couch_file:write_header(Fd, ?HEADER_SIG, Header2),
-        Db#db{header = Header2}
-    end.
-
-copy_raw_doc(SrcFd, SrcSp, DestFd, DestStream) ->
-    {ok, {BodyData, BinInfos}} = couch_stream:read_term(SrcFd, SrcSp),
-    % copy the bin values
-    NewBinInfos = lists:map(fun({Name, {Type, BinSp, Len}}) ->
-        {ok, NewBinSp} = couch_stream:copy_to_new_stream(SrcFd, BinSp, Len, DestFd),
-        {Name, {Type, NewBinSp, Len}}
-        end, BinInfos),
-    % now write the document summary
-    {ok, Sp} = couch_stream:write_term(DestStream, {BodyData, NewBinInfos}),
-    Sp.
-
-copy_rev_tree(_SrcFd, _DestFd, _DestStream, []) ->
-    [];
-copy_rev_tree(SrcFd, DestFd, DestStream, [{RevId, {IsDel, Sp}, []} | RestTree]) ->
-    % This is a leaf node, copy it over
-    NewSp = copy_raw_doc(SrcFd, Sp, DestFd, DestStream),
-    [{RevId, {IsDel, NewSp}, []} | copy_rev_tree(SrcFd, DestFd, DestStream, RestTree)];
-copy_rev_tree(SrcFd, DestFd, DestStream, [{RevId, _, SubTree} | RestTree]) ->
-    % inner node, only copy info/data from leaf nodes
-    [{RevId, ?REV_MISSING, copy_rev_tree(SrcFd, DestFd, DestStream, SubTree)} | copy_rev_tree(SrcFd, DestFd, DestStream, RestTree)].
-    
-copy_docs(#db{fd=SrcFd}=Db, #db{fd=DestFd,summary_stream=DestStream}=NewDb, InfoBySeq) ->
-    Ids = [Id || #doc_info{id=Id} <- InfoBySeq],
-    LookupResults = couch_btree:lookup(Db#db.fulldocinfo_by_id_btree, Ids),
-    NewFullDocInfos = lists:map(
-        fun({ok, #full_doc_info{rev_tree=RevTree}=Info}) ->
-            Info#full_doc_info{rev_tree=copy_rev_tree(SrcFd, DestFd, DestStream, RevTree)}
-        end, LookupResults),
-    NewDocInfos = [couch_doc:to_doc_info(FullDocInfo) || FullDocInfo <- NewFullDocInfos],
-    {ok, DocInfoBTree} =
-        couch_btree:add_remove(NewDb#db.docinfo_by_seq_btree, NewDocInfos, []),
-    {ok, FullDocInfoBTree} =
-        couch_btree:add_remove(NewDb#db.fulldocinfo_by_id_btree, NewFullDocInfos, []),
-    NewDb#db{fulldocinfo_by_id_btree=FullDocInfoBTree, docinfo_by_seq_btree=DocInfoBTree}.
-
-
-          
-copy_compact_docs(Db, NewDb) ->
-    EnumBySeqFun =
-    fun(#doc_info{update_seq=Seq}=DocInfo, _Offset, {AccNewDb, AccUncopied}) ->
-        case couch_util:should_flush() of
-        true ->
-            NewDb2 = copy_docs(Db, AccNewDb, lists:reverse([DocInfo | AccUncopied])),
-            {ok, {commit_data(NewDb2#db{update_seq=Seq}), []}};
-        false ->    
-            {ok, {AccNewDb, [DocInfo | AccUncopied]}}
-        end
-    end,
-    {ok, {NewDb2, Uncopied}} =
-        couch_btree:foldl(Db#db.docinfo_by_seq_btree, NewDb#db.update_seq + 1, EnumBySeqFun, {NewDb, []}),
-
-    case Uncopied of
-    [#doc_info{update_seq=LastSeq} | _] ->
-        commit_data( copy_docs(Db, NewDb2#db{update_seq=LastSeq},
-            lists:reverse(Uncopied)));
-    [] ->
-        NewDb2
-    end.
-
-start_copy_compact_int(#db{name=Name,filepath=Filepath}=Db, CopyLocal) ->
-    CompactFile = Filepath ++ ".compact",
-    ?LOG_DEBUG("Compaction process spawned for db \"~s\"", [Name]),
-    case couch_file:open(CompactFile) of
-    {ok, Fd} ->
-        ?LOG_DEBUG("Found existing compaction file for db \"~s\"", [Name]),
-        {ok, Header} = couch_file:read_header(Fd, ?HEADER_SIG);
-    {error, enoent} -> %
-        {ok, Fd} = couch_file:open(CompactFile, [create]),
-        Header =  #db_header{},
-        ok = couch_file:write_header(Fd, ?HEADER_SIG, Header)
-    end,
-    NewDb = init_db(Name, CompactFile, Fd, Header),
-    NewDb2 = copy_compact_docs(Db, NewDb),
-    NewDb3 =
-    case CopyLocal of
-    true ->
-        % suck up all the local docs into memory and write them to the new db
-        {ok, LocalDocs} = couch_btree:foldl(Db#db.local_docs_btree,
-                fun(Value, _Offset, Acc) -> {ok, [Value | Acc]} end, []),
-        {ok, NewLocalBtree} = couch_btree:add(NewDb2#db.local_docs_btree, LocalDocs),
-        commit_data(NewDb2#db{local_docs_btree=NewLocalBtree});
-    _ ->
-        NewDb2
-    end,
-    close_db(NewDb3),
-    Db#db.update_pid ! {compact_done, CompactFile}.
     
     
     
\ No newline at end of file

Modified: incubator/couchdb/trunk/src/couchdb/couch_db.hrl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_db.hrl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_db.hrl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_db.hrl Mon Aug  4 18:43:40 2008
@@ -69,3 +69,40 @@
     % couch_db:open_doc(Db, Id, Options).
     meta = []
     }).
+    
+    
+
+
+
+-record(db_header,
+    {write_version = 0,
+     update_seq = 0,
+     summary_stream_state = nil,
+     fulldocinfo_by_id_btree_state = nil,
+     docinfo_by_seq_btree_state = nil,
+     local_docs_btree_state = nil,
+     doc_count=0,
+     doc_del_count=0
+    }).
+
+-record(db,
+    {main_pid=nil,
+    update_pid=nil,
+    compactor_pid=nil,
+    fd,
+    header = #db_header{},
+    summary_stream,
+    fulldocinfo_by_id_btree,
+    docinfo_by_seq_btree,
+    local_docs_btree,
+    update_seq,
+    doc_count,
+    doc_del_count,
+    name,
+    filepath
+    }).
+    
+    
+
+% small value used in revision trees to indicate the revision isn't stored
+-define(REV_MISSING, []).

Added: incubator/couchdb/trunk/src/couchdb/couch_db_updater.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_db_updater.erl?rev=682560&view=auto
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_db_updater.erl (added)
+++ incubator/couchdb/trunk/src/couchdb/couch_db_updater.erl Mon Aug  4 18:43:40 2008
@@ -0,0 +1,499 @@
+% 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_db_updater).
+-behaviour(gen_server).
+
+-export([btree_by_id_reduce/2,btree_by_seq_reduce/2]).
+-export([init/1,terminate/2,handle_call/3,handle_cast/2,code_change/3,handle_info/2]).
+
+-include("couch_db.hrl").
+
+-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
+        Header =  #db_header{},
+        ok = couch_file:write_header(Fd, ?HEADER_SIG, Header),
+        % delete any old compaction files that might be hanging around
+        file:delete(Filepath ++ ".compact"),
+        file:delete(Filepath ++ ".old");
+    false ->
+        {ok, Header} = couch_file:read_header(Fd, ?HEADER_SIG)
+    end,
+    
+    Db = init_db(DbName, Filepath, Fd, Header),
+    {ok, Db#db{main_pid=MainPid}}.
+
+terminate(_Reason, Db) ->
+    close_db(Db).
+
+handle_call(get_db, _From, Db) ->
+    {reply, {ok, Db}, Db};
+handle_call({update_docs, DocActions, Options}, _From, Db) ->
+    try update_docs_int(Db, DocActions, Options) of
+    {ok, Db2} ->
+        ok = gen_server:call(Db#db.main_pid, {db_updated, Db2}),
+        couch_db_update_notifier:notify({updated, Db2#db.name}),
+        {reply, ok, Db2}
+    catch
+        throw: retry ->
+            {reply, retry, Db};
+        throw: conflict ->
+            {reply, conflict, Db}
+    end;
+handle_call(increment_update_seq, _From, Db) ->
+    Db2 = commit_data(Db#db{update_seq=Db#db.update_seq+1}),
+    ok = gen_server:call(Db2#db.main_pid, {db_updated, Db2}),
+    couch_db_update_notifier:notify({updated, Db#db.name}),
+    {reply, {ok, Db2#db.update_seq}, Db2}.
+
+    
+handle_cast(start_compact, Db) ->
+    case Db#db.compactor_pid of
+    nil ->
+        ?LOG_INFO("Starting compaction for db \"~s\"", [Db#db.name]),
+        Pid = spawn_link(fun() -> start_copy_compact_int(Db, true) end),
+        Db2 = Db#db{compactor_pid=Pid},
+        ok = gen_server:call(Db#db.main_pid, {db_updated, Db2}),
+        {noreply, Db2};
+    _ ->
+        % compact currently running, this is a no-op
+        {noreply, Db}
+    end;
+handle_cast({compact_done, CompactFilepath}, #db{filepath=Filepath}=Db) ->
+    {ok, NewFd} = couch_file:open(CompactFilepath),
+    {ok, NewHeader} = couch_file:read_header(NewFd, ?HEADER_SIG),
+    #db{update_seq=NewSeq} = NewDb =
+            init_db(Db#db.name, CompactFilepath, NewFd, NewHeader),
+    case Db#db.update_seq == NewSeq of
+    true ->
+        NewDb2 = commit_data(
+            NewDb#db{
+                main_pid = Db#db.main_pid,
+                doc_count = Db#db.doc_count,
+                doc_del_count = Db#db.doc_del_count,
+                filepath = Filepath}),
+            
+        ?LOG_DEBUG("CouchDB swapping files ~s and ~s.", [Filepath, CompactFilepath]),
+        file:delete(Filepath),
+        ok = file:rename(CompactFilepath, Filepath),
+        
+        couch_stream:close(Db#db.summary_stream),
+        couch_file:close_maybe(Db#db.fd),
+        file:delete(Filepath ++ ".old"),
+            
+        ok = gen_server:call(Db#db.main_pid, {db_updated, NewDb2}),
+        ?LOG_INFO("Compaction for db ~p completed.", [Db#db.name]),
+        {noreply, NewDb2#db{compactor_pid=nil}};
+    false ->
+        ?LOG_INFO("Compaction file still behind main file "
+            "(update seq=~p. compact update seq=~p). Retrying.",
+            [Db#db.update_seq, NewSeq]),
+        Pid = spawn_link(fun() -> start_copy_compact_int(Db, false) end),
+        Db2 = Db#db{compactor_pid=Pid},
+        couch_file:close(NewFd),
+        {noreply, Db2}
+    end.
+
+handle_info(Msg, Db) ->
+    ?LOG_ERROR("Bad message received for db ~s: ~p", [Db#db.name, Msg]),
+    exit({error, Msg}).
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+
+btree_by_seq_split(DocInfo) ->
+    #doc_info{
+        id = Id,
+        rev = Rev,
+        update_seq = Seq,
+        summary_pointer = Sp,
+        conflict_revs = Conflicts,
+        deleted_conflict_revs = DelConflicts,
+        deleted = Deleted} = DocInfo,
+    {Seq,{Id, Rev, Sp, Conflicts, DelConflicts, Deleted}}.
+    
+btree_by_seq_join(Seq,{Id, Rev, Sp, Conflicts, DelConflicts, Deleted}) ->
+    #doc_info{
+        id = Id,
+        rev = Rev,
+        update_seq = Seq,
+        summary_pointer = Sp,
+        conflict_revs = Conflicts,
+        deleted_conflict_revs = DelConflicts,
+        deleted = Deleted}.
+
+btree_by_id_split(#full_doc_info{id=Id, update_seq=Seq,
+        deleted=Deleted, rev_tree=Tree}) ->
+    {Id, {Seq, case Deleted of true -> 1; false-> 0 end, Tree}}.
+
+btree_by_id_join(Id, {Seq, Deleted, Tree}) ->
+    #full_doc_info{id=Id, update_seq=Seq, deleted=Deleted==1, rev_tree=Tree}.
+    
+
+
+btree_by_id_reduce(reduce, FullDocInfos) ->
+    % count the number of deleted documents
+    length([1 || #full_doc_info{deleted=false} <- FullDocInfos]);
+btree_by_id_reduce(rereduce, Reds) ->
+    lists:sum(Reds).
+            
+btree_by_seq_reduce(reduce, DocInfos) ->
+    % count the number of deleted documents
+    length(DocInfos);
+btree_by_seq_reduce(rereduce, Reds) ->
+    lists:sum(Reds).
+
+init_db(DbName, Filepath, Fd, Header) ->
+    {ok, SummaryStream} = couch_stream:open(Header#db_header.summary_stream_state, Fd),
+    ok = couch_stream:set_min_buffer(SummaryStream, 10000),
+    {ok, IdBtree} = couch_btree:open(Header#db_header.fulldocinfo_by_id_btree_state, Fd,
+        [{split, fun(X) -> btree_by_id_split(X) end},
+        {join, fun(X,Y) -> btree_by_id_join(X,Y) end},
+        {reduce, fun(X,Y) -> btree_by_id_reduce(X,Y) end}]),
+    {ok, SeqBtree} = couch_btree:open(Header#db_header.docinfo_by_seq_btree_state, Fd,
+            [{split, fun(X) -> btree_by_seq_split(X) end},
+            {join, fun(X,Y) -> btree_by_seq_join(X,Y) end},
+            {reduce, fun(X,Y) -> btree_by_seq_reduce(X,Y) end}]),
+    {ok, LocalDocsBtree} = couch_btree:open(Header#db_header.local_docs_btree_state, Fd),
+
+    #db{
+        update_pid=self(),
+        fd=Fd,
+        header=Header,
+        summary_stream = SummaryStream,
+        fulldocinfo_by_id_btree = IdBtree,
+        docinfo_by_seq_btree = SeqBtree,
+        local_docs_btree = LocalDocsBtree,
+        update_seq = Header#db_header.update_seq,
+        doc_count = Header#db_header.doc_count,
+        doc_del_count = Header#db_header.doc_del_count,
+        name = DbName,
+        filepath=Filepath }.
+
+close_db(#db{fd=Fd,summary_stream=Ss}) ->
+    couch_file:close(Fd),
+    couch_stream:close(Ss).
+
+% rev tree functions
+
+doc_to_tree(Doc) ->
+    doc_to_tree(Doc, lists:reverse(Doc#doc.revs)).
+
+doc_to_tree(Doc, [RevId]) ->
+    [{RevId, Doc, []}];
+doc_to_tree(Doc, [RevId | Rest]) ->
+    [{RevId, ?REV_MISSING, doc_to_tree(Doc, Rest)}].
+
+flush_trees(_Db, [], AccFlushedTrees) ->
+    {ok, lists:reverse(AccFlushedTrees)};
+flush_trees(#db{fd=Fd}=Db, [InfoUnflushed | RestUnflushed], AccFlushed) ->
+        #full_doc_info{rev_tree=Unflushed} = InfoUnflushed,
+        Flushed = couch_key_tree:map(
+        fun(_Rev, Value) ->
+            case Value of
+            #doc{attachments=Atts,deleted=IsDeleted}=Doc ->
+                % this node value is actually an unwritten document summary,
+                % write to disk.
+                % make sure the Fd in the written bins is the same Fd we are.
+                Bins =
+                case Atts of
+                [] -> [];
+                [{_BName, {_Type, {BinFd, _Sp, _Len}}} | _ ] when BinFd == Fd ->
+                    % convert bins, removing the FD.
+                    % All bins should have been flushed to disk already.
+                    [{BinName, {BinType, BinSp, BinLen}}
+                        || {BinName, {BinType, {_Fd, BinSp, BinLen}}}
+                        <- Atts];
+                _ ->
+                    % BinFd must not equal our Fd. This can happen when a database
+                    % is being switched out during a compaction
+                    ?LOG_DEBUG("File where the attachments are written has changed. Possibly retrying.", []),
+                    throw(retry)
+                end,
+                {ok, NewSummaryPointer} = couch_stream:write_term(Db#db.summary_stream, {Doc#doc.body, Bins}),
+                {IsDeleted, NewSummaryPointer};
+            _ ->
+                Value
+            end
+        end, Unflushed),
+    flush_trees(Db, RestUnflushed, [InfoUnflushed#full_doc_info{rev_tree=Flushed} | AccFlushed]).
+
+merge_rev_trees(_NoConflicts, [], [], AccNewInfos, AccSeq) ->
+    {ok, lists:reverse(AccNewInfos), AccSeq};
+merge_rev_trees(NoConflicts, [NewDocs|RestDocsList],
+        [OldDocInfo|RestOldInfo], AccNewInfos, AccSeq) ->
+    #full_doc_info{id=Id,rev_tree=OldTree}=OldDocInfo,
+    UpdatesRevTree = lists:foldl(
+        fun(NewDoc, AccTree) ->
+            couch_key_tree:merge(AccTree, doc_to_tree(NewDoc))
+        end,
+        [], NewDocs),
+    NewRevTree = couch_key_tree:merge(OldTree, UpdatesRevTree),
+    if NewRevTree == OldTree ->
+        % nothing changed
+        merge_rev_trees(NoConflicts, RestDocsList, RestOldInfo, AccNewInfos, AccSeq);
+    true ->
+        if NoConflicts andalso OldTree /= [] ->
+            OldConflicts = couch_key_tree:count_leafs(OldTree),
+            NewConflicts = couch_key_tree:count_leafs(NewRevTree),
+            if NewConflicts > OldConflicts ->
+                throw(conflict);
+            true -> ok
+            end;
+        true -> ok
+        end,
+        NewInfo = #full_doc_info{id=Id,update_seq=AccSeq+1,rev_tree=NewRevTree},
+        merge_rev_trees(NoConflicts, RestDocsList,RestOldInfo, 
+                [NewInfo|AccNewInfos],AccSeq+1)
+    end.
+
+new_index_entries([], DocCount, DelCount, AccById, AccBySeq) ->
+    {ok, DocCount, DelCount, AccById, AccBySeq};
+new_index_entries([FullDocInfo|RestInfos], DocCount, DelCount, AccById, AccBySeq) ->
+    #doc_info{deleted=Deleted} = DocInfo = couch_doc:to_doc_info(FullDocInfo),
+    {DocCount2, DelCount2} =
+    if Deleted -> {DocCount, DelCount + 1};
+    true -> {DocCount + 1, DelCount} 
+    end,
+    new_index_entries(RestInfos, DocCount2, DelCount2, 
+        [FullDocInfo#full_doc_info{deleted=Deleted}|AccById],
+        [DocInfo|AccBySeq]).
+
+update_docs_int(Db, DocsList, Options) ->
+    #db{
+        fulldocinfo_by_id_btree = DocInfoByIdBTree,
+        docinfo_by_seq_btree = DocInfoBySeqBTree,
+        update_seq = LastSeq,
+        doc_count = FullDocCount,
+        doc_del_count = FullDelCount
+        } = Db,
+
+    % separate out the NonRep documents from the rest of the documents
+    {DocsList2, NonRepDocs} = lists:foldl(
+        fun([#doc{id=Id}=Doc | Rest]=Docs, {DocsListAcc, NonRepDocsAcc}) ->
+            case Id of
+            ?LOCAL_DOC_PREFIX ++ _ when Rest==[] ->
+                % when saving NR (non rep) documents, you can only save a single rev
+                {DocsListAcc, [Doc | NonRepDocsAcc]};
+            Id->
+                {[Docs | DocsListAcc], NonRepDocsAcc}
+            end
+        end, {[], []}, DocsList),
+    
+    Ids = [Id || [#doc{id=Id}|_] <- DocsList2], 
+    
+    % lookup up the existing documents, if they exist.
+    OldDocLookups = couch_btree:lookup(DocInfoByIdBTree, Ids),
+    OldDocInfos = lists:zipwith(
+        fun(_Id, {ok, FullDocInfo}) ->
+            FullDocInfo;
+        (Id, not_found) ->
+            #full_doc_info{id=Id}
+        end,
+        Ids, OldDocLookups),
+    
+    {OldCount, OldDelCount} = lists:foldl(
+        fun({ok, FullDocInfo}, {OldCountAcc, OldDelCountAcc}) ->
+            case couch_doc:to_doc_info(FullDocInfo) of
+            #doc_info{deleted=false} ->
+                {OldCountAcc + 1, OldDelCountAcc};
+            _ ->
+                {OldCountAcc, OldDelCountAcc + 1}
+            end;
+        (not_found, Acc) ->
+            Acc
+        end, {0, 0}, OldDocLookups),
+    
+    % Merge the new docs into the revision trees.
+    NoConflicts = lists:member(new_edits, Options),
+    {ok, NewDocInfos, NewSeq} = merge_rev_trees(NoConflicts, DocsList2, OldDocInfos, [], LastSeq),
+    
+    RemoveSeqs =
+        [ OldSeq || {ok, #full_doc_info{update_seq=OldSeq}} <- OldDocLookups],
+    
+    % All regular documents are now ready to write.
+    
+    % Try to write the local documents first, a conflict might be generated
+    {ok, Db2}  = update_local_docs(Db, NonRepDocs),
+    
+    % Write out the documents summaries (they are stored in the nodes of the rev trees)
+    {ok, FlushedDocInfos} = flush_trees(Db2, NewDocInfos, []),
+    
+    {ok, NewDocsCount, NewDelCount, InfoById, InfoBySeq} =
+        new_index_entries(FlushedDocInfos, 0, 0, [], []),
+
+    % and the indexes to the documents
+    {ok, DocInfoBySeqBTree2} = couch_btree:add_remove(DocInfoBySeqBTree, InfoBySeq, RemoveSeqs),
+    {ok, DocInfoByIdBTree2} = couch_btree:add_remove(DocInfoByIdBTree, InfoById, []),
+
+    Db3 = Db2#db{
+        fulldocinfo_by_id_btree = DocInfoByIdBTree2,
+        docinfo_by_seq_btree = DocInfoBySeqBTree2,
+        update_seq = NewSeq,
+        doc_count = FullDocCount + NewDocsCount - OldCount,
+        doc_del_count = FullDelCount + NewDelCount - OldDelCount},
+
+    case lists:member(delay_commit, Options) of
+    true ->
+        {ok, Db3};
+    false ->
+        {ok, commit_data(Db3)}
+    end.
+
+update_local_docs(#db{local_docs_btree=Btree}=Db, Docs) ->
+    Ids = [Id || #doc{id=Id} <- Docs],
+    OldDocLookups = couch_btree:lookup(Btree, Ids),
+    BtreeEntries = lists:zipwith(
+        fun(#doc{id=Id,deleted=Delete,revs=Revs,body=Body}, OldDocLookup) ->
+            NewRev =
+            case Revs of
+                [] -> 0;
+                [RevStr|_] -> list_to_integer(RevStr)
+            end,
+            OldRev =
+            case OldDocLookup of
+                {ok, {_, {OldRev0, _}}} -> OldRev0;
+                not_found -> 0
+            end,
+            case OldRev + 1 == NewRev of
+            true ->
+                case Delete of
+                    false -> {update, {Id, {NewRev, Body}}};
+                    true  -> {remove, Id}
+                end;
+            false ->
+                throw(conflict)
+            end
+            
+        end, Docs, OldDocLookups),
+
+    BtreeIdsRemove = [Id || {remove, Id} <- BtreeEntries],
+    BtreeIdsUpdate = [ByIdDocInfo || {update, ByIdDocInfo} <- BtreeEntries],
+
+    {ok, Btree2} =
+        couch_btree:add_remove(Btree, BtreeIdsUpdate, BtreeIdsRemove),
+
+    {ok, Db#db{local_docs_btree = Btree2}}.
+
+
+
+commit_data(#db{fd=Fd, header=Header} = Db) ->
+    Header2 = Header#db_header{
+        update_seq = Db#db.update_seq,
+        summary_stream_state = couch_stream:get_state(Db#db.summary_stream),
+        docinfo_by_seq_btree_state = couch_btree:get_state(Db#db.docinfo_by_seq_btree),
+        fulldocinfo_by_id_btree_state = couch_btree:get_state(Db#db.fulldocinfo_by_id_btree),
+        local_docs_btree_state = couch_btree:get_state(Db#db.local_docs_btree),
+        doc_count = Db#db.doc_count,
+        doc_del_count = Db#db.doc_del_count
+        },
+    if Header == Header2 ->
+        Db; % unchanged. nothing to do
+    true ->
+        ok = couch_file:write_header(Fd, ?HEADER_SIG, Header2),
+        Db#db{header = Header2}
+    end.
+
+copy_raw_doc(SrcFd, SrcSp, DestFd, DestStream) ->
+    {ok, {BodyData, BinInfos}} = couch_stream:read_term(SrcFd, SrcSp),
+    % copy the bin values
+    NewBinInfos = lists:map(fun({Name, {Type, BinSp, Len}}) ->
+        {ok, NewBinSp} = couch_stream:copy_to_new_stream(SrcFd, BinSp, Len, DestFd),
+        {Name, {Type, NewBinSp, Len}}
+        end, BinInfos),
+    % now write the document summary
+    {ok, Sp} = couch_stream:write_term(DestStream, {BodyData, NewBinInfos}),
+    Sp.
+
+copy_rev_tree(_SrcFd, _DestFd, _DestStream, []) ->
+    [];
+copy_rev_tree(SrcFd, DestFd, DestStream, [{RevId, {IsDel, Sp}, []} | RestTree]) ->
+    % This is a leaf node, copy it over
+    NewSp = copy_raw_doc(SrcFd, Sp, DestFd, DestStream),
+    [{RevId, {IsDel, NewSp}, []} | copy_rev_tree(SrcFd, DestFd, DestStream, RestTree)];
+copy_rev_tree(SrcFd, DestFd, DestStream, [{RevId, _, SubTree} | RestTree]) ->
+    % inner node, only copy info/data from leaf nodes
+    [{RevId, ?REV_MISSING, copy_rev_tree(SrcFd, DestFd, DestStream, SubTree)} | copy_rev_tree(SrcFd, DestFd, DestStream, RestTree)].
+    
+copy_docs(#db{fd=SrcFd}=Db, #db{fd=DestFd,summary_stream=DestStream}=NewDb, InfoBySeq) ->
+    Ids = [Id || #doc_info{id=Id} <- InfoBySeq],
+    LookupResults = couch_btree:lookup(Db#db.fulldocinfo_by_id_btree, Ids),
+    NewFullDocInfos = lists:map(
+        fun({ok, #full_doc_info{rev_tree=RevTree}=Info}) ->
+            Info#full_doc_info{rev_tree=copy_rev_tree(SrcFd, DestFd, DestStream, RevTree)}
+        end, LookupResults),
+    NewDocInfos = [couch_doc:to_doc_info(FullDocInfo) || FullDocInfo <- NewFullDocInfos],
+    {ok, DocInfoBTree} =
+        couch_btree:add_remove(NewDb#db.docinfo_by_seq_btree, NewDocInfos, []),
+    {ok, FullDocInfoBTree} =
+        couch_btree:add_remove(NewDb#db.fulldocinfo_by_id_btree, NewFullDocInfos, []),
+    NewDb#db{fulldocinfo_by_id_btree=FullDocInfoBTree, docinfo_by_seq_btree=DocInfoBTree}.
+
+
+          
+copy_compact_docs(Db, NewDb) ->
+    EnumBySeqFun =
+    fun(#doc_info{update_seq=Seq}=DocInfo, _Offset, {AccNewDb, AccUncopied}) ->
+        case couch_util:should_flush() of
+        true ->
+            NewDb2 = copy_docs(Db, AccNewDb, lists:reverse([DocInfo | AccUncopied])),
+            {ok, {commit_data(NewDb2#db{update_seq=Seq}), []}};
+        false ->    
+            {ok, {AccNewDb, [DocInfo | AccUncopied]}}
+        end
+    end,
+    {ok, {NewDb2, Uncopied}} =
+        couch_btree:foldl(Db#db.docinfo_by_seq_btree, NewDb#db.update_seq + 1, EnumBySeqFun, {NewDb, []}),
+
+    case Uncopied of
+    [#doc_info{update_seq=LastSeq} | _] ->
+        commit_data( copy_docs(Db, NewDb2#db{update_seq=LastSeq},
+            lists:reverse(Uncopied)));
+    [] ->
+        NewDb2
+    end.
+
+start_copy_compact_int(#db{name=Name,filepath=Filepath}=Db, CopyLocal) ->
+    CompactFile = Filepath ++ ".compact",
+    ?LOG_DEBUG("Compaction process spawned for db \"~s\"", [Name]),
+    case couch_file:open(CompactFile) of
+    {ok, Fd} ->
+        ?LOG_DEBUG("Found existing compaction file for db \"~s\"", [Name]),
+        {ok, Header} = couch_file:read_header(Fd, ?HEADER_SIG);
+    {error, enoent} ->
+        receive after 1000 -> ok end,
+        {ok, Fd} = couch_file:open(CompactFile, [create]),
+        Header =  #db_header{},
+        ok = couch_file:write_header(Fd, ?HEADER_SIG, Header)
+    end,
+    NewDb = init_db(Name, CompactFile, Fd, Header),
+    NewDb2 = copy_compact_docs(Db, NewDb),
+    NewDb3 =
+    case CopyLocal of
+    true ->
+        % suck up all the local docs into memory and write them to the new db
+        {ok, LocalDocs} = couch_btree:foldl(Db#db.local_docs_btree,
+                fun(Value, _Offset, Acc) -> {ok, [Value | Acc]} end, []),
+        {ok, NewLocalBtree} = couch_btree:add(NewDb2#db.local_docs_btree, LocalDocs),
+        commit_data(NewDb2#db{local_docs_btree=NewLocalBtree});
+    _ ->
+        NewDb2
+    end,
+    close_db(NewDb3),
+    
+    gen_server:cast(Db#db.update_pid, {compact_done, CompactFile}).
\ No newline at end of file

Modified: incubator/couchdb/trunk/src/couchdb/couch_file.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_file.erl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_file.erl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_file.erl Mon Aug  4 18:43:40 2008
@@ -20,6 +20,7 @@
 -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].
@@ -164,7 +165,25 @@
 %%----------------------------------------------------------------------
 close(Fd) ->
     gen_server:cast(Fd, close).
+    
+close_maybe(Fd) ->
+    gen_server:cast(Fd, {close_maybe, self()}).
+
+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) ->
     TermBin = term_to_binary(Data),
@@ -267,7 +286,7 @@
 
 init_status_error(ReturnPid, Error) ->
     ReturnPid ! {self(), Error}, % signal back error status
-    self() ! self_close, % tell ourself to close async
+    gen_server:cast(self(), close), % tell ourself to close async
     {ok, nil}.
 
 % server functions
@@ -342,16 +361,57 @@
     {ok, <<TermLen:32>>}
         = file:pread(Fd, Pos, 4),
     {ok, Bin} = file:pread(Fd, Pos + 4, TermLen),
-    {reply, {ok, Bin}, Fd}.
+    {reply, {ok, Bin}, Fd};
+handle_call({add_ref, Pid},_From, Fd) ->
+    undefined = put(Pid, erlang:monitor(process, Pid)),
+    {reply, ok, Fd};
+handle_call(num_refs, _From, Fd) ->
+    {monitors, Monitors} =  process_info(self(), monitors),
+    {reply, length(Monitors), Fd}.
+
 
 
 handle_cast(close, Fd) ->
-    {stop,normal,Fd}. % causes terminate to be called
+    {stop,normal,Fd};
+handle_cast({close_maybe, Pid}, Fd) ->
+    catch unlink(Pid),
+    maybe_close_async(Fd);
+handle_cast({drop_ref, Pid}, Fd) ->
+    % don't check return of demonitor. The process could haved crashed causing
+    % the {'DOWN', ...} message to be sent and the process unmonitored.
+    erlang:demonitor(erase(Pid), [flush]),
+    maybe_close_async(Fd).
+
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-handle_info(self_close, State) ->
-    {stop,normal,State};
-handle_info(_Info, State) ->
-    {noreply, State}.
+handle_info({'DOWN', MonitorRef, _Type, Pid, _Info}, Fd) ->
+    MonitorRef = erase(Pid),
+    maybe_close_async(Fd);
+handle_info(Info, Fd) ->
+    exit({error, {Info, Fd}}).
+
+
+
+should_close(Fd) ->
+    case process_info(self(), links) of
+    {links, [Fd]} ->
+        % no linkers left (except our fd). What about monitors?
+        case process_info(self(), monitors) of
+        {monitors, []} ->
+            true;
+        _ ->
+            false
+        end;
+    {links,  Links} when length(Links) > 1 ->
+        false
+    end.
+
+maybe_close_async(Fd) ->
+    case should_close(Fd) of
+    true ->
+        {stop,normal,Fd};
+    false ->
+        {noreply,Fd}
+    end.

Modified: incubator/couchdb/trunk/src/couchdb/couch_httpd.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_httpd.erl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_httpd.erl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_httpd.erl Mon Aug  4 18:43:40 2008
@@ -156,7 +156,8 @@
 
 handle_db_request(Req, 'PUT', {DbName, []}) ->
     case couch_server:create(DbName, []) of
-        {ok, _Db} ->
+        {ok, Db} ->
+            couch_db:close(Db),
             send_json(Req, 201, {obj, [{ok, true}]});
         {error, database_already_exists} ->
             Msg = io_lib:format("Database ~p already exists.", [DbName]),
@@ -167,9 +168,13 @@
     end;
 
 handle_db_request(Req, Method, {DbName, Rest}) ->
-    case couch_server:open(DbName) of
+    case couch_db:open(DbName, []) of
         {ok, Db} ->
-            handle_db_request(Req, Method, {DbName, Db, Rest});
+            try 
+                handle_db_request(Req, Method, {DbName, Db, Rest})
+            after
+                couch_db:close(Db)
+            end;
         Error ->
             throw(Error)
     end;

Modified: incubator/couchdb/trunk/src/couchdb/couch_rep.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_rep.erl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_rep.erl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_rep.erl Mon Aug  4 18:43:40 2008
@@ -43,7 +43,18 @@
 
 replicate(Source, Target, Options) ->
     {ok, DbSrc} = open_db(Source),
-    {ok, DbTgt} = open_db(Target),
+    try
+        {ok, DbTgt} = open_db(Target),
+        try
+            replicate2(Source, DbSrc, Target, DbTgt, Options)
+        after
+            close_db(DbTgt)
+        end        
+    after
+        close_db(DbSrc)
+    end.
+    
+replicate2(Source, DbSrc, Target, DbTgt, Options) ->
     {ok, HostName} = inet:gethostname(),
 
     RepRecKey = ?LOCAL_DOC_PREFIX ++ HostName ++ ":" ++ Source ++ ":" ++ Target,
@@ -237,7 +248,12 @@
         {ok, "http" ++ DbName ++ "/"}
     end;
 open_db(DbName)->
-    couch_server:open(DbName).
+    couch_db:open(DbName, []).
+
+close_db("http" ++ _)->
+    ok;
+close_db(DbName)->
+    couch_db:close(DbName).
 
 
 enum_docs_since(DbUrl, StartSeq, InFun, InAcc) when is_list(DbUrl) ->

Modified: incubator/couchdb/trunk/src/couchdb/couch_server.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_server.erl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_server.erl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_server.erl Mon Aug  4 18:43:40 2008
@@ -15,7 +15,7 @@
 -behaviour(application).
 
 -export([start/0,start/1,start/2,stop/0,stop/1]).
--export([open/1,create/2,delete/1,all_databases/0,get_version/0]).
+-export([open/2,create/2,delete/1,all_databases/0,get_version/0]).
 -export([init/1, handle_call/3,sup_start_link/2]).
 -export([handle_cast/2,code_change/3,handle_info/2,terminate/2]).
 -export([dev_start/0,remote_restart/0]).
@@ -25,7 +25,9 @@
 -record(server,{
     root_dir = [],
     dbname_regexp,
-    options=[]
+    remote_restart=[],
+    max_dbs_open=100,
+    current_dbs_open=0
     }).
 
 start() ->
@@ -64,33 +66,41 @@
 sup_start_link(RootDir, Options) ->
     gen_server:start_link({local, couch_server}, couch_server, {RootDir, Options}, []).
 
-open(Filename) ->
-    gen_server:call(couch_server, {open, Filename}).
+open(DbName, Options) ->
+    gen_server:call(couch_server, {open, DbName, Options}).
 
-create(Filename, Options) ->
-    gen_server:call(couch_server, {create, Filename, Options}).
+create(DbName, Options) ->
+    gen_server:call(couch_server, {create, DbName, Options}).
 
-delete(Filename) ->
-    gen_server:call(couch_server, {delete, Filename}).
+delete(DbName) ->
+    gen_server:call(couch_server, {delete, DbName}).
 
 remote_restart() ->
     gen_server:call(couch_server, remote_restart).
 
-init({RootDir, Options}) ->
-    {ok, RegExp} = regexp:parse("^[a-z][a-z0-9\\_\\$()\\+\\-\\/]*$"),
-    {ok, #server{root_dir=RootDir, dbname_regexp=RegExp, options=Options}}.
-
-check_filename(#server{dbname_regexp=RegExp}, Filename) ->
-    case regexp:match(Filename, RegExp) of
+check_dbname(#server{dbname_regexp=RegExp}, DbName) ->
+    case regexp:match(DbName, RegExp) of
     nomatch ->
         {error, illegal_database_name};
     _Match ->
         ok
     end.
 
-get_full_filename(Server, Filename) ->
-    filename:join([Server#server.root_dir, "./" ++ Filename ++ ".couch"]).
+get_full_filename(Server, DbName) ->
+    filename:join([Server#server.root_dir, "./" ++ DbName ++ ".couch"]).
 
+init({RootDir, Options}) ->
+    {ok, RegExp} = regexp:parse("^[a-z][a-z0-9\\_\\$()\\+\\-\\/]*$"),
+    ets:new(couch_dbs_by_name, [set, private, named_table]),
+    ets:new(couch_dbs_by_pid, [set, private, named_table]),
+    ets:new(couch_dbs_by_lru, [ordered_set, private, named_table]),
+    process_flag(trap_exit, true),
+    MaxDbsOpen = proplists:get_value(max_dbs_open, Options),
+    RemoteRestart = proplists:get_value(remote_restart, Options),
+    {ok, #server{root_dir=RootDir,
+                dbname_regexp=RegExp,
+                max_dbs_open=MaxDbsOpen,
+                remote_restart=RemoteRestart}}.
 
 terminate(_Reason, _Server) ->
     ok.
@@ -109,107 +119,141 @@
     {ok, Filenames}.
 
 
+maybe_close_lru_db(#server{current_dbs_open=NumOpen, max_dbs_open=MaxOpen}=Server)
+        when NumOpen < MaxOpen ->
+    {ok, Server};
+maybe_close_lru_db(#server{current_dbs_open=NumOpen}=Server) ->
+    % must free up the lru db.
+    case try_close_lru(now()) of
+    ok -> {ok, Server#server{current_dbs_open=NumOpen-1}};
+    Error -> Error
+    end.
+
+try_close_lru(StartTime) ->
+    LruTime = ets:first(couch_dbs_by_lru),
+    if LruTime > StartTime ->
+        % this means we've looped through all our opened dbs and found them
+        % all in use.
+        {error, all_dbs_active};
+    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 ->
+            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 ->
+            % this still has referrers. Go ahead and give it a current lru time
+            % and try the next one in the table.
+            NewLruTime = now(),
+            true = ets:insert(couch_dbs_by_name, {DbName, {MainPid, NewLruTime}}),
+            true = ets:insert(couch_dbs_by_pid, {MainPid, DbName}),
+            true = ets:delete(couch_dbs_by_lru, LruTime),
+            true = ets:insert(couch_dbs_by_lru, {NewLruTime, DbName}),
+            try_close_lru(StartTime)
+        end
+    end.
+
+
 handle_call(get_root, _From, #server{root_dir=Root}=Server) ->
     {reply, {ok, Root}, Server};
-handle_call({open, Filename}, From, Server) ->
-    case check_filename(Server, Filename) of
-    {error, Error} ->
-        {reply, {error, Error}, Server};
-    ok ->
-        Filepath = get_full_filename(Server, Filename),
-        Result = supervisor:start_child(couch_server_sup,
-            {Filename,
-                {couch_db, open, [Filename, Filepath]},
-                transient ,
-                infinity,
-                supervisor,
-                [couch_db]}),
-        case Result of
-        {ok, Db} ->
-            {reply, {ok, Db}, Server};
-        {error, already_present} ->
-            ok = supervisor:delete_child(couch_server_sup, Filename),
-            % call self recursively
-            handle_call({open, Filename}, From, Server);
-        {error, {already_started, Db}} ->
-            {reply, {ok, Db}, Server};
-        {error, {not_found, _}} ->
-            {reply, not_found, Server};
-        {error, {Error, _}} ->
-            {reply, {error, Error}, Server}
-        end
+handle_call({open, DbName, Options}, {FromPid,_}, Server) ->
+    Filepath = get_full_filename(Server, DbName),
+    LruTime = now(),
+    case ets:lookup(couch_dbs_by_name, DbName) of
+    [] ->    
+        case maybe_close_lru_db(Server) of
+        {ok, Server2} ->
+            case couch_db:start_link(DbName, Filepath, Options) of
+            {ok, MainPid} ->
+                true = ets:insert(couch_dbs_by_name, {DbName, {MainPid, LruTime}}),
+                true = ets:insert(couch_dbs_by_pid, {MainPid, DbName}),
+                true = ets:insert(couch_dbs_by_lru, {LruTime, DbName}),
+                DbsOpen = Server2#server.current_dbs_open + 1,
+                {reply,
+                    couch_db:open_ref_counted(MainPid, FromPid),
+                    Server2#server{current_dbs_open=DbsOpen}};
+            CloseError ->
+                {reply, CloseError, Server2}
+            end;
+        Error ->
+            {reply, Error, Server}
+        end;
+    [{_, {MainPid, PrevLruTime}}] ->
+        true = ets:insert(couch_dbs_by_name, {DbName, {MainPid, LruTime}}),
+        true = ets:delete(couch_dbs_by_lru, PrevLruTime),
+        true = ets:insert(couch_dbs_by_lru, {LruTime, DbName}),
+        {reply, couch_db:open_ref_counted(MainPid, FromPid), Server}
     end;
-handle_call({create, Filename, Options}, _From, Server) ->
-    case check_filename(Server, Filename) of
-    {error, Error} ->
-        {reply, {error, Error}, Server};
+handle_call({create, DbName, Options}, {FromPid,_}, Server) ->
+    case check_dbname(Server, DbName) of
     ok ->
-        Filepath = get_full_filename(Server, Filename),
-        ChildSpec = {Filename,
-                {couch_db, create, [Filename, Filepath, Options]},
-                transient,
-                infinity,
-                supervisor,
-                [couch_db]},
-        Result =
-        case supervisor:delete_child(couch_server_sup, Filename) of
-        ok ->
-            sup_start_child(couch_server_sup, ChildSpec);
-        {error, not_found} ->
-            sup_start_child(couch_server_sup, ChildSpec);
-        {error, running} ->
-            % a server process for this database already started. Maybe kill it
-            case lists:member(overwrite, Options) of
-            true ->
-                supervisor:terminate_child(couch_server_sup, Filename),
-                ok = supervisor:delete_child(couch_server_sup, Filename),
-                sup_start_child(couch_server_sup, ChildSpec);
-            false ->
-                {error, database_already_exists}
-            end
-        end,
-        case Result of
-        {ok, _Db} -> couch_db_update_notifier:notify({created, Filename});
-        _ -> ok
-        end,
-        {reply, Result, Server}
+        Filepath = get_full_filename(Server, DbName),
+
+        case ets:lookup(couch_dbs_by_name, DbName) of
+        [] ->
+            case couch_db:start_link(DbName, Filepath, [create|Options]) of
+            {ok, MainPid} ->
+                LruTime = now(),
+                true = ets:insert(couch_dbs_by_name, {DbName, {MainPid, LruTime}}),
+                true = ets:insert(couch_dbs_by_pid, {MainPid, DbName}),
+                true = ets:insert(couch_dbs_by_lru, {LruTime, DbName}),
+                DbsOpen = Server#server.current_dbs_open + 1,
+                {reply,
+                    couch_db:open_ref_counted(MainPid, FromPid), 
+                    Server#server{current_dbs_open=DbsOpen}};
+            Error ->
+                {reply, Error, Server}
+            end;
+        [_AlreadyRunningDb] ->
+            {reply, {error, file_exists}, Server}
+        end;
+    Error ->
+        {reply, Error, Server}
     end;
-handle_call({delete, Filename}, _From, Server) ->
-    FullFilepath = get_full_filename(Server, Filename),
-    supervisor:terminate_child(couch_server_sup, Filename),
-    supervisor:delete_child(couch_server_sup, Filename),
+handle_call({delete, DbName}, _From, Server) ->
+    FullFilepath = get_full_filename(Server, DbName),
+    Server2 = 
+    case ets:lookup(couch_dbs_by_name, DbName) of
+    [] -> Server;
+    [{_, {Pid, LruTime}}] ->
+        exit(Pid, kill),
+        receive {'EXIT', Pid, _Reason} -> ok end,
+        true = ets:delete(couch_dbs_by_name, DbName),
+        true = ets:delete(couch_dbs_by_pid, Pid),
+        true = ets:delete(couch_dbs_by_lru, LruTime),
+        DbsOpen = Server#server.current_dbs_open - 1,
+        Server#server{current_dbs_open=DbsOpen}
+    end,
     case file:delete(FullFilepath) of
     ok ->
-        couch_db_update_notifier:notify({deleted, Filename}),
-        {reply, ok, Server};
+        couch_db_update_notifier:notify({deleted, DbName}),
+        {reply, ok, Server2};
     {error, enoent} ->
-        {reply, not_found, Server};
+        {reply, not_found, Server2};
     Else ->
-        {reply, Else, Server}
+        {reply, Else, Server2}
     end;
-handle_call(remote_restart, _From, #server{options=Options}=Server) ->
-    case proplists:get_value(remote_restart, Options) of
-    true ->
-        exit(self(), restart);
-    _ ->
-        ok
-    end,
+handle_call(remote_restart, _From, #server{remote_restart=false}=Server) ->
+    {reply, ok, Server};
+handle_call(remote_restart, _From, #server{remote_restart=true}=Server) ->
+    exit(couch_server_sup, restart),
     {reply, ok, Server}.
 
-% this function is just to strip out the child spec error stuff if hit
-sup_start_child(couch_server_sup, ChildSpec) ->
-    case supervisor:start_child(couch_server_sup, ChildSpec) of
-    {error, {Error, _ChildInfo}} ->
-        {error, Error};
-    Else ->
-        Else
-    end.
-
-handle_cast(_Msg, State) ->
-    {noreply,State}.
+handle_cast(Msg, _Server) ->
+    exit({unknown_cast_message, Msg}).
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-handle_info(_Info, State) ->
-    {noreply, State}.
+handle_info({'EXIT', Pid, _Reason}, Server) ->
+    [{Pid, DbName}] = ets:lookup(couch_dbs_by_pid, Pid),
+    true = ets:delete(couch_dbs_by_pid, Pid),
+    true = ets:delete(couch_dbs_by_name, DbName),
+    {noreply, Server};
+handle_info(Info, _Server) ->
+    exit({unknown_message, Info}).

Modified: incubator/couchdb/trunk/src/couchdb/couch_server_sup.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_server_sup.erl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_server_sup.erl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_server_sup.erl Mon Aug  4 18:43:40 2008
@@ -74,8 +74,9 @@
     UtilDriverDir = proplists:get_value({"Couch", "UtilDriverDir"}, Ini, ""),
     UpdateNotifierExes = proplists:get_all_values({"Couch", "DbUpdateNotificationProcess"}, Ini),
     FtSearchQueryServer = proplists:get_value({"Couch", "FullTextSearchQueryServer"}, Ini, ""),
-    RemoteRestart = list_to_atom(proplists:get_value({"Couch", "AllowRemoteRestart"}, Ini, "undefined")),
-    ServerOptions = [{remote_restart, RemoteRestart}],
+    RemoteRestart = list_to_atom(proplists:get_value({"Couch", "AllowRemoteRestart"}, Ini, "false")),
+    MaxDbsOpen = proplists:get_value({"Couch", "MaxDbsOpen"}, Ini, 100),
+    ServerOptions = [{remote_restart, RemoteRestart}, {max_dbs_open, MaxDbsOpen}],
     QueryServers = [{Lang, QueryExe} || {{"Couch Query Servers", Lang}, QueryExe} <- Ini],
 
     ChildProcesses =

Modified: incubator/couchdb/trunk/src/couchdb/couch_view.erl
URL: http://svn.apache.org/viewvc/incubator/couchdb/trunk/src/couchdb/couch_view.erl?rev=682560&r1=682559&r2=682560&view=diff
==============================================================================
--- incubator/couchdb/trunk/src/couchdb/couch_view.erl (original)
+++ incubator/couchdb/trunk/src/couchdb/couch_view.erl Mon Aug  4 18:43:40 2008
@@ -57,22 +57,22 @@
     Pid.
     
 get_updated_group(Pid) ->
-	Mref = erlang:monitor(process, Pid),
+    Mref = erlang:monitor(process, Pid),
     receive
-	{'DOWN', Mref, _, _, Reason} ->
-	    throw(Reason)
+    {'DOWN', Mref, _, _, Reason} ->
+        throw(Reason)
     after 0 ->
-	    Pid ! {self(), get_updated},
-	    receive
-    	{Pid, Response} ->
-    	    erlang:demonitor(Mref),
-    	    receive
-        		{'DOWN', Mref, _, _, _} -> ok
-        	    after 0 -> ok
-    	    end,
-    	    Response;
-    	{'DOWN', Mref, _, _, Reason} ->
-    	    throw(Reason)
+        Pid ! {self(), get_updated},
+        receive
+        {Pid, Response} ->
+            erlang:demonitor(Mref),
+            receive
+                {'DOWN', Mref, _, _, _} -> ok
+                after 0 -> ok
+            end,
+            Response;
+        {'DOWN', Mref, _, _, Reason} ->
+            throw(Reason)
         end
     end.
 
@@ -216,10 +216,7 @@
     {ok, #server{root_dir=RootDir}}.
 
 terminate(_Reason, _) ->
-    catch ets:delete(couch_views_by_name),
-    catch ets:delete(couch_views_by_updater),
-    catch ets:delete(couch_views_by_db),
-    catch ets:delete(couch_views_temp_fd_by_db).
+    ok.
 
 
 handle_call({start_temp_updater, DbName, Lang, MapSrc, RedSrc}, _From, #server{root_dir=Root}=Server) ->
@@ -317,7 +314,7 @@
 
 start_temp_update_loop(DbName, Fd, Lang, MapSrc, RedSrc) ->
     NotifyPids = get_notify_pids(1000),
-    case couch_server:open(DbName) of
+    case couch_db:open(DbName, []) of
     {ok, Db} ->
         View = #view{map_names=["_temp"],
             id_num=0,
@@ -331,16 +328,20 @@
             def_lang=Lang,
             id_btree=nil},
         Group2 = init_group(Db, Fd, Group,nil),
-        temp_update_loop(Group2, NotifyPids);
+        couch_db:monitor(Db),
+        couch_db:close(Db),
+        temp_update_loop(DbName, Group2, NotifyPids);
     Else ->
         exit(Else)
     end.
 
-temp_update_loop(Group, NotifyPids) ->
-    {ok, Group2} = update_group(Group),
+temp_update_loop(DbName, Group, NotifyPids) ->
+    {ok, Db} = couch_db:open(DbName, []),
+    {ok, Group2} = update_group(Group#group{db=Db}),
+    couch_db:close(Db),
     [Pid ! {self(), {ok, Group2}} || Pid <- NotifyPids],
     garbage_collect(),
-    temp_update_loop(Group2, get_notify_pids(10000)).
+    temp_update_loop(DbName, Group2, get_notify_pids(10000)).
 
 
 reset_group(#group{views=Views}=Group) ->
@@ -355,21 +356,21 @@
     
 start_update_loop(RootDir, DbName, GroupId, NotifyPids) ->
     {Db, Group} =
-    case (catch couch_server:open(DbName)) of
+    case (catch couch_db:open(DbName, [])) of
     {ok, Db0} ->
         case (catch couch_db:open_doc(Db0, GroupId)) of
         {ok, Doc} ->
             {Db0, design_doc_to_view_group(Doc)};
- 		Else ->
- 		    delete_index_file(RootDir, DbName, GroupId),
- 		    exit(Else)
- 		end;
- 	Else ->
- 	    delete_index_file(RootDir, DbName, GroupId),
- 	    exit(Else)
- 	end,
- 	FileName = RootDir ++ "/." ++ DbName ++ GroupId ++".view",
- 	Group2 =
+        Else ->
+            delete_index_file(RootDir, DbName, GroupId),
+            exit(Else)
+        end;
+    Else ->
+        delete_index_file(RootDir, DbName, GroupId),
+        exit(Else)
+    end,
+    FileName = RootDir ++ "/." ++ DbName ++ GroupId ++".view",
+    Group2 =
     case couch_file:open(FileName) of
     {ok, Fd} ->
         Sig = Group#group.sig,
@@ -386,7 +387,8 @@
         Error    -> throw(Error)
         end
     end,
-    
+    couch_db:monitor(Db),
+    couch_db:close(Db),
     update_loop(RootDir, DbName, GroupId, Group2, NotifyPids).
 
 reset_file(Db, Fd, DbName, #group{sig=Sig,name=Name} = Group) ->
@@ -396,14 +398,22 @@
     init_group(Db, Fd, reset_group(Group), nil).
 
 update_loop(RootDir, DbName, GroupId, #group{sig=Sig,fd=Fd}=Group, NotifyPids) ->
-    try update_group(Group) of
-    {ok, Group2} ->    
+    {ok, Db}= couch_db:open(DbName, []),
+    Result =
+    try
+        update_group(Group#group{db=Db})
+    catch
+        throw: restart -> restart
+    after
+        couch_db:close(Db)
+    end,
+    case Result of
+    {ok, Group2} ->
         HeaderData = {Sig, get_index_header_data(Group2)},
         ok = couch_file:write_header(Fd, <<$r, $c, $k, 0>>, HeaderData),
         [Pid ! {self(), {ok, Group2}} || Pid <- NotifyPids],
         garbage_collect(),
-        update_loop(RootDir, DbName, GroupId, Group2, get_notify_pids(100000))
-    catch
+        update_loop(RootDir, DbName, GroupId, Group2, get_notify_pids(100000));
     restart ->
         couch_file:close(Group#group.fd),
         start_update_loop(RootDir, DbName, GroupId, NotifyPids ++ get_notify_pids())
@@ -414,20 +424,23 @@
     receive
     {Pid, get_updated} ->
         [Pid | get_notify_pids()];
+    {'DOWN', _MonitorRef, _Type, _Pid, _Info} ->
+        ?LOG_DEBUG("View monitor received parent db shutdown notification. Shutting down instance.", []),
+        exit(normal);
     Else ->
         ?LOG_ERROR("Unexpected message in view updater: ~p", [Else]),
         exit({error, Else})
     after Wait ->
         exit(wait_timeout)
-	end.
+    end.
 % then keep getting all available and return.
 get_notify_pids() ->
     receive
     {Pid, get_updated} ->
         [Pid | get_notify_pids()]
-	after 0 ->
-	    []
-	end.
+    after 0 ->
+        []
+    end.
     
 update_group(#group{db=Db,current_seq=CurrentSeq, views=Views}=Group) ->
     ViewEmptyKVs = [{View, []} || View <- Views],



Mime
View raw message