couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vatam...@apache.org
Subject [3/4] fabric commit: updated refs/heads/3039-docid-nulls-in-all-docs-post to aa17ddf
Date Fri, 24 Jun 2016 16:22:56 GMT
Support update_seq=true in reduce views

COUCHDB-2991


Project: http://git-wip-us.apache.org/repos/asf/couchdb-fabric/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-fabric/commit/da85db2d
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-fabric/tree/da85db2d
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-fabric/diff/da85db2d

Branch: refs/heads/3039-docid-nulls-in-all-docs-post
Commit: da85db2dabe371296c54fded76d1de61289f3319
Parents: 37bcda9
Author: Robert Newson <rnewson@apache.org>
Authored: Wed Jun 22 13:34:14 2016 +0100
Committer: Robert Newson <rnewson@apache.org>
Committed: Wed Jun 22 13:44:12 2016 +0100

----------------------------------------------------------------------
 src/fabric_view_reduce.erl | 57 +++++++++++++++++++++++++----------------
 1 file changed, 35 insertions(+), 22 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/couchdb-fabric/blob/da85db2d/src/fabric_view_reduce.erl
----------------------------------------------------------------------
diff --git a/src/fabric_view_reduce.erl b/src/fabric_view_reduce.erl
index 7b16a0d..a638ff7 100644
--- a/src/fabric_view_reduce.erl
+++ b/src/fabric_view_reduce.erl
@@ -58,7 +58,7 @@ go(DbName, DDoc, VName, Args, Callback, Acc, VInfo) ->
     end.
 
 go2(DbName, Workers, {red, {_, Lang, View}, _}=VInfo, Args, Callback, Acc0) ->
-    #mrargs{limit = Limit, skip = Skip, keys = Keys} = Args,
+    #mrargs{limit = Limit, skip = Skip, keys = Keys, update_seq = UpdateSeq} = Args,
     RedSrc = couch_mrview_util:extract_view_reduce(VInfo),
     OsProc = case os_proc_needed(RedSrc) of
         true -> couch_query_servers:get_os_process(Lang);
@@ -77,7 +77,8 @@ go2(DbName, Workers, {red, {_, Lang, View}, _}=VInfo, Args, Callback, Acc0)
->
         reducer = RedSrc,
         collation = couch_util:get_value(<<"collation">>, View#mrview.options),
         rows = dict:new(),
-        user_acc = Acc0
+        user_acc = Acc0,
+        update_seq = case UpdateSeq of true -> []; false -> nil end
     },
     try rexi_utils:recv(Workers, #shard.ref, fun handle_message/3,
         State, infinity, 1000 * 60 * 60) of
@@ -99,30 +100,42 @@ handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _, State) ->
 handle_message({rexi_EXIT, Reason}, Worker, State) ->
     fabric_view:handle_worker_exit(State, Worker, Reason);
 
-%% HACK: this just sends meta once. Instead we should move the counter logic
-%% from the #view_row handle_message below into this function and and pass the
-%% meta call through maybe_send_row. This will also be more efficient doing it
-%% here as it's one less worker round trip reply.
-%% Prior to switching to couch_mrview, the fabric_view_reduce implementation
-%% did not get a total_and_offset call, whereas now we do. We now use this
-%% message as a clean way to indicate to couch_mrview_http:view_cb that the
-%% reduce response is starting.
-handle_message({meta, Meta}, {_Worker, From}, State) ->
-    rexi:stream_ack(From),
-
+handle_message({meta, Meta0}, {Worker, From}, State) ->
+    Seq = couch_util:get_value(update_seq, Meta0, 0),
     #collector{
         callback = Callback,
-        user_acc = AccIn
+        counters = Counters0,
+        user_acc = AccIn,
+        update_seq = UpdateSeq0
     } = State,
-
-    {Go, Acc} = case get(meta_sent) of
-        undefined ->
-            put(meta_sent, true),
-            Callback({meta, Meta}, AccIn);
-        _ ->
-            {ok, AccIn}
+    % Assert that we don't have other messages from this
+    % worker when the total_and_offset message arrives.
+    0 = fabric_dict:lookup_element(Worker, Counters0),
+    rexi:stream_ack(From),
+    Counters1 = fabric_dict:update_counter(Worker, 1, Counters0),
+    UpdateSeq = case UpdateSeq0 of
+        nil -> nil;
+        _   -> [{Worker, Seq} | UpdateSeq0]
     end,
-    {Go, State#collector{user_acc = Acc}};
+    case fabric_dict:any(0, Counters1) of
+    true ->
+        {ok, State#collector{
+            counters = Counters1,
+            update_seq = UpdateSeq
+        }};
+    false ->
+        Meta = case UpdateSeq of
+            nil ->
+                [];
+            _ ->
+                [{update_seq, fabric_view_changes:pack_seqs(UpdateSeq)}]
+        end,
+        {Go, Acc} = Callback({meta, Meta}, AccIn),
+        {Go, State#collector{
+            counters = fabric_dict:decrement_all(Counters1),
+            user_acc = Acc
+        }}
+    end;
 
 handle_message(#view_row{key=Key} = Row, {Worker, From}, State) ->
     #collector{counters = Counters0, rows = Rows0} = State,


Mime
View raw message