WIP

Author: Doug Rohrer

intercepts/riak_core_vnode_master_intercepts.erl

@@ -8,19 +8,6 @@
     forwardable :: boolean(),
     opts = [] :: list()}).
 
--record(fitting,
-{
-    pid :: pid(),
-    ref :: reference(),
-    chashfun,
-    nval
-}).
-
--record(cmd_enqueue, {fitting :: #fitting{},
-    input :: term(),
-    timeout,
-    usedpreflist}).
-
 -define(M, riak_core_vnode_master_orig).
 
 
@@ -47,31 +34,25 @@ stop_vnode_after_bloom_fold_request_succeeds(IndexNode, Req, Sender, VMaster) ->
         false -> ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster)
     end.
 
-stop_pipe_vnode_after_request_sent(IndexNode, Req, Sender, VMaster) ->
-    case Req of
-        #cmd_enqueue{} = _Req ->
-            %% ?I_INFO("Intercepting riak_core_vnode_master:command_returning_vnode"),
-            random:seed(os:timestamp()),
-            case random:uniform(20) of
-                5 ->
-                    %% Simulate what happens when a VNode completes handoff between command_returning_vnode
-                    %% and the fold attempting to start - other attempts to intercept and slow
-                    %% certain parts of Riak to invoke the particular race condition were unsuccessful
-                    ?I_INFO("Replaced VNode with spawned function in command_returning_vnode"),
-                    Runner = self(),
-                    VNodePid = spawn(fun() ->
-                                        Runner ! go,
-                                        exit(normal)
-                                     end),
-                    receive
-                        go -> ok
-                    end,
-                    %% Still need to send the work
-                    ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster),
-                    {ok, VNodePid};
-                _ ->
-                    ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster)
-            end;
+return_dead_process_pid_from_get_vnode_pid(Node, Index, VNodeMod = riak_pipe_vnode) ->
+    %% ?I_INFO("Intercepting riak_core_vnode_master:get_vnode_pid"),
+    random:seed(os:timestamp()),
+    case random:uniform(100) of
+        7 ->
+            %% Simulate what happens when a VNode completes handoff between get_vnode_pid
+            %% and the fold attempting to start - other attempts to intercept and slow
+            %% certain parts of Riak to invoke the particular race condition were unsuccessful
+            ?I_INFO("Replaced VNode with spawned function in get_vnode_pid"),
+            VNodePid = spawn(fun() ->
+                                ok
+                             end),
+            MonRef = erlang:monitor(VNodePid),
+            receive
+                {'DOWN', MonRef, process, VNodePid, _Reason} -> ok
+            end,
+            {ok, VNodePid};
         _ ->
-            ?M:command_return_vnode_orig(IndexNode, Req, Sender, VMaster)
-    end.
+            ?M:get_vnode_pid_orig(Node, Index, VNodeMod)
+    end;
+return_dead_process_pid_from_get_vnode_pid(Node, Index, VNodeMod) ->
+    ?M:get_vnode_pid_orig(Node, Index, VNodeMod).

tests/pipe_verify_handoff_blocking.erl

@@ -110,6 +110,7 @@ confirm() ->
 
     ok = rt:wait_until_transfers_complete(Nodes),
 
+    lager:info("Check input count"),
     FillerInputCount = stop_fillers(Fillers),
 
     %% if we make it this far, then no filler ever saw the vnode_down
@@ -118,7 +119,7 @@ confirm() ->
     %% with `{error,[{vnode_down,noproc}]}}` errors, as the `noproc` case
     %% should be handled similarly to the `normal` exit case in
     %% `riak_pipe_vnode:queue_work_wait`
-
+    lager:info("Check pipe status"),
     _Status2 = pipe_status(Primary, Pipe),
 
     lager:info("Send eoi and collect results"),
@@ -138,8 +139,8 @@ confirm() ->
 set_up_vnode_crashing_intercept(Primary) ->
     lager:info("Add intercept to kill vnode before calling the wait function"),
     rt_intercept:add(Primary, {riak_core_vnode_master,
-        [{{command_return_vnode, 4},
-            stop_pipe_vnode_after_request_sent}]}).
+        [{{get_vnode_pid, 3},
+            return_dead_process_pid_from_get_vnode_pid}]}).
 
 %% queue filling
 
@@ -177,8 +178,14 @@ queue_filler(Node, Pipe, Inputs, Count) ->
         {stop, Owner} -> Owner ! {done, Count}
     after 0 ->
             {{value, I}, Q} = queue:out(Inputs),
-            ok = rpc:call(Node, riak_pipe, queue_work, [Pipe, I]),
-            queue_filler(Node, Pipe, queue:in(I, Q), Count+1)
+            case rpc:call(Node, riak_pipe, queue_work, [Pipe, I], 40000) of
+                ok ->
+                    lager:info("Received response from queue_work"),
+                    queue_filler(Node, Pipe, queue:in(I, Q), Count+1);
+                _ ->
+                    lager:info("Timed out waiting for response from queue_work"),
+                    queue_filler(Node, Pipe, queue:in(I, Q), Count+1)
+            end
     end.
 
 %% @doc tell all fillers to stop and collect and sum their send counts