WIP Speculative Execution #2

Author: mpenick

src/io_worker.cpp

@@ -147,6 +147,8 @@ void IOWorker::retry(const SpeculativeExecution::Ptr& speculative_execution) {
     return;
   }
 
+  printf("Executing %s...\n", speculative_execution->current_host()->address_string().c_str());
+
   PoolMap::const_iterator it = pools_.find(speculative_execution->current_host()->address());
   if (it != pools_.end() && it->second->is_ready()) {
     const SharedRefPtr<Pool>& pool = it->second;

src/request_handler.cpp

@@ -111,6 +111,7 @@ RequestHandler::~RequestHandler() {
 }
 
 void RequestHandler::add_execution(SpeculativeExecution* speculative_execution) {
+  running_executions_++;
   speculative_execution->inc_ref();
   speculative_executions_.push_back(speculative_execution);
 }
@@ -127,26 +128,27 @@ void RequestHandler::schedule_next_execution(const Host::Ptr& current_host) {
 void RequestHandler::set_response(const Host::Ptr& host,
                                   const SharedRefPtr<Response>& response) {
   if (future_->set_response(host->address(), response)) {
+    printf("Response from %s\n", host->address_string().c_str());
     finish();
   }
 }
 
 void RequestHandler::set_error(CassError code,
                                const std::string& message) {
-  if (future_->set_error(code, message)) {
-    finish();
-  }
 }
 
 void RequestHandler::set_error(const Host::Ptr& host,
                                CassError code, const std::string& message) {
-  if (host) {
-    if (future_->set_error_with_address(host->address(), code, message)) {
-      finish();
-    }
-  } else {
-    if (future_->set_error(code, message)) {
-      finish();
+  bool skip = (code == CASS_ERROR_LIB_NO_HOSTS_AVAILABLE && --running_executions_ > 0);
+  if (!skip) {
+    if (host) {
+      if (future_->set_error_with_address(host->address(), code, message)) {
+        finish();
+      }
+    } else {
+      if (future_->set_error(code, message)) {
+        finish();
+      }
     }
   }
 }
@@ -236,6 +238,7 @@ void SpeculativeExecution::execute(int64_t timeout) {
   if (timeout > 0) {
     timer_.start(request_handler_->io_worker()->loop(), timeout, this, on_execute);
   } else {
+    // TODO: Need to move to next node when executing immediately (but not the first time :()
     if (request()->is_idempotent()) {
       request_handler_->schedule_next_execution(current_host_);
     }

src/request_handler.hpp

@@ -114,7 +114,8 @@ class RequestHandler : public RefCounted<RequestHandler> {
       , timestamp_(request->timestamp())
       , future_(future)
       , retry_policy_(retry_policy)
-      , io_worker_(NULL) { }
+      , io_worker_(NULL)
+      , running_executions_(0) { }
 
   ~RequestHandler();
 
@@ -170,6 +171,7 @@ class RequestHandler : public RefCounted<RequestHandler> {
   ScopedPtr<SpeculativeExecutionPlan> execution_plan_;
   Host::Ptr first_host_;
   IOWorker* io_worker_;
+  int running_executions_;
   SpeculativeExecutionVec speculative_executions_;
   Request::EncodingCache encoding_cache_;
 };

src/speculative_execution.hpp

@@ -66,10 +66,10 @@ class NoSpeculativeExecutionPolicy : public SpeculativeExecutionPolicy {
 class TestSpeculativeExecutionPlan : public SpeculativeExecutionPlan {
 public:
   TestSpeculativeExecutionPlan()
-    : count_(3) { }
+    : count_(2) { }
 
   virtual int64_t next_execution(const Host::Ptr& current_host) {
-    return --count_ > 0 ? 100 : -1;
+    return --count_ >= 0 ? 1 : -1;
   }
 
 private: