couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rnew...@apache.org
Subject [25/50] fabric commit: updated refs/heads/master to a71701c
Date Thu, 28 Aug 2014 12:21:00 GMT
Collect pending counts and report sum at end

BugzID: 24236


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

Branch: refs/heads/master
Commit: 7826ce5fa9058bbc1c85ac49be6c6fbb369f5606
Parents: 9de1096
Author: Adam Kocoloski <adam@cloudant.com>
Authored: Thu Oct 17 16:33:14 2013 -0400
Committer: Robert Newson <rnewson@apache.org>
Committed: Fri Aug 1 15:33:42 2014 +0100

----------------------------------------------------------------------
 src/fabric_view_changes.erl | 65 +++++++++++++++++++++++++++++++++-------
 1 file changed, 54 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/couchdb-fabric/blob/7826ce5f/src/fabric_view_changes.erl
----------------------------------------------------------------------
diff --git a/src/fabric_view_changes.erl b/src/fabric_view_changes.erl
index 36cb945..60dc54c 100644
--- a/src/fabric_view_changes.erl
+++ b/src/fabric_view_changes.erl
@@ -61,7 +61,7 @@ go(DbName, "normal", Options, Callback, Acc0) ->
     case validate_start_seq(DbName, Since) of
     ok ->
         {ok, Acc} = Callback(start, Acc0),
-        {ok, #collector{counters=Seqs, user_acc=AccOut}} = send_changes(
+        {ok, Collector} = send_changes(
             DbName,
             Args,
             Callback,
@@ -69,7 +69,8 @@ go(DbName, "normal", Options, Callback, Acc0) ->
             Acc,
             5000
         ),
-        Callback({stop, pack_seqs(Seqs)}, AccOut);
+        #collector{counters=Seqs, user_acc=AccOut, offset=Offset} = Collector,
+        Callback({stop, pack_seqs(Seqs), pending_count(Offset)}, AccOut);
     Error ->
         Callback(Error, Acc0)
     end.
@@ -77,12 +78,17 @@ go(DbName, "normal", Options, Callback, Acc0) ->
 keep_sending_changes(DbName, Args, Callback, Seqs, AccIn, Timeout, UpListen, T0) ->
     #changes_args{limit=Limit, feed=Feed, heartbeat=Heartbeat} = Args,
     {ok, Collector} = send_changes(DbName, Args, Callback, Seqs, AccIn, Timeout),
-    #collector{limit=Limit2, counters=NewSeqs, user_acc=AccOut} = Collector,
+    #collector{
+        limit = Limit2,
+        counters = NewSeqs,
+        offset = Offset,
+        user_acc = AccOut
+    } = Collector,
     LastSeq = pack_seqs(NewSeqs),
     MaintenanceMode = config:get("cloudant", "maintenance_mode"),
     if Limit > Limit2, Feed == "longpoll";
       MaintenanceMode == "true"; MaintenanceMode == "nolb" ->
-        Callback({stop, LastSeq}, AccOut);
+        Callback({stop, LastSeq, pending_count(Offset)}, AccOut);
     true ->
         WaitForUpdate = wait_db_updated(UpListen),
         AccumulatedTime = timer:now_diff(os:timestamp(), T0) div 1000,
@@ -94,9 +100,9 @@ keep_sending_changes(DbName, Args, Callback, Seqs, AccIn, Timeout, UpListen,
T0)
         end,
         case {Heartbeat, AccumulatedTime > Max, WaitForUpdate} of
         {undefined, _, timeout} ->
-            Callback({stop, LastSeq}, AccOut);
+            Callback({stop, LastSeq, pending_count(Offset)}, AccOut);
         {_, true, timeout} ->
-            Callback({stop, LastSeq}, AccOut);
+            Callback({stop, LastSeq, pending_count(Offset)}, AccOut);
         _ ->
             {ok, AccTimeout} = Callback(timeout, AccOut),
             ?MODULE:keep_sending_changes(
@@ -183,6 +189,7 @@ send_changes(DbName, Workers, Seqs, ChangesArgs, Callback, AccIn, Timeout)
->
         counters = orddict:from_list(Seqs),
         user_acc = AccIn,
         limit = ChangesArgs#changes_args.limit,
+        offset = fabric_dict:init(Workers, null),
         rows = Seqs % store sequence positions instead
     },
     %% TODO: errors need to be handled here
@@ -207,10 +214,29 @@ handle_message({rexi_EXIT, Reason}, Worker, State) ->
 
 % Temporary upgrade clause - Case 24236
 handle_message({complete, Key}, Worker, State) when is_tuple(Key) ->
-    handle_message({complete, [{seq, Key}]}, Worker, State);
+    handle_message({complete, [{seq, Key}, {pending, 0}]}, Worker, State);
+
+handle_message({change, Props}, {Worker, _}, #collector{limit=0} = State) ->
+    O0 = State#collector.offset,
+    O1 = case fabric_dict:lookup_element(Worker, O0) of
+        null ->
+            % Use Pending+1 because we're ignoring this row in the response
+            Pending = couch_util:get_value(pending, Props),
+            fabric_dict:store(Worker, Pending+1, O0);
+        _ ->
+            O0
+    end,
+    maybe_stop(State#collector{offset = O1});
 
-handle_message(_, _, #collector{limit=0} = State) ->
-    {stop, State};
+handle_message({complete, Props}, Worker, #collector{limit=0} = State) ->
+    O0 = State#collector.offset,
+    O1 = case fabric_dict:lookup_element(Worker, O0) of
+        null ->
+            fabric_dict:store(Worker, couch_util:get_value(pending,Props), O0);
+        _ ->
+            O0
+    end,
+    maybe_stop(State#collector{offset = O1});
 
 handle_message(#change{} = Row, {Worker, From}, St) ->
     Change = {change, [
@@ -226,11 +252,13 @@ handle_message({change, Props}, {Worker, From}, St) ->
     #collector{
         callback = Callback,
         counters = S0,
+        offset = O0,
         limit = Limit,
         user_acc = AccIn
     } = St,
     true = fabric_dict:is_key(Worker, S0),
     S1 = fabric_dict:store(Worker, couch_util:get_value(seq, Props), S0),
+    O1 = fabric_dict:store(Worker, couch_util:get_value(pending, Props), O0),
     % Temporary hack for FB 23637
     Interval = erlang:get(changes_seq_interval),
     if (Interval == undefined) orelse (Limit rem Interval == 0) ->
@@ -240,7 +268,7 @@ handle_message({change, Props}, {Worker, From}, St) ->
     end,
     {Go, Acc} = Callback(changes_row(Props2), AccIn),
     rexi:stream_ack(From),
-    {Go, St#collector{counters=S1, limit=Limit-1, user_acc=Acc}};
+    {Go, St#collector{counters=S1, offset=O1, limit=Limit-1, user_acc=Acc}};
 
 handle_message({no_pass, Seq}, {Worker, From}, St) ->
     #collector{counters = S0} = St,
@@ -253,11 +281,13 @@ handle_message({complete, Props}, Worker, State) ->
     Key = couch_util:get_value(seq, Props),
     #collector{
         counters = S0,
+        offset = O0,
         total_rows = Completed % override
     } = State,
     true = fabric_dict:is_key(Worker, S0),
     S1 = fabric_dict:store(Worker, Key, S0),
-    NewState = State#collector{counters=S1, total_rows=Completed+1},
+    O1 = fabric_dict:store(Worker, couch_util:get_value(pending, Props), O0),
+    NewState = State#collector{counters=S1, offset=O1, total_rows=Completed+1},
     % We're relying on S1 having exactly the numnber of workers that
     % are participtaing in this response. With the new stream_start
     % that's a bit more obvious but historically it wasn't quite
@@ -270,6 +300,7 @@ handle_message({complete, Props}, Worker, State) ->
     end,
     {Go, NewState}.
 
+
 make_replacement_arg(Node, {Seq, Uuid}) ->
     {replace, Node, Uuid, Seq};
 make_replacement_arg(Node, {Seq, Uuid, _}) ->
@@ -279,6 +310,15 @@ make_replacement_arg(Node, {Seq, Uuid, _}) ->
 make_replacement_arg(_, _) ->
     0.
 
+maybe_stop(#collector{offset = Offset} = State) ->
+    case fabric_dict:any(null, Offset) of
+        false ->
+            {stop, State};
+        true ->
+            % Wait till we've heard from everyone to compute pending count
+            {ok, State}
+    end.
+
 make_changes_args(#changes_args{style=Style, filter=undefined}=Args) ->
     Args#changes_args{filter = Style};
 make_changes_args(Args) ->
@@ -310,6 +350,9 @@ collect_update_seqs(Seq, Shard, Counters) when is_integer(Seq) ->
         end
     end.
 
+pending_count(Dict) ->
+    fabric_dict:fold(fun(_Worker, C, Acc) -> C+Acc end, 0, Dict).
+
 pack_seqs(Workers) ->
     SeqList = [{N,R,S} || {#shard{node=N, range=R}, S} <- Workers],
     SeqSum = lists:sum([seq(S) || {_,_,S} <- SeqList]),


Mime
View raw message