CPP-811 Requests won't complete if they exceed the number of streams (#293)

mpenick · web-flow · commit 36d054e40311 · 2019-10-10T10:50:58.000-04:00
* Fix null pointer dereferences in `RequestHandler`
* Fix connection write error handling for user requests
diff --git a/cpp-driver/gtests/src/unit/tests/test_pool.cpp b/cpp-driver/gtests/src/unit/tests/test_pool.cpp
@@ -324,7 +324,7 @@ class PoolUnitTest : public LoopTest {
     if (connection) {
       RequestStatus status(manager->loop(), 1);
       RequestCallback::Ptr callback(new RequestCallback(&status));
-      EXPECT_TRUE(connection->write(callback.get()))
+      EXPECT_TRUE(connection->write(callback.get()) > 0)
           << "Unable to write request to connection " << address.to_string();
       connection->flush(); // Flush requests to avoid unnecessary timeouts
       uv_run(loop(), UV_RUN_DEFAULT);
@@ -344,7 +344,7 @@ class PoolUnitTest : public LoopTest {
       PooledConnection::Ptr connection = manager->find_least_busy(generator.next());
       if (connection) {
         RequestCallback::Ptr callback(new RequestCallback(status));
-        if (!connection->write(callback.get())) {
+        if (connection->write(callback.get()) < 0) {
           status->error_failed_write();
         }
       } else {
diff --git a/cpp-driver/src/connection.cpp b/cpp-driver/src/connection.cpp
@@ -130,34 +130,12 @@ int32_t Connection::write(const RequestCallback::Ptr& callback) {
 
   int32_t request_size = socket_->write(callback.get());
 
-  if (request_size < 0) {
+  if (request_size <= 0) {
     stream_manager_.release(stream);
-
-    switch (request_size) {
-      case SocketRequest::SOCKET_REQUEST_ERROR_CLOSED:
-        callback->on_error(CASS_ERROR_LIB_WRITE_ERROR, "Unable to write to close socket");
-        break;
-
-      case SocketRequest::SOCKET_REQUEST_ERROR_NO_HANDLER:
-        callback->on_error(CASS_ERROR_LIB_WRITE_ERROR,
-                           "Socket is not properly configured with a handler");
-        break;
-
-      case Request::REQUEST_ERROR_BATCH_WITH_NAMED_VALUES:
-      case Request::REQUEST_ERROR_PARAMETER_UNSET:
-        // Already handled with a specific error.
-        break;
-
-      case Request::REQUEST_ERROR_UNSUPPORTED_PROTOCOL:
-        callback->on_error(CASS_ERROR_LIB_MESSAGE_ENCODE,
-                           "Operation unsupported by this protocol version");
-        break;
-
-      default:
-        callback->on_error(CASS_ERROR_LIB_WRITE_ERROR, "Unspecified write error occurred");
-        break;
+    if (request_size == 0) {
+      callback->on_error(CASS_ERROR_LIB_MESSAGE_ENCODE, "The encoded request had no data to write");
+      return Request::REQUEST_ERROR_NO_DATA_WRITTEN;
     }
-
     return request_size;
   }
 
@@ -340,7 +318,8 @@ void Connection::restart_heartbeat_timer() {
 
 void Connection::on_heartbeat(Timer* timer) {
   if (!heartbeat_outstanding_ && !socket_->is_closing()) {
-    if (!write_and_flush(RequestCallback::Ptr(new HeartbeatCallback(this)))) {
+    RequestCallback::Ptr callback(new HeartbeatCallback(this));
+    if (write_and_flush(callback) < 0) {
       // Recycling only this connection with a timeout error. This is unlikely and
       // it means the connection ran out of stream IDs as a result of requests
       // that never returned and as a result timed out.
diff --git a/cpp-driver/src/connection_pool.cpp b/cpp-driver/src/connection_pool.cpp
@@ -27,6 +27,13 @@ using namespace datastax;
 using namespace datastax::internal::core;
 
 static inline bool least_busy_comp(const PooledConnection::Ptr& a, const PooledConnection::Ptr& b) {
+  // Don't consider closed connections to be the least busy.
+  if (a->is_closing()) { // "a" is closed so it can't be the least busy.
+    return false;
+  } else if (b->is_closing()) { // "a" is not close, but "b" is closed so "a" is less busy.
+    return true;
+  }
+  // Both "a" and "b" are not closed so compare their inflight request counts.
   return a->inflight_request_count() < b->inflight_request_count();
 }
 
@@ -89,10 +96,12 @@ ConnectionPool::ConnectionPool(const Connection::Vec& connections, ConnectionPoo
 }
 
 PooledConnection::Ptr ConnectionPool::find_least_busy() const {
-  if (connections_.empty()) {
+  PooledConnection::Vec::const_iterator it =
+      std::min_element(connections_.begin(), connections_.end(), least_busy_comp);
+  if (it == connections_.end() || (*it)->is_closing()) {
     return PooledConnection::Ptr();
   }
-  return *std::min_element(connections_.begin(), connections_.end(), least_busy_comp);
+  return *it;
 }
 
 bool ConnectionPool::has_connections() const { return !connections_.empty(); }
diff --git a/cpp-driver/src/connection_pool.hpp b/cpp-driver/src/connection_pool.hpp
@@ -131,7 +131,7 @@ class ConnectionPool : public RefCounted<ConnectionPool> {
 
   /**
    * Find the least busy connection for the pool. The least busy connection has
-   * the lowest number of outstanding requests.
+   * the lowest number of outstanding requests and is not closed.
    *
    * @return The least busy connection or null if no connection is available.
    */
diff --git a/cpp-driver/src/control_connection.cpp b/cpp-driver/src/control_connection.cpp
@@ -378,8 +378,8 @@ void ControlConnection::refresh_node(RefreshNodeType type, const Address& addres
 
   LOG_DEBUG("Refresh node: %s", query.c_str());
 
-  if (write_and_flush(RequestCallback::Ptr(
-          new RefreshNodeCallback(address, type, is_all_peers, query, this))) < 0) {
+  RequestCallback::Ptr callback(new RefreshNodeCallback(address, type, is_all_peers, query, this));
+  if (write_and_flush(callback) < 0) {
     LOG_ERROR("No more stream available while attempting to refresh node info");
     defunct();
   }
@@ -448,8 +448,9 @@ void ControlConnection::refresh_keyspace(const StringRef& keyspace_name) {
 
   LOG_DEBUG("Refreshing keyspace %s", query.c_str());
 
-  if (write_and_flush(RequestCallback::Ptr(
-          new RefreshKeyspaceCallback(keyspace_name.to_string(), query, this))) < 0) {
+  RequestCallback::Ptr callback(
+      new RefreshKeyspaceCallback(keyspace_name.to_string(), query, this));
+  if (write_and_flush(callback) < 0) {
     LOG_ERROR("No more stream available while attempting to refresh keyspace info");
     defunct();
   }
@@ -594,8 +595,9 @@ void ControlConnection::refresh_type(const StringRef& keyspace_name, const Strin
 
   LOG_DEBUG("Refreshing type %s", query.c_str());
 
-  if (!write_and_flush(RequestCallback::Ptr(new RefreshTypeCallback(
-          keyspace_name.to_string(), type_name.to_string(), query, this)))) {
+  RequestCallback::Ptr callback(
+      new RefreshTypeCallback(keyspace_name.to_string(), type_name.to_string(), query, this));
+  if (write_and_flush(callback) < 0) {
     LOG_ERROR("No more stream available while attempting to refresh type info");
     defunct();
   }
@@ -654,9 +656,10 @@ void ControlConnection::refresh_function(const StringRef& keyspace_name,
   request->set(1, CassString(function_name.data(), function_name.size()));
   request->set(2, signature.get());
 
-  if (!write_and_flush(RequestCallback::Ptr(
-          new RefreshFunctionCallback(keyspace_name.to_string(), function_name.to_string(),
-                                      to_strings(arg_types), is_aggregate, request, this)))) {
+  RequestCallback::Ptr callback(
+      new RefreshFunctionCallback(keyspace_name.to_string(), function_name.to_string(),
+                                  to_strings(arg_types), is_aggregate, request, this));
+  if (write_and_flush(callback) < 0) {
     LOG_ERROR("No more stream available while attempting to refresh function info");
     defunct();
   }
diff --git a/cpp-driver/src/pooled_connection.cpp b/cpp-driver/src/pooled_connection.cpp
@@ -88,7 +88,7 @@ void ChainedSetKeyspaceCallback::on_internal_timeout() { chained_callback_->on_r
 void ChainedSetKeyspaceCallback::on_result_response(ResponseMessage* response) {
   ResultResponse* result = static_cast<ResultResponse*>(response->response_body().get());
   if (result->kind() == CASS_RESULT_KIND_SET_KEYSPACE) {
-    if (!connection_->write_and_flush(chained_callback_)) {
+    if (connection_->write_and_flush(chained_callback_) < 0) {
       // Try on the same host but a different connection
       chained_callback_->on_retry_current_host();
     }
@@ -107,8 +107,8 @@ PooledConnection::PooledConnection(ConnectionPool* pool, const Connection::Ptr&
   connection_->set_listener(this);
 }
 
-bool PooledConnection::write(RequestCallback* callback) {
-  bool result = false;
+int32_t PooledConnection::write(RequestCallback* callback) {
+  int32_t result;
   const String& keyspace(pool_->keyspace());
   if (keyspace != connection_->keyspace()) {
     LOG_DEBUG("Setting keyspace %s on connection(%p) pool(%p)", keyspace.c_str(),
@@ -119,9 +119,10 @@ bool PooledConnection::write(RequestCallback* callback) {
     result = connection_->write(RequestCallback::Ptr(callback));
   }
 
-  if (result) {
+  if (result > 0) {
     pool_->requires_flush(this, ConnectionPool::Protected());
   }
+
   return result;
 }
 
@@ -142,6 +143,8 @@ int PooledConnection::inflight_request_count() const {
   return connection_->inflight_request_count();
 }
 
+bool PooledConnection::is_closing() const { return connection_->is_closing(); }
+
 void PooledConnection::on_read() {
   if (event_loop_) {
     event_loop_->maybe_start_io_time();
diff --git a/cpp-driver/src/pooled_connection.hpp b/cpp-driver/src/pooled_connection.hpp
@@ -51,10 +51,9 @@ class PooledConnection
    * the connection pool manager flushes the request.
    *
    * @param callback A request callback that handles the request.
-   * @return Returns true if the request was written, otherwise, an error
-   * occurred.
+   * @return The number of bytes written, or negative if an error occurred.
    */
-  bool write(RequestCallback* callback);
+  int32_t write(RequestCallback* callback);
 
   /**
    * Flush pending writes.
@@ -73,6 +72,13 @@ class PooledConnection
    */
   int inflight_request_count() const;
 
+  /**
+   * Determine if the connection is closing.
+   *
+   * @return Returns true if closing.
+   */
+  bool is_closing() const;
+
 public:
   const String& keyspace() const { return connection_->keyspace(); } // Test only
 
diff --git a/cpp-driver/src/prepare_host_handler.cpp b/cpp-driver/src/prepare_host_handler.cpp
@@ -111,8 +111,8 @@ void PrepareHostHandler::prepare_next() {
     // Set the keyspace in case per request keyspaces are supported
     prepare_request->set_keyspace(current_keyspace_);
 
-    if (!connection_->write(
-            PrepareCallback::Ptr(new PrepareCallback(prepare_request, Ptr(this))))) {
+    PrepareCallback::Ptr callback(new PrepareCallback(prepare_request, Ptr(this)));
+    if (connection_->write(callback) < 0) {
       LOG_WARN("Failed to write prepare request while preparing all queries on host %s",
                host_->address_string().c_str());
       close();
@@ -134,8 +134,8 @@ bool PrepareHostHandler::check_and_set_keyspace() {
   const String& keyspace((*current_entry_it_)->keyspace());
 
   if (keyspace != current_keyspace_) {
-    if (!connection_->write_and_flush(
-            PrepareCallback::Ptr(new SetKeyspaceCallback(keyspace, Ptr(this))))) {
+    PrepareCallback::Ptr callback(new SetKeyspaceCallback(keyspace, Ptr(this)));
+    if (connection_->write_and_flush(callback) < 0) {
       LOG_WARN("Failed to write \"USE\" keyspace request while preparing all queries on host %s",
                host_->address_string().c_str());
       close();
diff --git a/cpp-driver/src/request.hpp b/cpp-driver/src/request.hpp
@@ -81,7 +81,8 @@ class Request : public RefCounted<Request> {
     REQUEST_ERROR_UNSUPPORTED_PROTOCOL = SocketRequest::SOCKET_REQUEST_ERROR_LAST_ENTRY,
     REQUEST_ERROR_BATCH_WITH_NAMED_VALUES,
     REQUEST_ERROR_PARAMETER_UNSET,
-    REQUEST_ERROR_NO_AVAILABLE_STREAM_IDS
+    REQUEST_ERROR_NO_AVAILABLE_STREAM_IDS,
+    REQUEST_ERROR_NO_DATA_WRITTEN
   };
 
   Request(uint8_t opcode)
diff --git a/cpp-driver/src/request_callback.cpp b/cpp-driver/src/request_callback.cpp
@@ -58,6 +58,7 @@ bool RequestCallback::skip_metadata() const {
 int32_t RequestCallback::encode(BufferVec* bufs) {
   const ProtocolVersion version = protocol_version_;
   if (version < CASS_LOWEST_SUPPORTED_PROTOCOL_VERSION) {
+    on_error(CASS_ERROR_LIB_MESSAGE_ENCODE, "Operation unsupported by this protocol version");
     return Request::REQUEST_ERROR_UNSUPPORTED_PROTOCOL;
   }
 
@@ -245,8 +246,7 @@ ResultResponse::Ptr ChainedRequestCallback::result(const String& key) const {
 
 void ChainedRequestCallback::on_internal_write(Connection* connection) {
   if (chain_) {
-    int32_t result = connection->write_and_flush(chain_);
-    if (result == Request::REQUEST_ERROR_NO_AVAILABLE_STREAM_IDS) {
+    if (connection->write_and_flush(chain_) < 0) {
       on_error(CASS_ERROR_LIB_NO_STREAMS,
                "No streams available when attempting to write chained request");
     }
diff --git a/cpp-driver/src/request_handler.cpp b/cpp-driver/src/request_handler.cpp
@@ -306,18 +306,54 @@ void RequestHandler::internal_retry(RequestExecution* request_execution) {
     return;
   }
 
-  bool is_successful = false;
-  while (request_execution->current_host()) {
+  bool is_done = false;
+  while (!is_done && request_execution->current_host()) {
     PooledConnection::Ptr connection =
         manager_->find_least_busy(request_execution->current_host()->address());
-    if (connection && connection->write(request_execution)) {
-      is_successful = true;
-      break;
+    if (connection) {
+      int32_t result = connection->write(request_execution);
+
+      if (result > 0) {
+        is_done = true;
+      } else {
+        switch (result) {
+          case SocketRequest::SOCKET_REQUEST_ERROR_CLOSED:
+            // This should never happen, but retry with next host if it does.
+            request_execution->next_host();
+            break;
+
+          case SocketRequest::SOCKET_REQUEST_ERROR_NO_HANDLER:
+            set_error(CASS_ERROR_LIB_WRITE_ERROR,
+                      "Socket is not properly configured with a handler");
+            is_done = true;
+            break;
+
+          case Request::REQUEST_ERROR_NO_AVAILABLE_STREAM_IDS:
+            // Retry with next host
+            request_execution->next_host();
+            break;
+
+          case Request::REQUEST_ERROR_BATCH_WITH_NAMED_VALUES:
+          case Request::REQUEST_ERROR_PARAMETER_UNSET:
+          case Request::REQUEST_ERROR_UNSUPPORTED_PROTOCOL:
+          case Request::REQUEST_ERROR_NO_DATA_WRITTEN:
+            // Already handled with a specific error.
+            is_done = true;
+            break;
+
+          default:
+            set_error(CASS_ERROR_LIB_WRITE_ERROR, "Unspecified write error occurred");
+            is_done = true;
+            break;
+        }
+      }
+    } else {
+      // No connection available on the current host, move to the next host.
+      request_execution->next_host();
     }
-    request_execution->next_host();
   }
 
-  if (!is_successful) {
+  if (!request_execution->current_host()) {
     set_error(CASS_ERROR_LIB_NO_HOSTS_AVAILABLE, "All hosts in current policy attempted "
                                                  "and were either unavailable or failed");
   }
@@ -335,7 +371,7 @@ void RequestExecution::on_execute_next(Timer* timer) { request_handler_->execute
 void RequestExecution::on_retry_current_host() { retry_current_host(); }
 
 void RequestExecution::on_retry_next_host() {
-  current_host_->decrement_inflight_requests();
+  if (current_host_) current_host_->decrement_inflight_requests();
   retry_next_host();
 }
 
@@ -392,14 +428,8 @@ void RequestExecution::on_set(ResponseMessage* response) {
 }
 
 void RequestExecution::on_error(CassError code, const String& message) {
-  current_host_->decrement_inflight_requests();
-
-  // Handle recoverable errors by retrying with the next host
-  if (code == CASS_ERROR_LIB_WRITE_ERROR || code == CASS_ERROR_LIB_UNABLE_TO_SET_KEYSPACE) {
-    retry_next_host();
-  } else {
-    set_error(code, message);
-  }
+  if (current_host_) current_host_->decrement_inflight_requests();
+  set_error(code, message);
 }
 
 void RequestExecution::notify_result_metadata_changed(const Request* request,
@@ -591,7 +621,8 @@ void RequestExecution::on_error_unprepared(Connection* connection, ErrorResponse
     return;
   }
 
-  if (!connection->write_and_flush(RequestCallback::Ptr(new PrepareCallback(query, this)))) {
+  RequestCallback::Ptr callback(new PrepareCallback(query, this));
+  if (connection->write_and_flush(callback) < 0) {
     // Try to prepare on the same host but on a different connection
     retry_current_host();
   }
diff --git a/cpp-driver/src/request_processor.cpp b/cpp-driver/src/request_processor.cpp
@@ -583,7 +583,7 @@ bool RequestProcessor::write_wait_callback(const RequestHandler::Ptr& request_ha
                                            const RequestCallback::Ptr& callback) {
   PooledConnection::Ptr connection(
       connection_pool_manager_->find_least_busy(current_host->address()));
-  if (connection && connection->write(callback.get())) {
+  if (connection && connection->write(callback.get()) > 0) {
     // Stop the original request timer now that we have a response and
     // are waiting for the maximum wait time of the handler.
     request_handler->stop_timer();
diff --git a/cpp-driver/src/socket.cpp b/cpp-driver/src/socket.cpp
@@ -234,7 +234,7 @@ void SocketWriteBase::on_close() {
 int32_t SocketWriteBase::write(SocketRequest* request) {
   size_t last_buffer_size = buffers_.size();
   int32_t request_size = request->encode(&buffers_);
-  if (request_size < 0) {
+  if (request_size <= 0) {
     buffers_.resize(last_buffer_size); // Rollback
     return request_size;
   }
diff --git a/cpp-driver/src/wait_for_handler.cpp b/cpp-driver/src/wait_for_handler.cpp
@@ -127,7 +127,7 @@ void WaitForHandler::on_retry_timeout(Timer* timer) {
     finish();
   } else if (connection_->write_and_flush(callback(requests_)) ==
              core::Request::REQUEST_ERROR_NO_AVAILABLE_STREAM_IDS) {
-    on_error(WaitForHandler::WAIT_FOR_ERROR_NO_STREAMS, "Connection closed");
+    on_error(WaitForHandler::WAIT_FOR_ERROR_NO_STREAMS, "No streams available");
     finish();
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -27,6 +27,13 @@ using namespace datastax;`
`27`	`27`	`using namespace datastax::internal::core;`
`28`	`28`
`29`	`29`	`static inline bool least_busy_comp(const PooledConnection::Ptr& a, const PooledConnection::Ptr& b) {`
	`30`	`+ // Don't consider closed connections to be the least busy.`
	`31`	`+ if (a->is_closing()) { // "a" is closed so it can't be the least busy.`
	`32`	`+ return false;`
	`33`	`+ } else if (b->is_closing()) { // "a" is not close, but "b" is closed so "a" is less busy.`
	`34`	`+ return true;`
	`35`	`+ }`
	`36`	`+ // Both "a" and "b" are not closed so compare their inflight request counts.`
`30`	`37`	`return a->inflight_request_count() < b->inflight_request_count();`
`31`	`38`	`}`
`32`	`39`
`@@ -89,10 +96,12 @@ ConnectionPool::ConnectionPool(const Connection::Vec& connections, ConnectionPoo`
`89`	`96`	`}`
`90`	`97`
`91`	`98`	`PooledConnection::Ptr ConnectionPool::find_least_busy() const {`
`92`		`- if (connections_.empty()) {`
	`99`	`+ PooledConnection::Vec::const_iterator it =`
	`100`	`+ std::min_element(connections_.begin(), connections_.end(), least_busy_comp);`
	`101`	`+ if (it == connections_.end() \|\| (*it)->is_closing()) {`
`93`	`102`	`return PooledConnection::Ptr();`
`94`	`103`	`}`
`95`		`- return *std::min_element(connections_.begin(), connections_.end(), least_busy_comp);`
	`104`	`+ return *it;`
`96`	`105`	`}`
`97`	`106`
`98`	`107`	`bool ConnectionPool::has_connections() const { return !connections_.empty(); }`
Original file line number	Diff line number	Diff line change
`@@ -131,7 +131,7 @@ class ConnectionPool : public RefCounted<ConnectionPool> {`
`131`	`131`
`132`	`132`	`/**`
`133`	`133`	`* Find the least busy connection for the pool. The least busy connection has`
`134`		`- * the lowest number of outstanding requests.`
	`134`	`+ * the lowest number of outstanding requests and is not closed.`
`135`	`135`	`*`
`136`	`136`	`* @return The least busy connection or null if no connection is available.`
`137`	`137`	`*/`
Original file line number	Diff line number	Diff line change
`@@ -88,7 +88,7 @@ void ChainedSetKeyspaceCallback::on_internal_timeout() { chained_callback_->on_r`
`88`	`88`	`void ChainedSetKeyspaceCallback::on_result_response(ResponseMessage* response) {`
`89`	`89`	`ResultResponse* result = static_cast<ResultResponse*>(response->response_body().get());`
`90`	`90`	`if (result->kind() == CASS_RESULT_KIND_SET_KEYSPACE) {`
`91`		`- if (!connection_->write_and_flush(chained_callback_)) {`
	`91`	`+ if (connection_->write_and_flush(chained_callback_) < 0) {`
`92`	`92`	`// Try on the same host but a different connection`
`93`	`93`	`chained_callback_->on_retry_current_host();`
`94`	`94`	`}`
`@@ -107,8 +107,8 @@ PooledConnection::PooledConnection(ConnectionPool* pool, const Connection::Ptr&`
`107`	`107`	`connection_->set_listener(this);`
`108`	`108`	`}`
`109`	`109`
`110`		`-bool PooledConnection::write(RequestCallback* callback) {`
`111`		`- bool result = false;`
	`110`	`+int32_t PooledConnection::write(RequestCallback* callback) {`
	`111`	`+ int32_t result;`
`112`	`112`	`const String& keyspace(pool_->keyspace());`
`113`	`113`	`if (keyspace != connection_->keyspace()) {`
`114`	`114`	`LOG_DEBUG("Setting keyspace %s on connection(%p) pool(%p)", keyspace.c_str(),`
`@@ -119,9 +119,10 @@ bool PooledConnection::write(RequestCallback* callback) {`
`119`	`119`	`result = connection_->write(RequestCallback::Ptr(callback));`
`120`	`120`	`}`
`121`	`121`
`122`		`- if (result) {`
	`122`	`+ if (result > 0) {`
`123`	`123`	`pool_->requires_flush(this, ConnectionPool::Protected());`
`124`	`124`	`}`
	`125`	`+`
`125`	`126`	`return result;`
`126`	`127`	`}`
`127`	`128`
`@@ -142,6 +143,8 @@ int PooledConnection::inflight_request_count() const {`
`142`	`143`	`return connection_->inflight_request_count();`
`143`	`144`	`}`
`144`	`145`
	`146`	`+bool PooledConnection::is_closing() const { return connection_->is_closing(); }`
	`147`	`+`
`145`	`148`	`void PooledConnection::on_read() {`
`146`	`149`	`if (event_loop_) {`
`147`	`150`	`event_loop_->maybe_start_io_time();`