couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kocol...@apache.org
Subject svn commit: r794843 - in /couchdb/trunk/src/couchdb: couch_view_group.erl couch_view_updater.erl
Date Thu, 16 Jul 2009 20:40:59 GMT
Author: kocolosk
Date: Thu Jul 16 20:40:59 2009
New Revision: 794843

URL: http://svn.apache.org/viewvc?rev=794843&view=rev
Log:
checkpoint long-running view updates

Improves stale=ok view queries, which will now see gradually updating indexes.
Also reduces the consequences of an updater crash because we can restart from
the latest checkpoint.

Modified:
    couchdb/trunk/src/couchdb/couch_view_group.erl
    couchdb/trunk/src/couchdb/couch_view_updater.erl

Modified: couchdb/trunk/src/couchdb/couch_view_group.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_view_group.erl?rev=794843&r1=794842&r2=794843&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_view_group.erl (original)
+++ couchdb/trunk/src/couchdb/couch_view_group.erl Thu Jul 16 20:40:59 2009
@@ -89,7 +89,8 @@
     case prepare_group(InitArgs, false) of
     {ok, #group{db=Db, fd=Fd}=Group} ->
         couch_db:monitor(Db),
-        Pid = spawn_link(fun()-> couch_view_updater:update(Group) end),
+        Owner = self(),
+        Pid = spawn_link(fun()-> couch_view_updater:update(Owner, Group) end),
         {ok, RefCounter} = couch_ref_counter:start([Fd]),
         {ok, #group_state{
                 db_name=couch_db:name(Db),
@@ -127,7 +128,8 @@
             }=State) when RequestSeq > Seq ->
     {ok, Db} = couch_db:open(DbName, []),
     Group2 = Group#group{db=Db},
-    Pid = spawn_link(fun()-> couch_view_updater:update(Group2) end),
+    Owner = self(),
+    Pid = spawn_link(fun()-> couch_view_updater:update(Owner, Group2) end),
 
     {noreply, State#group_state{
         updater_pid=Pid,
@@ -205,7 +207,7 @@
     {ok, Db} = couch_db:open(DbName, []),
     Pid = spawn_link(fun() ->
         {_,Ref} = erlang:spawn_monitor(fun() ->
-            couch_view_updater:update(NewGroup#group{db = Db})
+            couch_view_updater:update(nil, NewGroup#group{db = Db})
         end),
         receive
             {'DOWN', Ref, _, _, {new_group, NewGroup2}} ->
@@ -214,7 +216,21 @@
                 gen_server:cast(Pid2, {compact_done, NewGroup2})
         end
     end),
-    {noreply, State#group_state{compactor_pid = Pid}}.
+    {noreply, State#group_state{compactor_pid = Pid}};
+
+handle_cast({partial_update, NewGroup}, State) ->
+    #group_state{
+        db_name = DbName,
+        waiting_commit = WaitingCommit
+    } = State,
+    NewSeq = NewGroup#group.current_seq,
+    ?LOG_INFO("checkpointing view update at seq ~p for ~s ~s", [NewSeq,
+        DbName, NewGroup#group.name]),
+    if not WaitingCommit ->
+        erlang:send_after(1000, self(), delayed_commit);
+    true -> ok
+    end,
+    {noreply, State#group_state{group=NewGroup, waiting_commit=true}}.
 
 handle_info(delayed_commit, #group_state{db_name=DbName,group=Group}=State) ->
     {ok, Db} = couch_db:open(DbName, []),
@@ -254,7 +270,8 @@
         % we still have some waiters, reopen the database and reupdate the index
         {ok, Db2} = couch_db:open(DbName, []),
         Group2 = Group#group{db=Db2},
-        Pid = spawn_link(fun() -> couch_view_updater:update(Group2) end),
+        Owner = self(),
+        Pid = spawn_link(fun() -> couch_view_updater:update(Owner, Group2) end),
         {noreply, State#group_state{waiting_commit=true,
                 waiting_list=StillWaiting, group=Group2, updater_pid=Pid}}
     end;
@@ -267,7 +284,8 @@
     ok = couch_db:close(Group#group.db),
     case prepare_group(InitArgs, true) of
     {ok, ResetGroup} ->
-        Pid = spawn_link(fun()-> couch_view_updater:update(ResetGroup) end),
+        Owner = self(),
+        Pid = spawn_link(fun()-> couch_view_updater:update(Owner, ResetGroup) end),
         {noreply, State#group_state{
                 updater_pid=Pid,
                 group=ResetGroup}};

Modified: couchdb/trunk/src/couchdb/couch_view_updater.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_view_updater.erl?rev=794843&r1=794842&r2=794843&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_view_updater.erl (original)
+++ couchdb/trunk/src/couchdb/couch_view_updater.erl Thu Jul 16 20:40:59 2009
@@ -12,11 +12,17 @@
 
 -module(couch_view_updater).
 
--export([update/1]).
+-export([update/2]).
 
 -include("couch_db.hrl").
 
-update(#group{db=#db{name=DbName}=Db,name=GroupName,current_seq=Seq,purge_seq=PurgeSeq}=Group)
->
+update(Owner, Group) ->
+    #group{
+        db = #db{name=DbName} = Db,
+        name = GroupName,
+        current_seq = Seq,
+        purge_seq = PurgeSeq
+    } = Group,
     couch_task_status:add_task(<<"View Group Indexer">>, <<DbName/binary,"
",GroupName/binary>>, <<"Starting index update">>),
 
     DbPurgeSeq = couch_db:get_purge_seq(Db),
@@ -43,7 +49,7 @@
             fun(DocInfo, _, {ChangesProcessed, Acc}) ->
                 couch_task_status:update("Processed ~p of ~p changes (~p%)",
                         [ChangesProcessed, TotalChanges, (ChangesProcessed*100) div TotalChanges]),
-                {ok, {ChangesProcessed+1, process_doc(Db, DocInfo, Acc)}}
+                {ok, {ChangesProcessed+1, process_doc(Db, Owner, DocInfo, Acc)}}
             end,
             {0, {[], Group2, ViewEmptyKVs, []}}
             ),
@@ -90,9 +96,9 @@
             views=Views2,
             purge_seq=couch_db:get_purge_seq(Db)}.
 
-process_doc(Db, DocInfo, {Docs, #group{sig=Sig,name=GroupId,design_options=DesignOptions}=Group,
ViewKVs,
-        DocIdViewIdKeys}) ->
-    % This fun computes once for each document
+% This fun computes once for each document
+process_doc(Db, Owner, DocInfo, {Docs, Group, ViewKVs, DocIdViewIdKeys}) ->
+    #group{ design_options = DesignOptions } = Group,
 
     #doc_info{id=DocId, revs=[#rev_info{deleted=Deleted}|_]} = DocInfo,
     IncludeDesign = proplists:get_value(<<"include_design">>,
@@ -125,6 +131,9 @@
                 Results, ViewKVs, DocIdViewIdKeys2),
             {ok, Group2} = write_changes(Group1, ViewKVs3, DocIdViewIdKeys3,
                 DocInfo#doc_info.high_seq),
+            if is_pid(Owner) ->
+                ok = gen_server:cast(Owner, {partial_update, Group2});
+            true -> ok end,
             garbage_collect(),
             ViewEmptyKeyValues = [{View, []} || View <- Group2#group.views],
             {[], Group2, ViewEmptyKeyValues, []};



Mime
View raw message