| /** |
| * 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 "query_execution/ForemanDistributed.hpp" |
| |
| #include <cstddef> |
| #include <cstdio> |
| #include <cstdlib> |
| #include <memory> |
| #include <unordered_map> |
| #include <unordered_set> |
| #include <utility> |
| #include <vector> |
| |
| #include "catalog/Catalog.pb.h" |
| #include "catalog/CatalogDatabase.hpp" |
| #include "catalog/CatalogRelation.hpp" |
| #include "catalog/CatalogTypedefs.hpp" |
| #include "cli/Flags.hpp" |
| #include "query_execution/AdmitRequestMessage.hpp" |
| #include "query_execution/BlockLocator.hpp" |
| #include "query_execution/BlockLocatorUtil.hpp" |
| #include "query_execution/PolicyEnforcerBase.hpp" |
| #include "query_execution/PolicyEnforcerDistributed.hpp" |
| #include "query_execution/QueryContext.hpp" |
| #include "query_execution/QueryExecutionMessages.pb.h" |
| #include "query_execution/QueryExecutionTypedefs.hpp" |
| #include "query_execution/QueryExecutionUtil.hpp" |
| #include "query_execution/ShiftbossDirectory.hpp" |
| #include "relational_operators/WorkOrder.pb.h" |
| #include "storage/DataExchangerAsync.hpp" |
| #include "storage/StorageBlockInfo.hpp" |
| #include "storage/StorageManager.hpp" |
| #include "threading/ThreadUtil.hpp" |
| #include "utility/EqualsAnyConstant.hpp" |
| |
| #include "glog/logging.h" |
| |
| #include "tmb/address.h" |
| #include "tmb/id_typedefs.h" |
| #include "tmb/message_bus.h" |
| #include "tmb/message_style.h" |
| #include "tmb/tagged_message.h" |
| |
| using std::make_unique; |
| using std::move; |
| using std::size_t; |
| using std::unique_ptr; |
| using std::vector; |
| |
| using tmb::AnnotatedMessage; |
| using tmb::MessageBus; |
| using tmb::TaggedMessage; |
| using tmb::client_id; |
| |
| namespace quickstep { |
| |
| namespace S = serialization; |
| |
| class QueryHandle; |
| |
| ForemanDistributed::ForemanDistributed( |
| const BlockLocator &block_locator, |
| MessageBus *bus, |
| CatalogDatabaseLite *catalog_database, |
| QueryProcessor *query_processor, |
| const int cpu_id) |
| : ForemanBase(bus, cpu_id), |
| block_locator_(block_locator), |
| catalog_database_(DCHECK_NOTNULL(catalog_database)) { |
| const std::vector<QueryExecutionMessageType> sender_message_types{ |
| kBlockDomainRegistrationMessage, |
| kShiftbossRegistrationResponseMessage, |
| kQueryInitiateMessage, |
| kWorkOrderMessage, |
| kInitiateRebuildMessage, |
| kQueryTeardownMessage, |
| kQueryExecutionSuccessMessage, |
| kCommandResponseMessage, |
| kPoisonMessage}; |
| |
| for (const auto message_type : sender_message_types) { |
| bus_->RegisterClientAsSender(foreman_client_id_, message_type); |
| } |
| |
| const std::vector<QueryExecutionMessageType> receiver_message_types{ |
| kBlockDomainRegistrationResponseMessage, |
| kShiftbossRegistrationMessage, |
| kAdmitRequestMessage, |
| kQueryInitiateResponseMessage, |
| kCatalogRelationNewBlockMessage, |
| kDataPipelineMessage, |
| kInitiateRebuildResponseMessage, |
| kWorkOrderCompleteMessage, |
| kRebuildWorkOrderCompleteMessage, |
| kWorkOrderFeedbackMessage, |
| kPoisonMessage}; |
| |
| for (const auto message_type : receiver_message_types) { |
| bus_->RegisterClientAsReceiver(foreman_client_id_, message_type); |
| } |
| |
| client_id locator_client_id; |
| storage_manager_ = make_unique<StorageManager>( |
| FLAGS_storage_path, |
| block_locator::getBlockDomain(data_exchanger_.network_address(), foreman_client_id_, &locator_client_id, bus_), |
| locator_client_id, bus_); |
| |
| data_exchanger_.set_storage_manager(storage_manager_.get()); |
| data_exchanger_.start(); |
| |
| policy_enforcer_ = make_unique<PolicyEnforcerDistributed>( |
| foreman_client_id_, catalog_database_, query_processor, storage_manager_.get(), &shiftboss_directory_, bus_); |
| } |
| |
| void ForemanDistributed::run() { |
| if (cpu_id_ >= 0) { |
| // We can pin the foreman thread to a CPU if specified. |
| ThreadUtil::BindToCPU(cpu_id_); |
| } |
| |
| // Ensure that at least one Shiftboss to register. |
| if (shiftboss_directory_.empty()) { |
| const AnnotatedMessage annotated_message = bus_->Receive(foreman_client_id_, 0, true); |
| const TaggedMessage &tagged_message = annotated_message.tagged_message; |
| DCHECK_EQ(kShiftbossRegistrationMessage, tagged_message.message_type()); |
| DLOG(INFO) << "ForemanDistributed received ShiftbossRegistrationMessage from Client " << annotated_message.sender; |
| |
| S::ShiftbossRegistrationMessage proto; |
| CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes())); |
| |
| processShiftbossRegistrationMessage(annotated_message.sender, proto.work_order_capacity()); |
| DCHECK_EQ(1u, shiftboss_directory_.size()); |
| } |
| |
| // Event loop |
| for (;;) { |
| // Receive() causes this thread to sleep until next message is received. |
| const AnnotatedMessage annotated_message = |
| bus_->Receive(foreman_client_id_, 0, true); |
| const TaggedMessage &tagged_message = annotated_message.tagged_message; |
| const tmb::message_type_id message_type = tagged_message.message_type(); |
| DLOG(INFO) << "ForemanDistributed received " << QueryExecutionUtil::MessageTypeToString(message_type) |
| << " from Client " << annotated_message.sender; |
| switch (message_type) { |
| case kShiftbossRegistrationMessage: { |
| S::ShiftbossRegistrationMessage proto; |
| CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes())); |
| |
| processShiftbossRegistrationMessage(annotated_message.sender, proto.work_order_capacity()); |
| break; |
| } |
| case kAdmitRequestMessage: { |
| const AdmitRequestMessage *request_message = |
| static_cast<const AdmitRequestMessage*>(tagged_message.message()); |
| |
| if (!policy_enforcer_->admitQueries(request_message->getQueryHandles())) { |
| LOG(WARNING) << "The scheduler could not admit all the queries"; |
| } |
| break; |
| } |
| case kQueryInitiateResponseMessage: { |
| S::QueryInitiateResponseMessage proto; |
| CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes())); |
| break; |
| } |
| case kCatalogRelationNewBlockMessage: // Fall through |
| case kDataPipelineMessage: |
| case kRebuildWorkOrderCompleteMessage: |
| case kWorkOrderCompleteMessage: |
| case kWorkOrderFeedbackMessage: { |
| policy_enforcer_->processMessage(tagged_message); |
| break; |
| } |
| case kInitiateRebuildResponseMessage: { |
| // A unique case in the distributed version. |
| static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())-> |
| processInitiateRebuildResponseMessage(tagged_message); |
| break; |
| } |
| case kPoisonMessage: { |
| if (policy_enforcer_->hasQueries()) { |
| LOG(WARNING) << "ForemanDistributed thread exiting while some queries are " |
| "under execution or waiting to be admitted"; |
| } |
| |
| // Shutdown all Shiftbosses. |
| tmb::Address shiftboss_addresses; |
| for (std::size_t i = 0; i < shiftboss_directory_.size(); ++i) { |
| shiftboss_addresses.AddRecipient(shiftboss_directory_.getClientId(i)); |
| } |
| |
| tmb::MessageStyle broadcast_style; |
| broadcast_style.Broadcast(true); |
| |
| TaggedMessage poison_message(kPoisonMessage); |
| |
| const MessageBus::SendStatus send_status = |
| bus_->Send(foreman_client_id_, |
| shiftboss_addresses, |
| broadcast_style, |
| move(poison_message)); |
| DCHECK(send_status == MessageBus::SendStatus::kOK); |
| return; |
| } |
| default: |
| LOG(FATAL) << "Unknown message type to ForemanDistributed"; |
| } |
| |
| if (canCollectNewMessages(message_type)) { |
| vector<unique_ptr<S::WorkOrderMessage>> new_messages; |
| static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())-> |
| getWorkOrderProtoMessages(&new_messages); |
| dispatchWorkOrderMessages(new_messages); |
| } |
| } |
| } |
| |
| bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id message_type) { |
| return !QUICKSTEP_EQUALS_ANY_CONSTANT(message_type, |
| kCatalogRelationNewBlockMessage, |
| kWorkOrderFeedbackMessage); |
| } |
| |
| bool ForemanDistributed::isAggregationRelatedWorkOrder(const S::WorkOrderMessage &proto, |
| const size_t next_shiftboss_index_to_schedule, |
| size_t *shiftboss_index_for_aggregation) { |
| const S::WorkOrder &work_order_proto = proto.work_order(); |
| QueryContext::aggregation_state_id aggr_state_index; |
| partition_id part_id; |
| vector<QueryContext::lip_filter_id> lip_filter_indexes; |
| block_id block = kInvalidBlockId; |
| |
| switch (work_order_proto.work_order_type()) { |
| case S::AGGREGATION: |
| aggr_state_index = work_order_proto.GetExtension(S::AggregationWorkOrder::aggr_state_index); |
| part_id = work_order_proto.GetExtension(S::AggregationWorkOrder::partition_id); |
| |
| for (int i = 0; i < work_order_proto.ExtensionSize(S::AggregationWorkOrder::lip_filter_indexes); ++i) { |
| lip_filter_indexes.push_back(work_order_proto.GetExtension(S::AggregationWorkOrder::lip_filter_indexes, i)); |
| } |
| |
| block = work_order_proto.GetExtension(S::AggregationWorkOrder::block_id); |
| break; |
| case S::BUILD_AGGREGATION_EXISTENCE_MAP: |
| aggr_state_index = work_order_proto.GetExtension(S::BuildAggregationExistenceMapWorkOrder::aggr_state_index); |
| part_id = work_order_proto.GetExtension(S::BuildAggregationExistenceMapWorkOrder::partition_id); |
| break; |
| case S::INITIALIZE_AGGREGATION: |
| aggr_state_index = work_order_proto.GetExtension(S::InitializeAggregationWorkOrder::aggr_state_index); |
| part_id = work_order_proto.GetExtension(S::InitializeAggregationWorkOrder::partition_id); |
| break; |
| case S::FINALIZE_AGGREGATION: |
| aggr_state_index = work_order_proto.GetExtension(S::FinalizeAggregationWorkOrder::aggr_state_index); |
| part_id = work_order_proto.GetExtension(S::FinalizeAggregationWorkOrder::partition_id); |
| break; |
| case S::DESTROY_AGGREGATION_STATE: |
| aggr_state_index = work_order_proto.GetExtension(S::DestroyAggregationStateWorkOrder::aggr_state_index); |
| part_id = work_order_proto.GetExtension(S::DestroyAggregationStateWorkOrder::partition_id); |
| break; |
| default: |
| return false; |
| } |
| |
| static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->getShiftbossIndexForAggregation( |
| proto.query_id(), aggr_state_index, part_id, lip_filter_indexes, block_locator_, block, |
| next_shiftboss_index_to_schedule, shiftboss_index_for_aggregation); |
| |
| return true; |
| } |
| |
| bool ForemanDistributed::isHashJoinRelatedWorkOrder(const S::WorkOrderMessage &proto, |
| const size_t next_shiftboss_index_to_schedule, |
| size_t *shiftboss_index_for_hash_join) { |
| const S::WorkOrder &work_order_proto = proto.work_order(); |
| QueryContext::join_hash_table_id join_hash_table_index; |
| partition_id part_id; |
| vector<QueryContext::lip_filter_id> lip_filter_indexes; |
| block_id block = kInvalidBlockId; |
| |
| switch (work_order_proto.work_order_type()) { |
| case S::BUILD_HASH: |
| join_hash_table_index = work_order_proto.GetExtension(S::BuildHashWorkOrder::join_hash_table_index); |
| part_id = work_order_proto.GetExtension(S::BuildHashWorkOrder::partition_id); |
| |
| for (int i = 0; i < work_order_proto.ExtensionSize(S::BuildHashWorkOrder::lip_filter_indexes); ++i) { |
| lip_filter_indexes.push_back(work_order_proto.GetExtension(S::BuildHashWorkOrder::lip_filter_indexes, i)); |
| } |
| |
| block = work_order_proto.GetExtension(S::BuildHashWorkOrder::block_id); |
| break; |
| case S::HASH_JOIN: |
| join_hash_table_index = work_order_proto.GetExtension(S::HashJoinWorkOrder::join_hash_table_index); |
| part_id = work_order_proto.GetExtension(S::HashJoinWorkOrder::partition_id); |
| break; |
| case S::DESTROY_HASH: |
| join_hash_table_index = work_order_proto.GetExtension(S::DestroyHashWorkOrder::join_hash_table_index); |
| part_id = work_order_proto.GetExtension(S::DestroyHashWorkOrder::partition_id); |
| break; |
| default: |
| return false; |
| } |
| |
| static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->getShiftbossIndexForHashJoin( |
| proto.query_id(), join_hash_table_index, part_id, lip_filter_indexes, block_locator_, block, |
| next_shiftboss_index_to_schedule, shiftboss_index_for_hash_join); |
| |
| return true; |
| } |
| |
| bool ForemanDistributed::isLipRelatedWorkOrder(const S::WorkOrderMessage &proto, |
| const size_t next_shiftboss_index_to_schedule, |
| size_t *shiftboss_index_for_lip) { |
| const S::WorkOrder &work_order_proto = proto.work_order(); |
| vector<QueryContext::lip_filter_id> lip_filter_indexes; |
| partition_id part_id; |
| block_id block = kInvalidBlockId; |
| |
| switch (work_order_proto.work_order_type()) { |
| case S::BUILD_LIP_FILTER: |
| for (int i = 0; i < work_order_proto.ExtensionSize(S::BuildLIPFilterWorkOrder::lip_filter_indexes); ++i) { |
| lip_filter_indexes.push_back(work_order_proto.GetExtension(S::BuildLIPFilterWorkOrder::lip_filter_indexes, i)); |
| } |
| part_id = work_order_proto.GetExtension(S::BuildLIPFilterWorkOrder::partition_id); |
| block = work_order_proto.GetExtension(S::BuildLIPFilterWorkOrder::build_block_id); |
| break; |
| case S::SELECT: |
| for (int i = 0; i < work_order_proto.ExtensionSize(S::SelectWorkOrder::lip_filter_indexes); ++i) { |
| lip_filter_indexes.push_back(work_order_proto.GetExtension(S::SelectWorkOrder::lip_filter_indexes, i)); |
| } |
| part_id = work_order_proto.GetExtension(S::SelectWorkOrder::partition_id); |
| block = work_order_proto.GetExtension(S::SelectWorkOrder::block_id); |
| break; |
| default: |
| return false; |
| } |
| |
| static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->getShiftbossIndexForLip( |
| proto.query_id(), lip_filter_indexes, part_id, block_locator_, block, next_shiftboss_index_to_schedule, |
| shiftboss_index_for_lip); |
| |
| return true; |
| } |
| |
| namespace { |
| |
| constexpr size_t kDefaultShiftbossIndex = 0u; |
| |
| bool hasBlockLocalityInfo(const serialization::WorkOrder &work_order_proto, |
| const BlockLocator &block_locator, |
| std::size_t *shiftboss_index_for_block) { |
| block_id block = kInvalidBlockId; |
| switch (work_order_proto.work_order_type()) { |
| case S::SAVE_BLOCKS: { |
| block = work_order_proto.GetExtension(S::SaveBlocksWorkOrder::block_id); |
| break; |
| } |
| case S::SORT_RUN_GENERATION: { |
| block = work_order_proto.GetExtension(S::SortRunGenerationWorkOrder::block_id); |
| break; |
| } |
| default: |
| return false; |
| } |
| |
| DCHECK_NE(block, kInvalidBlockId); |
| return block_locator.getBlockLocalityInfo(block, shiftboss_index_for_block); |
| } |
| |
| } // namespace |
| |
| void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::WorkOrderMessage>> &messages) { |
| static size_t shiftboss_index = kDefaultShiftbossIndex; |
| |
| PolicyEnforcerDistributed* policy_enforcer_dist = static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get()); |
| for (const auto &message : messages) { |
| DCHECK(message != nullptr); |
| const S::WorkOrderMessage &proto = *message; |
| const S::WorkOrder &work_order_proto = proto.work_order(); |
| size_t shiftboss_index_for_particular_work_order_type; |
| if (policy_enforcer_dist->isSingleNodeQuery(proto.query_id())) { |
| // Always schedule the single-node query to the same Shiftboss. |
| shiftboss_index_for_particular_work_order_type = kDefaultShiftbossIndex; |
| } else if (isLipRelatedWorkOrder(proto, shiftboss_index, &shiftboss_index_for_particular_work_order_type)) { |
| } else if (isAggregationRelatedWorkOrder(proto, shiftboss_index, &shiftboss_index_for_particular_work_order_type)) { |
| } else if (isHashJoinRelatedWorkOrder(proto, shiftboss_index, &shiftboss_index_for_particular_work_order_type)) { |
| } else if (work_order_proto.work_order_type() == S::NESTED_LOOP_JOIN) { |
| static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->getShiftbossIndexForNestedLoopsJoin( |
| proto.query_id(), work_order_proto.GetExtension(S::NestedLoopsJoinWorkOrder::nested_loops_join_index), |
| work_order_proto.GetExtension(S::NestedLoopsJoinWorkOrder::partition_id), block_locator_, |
| work_order_proto.GetExtension(S::NestedLoopsJoinWorkOrder::left_block_id), |
| work_order_proto.GetExtension(S::NestedLoopsJoinWorkOrder::right_block_id), |
| shiftboss_index, &shiftboss_index_for_particular_work_order_type); |
| } else if (hasBlockLocalityInfo(work_order_proto, block_locator_, |
| &shiftboss_index_for_particular_work_order_type)) { |
| } else { |
| shiftboss_index_for_particular_work_order_type = shiftboss_index; |
| } |
| |
| sendWorkOrderMessage(shiftboss_index_for_particular_work_order_type, proto); |
| shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index_for_particular_work_order_type); |
| |
| if (shiftboss_index == shiftboss_index_for_particular_work_order_type && |
| shiftboss_directory_.hasReachedCapacity(shiftboss_index)) { |
| shiftboss_index = (shiftboss_index + 1) % shiftboss_directory_.size(); |
| } else { |
| // NOTE(zuyu): This is not the exact round-robin scheduling, as in this case, |
| // <shiftboss_index_for_particular_work_order_type> might be scheduled one |
| // more WorkOrder for an Aggregation or a HashJoin. |
| } |
| } |
| } |
| |
| void ForemanDistributed::sendWorkOrderMessage(const size_t shiftboss_index, |
| const S::WorkOrderMessage &proto) { |
| const size_t proto_length = proto.ByteSize(); |
| char *proto_bytes = static_cast<char*>(malloc(proto_length)); |
| CHECK(proto.SerializeToArray(proto_bytes, proto_length)); |
| |
| TaggedMessage message(static_cast<const void*>(proto_bytes), |
| proto_length, |
| kWorkOrderMessage); |
| free(proto_bytes); |
| |
| const client_id shiftboss_client_id = shiftboss_directory_.getClientId(shiftboss_index); |
| DLOG(INFO) << "ForemanDistributed sent WorkOrderMessage to Shiftboss with Client " << shiftboss_client_id; |
| const MessageBus::SendStatus send_status = |
| QueryExecutionUtil::SendTMBMessage(bus_, |
| foreman_client_id_, |
| shiftboss_client_id, |
| move(message)); |
| CHECK(send_status == MessageBus::SendStatus::kOK); |
| } |
| |
| void ForemanDistributed::printWorkOrderProfilingResults(const std::size_t query_id, |
| std::FILE *out) const { |
| const std::vector<WorkOrderTimeEntry> &recorded_times = |
| policy_enforcer_->getProfilingResults(query_id); |
| fputs("Query ID,Worker ID,Operator ID,Time (microseconds)\n", out); |
| for (const auto &workorder_entry : recorded_times) { |
| const std::size_t worker_id = workorder_entry.worker_id; |
| fprintf(out, |
| "%lu,%lu,%lu,%lu\n", |
| query_id, |
| worker_id, |
| workorder_entry.operator_id, // Operator ID. |
| workorder_entry.end_time - workorder_entry.start_time); // Time. |
| } |
| } |
| |
| void ForemanDistributed::processShiftbossRegistrationMessage(const client_id shiftboss_client_id, |
| const std::size_t work_order_capacity) { |
| S::ShiftbossRegistrationResponseMessage proto; |
| proto.set_shiftboss_index(shiftboss_directory_.size()); |
| proto.mutable_catalog_database()->MergeFrom(static_cast<CatalogDatabase*>(catalog_database_)->getProto()); |
| |
| const size_t proto_length = proto.ByteSize(); |
| char *proto_bytes = static_cast<char*>(malloc(proto_length)); |
| CHECK(proto.SerializeToArray(proto_bytes, proto_length)); |
| |
| TaggedMessage message(static_cast<const void*>(proto_bytes), |
| proto_length, |
| kShiftbossRegistrationResponseMessage); |
| free(proto_bytes); |
| |
| shiftboss_directory_.addShiftboss(shiftboss_client_id, work_order_capacity); |
| |
| DLOG(INFO) << "ForemanDistributed sent ShiftbossRegistrationResponseMessage to Shiftboss with Client " |
| << shiftboss_client_id; |
| const MessageBus::SendStatus send_status = |
| QueryExecutionUtil::SendTMBMessage(bus_, |
| foreman_client_id_, |
| shiftboss_client_id, |
| move(message)); |
| CHECK(send_status == MessageBus::SendStatus::kOK); |
| } |
| |
| } // namespace quickstep |
