src/request_processor.cpp - cassandra-cpp-driver - Git at Google

 /*
   Licensed to the Apache Software Foundation (ASF) under one
   or more contributor license agreements.  See the NOTICE file
   distributed with this work for additional information
   regarding copyright ownership.  The ASF licenses this file
   to you under the Apache License, Version 2.0 (the
   "License"); you may not use this file except in compliance
   with the License.  You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
 */

 #include "request_processor.hpp"

 #include "connection_pool_manager_initializer.hpp"
 #include "prepare_all_handler.hpp"
 #include "request_processor.hpp"
 #include "session.hpp"
 #include "tracing_data_handler.hpp"
 #include "utils.hpp"

 using namespace datastax;
 using namespace datastax::internal;
 using namespace datastax::internal::core;

 namespace datastax { namespace internal { namespace core {

 class ProcessorRunClose : public Task {
 public:
   ProcessorRunClose(const RequestProcessor::Ptr& processor)
       : processor_(processor) {}

   virtual void run(EventLoop* event_loop) { processor_->internal_close(); }

 private:
   const RequestProcessor::Ptr processor_;
 };

 class ProcessorNotifyHostAdd : public Task {
 public:
   ProcessorNotifyHostAdd(const Host::Ptr host, const RequestProcessor::Ptr& request_processor)
       : request_processor_(request_processor)
       , host_(host) {}
   virtual void run(EventLoop* event_loop) { request_processor_->internal_host_add(host_); }

 private:
   const RequestProcessor::Ptr request_processor_;
   const Host::Ptr host_;
 };

 class ProcessorNotifyHostRemove : public Task {
 public:
   ProcessorNotifyHostRemove(const Host::Ptr host, const RequestProcessor::Ptr& request_processor)
       : request_processor_(request_processor)
       , host_(host) {}
   virtual void run(EventLoop* event_loop) { request_processor_->internal_host_remove(host_); }

 private:
   RequestProcessor::Ptr request_processor_;
   const Host::Ptr host_;
 };

 class ProcessorNotifyHostReady : public Task {
 public:
   ProcessorNotifyHostReady(const Host::Ptr& host, const RequestProcessor::Ptr& request_processor)
       : request_processor_(request_processor)
       , host_(host) {}
   virtual void run(EventLoop* event_loop) { request_processor_->internal_host_ready(host_); }

 private:
   const RequestProcessor::Ptr request_processor_;
   const Host::Ptr host_;
 };

 class ProcessorNotifyMaybeHostUp : public Task {
 public:
   ProcessorNotifyMaybeHostUp(const Address& address, const RequestProcessor::Ptr& request_processor)
       : request_processor_(request_processor)
       , address_(address) {}
   virtual void run(EventLoop* event_loop) { request_processor_->internal_host_maybe_up(address_); }

 private:
   const RequestProcessor::Ptr request_processor_;
   const Address address_;
 };

 class ProcessorNotifyTokenMapUpdate : public Task {
 public:
   ProcessorNotifyTokenMapUpdate(const TokenMap::Ptr& token_map,
                                 const RequestProcessor::Ptr& request_processor)
       : request_processor_(request_processor)
       , token_map_(token_map) {}

   virtual void run(EventLoop* event_loop) { request_processor_->token_map_ = token_map_; }

 private:
   const RequestProcessor::Ptr request_processor_;
   const TokenMap::Ptr token_map_;
 };

 class SetKeyspaceProcessor : public Task {
 public:
   SetKeyspaceProcessor(const ConnectionPoolManager::Ptr& manager, const String& keyspace,
                        const KeyspaceChangedHandler::Ptr& handler)
       : manager_(manager)
       , keyspace_(keyspace)
       , handler_(handler) {}

   virtual void run(EventLoop* event_loop) { manager_->set_keyspace(keyspace_); }

 private:
   ConnectionPoolManager::Ptr manager_;
   const String keyspace_;
   KeyspaceChangedHandler::Ptr handler_;
 };

 class NopRequestProcessorListener : public RequestProcessorListener {
 public:
   virtual void on_pool_up(const Address& address) {}
   virtual void on_pool_down(const Address& address) {}
   virtual void on_pool_critical_error(const Address& address, Connector::ConnectionError code,
                                       const String& message) {}
   virtual void on_keyspace_changed(const String& keyspace,
                                    const KeyspaceChangedHandler::Ptr& handler) {}
   virtual void on_prepared_metadata_changed(const String& id,
                                             const PreparedMetadata::Entry::Ptr& entry) {}
   virtual void on_close(RequestProcessor* processor) {}
 };

 }}} // namespace datastax::internal::core

 static NopRequestProcessorListener nop_request_processor_listener__;

 RequestProcessorSettings::RequestProcessorSettings()
     : max_schema_wait_time_ms(10000)
     , prepare_on_all_hosts(true)
     , timestamp_generator(new ServerSideTimestampGenerator())
     , default_profile(Config().default_profile())
     , request_queue_size(8192)
     , coalesce_delay_us(CASS_DEFAULT_COALESCE_DELAY)
     , new_request_ratio(CASS_DEFAULT_NEW_REQUEST_RATIO)
     , max_tracing_wait_time_ms(CASS_DEFAULT_MAX_TRACING_DATA_WAIT_TIME_MS)
     , retry_tracing_wait_time_ms(CASS_DEFAULT_RETRY_TRACING_DATA_WAIT_TIME_MS)
     , tracing_consistency(CASS_DEFAULT_TRACING_CONSISTENCY)
     , address_factory(new AddressFactory()) {
   profiles.set_empty_key("");
 }

 RequestProcessorSettings::RequestProcessorSettings(const Config& config)
     : connection_pool_settings(config)
     , max_schema_wait_time_ms(config.max_schema_wait_time_ms())
     , prepare_on_all_hosts(config.prepare_on_all_hosts())
     , timestamp_generator(config.timestamp_gen())
     , default_profile(config.default_profile())
     , profiles(config.profiles())
     , request_queue_size(config.queue_size_io())
     , coalesce_delay_us(config.coalesce_delay_us())
     , new_request_ratio(config.new_request_ratio())
     , max_tracing_wait_time_ms(config.max_tracing_wait_time_ms())
     , retry_tracing_wait_time_ms(config.retry_tracing_wait_time_ms())
     , tracing_consistency(config.tracing_consistency())
     , address_factory(create_address_factory_from_config(config)) {}

 RequestProcessor::RequestProcessor(RequestProcessorListener* listener, EventLoop* event_loop,
                                    const ConnectionPoolManager::Ptr& connection_pool_manager,
                                    const Host::Ptr& connected_host, const HostMap& hosts,
                                    const TokenMap::Ptr& token_map,
                                    const RequestProcessorSettings& settings, Random* random,
                                    const String& local_dc)
     : connection_pool_manager_(connection_pool_manager)
     , listener_(listener ? listener : &nop_request_processor_listener__)
     , event_loop_(event_loop)
     , settings_(settings)
     , default_profile_(settings.default_profile)
     , profiles_(settings.profiles)
     , request_count_(0)
     , request_queue_(new MPMCQueue<RequestHandler*>(settings.request_queue_size))
     , is_closing_(false)
     , is_processing_(false)
     , attempts_without_requests_(0)
     , io_time_during_coalesce_(0)
 #ifdef CASS_INTERNAL_DIAGNOSTICS
     , reads_during_coalesce_(0)
     , writes_during_coalesce_(0)
     , writes_per_("writes")
     , reads_per_("reads")
 #endif
 {
   inc_ref(); // For the connection pool manager
   connection_pool_manager_->set_listener(this);

   // Build/Assign the load balancing policies from the execution profiles
   default_profile_.build_load_balancing_policy();
   load_balancing_policies_.push_back(default_profile_.load_balancing_policy());
   for (ExecutionProfile::Map::iterator it = profiles_.begin(), end = profiles_.end(); it != end;
        ++it) {
     it->second.build_load_balancing_policy();
     const LoadBalancingPolicy::Ptr& load_balancing_policy = it->second.load_balancing_policy();
     if (load_balancing_policy) {
       LOG_TRACE("Built load balancing policy for '%s' execution profile", it->first.c_str());
       load_balancing_policies_.push_back(load_balancing_policy);
     } else {
       it->second.use_load_balancing_policy(default_profile_.load_balancing_policy());
     }
   }

   token_map_ = token_map;

   LoadBalancingPolicy::Vec policies = load_balancing_policies();
   for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end(); ++it) {
     // Initialize the load balancing policies
     (*it)->init(connected_host, hosts, random, local_dc);
     (*it)->register_handles(event_loop_->loop());
   }

   listener_->on_connect(this);
 }

 void RequestProcessor::close() { event_loop_->add(new ProcessorRunClose(Ptr(this))); }

 void RequestProcessor::set_listener(RequestProcessorListener* listener) {
   listener_ = listener ? listener : &nop_request_processor_listener__;
 }

 void RequestProcessor::set_keyspace(const String& keyspace,
                                     const KeyspaceChangedHandler::Ptr& handler) {
   // If running on the the current event loop then just set the keyspace,
   // otherwise we're on a different thread so we'll wait for the keyspace
   // to be set in a task on the event loop thread.
   if (event_loop_->is_running_on()) {
     connection_pool_manager_->set_keyspace(keyspace);
   } else {
     event_loop_->add(new SetKeyspaceProcessor(connection_pool_manager_, keyspace, handler));
   }
 }

 void RequestProcessor::notify_host_added(const Host::Ptr& host) {
   event_loop_->add(new ProcessorNotifyHostAdd(host, Ptr(this)));
 }

 void RequestProcessor::notify_host_removed(const Host::Ptr& host) {
   event_loop_->add(new ProcessorNotifyHostRemove(host, Ptr(this)));
 }

 void RequestProcessor::notify_host_ready(const Host::Ptr& host) {
   event_loop_->add(new ProcessorNotifyHostReady(host, Ptr(this)));
 }

 void RequestProcessor::notify_host_maybe_up(const Address& address) {
   event_loop_->add(new ProcessorNotifyMaybeHostUp(address, Ptr(this)));
 }

 void RequestProcessor::notify_token_map_updated(const TokenMap::Ptr& token_map) {
   event_loop_->add(new ProcessorNotifyTokenMapUpdate(token_map, Ptr(this)));
 }

 void RequestProcessor::process_request(const RequestHandler::Ptr& request_handler) {
   request_handler->inc_ref(); // Queue reference

   if (request_queue_->enqueue(request_handler.get())) {
     request_count_.fetch_add(1);
     // Only signal the request queue if it's not already processing requests.
     bool expected = false;
     if (!is_processing_.load(MEMORY_ORDER_RELAXED) &&
         is_processing_.compare_exchange_strong(expected, true)) {
       async_.send();
     }
   } else {
     request_handler->dec_ref();
     request_handler->set_error(CASS_ERROR_LIB_REQUEST_QUEUE_FULL,
                                "The request queue has reached capacity");
   }
 }

 int RequestProcessor::init(Protected) {
   int rc = async_.start(event_loop_->loop(), bind_callback(&RequestProcessor::on_async, this));
   if (rc != 0) return rc;
   return prepare_.start(event_loop_->loop(), bind_callback(&RequestProcessor::on_prepare, this));
 }

 void RequestProcessor::on_pool_up(const Address& address) {
   // Don't immediately update the load balancing policies. Give the listener
   // a chance to process the up status and it should call `notify_host_ready()`
   // when it's ready.
   listener_->on_pool_up(address);
 }

 void RequestProcessor::on_pool_down(const Address& address) {
   internal_pool_down(address);
   listener_->on_pool_down(address);
 }

 void RequestProcessor::on_pool_critical_error(const Address& address,
                                               Connector::ConnectionError code,
                                               const String& message) {
   internal_pool_down(address);
   listener_->on_pool_critical_error(address, code, message);
 }

 void RequestProcessor::on_requires_flush() {
   if (!timer_.is_running()) {
     is_processing_.store(true);
     start_coalescing();
   }
 }

 void RequestProcessor::on_close(ConnectionPoolManager* manager) {
   for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin();
        it != load_balancing_policies_.end(); ++it) {
     (*it)->close_handles();
   }
   async_.close_handle();
   prepare_.close_handle();
   timer_.stop();
   connection_pool_manager_.reset();
   listener_->on_close(this);
   dec_ref();
 }

 void RequestProcessor::on_prepared_metadata_changed(const String& id,
                                                     const PreparedMetadata::Entry::Ptr& entry) {
   listener_->on_prepared_metadata_changed(id, entry);
 }

 void RequestProcessor::on_keyspace_changed(const String& keyspace,
                                            KeyspaceChangedResponse response) {
   listener_->on_keyspace_changed(
       keyspace, KeyspaceChangedHandler::Ptr(new KeyspaceChangedHandler(event_loop_, response)));
 }

 bool RequestProcessor::on_wait_for_tracing_data(const RequestHandler::Ptr& request_handler,
                                                 const Host::Ptr& current_host,
                                                 const Response::Ptr& response) {
   TracingDataHandler::Ptr handler(new TracingDataHandler(
       request_handler, current_host, response, settings_.tracing_consistency,
       settings_.max_tracing_wait_time_ms, settings_.retry_tracing_wait_time_ms));

   return write_wait_callback(request_handler, current_host, handler->callback());
 }

 bool RequestProcessor::on_wait_for_schema_agreement(const RequestHandler::Ptr& request_handler,
                                                     const Host::Ptr& current_host,
                                                     const Response::Ptr& response) {
   SchemaAgreementHandler::Ptr handler(
       new SchemaAgreementHandler(request_handler, current_host, response, this,
                                  settings_.max_schema_wait_time_ms, settings_.address_factory));

   return write_wait_callback(request_handler, current_host, handler->callback());
 }

 bool RequestProcessor::on_prepare_all(const RequestHandler::Ptr& request_handler,
                                       const Host::Ptr& current_host,
                                       const Response::Ptr& response) {
   if (!settings_.prepare_on_all_hosts) {
     return false;
   }

   AddressVec addresses = connection_pool_manager_->available();
   if (addresses.empty() || (addresses.size() == 1 && addresses[0] == current_host->address())) {
     return false;
   }

   PrepareAllHandler::Ptr prepare_all_handler(
       new PrepareAllHandler(current_host, response, request_handler,
                             // Subtract the node that's already been prepared
                             addresses.size() - 1));

   for (AddressVec::const_iterator it = addresses.begin(), end = addresses.end(); it != end; ++it) {
     const Address& address(*it);

     // Skip over the node we've already prepared
     if (address == current_host->address()) {
       continue;
     }

     // The destructor of `PrepareAllCallback` will decrement the remaining
     // count in `PrepareAllHandler` even if this is unable to write to a
     // connection successfully.
     PrepareAllCallback::Ptr prepare_all_callback(
         new PrepareAllCallback(address, prepare_all_handler));

     PooledConnection::Ptr connection(connection_pool_manager_->find_least_busy(address));
     if (connection) {
       connection->write(prepare_all_callback.get());
     }
   }

   return true;
 }

 void RequestProcessor::on_done() {
 #ifdef CASS_INTERNAL_DIAGNOSTICS
   reads_during_coalesce_++;
 #endif
   maybe_close(request_count_.fetch_sub(1) - 1);
 }

 bool RequestProcessor::on_is_host_up(const Address& address) {
   return default_profile_.load_balancing_policy()->is_host_up(address);
 }

 void RequestProcessor::internal_close() {
   is_closing_ = true;
   maybe_close(request_count_.load());
 }

 void RequestProcessor::internal_pool_down(const Address& address) {
   LoadBalancingPolicy::Vec policies = load_balancing_policies();
   for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end(); ++it) {
     (*it)->on_host_down(address);
   }
 }

 const ExecutionProfile* RequestProcessor::execution_profile(const String& name) const {
   // Determine if cluster profile should be used
   if (name.empty()) {
     return &default_profile_;
   }

   // Handle profile lookup
   ExecutionProfile::Map::const_iterator it = profiles_.find(name);
   if (it != profiles_.end()) {
     return &it->second;
   }
   return NULL;
 }

 const LoadBalancingPolicy::Vec& RequestProcessor::load_balancing_policies() const {
   return load_balancing_policies_;
 }

 void RequestProcessor::internal_host_add(const Host::Ptr& host) {
   if (connection_pool_manager_) {
     LoadBalancingPolicy::Vec policies = load_balancing_policies();
     if (!is_host_ignored(policies, host)) {
       connection_pool_manager_->add(host);
       for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end();
            ++it) {
         if ((*it)->distance(host) != CASS_HOST_DISTANCE_IGNORE) {
           (*it)->on_host_added(host);
         }
       }
     } else {
       LOG_DEBUG("Host %s will be ignored by all query plans", host->address_string().c_str());
     }
   }
 }

 void RequestProcessor::internal_host_remove(const Host::Ptr& host) {
   if (connection_pool_manager_) {
     connection_pool_manager_->remove(host->address());
     LoadBalancingPolicy::Vec policies = load_balancing_policies();
     for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end();
          ++it) {
       (*it)->on_host_removed(host);
     }
   }
 }

 void RequestProcessor::internal_host_ready(const Host::Ptr& host) {
   // Only mark the host as up if it has connections.
   if (connection_pool_manager_ && connection_pool_manager_->has_connections(host->address())) {
     LoadBalancingPolicy::Vec policies = load_balancing_policies();
     for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end();
          ++it) {
       (*it)->on_host_up(host);
     }
   }
 }

 void RequestProcessor::internal_host_maybe_up(const Address& address) {
   if (connection_pool_manager_) {
     connection_pool_manager_->attempt_immediate_connect(address);
   }
 }

 void RequestProcessor::start_coalescing() {
   io_time_during_coalesce_ = 0;
   timer_.start(event_loop_->loop(), settings_.coalesce_delay_us,
                bind_callback(&RequestProcessor::on_timeout, this));
 }

 void RequestProcessor::on_timeout(MicroTimer* timer) {
   // Don't process for more time than the coalesce delay.
   uint64_t processing_time =
       std::min((io_time_during_coalesce_ * settings_.new_request_ratio) / 100,
                settings_.coalesce_delay_us * 1000);
   int processed = process_requests(processing_time);

   connection_pool_manager_->flush();

   if (processed > 0) {
     attempts_without_requests_ = 0;

 #ifdef CASS_INTERNAL_DIAGNOSTICS
     reads_per_.record_value(reads_during_coalesce_);
     writes_per_.record_value(writes_during_coalesce_);
     reads_during_coalesce_ = 0;
     writes_during_coalesce_ = 0;
 #endif
   } else {
     // Keep trying to process more requests before for a few iterations before
     // putting the loop back to sleep.
     attempts_without_requests_++;
     if (attempts_without_requests_ > 5) {
       attempts_without_requests_ = 0;
       is_processing_.store(false);
       bool expected = false;
       if (request_queue_->is_empty() || !is_processing_.compare_exchange_strong(expected, true)) {
         return;
       }
     }
   }

   if (!timer_.is_running()) {
     start_coalescing();
   }
 }

 void RequestProcessor::on_async(Async* async) {
   process_requests(0);

   connection_pool_manager_->flush();

   // Always attempt to coalesce even if no requests are written so that
   // processing is properly terminated.
   if (!timer_.is_running()) {
     start_coalescing();
   }
 }

 void RequestProcessor::on_prepare(Prepare* prepare) {
   io_time_during_coalesce_ += event_loop_->io_time_elapsed();
 }

 void RequestProcessor::maybe_close(int request_count) {
   if (is_closing_ && request_count <= 0 && request_queue_->is_empty()) {
     if (connection_pool_manager_) connection_pool_manager_->close();
   }
 }

 int RequestProcessor::process_requests(uint64_t processing_time) {
   uint64_t finish_time = uv_hrtime() + processing_time;

   int processed = 0;
   RequestHandler* request_handler = NULL;
   while (request_queue_->dequeue(request_handler)) {
     if (request_handler) {
       const String& profile_name = request_handler->request()->execution_profile_name();
       const ExecutionProfile* profile(execution_profile(profile_name));
       if (profile) {
         if (!profile_name.empty()) {
           LOG_TRACE("Using execution profile '%s'", profile_name.c_str());
         }
         request_handler->init(*profile, connection_pool_manager_.get(), token_map_.get(),
                               settings_.timestamp_generator.get(), this);
         request_handler->execute();
         processed++;
       } else {
         maybe_close(request_count_.fetch_sub(1) - 1);
         request_handler->set_error(CASS_ERROR_LIB_EXECUTION_PROFILE_INVALID,
                                    profile_name + " does not exist");
       }
       request_handler->dec_ref();
     }

     if ((processed & 0x3F) == 0 && // Check the finish time every 64 requests
         uv_hrtime() >= finish_time) {
       break;
     }
   }

 #ifdef CASS_INTERNAL_DIAGNOSTICS
   writes_during_coalesce_ += processed;
 #endif

   return processed;
 }

 bool RequestProcessor::write_wait_callback(const RequestHandler::Ptr& request_handler,
                                            const Host::Ptr& current_host,
                                            const RequestCallback::Ptr& callback) {
   PooledConnection::Ptr connection(
       connection_pool_manager_->find_least_busy(current_host->address()));
   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();
     return true;
   }
   return false;
 }
	/*
	Licensed to the Apache Software Foundation (ASF) under one
	or more contributor license agreements. See the NOTICE file
	distributed with this work for additional information
	regarding copyright ownership. The ASF licenses this file
	to you under the Apache License, Version 2.0 (the
	"License"); you may not use this file except in compliance
	with the License. You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.
	*/

	#include "request_processor.hpp"

	#include "connection_pool_manager_initializer.hpp"
	#include "prepare_all_handler.hpp"
	#include "request_processor.hpp"
	#include "session.hpp"
	#include "tracing_data_handler.hpp"
	#include "utils.hpp"

	using namespace datastax;
	using namespace datastax::internal;
	using namespace datastax::internal::core;

	namespace datastax { namespace internal { namespace core {

	class ProcessorRunClose : public Task {
	public:
	ProcessorRunClose(const RequestProcessor::Ptr& processor)
	: processor_(processor) {}

	virtual void run(EventLoop* event_loop) { processor_->internal_close(); }

	private:
	const RequestProcessor::Ptr processor_;
	};

	class ProcessorNotifyHostAdd : public Task {
	public:
	ProcessorNotifyHostAdd(const Host::Ptr host, const RequestProcessor::Ptr& request_processor)
	: request_processor_(request_processor)
	, host_(host) {}
	virtual void run(EventLoop* event_loop) { request_processor_->internal_host_add(host_); }

	private:
	const RequestProcessor::Ptr request_processor_;
	const Host::Ptr host_;
	};

	class ProcessorNotifyHostRemove : public Task {
	public:
	ProcessorNotifyHostRemove(const Host::Ptr host, const RequestProcessor::Ptr& request_processor)
	: request_processor_(request_processor)
	, host_(host) {}
	virtual void run(EventLoop* event_loop) { request_processor_->internal_host_remove(host_); }

	private:
	RequestProcessor::Ptr request_processor_;
	const Host::Ptr host_;
	};

	class ProcessorNotifyHostReady : public Task {
	public:
	ProcessorNotifyHostReady(const Host::Ptr& host, const RequestProcessor::Ptr& request_processor)
	: request_processor_(request_processor)
	, host_(host) {}
	virtual void run(EventLoop* event_loop) { request_processor_->internal_host_ready(host_); }

	private:
	const RequestProcessor::Ptr request_processor_;
	const Host::Ptr host_;
	};

	class ProcessorNotifyMaybeHostUp : public Task {
	public:
	ProcessorNotifyMaybeHostUp(const Address& address, const RequestProcessor::Ptr& request_processor)
	: request_processor_(request_processor)
	, address_(address) {}
	virtual void run(EventLoop* event_loop) { request_processor_->internal_host_maybe_up(address_); }

	private:
	const RequestProcessor::Ptr request_processor_;
	const Address address_;
	};

	class ProcessorNotifyTokenMapUpdate : public Task {
	public:
	ProcessorNotifyTokenMapUpdate(const TokenMap::Ptr& token_map,
	const RequestProcessor::Ptr& request_processor)
	: request_processor_(request_processor)
	, token_map_(token_map) {}

	virtual void run(EventLoop* event_loop) { request_processor_->token_map_ = token_map_; }

	private:
	const RequestProcessor::Ptr request_processor_;
	const TokenMap::Ptr token_map_;
	};

	class SetKeyspaceProcessor : public Task {
	public:
	SetKeyspaceProcessor(const ConnectionPoolManager::Ptr& manager, const String& keyspace,
	const KeyspaceChangedHandler::Ptr& handler)
	: manager_(manager)
	, keyspace_(keyspace)
	, handler_(handler) {}

	virtual void run(EventLoop* event_loop) { manager_->set_keyspace(keyspace_); }

	private:
	ConnectionPoolManager::Ptr manager_;
	const String keyspace_;
	KeyspaceChangedHandler::Ptr handler_;
	};

	class NopRequestProcessorListener : public RequestProcessorListener {
	public:
	virtual void on_pool_up(const Address& address) {}
	virtual void on_pool_down(const Address& address) {}
	virtual void on_pool_critical_error(const Address& address, Connector::ConnectionError code,
	const String& message) {}
	virtual void on_keyspace_changed(const String& keyspace,
	const KeyspaceChangedHandler::Ptr& handler) {}
	virtual void on_prepared_metadata_changed(const String& id,
	const PreparedMetadata::Entry::Ptr& entry) {}
	virtual void on_close(RequestProcessor* processor) {}
	};

	}}} // namespace datastax::internal::core

	static NopRequestProcessorListener nop_request_processor_listener__;

	RequestProcessorSettings::RequestProcessorSettings()
	: max_schema_wait_time_ms(10000)
	, prepare_on_all_hosts(true)
	, timestamp_generator(new ServerSideTimestampGenerator())
	, default_profile(Config().default_profile())
	, request_queue_size(8192)
	, coalesce_delay_us(CASS_DEFAULT_COALESCE_DELAY)
	, new_request_ratio(CASS_DEFAULT_NEW_REQUEST_RATIO)
	, max_tracing_wait_time_ms(CASS_DEFAULT_MAX_TRACING_DATA_WAIT_TIME_MS)
	, retry_tracing_wait_time_ms(CASS_DEFAULT_RETRY_TRACING_DATA_WAIT_TIME_MS)
	, tracing_consistency(CASS_DEFAULT_TRACING_CONSISTENCY)
	, address_factory(new AddressFactory()) {
	profiles.set_empty_key("");
	}

	RequestProcessorSettings::RequestProcessorSettings(const Config& config)
	: connection_pool_settings(config)
	, max_schema_wait_time_ms(config.max_schema_wait_time_ms())
	, prepare_on_all_hosts(config.prepare_on_all_hosts())
	, timestamp_generator(config.timestamp_gen())
	, default_profile(config.default_profile())
	, profiles(config.profiles())
	, request_queue_size(config.queue_size_io())
	, coalesce_delay_us(config.coalesce_delay_us())
	, new_request_ratio(config.new_request_ratio())
	, max_tracing_wait_time_ms(config.max_tracing_wait_time_ms())
	, retry_tracing_wait_time_ms(config.retry_tracing_wait_time_ms())
	, tracing_consistency(config.tracing_consistency())
	, address_factory(create_address_factory_from_config(config)) {}

	RequestProcessor::RequestProcessor(RequestProcessorListener* listener, EventLoop* event_loop,
	const ConnectionPoolManager::Ptr& connection_pool_manager,
	const Host::Ptr& connected_host, const HostMap& hosts,
	const TokenMap::Ptr& token_map,
	const RequestProcessorSettings& settings, Random* random,
	const String& local_dc)
	: connection_pool_manager_(connection_pool_manager)
	, listener_(listener ? listener : &nop_request_processor_listener__)
	, event_loop_(event_loop)
	, settings_(settings)
	, default_profile_(settings.default_profile)
	, profiles_(settings.profiles)
	, request_count_(0)
	, request_queue_(new MPMCQueue<RequestHandler*>(settings.request_queue_size))
	, is_closing_(false)
	, is_processing_(false)
	, attempts_without_requests_(0)
	, io_time_during_coalesce_(0)
	#ifdef CASS_INTERNAL_DIAGNOSTICS
	, reads_during_coalesce_(0)
	, writes_during_coalesce_(0)
	, writes_per_("writes")
	, reads_per_("reads")
	#endif
	{
	inc_ref(); // For the connection pool manager
	connection_pool_manager_->set_listener(this);

	// Build/Assign the load balancing policies from the execution profiles
	default_profile_.build_load_balancing_policy();
	load_balancing_policies_.push_back(default_profile_.load_balancing_policy());
	for (ExecutionProfile::Map::iterator it = profiles_.begin(), end = profiles_.end(); it != end;
	++it) {
	it->second.build_load_balancing_policy();
	const LoadBalancingPolicy::Ptr& load_balancing_policy = it->second.load_balancing_policy();
	if (load_balancing_policy) {
	LOG_TRACE("Built load balancing policy for '%s' execution profile", it->first.c_str());
	load_balancing_policies_.push_back(load_balancing_policy);
	} else {
	it->second.use_load_balancing_policy(default_profile_.load_balancing_policy());
	}
	}

	token_map_ = token_map;

	LoadBalancingPolicy::Vec policies = load_balancing_policies();
	for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end(); ++it) {
	// Initialize the load balancing policies
	(*it)->init(connected_host, hosts, random, local_dc);
	(*it)->register_handles(event_loop_->loop());
	}

	listener_->on_connect(this);
	}

	void RequestProcessor::close() { event_loop_->add(new ProcessorRunClose(Ptr(this))); }

	void RequestProcessor::set_listener(RequestProcessorListener* listener) {
	listener_ = listener ? listener : &nop_request_processor_listener__;
	}

	void RequestProcessor::set_keyspace(const String& keyspace,
	const KeyspaceChangedHandler::Ptr& handler) {
	// If running on the the current event loop then just set the keyspace,
	// otherwise we're on a different thread so we'll wait for the keyspace
	// to be set in a task on the event loop thread.
	if (event_loop_->is_running_on()) {
	connection_pool_manager_->set_keyspace(keyspace);
	} else {
	event_loop_->add(new SetKeyspaceProcessor(connection_pool_manager_, keyspace, handler));
	}
	}

	void RequestProcessor::notify_host_added(const Host::Ptr& host) {
	event_loop_->add(new ProcessorNotifyHostAdd(host, Ptr(this)));
	}

	void RequestProcessor::notify_host_removed(const Host::Ptr& host) {
	event_loop_->add(new ProcessorNotifyHostRemove(host, Ptr(this)));
	}

	void RequestProcessor::notify_host_ready(const Host::Ptr& host) {
	event_loop_->add(new ProcessorNotifyHostReady(host, Ptr(this)));
	}

	void RequestProcessor::notify_host_maybe_up(const Address& address) {
	event_loop_->add(new ProcessorNotifyMaybeHostUp(address, Ptr(this)));
	}

	void RequestProcessor::notify_token_map_updated(const TokenMap::Ptr& token_map) {
	event_loop_->add(new ProcessorNotifyTokenMapUpdate(token_map, Ptr(this)));
	}

	void RequestProcessor::process_request(const RequestHandler::Ptr& request_handler) {
	request_handler->inc_ref(); // Queue reference

	if (request_queue_->enqueue(request_handler.get())) {
	request_count_.fetch_add(1);
	// Only signal the request queue if it's not already processing requests.
	bool expected = false;
	if (!is_processing_.load(MEMORY_ORDER_RELAXED) &&
	is_processing_.compare_exchange_strong(expected, true)) {
	async_.send();
	}
	} else {
	request_handler->dec_ref();
	request_handler->set_error(CASS_ERROR_LIB_REQUEST_QUEUE_FULL,
	"The request queue has reached capacity");
	}
	}

	int RequestProcessor::init(Protected) {
	int rc = async_.start(event_loop_->loop(), bind_callback(&RequestProcessor::on_async, this));
	if (rc != 0) return rc;
	return prepare_.start(event_loop_->loop(), bind_callback(&RequestProcessor::on_prepare, this));
	}

	void RequestProcessor::on_pool_up(const Address& address) {
	// Don't immediately update the load balancing policies. Give the listener
	// a chance to process the up status and it should call `notify_host_ready()`
	// when it's ready.
	listener_->on_pool_up(address);
	}

	void RequestProcessor::on_pool_down(const Address& address) {
	internal_pool_down(address);
	listener_->on_pool_down(address);
	}

	void RequestProcessor::on_pool_critical_error(const Address& address,
	Connector::ConnectionError code,
	const String& message) {
	internal_pool_down(address);
	listener_->on_pool_critical_error(address, code, message);
	}

	void RequestProcessor::on_requires_flush() {
	if (!timer_.is_running()) {
	is_processing_.store(true);
	start_coalescing();
	}
	}

	void RequestProcessor::on_close(ConnectionPoolManager* manager) {
	for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin();
	it != load_balancing_policies_.end(); ++it) {
	(*it)->close_handles();
	}
	async_.close_handle();
	prepare_.close_handle();
	timer_.stop();
	connection_pool_manager_.reset();
	listener_->on_close(this);
	dec_ref();
	}

	void RequestProcessor::on_prepared_metadata_changed(const String& id,
	const PreparedMetadata::Entry::Ptr& entry) {
	listener_->on_prepared_metadata_changed(id, entry);
	}

	void RequestProcessor::on_keyspace_changed(const String& keyspace,
	KeyspaceChangedResponse response) {
	listener_->on_keyspace_changed(
	keyspace, KeyspaceChangedHandler::Ptr(new KeyspaceChangedHandler(event_loop_, response)));
	}

	bool RequestProcessor::on_wait_for_tracing_data(const RequestHandler::Ptr& request_handler,
	const Host::Ptr& current_host,
	const Response::Ptr& response) {
	TracingDataHandler::Ptr handler(new TracingDataHandler(
	request_handler, current_host, response, settings_.tracing_consistency,
	settings_.max_tracing_wait_time_ms, settings_.retry_tracing_wait_time_ms));

	return write_wait_callback(request_handler, current_host, handler->callback());
	}

	bool RequestProcessor::on_wait_for_schema_agreement(const RequestHandler::Ptr& request_handler,
	const Host::Ptr& current_host,
	const Response::Ptr& response) {
	SchemaAgreementHandler::Ptr handler(
	new SchemaAgreementHandler(request_handler, current_host, response, this,
	settings_.max_schema_wait_time_ms, settings_.address_factory));

	return write_wait_callback(request_handler, current_host, handler->callback());
	}

	bool RequestProcessor::on_prepare_all(const RequestHandler::Ptr& request_handler,
	const Host::Ptr& current_host,
	const Response::Ptr& response) {
	if (!settings_.prepare_on_all_hosts) {
	return false;
	}

	AddressVec addresses = connection_pool_manager_->available();
	if (addresses.empty() \|\| (addresses.size() == 1 && addresses[0] == current_host->address())) {
	return false;
	}

	PrepareAllHandler::Ptr prepare_all_handler(
	new PrepareAllHandler(current_host, response, request_handler,
	// Subtract the node that's already been prepared
	addresses.size() - 1));

	for (AddressVec::const_iterator it = addresses.begin(), end = addresses.end(); it != end; ++it) {
	const Address& address(*it);

	// Skip over the node we've already prepared
	if (address == current_host->address()) {
	continue;
	}

	// The destructor of `PrepareAllCallback` will decrement the remaining
	// count in `PrepareAllHandler` even if this is unable to write to a
	// connection successfully.
	PrepareAllCallback::Ptr prepare_all_callback(
	new PrepareAllCallback(address, prepare_all_handler));

	PooledConnection::Ptr connection(connection_pool_manager_->find_least_busy(address));
	if (connection) {
	connection->write(prepare_all_callback.get());
	}
	}

	return true;
	}

	void RequestProcessor::on_done() {
	#ifdef CASS_INTERNAL_DIAGNOSTICS
	reads_during_coalesce_++;
	#endif
	maybe_close(request_count_.fetch_sub(1) - 1);
	}

	bool RequestProcessor::on_is_host_up(const Address& address) {
	return default_profile_.load_balancing_policy()->is_host_up(address);
	}

	void RequestProcessor::internal_close() {
	is_closing_ = true;
	maybe_close(request_count_.load());
	}

	void RequestProcessor::internal_pool_down(const Address& address) {
	LoadBalancingPolicy::Vec policies = load_balancing_policies();
	for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end(); ++it) {
	(*it)->on_host_down(address);
	}
	}

	const ExecutionProfile* RequestProcessor::execution_profile(const String& name) const {
	// Determine if cluster profile should be used
	if (name.empty()) {
	return &default_profile_;
	}

	// Handle profile lookup
	ExecutionProfile::Map::const_iterator it = profiles_.find(name);
	if (it != profiles_.end()) {
	return &it->second;
	}
	return NULL;
	}

	const LoadBalancingPolicy::Vec& RequestProcessor::load_balancing_policies() const {
	return load_balancing_policies_;
	}

	void RequestProcessor::internal_host_add(const Host::Ptr& host) {
	if (connection_pool_manager_) {
	LoadBalancingPolicy::Vec policies = load_balancing_policies();
	if (!is_host_ignored(policies, host)) {
	connection_pool_manager_->add(host);
	for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end();
	++it) {
	if ((*it)->distance(host) != CASS_HOST_DISTANCE_IGNORE) {
	(*it)->on_host_added(host);
	}
	}
	} else {
	LOG_DEBUG("Host %s will be ignored by all query plans", host->address_string().c_str());
	}
	}
	}

	void RequestProcessor::internal_host_remove(const Host::Ptr& host) {
	if (connection_pool_manager_) {
	connection_pool_manager_->remove(host->address());
	LoadBalancingPolicy::Vec policies = load_balancing_policies();
	for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end();
	++it) {
	(*it)->on_host_removed(host);
	}
	}
	}

	void RequestProcessor::internal_host_ready(const Host::Ptr& host) {
	// Only mark the host as up if it has connections.
	if (connection_pool_manager_ && connection_pool_manager_->has_connections(host->address())) {
	LoadBalancingPolicy::Vec policies = load_balancing_policies();
	for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(); it != policies.end();
	++it) {
	(*it)->on_host_up(host);
	}
	}
	}

	void RequestProcessor::internal_host_maybe_up(const Address& address) {
	if (connection_pool_manager_) {
	connection_pool_manager_->attempt_immediate_connect(address);
	}
	}

	void RequestProcessor::start_coalescing() {
	io_time_during_coalesce_ = 0;
	timer_.start(event_loop_->loop(), settings_.coalesce_delay_us,
	bind_callback(&RequestProcessor::on_timeout, this));
	}

	void RequestProcessor::on_timeout(MicroTimer* timer) {
	// Don't process for more time than the coalesce delay.
	uint64_t processing_time =
	std::min((io_time_during_coalesce_ * settings_.new_request_ratio) / 100,
	settings_.coalesce_delay_us * 1000);
	int processed = process_requests(processing_time);

	connection_pool_manager_->flush();

	if (processed > 0) {
	attempts_without_requests_ = 0;

	#ifdef CASS_INTERNAL_DIAGNOSTICS
	reads_per_.record_value(reads_during_coalesce_);
	writes_per_.record_value(writes_during_coalesce_);
	reads_during_coalesce_ = 0;
	writes_during_coalesce_ = 0;
	#endif
	} else {
	// Keep trying to process more requests before for a few iterations before
	// putting the loop back to sleep.
	attempts_without_requests_++;
	if (attempts_without_requests_ > 5) {
	attempts_without_requests_ = 0;
	is_processing_.store(false);
	bool expected = false;
	if (request_queue_->is_empty() \|\| !is_processing_.compare_exchange_strong(expected, true)) {
	return;
	}
	}
	}

	if (!timer_.is_running()) {
	start_coalescing();
	}
	}

	void RequestProcessor::on_async(Async* async) {
	process_requests(0);

	connection_pool_manager_->flush();

	// Always attempt to coalesce even if no requests are written so that
	// processing is properly terminated.
	if (!timer_.is_running()) {
	start_coalescing();
	}
	}

	void RequestProcessor::on_prepare(Prepare* prepare) {
	io_time_during_coalesce_ += event_loop_->io_time_elapsed();
	}

	void RequestProcessor::maybe_close(int request_count) {
	if (is_closing_ && request_count <= 0 && request_queue_->is_empty()) {
	if (connection_pool_manager_) connection_pool_manager_->close();
	}
	}

	int RequestProcessor::process_requests(uint64_t processing_time) {
	uint64_t finish_time = uv_hrtime() + processing_time;

	int processed = 0;
	RequestHandler* request_handler = NULL;
	while (request_queue_->dequeue(request_handler)) {
	if (request_handler) {
	const String& profile_name = request_handler->request()->execution_profile_name();
	const ExecutionProfile* profile(execution_profile(profile_name));
	if (profile) {
	if (!profile_name.empty()) {
	LOG_TRACE("Using execution profile '%s'", profile_name.c_str());
	}
	request_handler->init(*profile, connection_pool_manager_.get(), token_map_.get(),
	settings_.timestamp_generator.get(), this);
	request_handler->execute();
	processed++;
	} else {
	maybe_close(request_count_.fetch_sub(1) - 1);
	request_handler->set_error(CASS_ERROR_LIB_EXECUTION_PROFILE_INVALID,
	profile_name + " does not exist");
	}
	request_handler->dec_ref();
	}

	if ((processed & 0x3F) == 0 && // Check the finish time every 64 requests
	uv_hrtime() >= finish_time) {
	break;
	}
	}

	#ifdef CASS_INTERNAL_DIAGNOSTICS
	writes_during_coalesce_ += processed;
	#endif

	return processed;
	}

	bool RequestProcessor::write_wait_callback(const RequestHandler::Ptr& request_handler,
	const Host::Ptr& current_host,
	const RequestCallback::Ptr& callback) {
	PooledConnection::Ptr connection(
	connection_pool_manager_->find_least_busy(current_host->address()));
	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();
	return true;
	}
	return false;
	}