You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by zu...@apache.org on 2016/11/28 01:05:03 UTC
[1/4] incubator-quickstep git commit: A better debug log for Worker.
[Forced Update!]
Repository: incubator-quickstep
Updated Branches:
refs/heads/multi-shiftboss-test d2862e578 -> c64d60f7a (forced update)
A better debug log for Worker.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/4286d75e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/4286d75e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/4286d75e
Branch: refs/heads/multi-shiftboss-test
Commit: 4286d75ec622ac6e60813c6892b131660ce3a931
Parents: a31dde4
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 27 11:45:28 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 11:45:28 2016 -0800
----------------------------------------------------------------------
query_execution/Worker.cpp | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/4286d75e/query_execution/Worker.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Worker.cpp b/query_execution/Worker.cpp
index 31eb964..9a548fd 100644
--- a/query_execution/Worker.cpp
+++ b/query_execution/Worker.cpp
@@ -107,7 +107,14 @@ void Worker::sendWorkOrderCompleteMessage(const tmb::client_id receiver,
static_cast<const void *>(proto_bytes), proto_length, message_type);
std::free(proto_bytes);
- DLOG(INFO) << "Worker sent WorkOrderCompleteMessage (typed '" << message_type
+ DLOG(INFO) << "Worker " << worker_thread_index_
+#ifdef QUICKSTEP_DISTRIBUTED
+ << " in Shiftboss " << shiftboss_index_
+#endif // QUICKSTEP_DISTRIBUTED
+ << " sent "
+ << (message_type == kWorkOrderCompleteMessage ? "WorkOrderCompleteMessage"
+ : "RebuildWorkOrderCompleteMessage")
+ << " (typed '" << message_type
<< "') to Scheduler with TMB client ID " << receiver;
const tmb::MessageBus::SendStatus send_status =
QueryExecutionUtil::SendTMBMessage(
[2/4] incubator-quickstep git commit: Used multiple Shiftbosses in
the distributed unit tests.
Posted by zu...@apache.org.
Used multiple Shiftbosses in the distributed unit tests.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/c4f4b285
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/c4f4b285
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/c4f4b285
Branch: refs/heads/multi-shiftboss-test
Commit: c4f4b285b63e8a3d6bb8fd312f6d00cc8f1cbd45
Parents: 4286d75
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sat Nov 12 22:31:42 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 11:46:23 2016 -0800
----------------------------------------------------------------------
query_execution/ForemanDistributed.cpp | 14 +++---
query_execution/PolicyEnforcerDistributed.cpp | 6 ++-
query_execution/QueryExecutionState.hpp | 53 ++++++++++++++++++++++
query_execution/QueryManagerDistributed.cpp | 35 ++++++++------
query_execution/QueryManagerDistributed.hpp | 4 +-
5 files changed, 88 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c4f4b285/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index d619657..7dccce4 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -238,17 +238,19 @@ void ForemanDistributed::run() {
bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id message_type) {
return !QUICKSTEP_EQUALS_ANY_CONSTANT(message_type,
kCatalogRelationNewBlockMessage,
- kWorkOrderFeedbackMessage) &&
- // TODO(zuyu): Multiple Shiftbosses support.
- !shiftboss_directory_.hasReachedCapacity(0);
+ kWorkOrderFeedbackMessage);
}
void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::WorkOrderMessage>> &messages) {
+ const size_t num_shiftbosses = shiftboss_directory_.size();
+ size_t shiftboss_index = 0u;
for (const auto &message : messages) {
DCHECK(message != nullptr);
- // TODO(zuyu): Multiple Shiftbosses support.
- sendWorkOrderMessage(0, *message);
- shiftboss_directory_.incrementNumQueuedWorkOrders(0);
+ sendWorkOrderMessage(shiftboss_index, *message);
+ shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
+
+ // TO(zuyu): Take data-locality into account for scheduling.
+ shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c4f4b285/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index c06fd86..6e09ea8 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -140,8 +140,10 @@ void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb:
QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
const std::size_t num_rebuild_work_orders = proto.num_rebuild_work_orders();
- query_manager->processInitiateRebuildResponseMessage(proto.operator_index(), num_rebuild_work_orders);
- shiftboss_directory_->addNumQueuedWorkOrders(proto.shiftboss_index(), num_rebuild_work_orders);
+ const size_t shiftboss_index = proto.shiftboss_index();
+ query_manager->processInitiateRebuildResponseMessage(
+ proto.operator_index(), num_rebuild_work_orders, shiftboss_index);
+ shiftboss_directory_->addNumQueuedWorkOrders(shiftboss_index, num_rebuild_work_orders);
if (query_manager->getQueryExecutionState().hasQueryExecutionFinished()) {
onQueryCompletion(query_manager);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c4f4b285/query_execution/QueryExecutionState.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryExecutionState.hpp b/query_execution/QueryExecutionState.hpp
index f5281d5..a4273dc 100644
--- a/query_execution/QueryExecutionState.hpp
+++ b/query_execution/QueryExecutionState.hpp
@@ -22,6 +22,12 @@
#include <cstddef>
#include <unordered_map>
+
+#include "query_optimizer/QueryOptimizerConfig.h" // For QUICKSTEP_DISTRIBUTED.
+#ifdef QUICKSTEP_DISTRIBUTED
+#include <unordered_set>
+#endif // QUICKSTEP_DISTRIBUTED
+
#include <utility>
#include <vector>
@@ -103,6 +109,49 @@ class QueryExecutionState {
}
}
+#ifdef QUICKSTEP_DISTRIBUTED
+ /**
+ * @brief Update the rebuild status of the given operator the number of
+ * pending rebuild work orders, after the rebuild has been initiated.
+ *
+ * @param operator_index The index of the given operator.
+ * @param num_rebuild_workorders The number of rebuild workorders of the given
+ * operator.
+ * @param shiftboss_index The index of the Shiftboss that rebuilt.
+ **/
+ void updateRebuildStatus(const std::size_t operator_index,
+ const std::size_t num_rebuild_workorders,
+ const std::size_t shiftboss_index) {
+ DCHECK_LT(operator_index, num_operators_);
+ auto search_res = rebuild_status_.find(operator_index);
+ DCHECK(search_res != rebuild_status_.end() && search_res->second.has_initiated);
+ search_res->second.num_pending_workorders += num_rebuild_workorders;
+ search_res->second.rebuilt_shiftboss_indexes.insert(shiftboss_index);
+ }
+
+ /**
+ * @brief Check if the rebuild has been finished for the given operator.
+ *
+ * @param operator_index The index of the given operator.
+ * @param num_shiftbosses The number of the Shiftbosses for rebuilt.
+ *
+ * @return True if the rebuild has been finished, false otherwise.
+ **/
+ inline bool hasRebuildFinished(const std::size_t operator_index,
+ const std::size_t num_shiftbosses) const {
+ DCHECK_LT(operator_index, num_operators_);
+ const auto search_res = rebuild_status_.find(operator_index);
+ DCHECK(search_res != rebuild_status_.end());
+
+ const auto &rebuild_status = search_res->second;
+ DCHECK(rebuild_status.has_initiated);
+
+ return rebuild_status.rebuilt_shiftboss_indexes.size() == num_shiftbosses &&
+ rebuild_status.num_pending_workorders == 0u;
+ }
+
+#endif // QUICKSTEP_DISTRIBUTED
+
/**
* @brief Check if the rebuild has been initiated for the given operator.
*
@@ -314,6 +363,10 @@ class QueryExecutionState {
// The number of pending rebuild workorders for the operator.
// Valid if and only if 'has_initiated' is true.
std::size_t num_pending_workorders;
+
+#ifdef QUICKSTEP_DISTRIBUTED
+ std::unordered_set<std::size_t> rebuilt_shiftboss_indexes;
+#endif // QUICKSTEP_DISTRIBUTED
};
// Key is dag_node_index for which rebuild is required.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c4f4b285/query_execution/QueryManagerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.cpp b/query_execution/QueryManagerDistributed.cpp
index 7d45933..20650d0 100644
--- a/query_execution/QueryManagerDistributed.cpp
+++ b/query_execution/QueryManagerDistributed.cpp
@@ -36,7 +36,9 @@
#include "glog/logging.h"
+#include "tmb/address.h"
#include "tmb/id_typedefs.h"
+#include "tmb/tagged_message.h"
using std::free;
using std::malloc;
@@ -125,16 +127,16 @@ bool QueryManagerDistributed::fetchNormalWorkOrders(const dag_node_index index)
}
void QueryManagerDistributed::processInitiateRebuildResponseMessage(const dag_node_index op_index,
- const std::size_t num_rebuild_work_orders) {
- // TODO(zuyu): Multiple Shiftbosses support.
- query_exec_state_->setRebuildStatus(op_index, num_rebuild_work_orders, true);
+ const std::size_t num_rebuild_work_orders,
+ const std::size_t shiftboss_index) {
+ query_exec_state_->updateRebuildStatus(op_index, num_rebuild_work_orders, shiftboss_index);
- if (num_rebuild_work_orders != 0u) {
+ if (!query_exec_state_->hasRebuildFinished(op_index, shiftboss_directory_->size())) {
// Wait for the rebuild work orders to finish.
return;
}
- // No needs for rebuilds.
+ // No needs for rebuilds, or the rebuild has finished.
markOperatorFinished(op_index);
for (const std::pair<dag_node_index, bool> &dependent_link :
@@ -168,17 +170,20 @@ bool QueryManagerDistributed::initiateRebuild(const dag_node_index index) {
kInitiateRebuildMessage);
free(proto_bytes);
+ // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
+ tmb::Address shiftboss_addresses;
+ for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
+ shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
+ }
+
LOG(INFO) << "ForemanDistributed sent InitiateRebuildMessage (typed '" << kInitiateRebuildMessage
- << "') to Shiftboss";
- // TODO(zuyu): Multiple workers support.
- QueryExecutionUtil::SendTMBMessage(bus_,
- foreman_client_id_,
- shiftboss_directory_->getClientId(0),
- move(tagged_msg));
-
- // The negative value indicates that the number of rebuild work orders is to be
- // determined.
- query_exec_state_->setRebuildStatus(index, -1, true);
+ << "') to all Shiftbosses";
+ QueryExecutionUtil::BroadcastMessage(foreman_client_id_,
+ shiftboss_addresses,
+ move(tagged_msg),
+ bus_);
+
+ query_exec_state_->setRebuildStatus(index, 0, true);
// Wait for Shiftbosses to report the number of rebuild work orders.
return false;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c4f4b285/query_execution/QueryManagerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.hpp b/query_execution/QueryManagerDistributed.hpp
index e609ab8..f8ac53c 100644
--- a/query_execution/QueryManagerDistributed.hpp
+++ b/query_execution/QueryManagerDistributed.hpp
@@ -73,9 +73,11 @@ class QueryManagerDistributed final : public QueryManagerBase {
* for initiating the rebuild work order.
* @param num_rebuild_work_orders The number of the rebuild work orders
* generated for the operator indexed by 'op_index'.
+ * @param shiftboss_index The index of the Shiftboss that sends the message.
**/
void processInitiateRebuildResponseMessage(const dag_node_index op_index,
- const std::size_t num_rebuild_work_orders);
+ const std::size_t num_rebuild_work_orders,
+ const std::size_t shiftboss_index);
/**
* @brief Get the next normal workorder to be excuted, wrapped in a
[4/4] incubator-quickstep git commit: Scheduled WorkOrders w/ the
same aggr_state_index on the same Shiftboss.
Posted by zu...@apache.org.
Scheduled WorkOrders w/ the same aggr_state_index on the same Shiftboss.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/c64d60f7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/c64d60f7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/c64d60f7
Branch: refs/heads/multi-shiftboss-test
Commit: c64d60f7a1844bd1e15de2decfd40cb434579f74
Parents: e75c265
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 27 17:04:30 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 17:04:30 2016 -0800
----------------------------------------------------------------------
query_execution/ForemanDistributed.cpp | 61 ++++++++++++++++------
query_execution/ForemanDistributed.hpp | 4 ++
query_execution/PolicyEnforcerDistributed.cpp | 12 +++++
query_execution/PolicyEnforcerDistributed.hpp | 18 +++++++
query_execution/QueryManagerDistributed.hpp | 23 ++++++++
5 files changed, 102 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c64d60f7/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index fc9cd3c..61f0603 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -243,6 +243,32 @@ bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id messag
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;
+
+ switch (work_order_proto.work_order_type()) {
+ case S::AGGREGATION:
+ aggr_state_index = work_order_proto.GetExtension(S::AggregationWorkOrder::aggr_state_index);
+ break;
+ case S::FINALIZE_AGGREGATION:
+ aggr_state_index = work_order_proto.GetExtension(S::FinalizeAggregationWorkOrder::aggr_state_index);
+ break;
+ case S::DESTROY_AGGREGATION_STATE:
+ aggr_state_index = work_order_proto.GetExtension(S::DestroyAggregationStateWorkOrder::aggr_state_index);
+ break;
+ default:
+ return false;
+ }
+
+ static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->getShiftbossIndexForAggregation(
+ proto.query_id(), aggr_state_index, 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) {
@@ -251,13 +277,13 @@ bool ForemanDistributed::isHashJoinRelatedWorkOrder(const S::WorkOrderMessage &p
switch (work_order_proto.work_order_type()) {
case S::BUILD_HASH:
- join_hash_table_index = work_order_proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index);
- break;
- case S::DESTROY_HASH:
- join_hash_table_index = work_order_proto.GetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index);
+ join_hash_table_index = work_order_proto.GetExtension(S::BuildHashWorkOrder::join_hash_table_index);
break;
case S::HASH_JOIN:
- join_hash_table_index = work_order_proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index);
+ join_hash_table_index = work_order_proto.GetExtension(S::HashJoinWorkOrder::join_hash_table_index);
+ break;
+ case S::DESTROY_HASH:
+ join_hash_table_index = work_order_proto.GetExtension(S::DestroyHashWorkOrder::join_hash_table_index);
break;
default:
return false;
@@ -275,20 +301,23 @@ void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::Wo
for (const auto &message : messages) {
DCHECK(message != nullptr);
const S::WorkOrderMessage &proto = *message;
- size_t shiftboss_index_for_hash_join;
- if (isHashJoinRelatedWorkOrder(proto, shiftboss_index, &shiftboss_index_for_hash_join)) {
- sendWorkOrderMessage(shiftboss_index_for_hash_join, proto);
- shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index_for_hash_join);
-
- if (shiftboss_index == shiftboss_index_for_hash_join) {
- shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
- }
+ size_t shiftboss_index_for_particular_work_order_type;
+ 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 {
- sendWorkOrderMessage(shiftboss_index, proto);
- shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
-
// TODO(zuyu): Take data-locality into account for scheduling.
+ 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_index = (shiftboss_index + 1) % num_shiftbosses;
+ } 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.
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c64d60f7/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index 0616f30..34bac07 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -71,6 +71,10 @@ class ForemanDistributed final : public ForemanBase {
void run() override;
private:
+ bool isAggregationRelatedWorkOrder(const serialization::WorkOrderMessage &proto,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index_for_aggregation);
+
bool isHashJoinRelatedWorkOrder(const serialization::WorkOrderMessage &proto,
const std::size_t next_shiftboss_index_to_schedule,
std::size_t *shiftboss_index_for_hash_join);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c64d60f7/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index 86b36c8..c5642bc 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -158,6 +158,18 @@ void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb:
}
}
+void PolicyEnforcerDistributed::getShiftbossIndexForAggregation(
+ const std::size_t query_id,
+ const QueryContext::aggregation_state_id aggr_state_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+ QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
+ query_manager->getShiftbossIndexForAggregation(aggr_state_index,
+ next_shiftboss_index_to_schedule,
+ shiftboss_index);
+}
+
void PolicyEnforcerDistributed::getShiftbossIndexForHashJoin(
const std::size_t query_id,
const QueryContext::join_hash_table_id join_hash_table_index,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c64d60f7/query_execution/PolicyEnforcerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.hpp b/query_execution/PolicyEnforcerDistributed.hpp
index 37326bd..e8bc394 100644
--- a/query_execution/PolicyEnforcerDistributed.hpp
+++ b/query_execution/PolicyEnforcerDistributed.hpp
@@ -90,6 +90,24 @@ class PolicyEnforcerDistributed final : public PolicyEnforcerBase {
void processInitiateRebuildResponseMessage(const tmb::TaggedMessage &tagged_message);
/**
+ * @brief Get or set the index of Shiftboss for an Aggregation related
+ * WorkOrder. If it is the first Aggregation on <aggr_state_index>,
+ * <shiftboss_index> will be set to <next_shiftboss_index_to_schedule>.
+ * Otherwise, <shiftboss_index> will be set to the index of the Shiftboss that
+ * has executed the first Aggregation.
+ *
+ * @param query_id The query id.
+ * @param aggr_state_index The Hash Table for the Aggregation.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForAggregation(
+ const std::size_t query_id,
+ const QueryContext::aggregation_state_id aggr_state_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index);
+
+ /**
* @brief Get or set the index of Shiftboss for a HashJoin related WorkOrder.
* If it is the first BuildHash on <join_hash_table_index>, <shiftboss_index>
* will be set to <next_shiftboss_index_to_schedule>. Otherwise,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c64d60f7/query_execution/QueryManagerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.hpp b/query_execution/QueryManagerDistributed.hpp
index 2b21303..6455bf7 100644
--- a/query_execution/QueryManagerDistributed.hpp
+++ b/query_execution/QueryManagerDistributed.hpp
@@ -95,6 +95,26 @@ class QueryManagerDistributed final : public QueryManagerBase {
const dag_node_index start_operator_index);
/**
+ * @brief Get the index of Shiftboss for an Aggregation related WorkOrder. If
+ * the Shiftboss index is not found, set using <next_shiftboss_index_to_schedule>.
+ *
+ * @param aggr_state_index The Hash Table for the Aggregation.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForAggregation(const QueryContext::aggregation_state_id aggr_state_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ const auto cit = shiftboss_indexes_for_aggrs_.find(aggr_state_index);
+ if (cit != shiftboss_indexes_for_aggrs_.end()) {
+ *shiftboss_index = cit->second;
+ } else {
+ shiftboss_indexes_for_aggrs_.emplace(aggr_state_index, next_shiftboss_index_to_schedule);
+ *shiftboss_index = next_shiftboss_index_to_schedule;
+ }
+ }
+
+ /**
* @brief Get the index of Shiftboss for a HashJoin related WorkOrder. If the
* Shiftboss index is not found, set using <next_shiftboss_index_to_schedule>.
*
@@ -136,6 +156,9 @@ class QueryManagerDistributed final : public QueryManagerBase {
std::unique_ptr<WorkOrderProtosContainer> normal_workorder_protos_container_;
+ // A map from a aggregation to its scheduled Shiftboss index.
+ std::unordered_map<QueryContext::aggregation_state_id, std::size_t> shiftboss_indexes_for_aggrs_;
+
// A map from a join hash table to its scheduled Shiftboss index.
std::unordered_map<QueryContext::join_hash_table_id, std::size_t> shiftboss_indexes_for_hash_joins_;
[3/4] incubator-quickstep git commit: Scheduled HashJoinWorkOrder on
the same Shiftboss of BuildHashWorkOrder.
Posted by zu...@apache.org.
Scheduled HashJoinWorkOrder on the same Shiftboss of BuildHashWorkOrder.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/e75c265e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/e75c265e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/e75c265e
Branch: refs/heads/multi-shiftboss-test
Commit: e75c265eb279256d89f4fa634c542d566ce3cd7a
Parents: c4f4b28
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 13 20:20:56 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 27 16:45:22 2016 -0800
----------------------------------------------------------------------
query_execution/CMakeLists.txt | 3 ++
query_execution/ForemanDistributed.cpp | 47 ++++++++++++++++++++--
query_execution/ForemanDistributed.hpp | 4 ++
query_execution/PolicyEnforcerDistributed.cpp | 12 ++++++
query_execution/PolicyEnforcerDistributed.hpp | 19 +++++++++
query_execution/QueryManagerDistributed.hpp | 25 ++++++++++++
6 files changed, 106 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e75c265e/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 719d9f3..1f7add8 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -108,10 +108,12 @@ if (ENABLE_DISTRIBUTED)
quickstep_queryexecution_ForemanBase
quickstep_queryexecution_PolicyEnforcerBase
quickstep_queryexecution_PolicyEnforcerDistributed
+ quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
quickstep_queryexecution_ShiftbossDirectory
+ quickstep_relationaloperators_WorkOrder_proto
quickstep_threading_ThreadUtil
quickstep_utility_EqualsAnyConstant
quickstep_utility_Macros
@@ -153,6 +155,7 @@ if (ENABLE_DISTRIBUTED)
quickstep_catalog_CatalogRelation
quickstep_catalog_Catalog_proto
quickstep_queryexecution_PolicyEnforcerBase
+ quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryContext_proto
quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionState
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e75c265e/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 7dccce4..fc9cd3c 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -30,10 +30,12 @@
#include "query_execution/AdmitRequestMessage.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 "threading/ThreadUtil.hpp"
#include "utility/EqualsAnyConstant.hpp"
@@ -241,16 +243,53 @@ bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id messag
kWorkOrderFeedbackMessage);
}
+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;
+
+ switch (work_order_proto.work_order_type()) {
+ case S::BUILD_HASH:
+ join_hash_table_index = work_order_proto.GetExtension(serialization::BuildHashWorkOrder::join_hash_table_index);
+ break;
+ case S::DESTROY_HASH:
+ join_hash_table_index = work_order_proto.GetExtension(serialization::DestroyHashWorkOrder::join_hash_table_index);
+ break;
+ case S::HASH_JOIN:
+ join_hash_table_index = work_order_proto.GetExtension(serialization::HashJoinWorkOrder::join_hash_table_index);
+ break;
+ default:
+ return false;
+ }
+
+ static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->getShiftbossIndexForHashJoin(
+ proto.query_id(), join_hash_table_index, next_shiftboss_index_to_schedule, shiftboss_index_for_hash_join);
+
+ return true;
+}
+
void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::WorkOrderMessage>> &messages) {
const size_t num_shiftbosses = shiftboss_directory_.size();
size_t shiftboss_index = 0u;
for (const auto &message : messages) {
DCHECK(message != nullptr);
- sendWorkOrderMessage(shiftboss_index, *message);
- shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
+ const S::WorkOrderMessage &proto = *message;
+ size_t shiftboss_index_for_hash_join;
+ if (isHashJoinRelatedWorkOrder(proto, shiftboss_index, &shiftboss_index_for_hash_join)) {
+ sendWorkOrderMessage(shiftboss_index_for_hash_join, proto);
+ shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index_for_hash_join);
+
+ if (shiftboss_index == shiftboss_index_for_hash_join) {
+ shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
+ }
+ } else {
+ sendWorkOrderMessage(shiftboss_index, proto);
+ shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
- // TO(zuyu): Take data-locality into account for scheduling.
- shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
+ // TODO(zuyu): Take data-locality into account for scheduling.
+ shiftboss_index = (shiftboss_index + 1) % num_shiftbosses;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e75c265e/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index ccdd0ae..0616f30 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -71,6 +71,10 @@ class ForemanDistributed final : public ForemanBase {
void run() override;
private:
+ bool isHashJoinRelatedWorkOrder(const serialization::WorkOrderMessage &proto,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index_for_hash_join);
+
/**
* @brief Dispatch schedulable WorkOrders, wrapped in WorkOrderMessages to the
* worker threads.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e75c265e/query_execution/PolicyEnforcerDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.cpp b/query_execution/PolicyEnforcerDistributed.cpp
index 6e09ea8..86b36c8 100644
--- a/query_execution/PolicyEnforcerDistributed.cpp
+++ b/query_execution/PolicyEnforcerDistributed.cpp
@@ -158,6 +158,18 @@ void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb:
}
}
+void PolicyEnforcerDistributed::getShiftbossIndexForHashJoin(
+ const std::size_t query_id,
+ const QueryContext::join_hash_table_id join_hash_table_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
+ QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
+ query_manager->getShiftbossIndexForHashJoin(join_hash_table_index,
+ next_shiftboss_index_to_schedule,
+ shiftboss_index);
+}
+
void PolicyEnforcerDistributed::initiateQueryInShiftboss(QueryHandle *query_handle) {
S::QueryInitiateMessage proto;
proto.set_query_id(query_handle->query_id());
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e75c265e/query_execution/PolicyEnforcerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.hpp b/query_execution/PolicyEnforcerDistributed.hpp
index 146e9af..37326bd 100644
--- a/query_execution/PolicyEnforcerDistributed.hpp
+++ b/query_execution/PolicyEnforcerDistributed.hpp
@@ -20,6 +20,7 @@
#include <vector>
#include "query_execution/PolicyEnforcerBase.hpp"
+#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/ShiftbossDirectory.hpp"
#include "utility/Macros.hpp"
@@ -88,6 +89,24 @@ class PolicyEnforcerDistributed final : public PolicyEnforcerBase {
**/
void processInitiateRebuildResponseMessage(const tmb::TaggedMessage &tagged_message);
+ /**
+ * @brief Get or set the index of Shiftboss for a HashJoin related WorkOrder.
+ * If it is the first BuildHash on <join_hash_table_index>, <shiftboss_index>
+ * will be set to <next_shiftboss_index_to_schedule>. Otherwise,
+ * <shiftboss_index> will be set to the index of the Shiftboss that has
+ * executed the first BuildHash.
+ *
+ * @param query_id The query id.
+ * @param join_hash_table_index The Hash Table for the Join.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForHashJoin(
+ const std::size_t query_id,
+ const QueryContext::join_hash_table_id join_hash_table_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index);
+
private:
void decrementNumQueuedWorkOrders(const serialization::WorkOrderCompletionMessage &proto) override {
shiftboss_directory_->decrementNumQueuedWorkOrders(proto.shiftboss_index());
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e75c265e/query_execution/QueryManagerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryManagerDistributed.hpp b/query_execution/QueryManagerDistributed.hpp
index f8ac53c..2b21303 100644
--- a/query_execution/QueryManagerDistributed.hpp
+++ b/query_execution/QueryManagerDistributed.hpp
@@ -22,7 +22,9 @@
#include <cstddef>
#include <memory>
+#include <unordered_map>
+#include "query_execution/QueryContext.hpp"
#include "query_execution/QueryExecutionState.hpp"
#include "query_execution/QueryManagerBase.hpp"
#include "query_execution/WorkOrderProtosContainer.hpp"
@@ -92,6 +94,26 @@ class QueryManagerDistributed final : public QueryManagerBase {
serialization::WorkOrderMessage* getNextWorkOrderMessage(
const dag_node_index start_operator_index);
+ /**
+ * @brief Get the index of Shiftboss for a HashJoin related WorkOrder. If the
+ * Shiftboss index is not found, set using <next_shiftboss_index_to_schedule>.
+ *
+ * @param join_hash_table_index The Hash Table for the Join.
+ * @param next_shiftboss_index The index of Shiftboss to schedule a next WorkOrder.
+ * @param shiftboss_index The index of Shiftboss to schedule the WorkOrder.
+ **/
+ void getShiftbossIndexForHashJoin(const QueryContext::join_hash_table_id join_hash_table_index,
+ const std::size_t next_shiftboss_index_to_schedule,
+ std::size_t *shiftboss_index) {
+ const auto cit = shiftboss_indexes_for_hash_joins_.find(join_hash_table_index);
+ if (cit != shiftboss_indexes_for_hash_joins_.end()) {
+ *shiftboss_index = cit->second;
+ } else {
+ shiftboss_indexes_for_hash_joins_.emplace(join_hash_table_index, next_shiftboss_index_to_schedule);
+ *shiftboss_index = next_shiftboss_index_to_schedule;
+ }
+ }
+
private:
bool checkNormalExecutionOver(const dag_node_index index) const override {
return (checkAllDependenciesMet(index) &&
@@ -114,6 +136,9 @@ class QueryManagerDistributed final : public QueryManagerBase {
std::unique_ptr<WorkOrderProtosContainer> normal_workorder_protos_container_;
+ // A map from a join hash table to its scheduled Shiftboss index.
+ std::unordered_map<QueryContext::join_hash_table_id, std::size_t> shiftboss_indexes_for_hash_joins_;
+
DISALLOW_COPY_AND_ASSIGN(QueryManagerDistributed);
};