The ProtoBuffs client is using {auto_reconnect,true}. Retry each op i…

…f we get {error,disconnected}
SyncFree · Dec 10, 2013 · c8ef577 · c8ef577
1 parent b1141e3
commit c8ef577
Showing 1 changed file with 30 additions and 0 deletions.
diff --git a/src/basho_bench_driver_riakc_pb.erl b/src/basho_bench_driver_riakc_pb.erl
@@ -151,6 +151,8 @@ run(get, KeyGen, _ValueGen, State) ->
             {ok, State};
         {error, notfound} ->
             {ok, State};
+        {error, disconnected} ->
+            run(get, KeyGen, _ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -162,6 +164,8 @@ run(get_existing, KeyGen, _ValueGen, State) ->
             {ok, State};
         {error, notfound} ->
             {error, {not_found, Key}, State};
+        {error, disconnected} ->
+            run(get_existing, KeyGen, _ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -172,6 +176,8 @@ run(put, KeyGen, ValueGen, State) ->
                                                      {dw, State#state.dw}], State#state.timeout_write) of
         ok ->
             {ok, State};
+        {error, disconnected} ->
+            run(put, KeyGen, ValueGen, State);  % suboptimal, but ...
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -185,6 +191,8 @@ run(update, KeyGen, ValueGen, State) ->
                                                               {dw, State#state.dw}], State#state.timeout_write) of
                 ok ->
                     {ok, State};
+                {error, disconnected} ->
+                    run(update, KeyGen, ValueGen, State);  % suboptimal, but ...
                 {error, Reason} ->
                     {error, Reason, State}
             end;
@@ -195,9 +203,13 @@ run(update, KeyGen, ValueGen, State) ->
                                                              {dw, State#state.dw}], State#state.timeout_write) of
                 ok ->
                     {ok, State};
+                {error, disconnected} ->
+                    run(update, KeyGen, ValueGen, State);  % suboptimal, but ...
                 {error, Reason} ->
                     {error, Reason, State}
             end;
+        {error, disconnected} ->
+            run(update, KeyGen, ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -211,11 +223,15 @@ run(update_existing, KeyGen, ValueGen, State) ->
                                                               {dw, State#state.dw}], State#state.timeout_write) of
                 ok ->
                     {ok, State};
+                {error, disconnected} ->
+                    run(update_existing, KeyGen, ValueGen, State);  % suboptimal, but ...
                 {error, Reason} ->
                     {error, Reason, State}
             end;
         {error, notfound} ->
             {error, {not_found, Key}, State};
+        {error, disconnected} ->
+            run(update_existing, KeyGen, ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -227,6 +243,8 @@ run(delete, KeyGen, _ValueGen, State) ->
             {ok, State};
         {error, notfound} ->
             {ok, State};
+        {error, disconnected} ->
+            run(delete, KeyGen, _ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -235,6 +253,8 @@ run(listkeys, _KeyGen, _ValueGen, State) ->
     case riakc_pb_socket:list_keys(State#state.pid, State#state.bucket, State#state.timeout_listkeys) of
         {ok, _Keys} ->
             {ok, State};
+        {error, disconnected} ->
+            run(listkeys, _KeyGen, _ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -246,6 +266,8 @@ run(search, _KeyGen, _ValueGen, #state{search_queries=SearchQs}=State) ->
     case riakc_pb_socket:search(NewState#state.pid, Index, Query, Options, NewState#state.timeout_read) of
           {ok, _Results} ->
               {ok, NewState};
+          {error, disconnected} ->
+              run(search, _KeyGen, _ValueGen, State);
           {error, Reason} ->
               {error, Reason, NewState}
     end;
@@ -263,6 +285,8 @@ run(search_interval, _KeyGen, _ValueGen, #state{search_queries=SearchQs, start_t
     case riakc_pb_socket:search(NewState#state.pid, Index, Query, Options, NewState#state.timeout_read) of
           {ok, _Results} ->
               {ok, NewState};
+          {error, disconnected} ->
+              run(search_interval, _KeyGen, _ValueGen, State);
           {error, Reason} ->
               {error, Reason, NewState}
     end;
@@ -287,6 +311,8 @@ run(counter_incr, KeyGen, ValueGen, State) ->
                                        {pw, State#state.pw}]) of
         ok ->
             {ok, State};
+        {error, disconnected} ->
+            run(counter_incr, KeyGen, ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end;
@@ -298,6 +324,8 @@ run(counter_val, KeyGen, _ValueGen, State) ->
             {ok, State};
         {error, notfound} ->
             {ok, State};
+        {error, disconnected} ->
+            run(counter_val, KeyGen, _ValueGen, State);
         {error, Reason} ->
             {error, Reason, State}
     end.
@@ -316,6 +344,8 @@ mapred(State, Input, Query) ->
                 {error, Reason} ->
                     {error, Reason, State}
             end;
+        {error, disconnected} ->
+            mapred(State, Input, Query);
         {error, Reason} ->
             {error, Reason, State}
     end.