storage_proxy: do not ignore connection errors

Do nothing about them for now. Read will eventually fail on timeout.
scylladb · Jul 22, 2015 · 150c28e · 150c28e
1 parent 98fae1a
commit 150c28e
Showing 1 changed file with 21 additions and 6 deletions.
diff --git a/service/storage_proxy.cc b/service/storage_proxy.cc
@@ -1388,6 +1388,9 @@ class abstract_read_resolver {
     future<> done() {
         return _done_promise.get_future();
     }
+    virtual void error(gms::inet_address ep) {
+        // do nothing for now, request will timeout eventually
+    }
 };
 
 class digest_read_resolver : public abstract_read_resolver {
@@ -1596,22 +1599,34 @@ class abstract_read_executor : public enable_shared_from_this<abstract_read_exec
     }
     future<> make_mutation_data_requests(data_resolver_ptr resolver, targets_iterator begin, targets_iterator end) {
         return parallel_for_each(begin, end, [this, resolver = std::move(resolver)] (gms::inet_address ep) {
-            return make_mutation_data_request(ep).then([resolver, ep] (foreign_ptr<lw_shared_ptr<reconcilable_result>> result) {
-                resolver->add_mutate_data(ep, std::move(result));
+            return make_mutation_data_request(ep).then_wrapped([resolver, ep] (future<foreign_ptr<lw_shared_ptr<reconcilable_result>>> f) {
+                try {
+                    resolver->add_mutate_data(ep, f.get0());
+                } catch(...) {
+                    resolver->error(ep);
+                }
             });
         });
     }
     future<> make_data_requests(digest_resolver_ptr resolver, targets_iterator begin, targets_iterator end) {
         return parallel_for_each(begin, end, [this, resolver = std::move(resolver)] (gms::inet_address ep) {
-            return make_data_request(ep).then([resolver, ep] (foreign_ptr<lw_shared_ptr<query::result>> result) {
-                resolver->add_data(ep, std::move(result));
+            return make_data_request(ep).then_wrapped([resolver, ep] (future<foreign_ptr<lw_shared_ptr<query::result>>> f) {
+                try {
+                    resolver->add_data(ep, f.get0());
+                } catch(...) {
+                    resolver->error(ep);
+                }
             });
         });
     }
     future<> make_digest_requests(digest_resolver_ptr resolver, targets_iterator begin, targets_iterator end) {
         return parallel_for_each(begin, end, [this, resolver = std::move(resolver)] (gms::inet_address ep) {
-            return make_digest_request(ep).then([resolver, ep] (query::result_digest&& digest) {
-                resolver->add_digest(ep, std::move(digest));
+            return make_digest_request(ep).then_wrapped([resolver, ep] (future<query::result_digest> f) {
+                try {
+                    resolver->add_digest(ep, f.get0());
+                } catch(...) {
+                    resolver->error(ep);
+                }
             });
         });
     }