You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by hb...@apache.org on 2016/06/09 16:13:07 UTC
[4/4] incubator-quickstep git commit: Created QueryManager class and
tests.
Created QueryManager class and tests.
- Responsible for managing execution state, generation of work orders of
the query and providing work orders for execution.
- Tests for QueryManager, which are simplified versions of the Foreman
tests.
WorkerMessage enum type change in Foreman.
Replaced DCHECK_NOTNULL with DCHECK.
Query ID can be set for every WorkOrder.
Query ID can now be added to WorkOrders in WorkOrdersContainer.
Added query ID field in Work order completion message.
- Set before the worker sends the completion message.
QueryManager unittest fix.
Added query ID field to messages, query context and InsertDestination.
- WorkOrderCompletion and DataPipelineMessage have query ID field.
- InsertDestination proto and query context have query ID field.
- ExecutionGenerator sets the query ID in InsertDestination.
Added query ID to WorkOrdersAvailable message.
Added query ID to CatalogRelationNewBlockMessage.
Minor style fix in CMakeLists.
Created PolicyEnforcer class.
- First point of entry for queries in the scheduler.
- Can perform admission control.
- Can talk to the QueryManagers of the active queries to provide them
messages to process and collect work orders for execution from them.
Support for admitting multiple queries to the PolicyEnforcer.
- Function to remove a query made public.
Added a new class for sending queries to the Foreman.
Minor error fix in PolicyEnforcer
Long lived Foreman thread
- Foreman thread lives through the lifetime of the Quickstep process.
- Foreman and main thread communicate through TMB messages.
- Foreman admits queries and routes them to PolicyEnforcer.
- Foreman relays messages to policy enforcer which in turn processes it,
based on the query ID of the message.
- All the tests modified accordingly.
IWYU and lint fixes.
Code review changes.
Partition aware selection using NUMA-awareness (#175)
Link: https://github.com/pivotalsoftware/quickstep/pull/175
Fixes bug which disallows the creation of BITWEAVING_H indices. (#204)
Fixes a bug where numa settings were not properly set (not compiled in) (#205)
Change default aggregate_hashtable_type from LinearOpenAddressing to SeparateChaining (#207)
Revert "Change default aggregate_hashtable_type from LinearOpenAddressing to SeparateChaining" (#208)
Transaction Part 4: LockManager, CycleDetector and DeadLockDetector. (#187)
Change default aggregate_hashtable_type from LinearOpenAddressing to SeparateChaining (#210)
Added support for the substring function. (#211)
Cleanup in Preloader class (#209)
Added print_query gflag (#213)
Refactor hasNUMAPlacementScheme. (#215)
Added a method for double parsing with significant number argument. (#216)
Leave 10% more memory for the OS. (#218)
Added a quick fix to the count(*) problem (#220)
Fixed a g++ compiler warning (#221)
Minor improvements in FileManagers. (#224)
Quickstep print catalog (#222)
Change the default value of Joined Tuple Collector (#226)
On NUMA boxes, the vector based joined tuple collector slows down queries.
Initial support for collecting table statistics: number of distinct values (#227)
Added hash join order optimization for star schema queries. (#229)
In ScopedBuffer, initialize the allocated memory bytes to 0 (#230)
Quickstep gen stats (#225)
Revert "Quickstep gen stats" (#231)
Reverting the PR that gets the number of rows (#233)
Resolves SMA bugs revealed by TPCH. (#234)
Revert "Explicitly specify where tcmalloc comes from" (#237)
Groupby hashtable pool (#236)
- Created a HashTablePool class for group by clause.
- Each thread can checkout it's own hash table while doing group by
aggregation.
- AggregationOperationState uses one hash table pool per group by
clause.
QUICKSTEP-1: Add incubation disclaimer to README.md.
This closes #2
Fixed Travis CI link in Readme.
Added BlockLocator.
Added Async DataExchange Service.
Bug fixed in isFull() function.
- Earlier check for isFull() was that whether there is at least one
empty bucket available for insertion.
- Now isFull() can check if there is enough space adding specified
number of buckets.
- To reproduce the bug behind this fix: Run TPC-H Q2 on SF100 dataset after
running the \analyze command. The optimizer allocates smaller than
necessary space for one of the hash joins. The resize() is triggered
but never gets executed because of the restrictive isFull() condition.
Added query ID to relational operators and Aggregation op.
Added query ID to WorkOrder base class and Agg work order.
Added query ID to BuildHash operator and WorkOrders.
Checking if Travis Ubuntu uses pre-built clang-3.7 instead of downloading.
Added query ID to hash join operator.
Query ID added in CreateTable, Index, Delete and DestroyHash
Added query ID to all the remaining operators and work orders.
Doxygen and minor bug fixes.
Manually resetting the changes in the travis YAML.
Reordered query ID in aggregation operator
Reordered query ID in operators and work orders.
Reordered query ID in CreateIndex.
Reordered query ID in delete.
Reordered query ID in destroy hash.
Reordered query ID in Drop table.
Reordered query ID in Finalize aggregate
Reordered query ID in hash join
Reordered query ID in insert op
Reordered query ID in nested loops
Reordered query ID in rebuild
Reordered query ID in sample
Reordered query ID in save blocks
Reordered query ID in select
Reordered query ID in sort merge
Reordered query ID in sort run
Reordered query ID in table gen
Reorder query ID in text scan
Reordered query ID in update
Changes during the rebase with origin master.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/a690455e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a690455e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a690455e
Branch: refs/heads/query-manager-used-in-foreman
Commit: a690455ec963d670a3770a3238cafee3e66fafb7
Parents: fd75e17
Author: Harshad Deshmukh <ha...@cs.wisc.edu>
Authored: Sat Apr 9 15:08:40 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Thu Jun 9 11:09:38 2016 -0500
----------------------------------------------------------------------
CMakeLists.txt | 2 +
cli/CommandExecutor.cpp | 53 +-
cli/CommandExecutor.hpp | 11 +-
cli/QuickstepCli.cpp | 70 +--
cli/tests/CMakeLists.txt | 2 +
cli/tests/CommandExecutorTestRunner.cpp | 29 +-
cli/tests/CommandExecutorTestRunner.hpp | 37 +-
query_execution/AdmitRequestMessage.hpp | 73 +++
query_execution/CMakeLists.txt | 68 +--
query_execution/Foreman.cpp | 578 +++++--------------
query_execution/Foreman.hpp | 394 +------------
query_execution/PolicyEnforcer.cpp | 177 ++++++
query_execution/PolicyEnforcer.hpp | 168 ++++++
query_execution/QueryContext.cpp | 3 +-
query_execution/QueryContext.hpp | 9 +
query_execution/QueryContext.proto | 2 +
query_execution/QueryExecutionMessages.proto | 4 +
query_execution/QueryExecutionTypedefs.hpp | 4 +-
query_execution/QueryExecutionUtil.hpp | 52 ++
query_execution/QueryManager.hpp | 5 +-
query_execution/WorkOrdersContainer.hpp | 70 +--
query_execution/Worker.cpp | 11 +-
query_execution/Worker.hpp | 2 +
query_execution/WorkerMessage.hpp | 22 +-
query_execution/tests/QueryManager_unittest.cpp | 6 +
.../tests/WorkOrdersContainer_unittest.cpp | 26 +
query_optimizer/ExecutionGenerator.cpp | 32 +-
query_optimizer/ExecutionGenerator.hpp | 8 +
query_optimizer/tests/CMakeLists.txt | 2 +
.../tests/ExecutionGeneratorTestRunner.cpp | 23 +-
.../tests/ExecutionGeneratorTestRunner.hpp | 33 +-
query_optimizer/tests/TestDatabaseLoader.cpp | 1 +
relational_operators/AggregationOperator.hpp | 2 +
relational_operators/DeleteOperator.cpp | 1 +
relational_operators/DeleteOperator.hpp | 1 +
relational_operators/DestroyHashOperator.hpp | 4 +
relational_operators/DropTableOperator.hpp | 2 +
relational_operators/HashJoinOperator.hpp | 1 +
relational_operators/RebuildWorkOrder.hpp | 1 +
relational_operators/SortMergeRunOperator.cpp | 1 +
relational_operators/TableGeneratorOperator.hpp | 2 +
relational_operators/TextScanOperator.cpp | 3 +
relational_operators/TextScanOperator.hpp | 1 +
relational_operators/UpdateOperator.cpp | 1 +
relational_operators/UpdateOperator.hpp | 1 +
relational_operators/WorkOrder.hpp | 20 +-
.../tests/AggregationOperator_unittest.cpp | 7 +
.../tests/HashJoinOperator_unittest.cpp | 12 +
.../tests/SortMergeRunOperator_unittest.cpp | 6 +
.../SortRunGenerationOperator_unittest.cpp | 2 +
.../tests/TextScanOperator_unittest.cpp | 2 +
storage/InsertDestination.cpp | 49 +-
storage/InsertDestination.hpp | 58 +-
storage/InsertDestination.proto | 1 +
54 files changed, 1107 insertions(+), 1048 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index ef7fd50..f674a78 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -711,9 +711,11 @@ target_link_libraries(quickstep_cli_shell
quickstep_cli_PrintToScreen
quickstep_parser_ParseStatement
quickstep_parser_SqlParserWrapper
+ quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_Foreman
quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryExecutionTypedefs
+ quickstep_queryexecution_QueryExecutionUtil
quickstep_queryexecution_Worker
quickstep_queryexecution_WorkerDirectory
quickstep_queryexecution_WorkerMessage
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/cli/CommandExecutor.cpp
----------------------------------------------------------------------
diff --git a/cli/CommandExecutor.cpp b/cli/CommandExecutor.cpp
index ddcd38f..dc14741 100644
--- a/cli/CommandExecutor.cpp
+++ b/cli/CommandExecutor.cpp
@@ -51,6 +51,8 @@
#include "glog/logging.h"
+#include "tmb/id_typedefs.h"
+
using std::fprintf;
using std::fputc;
using std::fputs;
@@ -58,6 +60,8 @@ using std::size_t;
using std::string;
using std::vector;
+namespace tmb { class MessageBus; }
+
namespace quickstep {
namespace cli {
namespace {
@@ -194,11 +198,14 @@ void executeDescribeTable(
/**
* @brief A helper function that executes a SQL query to obtain a scalar result.
*/
-inline TypedValue executeQueryForSingleResult(const std::string &query_string,
- StorageManager *storage_manager,
- QueryProcessor *query_processor,
- SqlParserWrapper *parser_wrapper,
- Foreman *foreman) {
+inline TypedValue executeQueryForSingleResult(
+ const tmb::client_id main_thread_client_id,
+ const tmb::client_id foreman_client_id,
+ const std::string &query_string,
+ tmb::MessageBus *bus,
+ StorageManager *storage_manager,
+ QueryProcessor *query_processor,
+ SqlParserWrapper *parser_wrapper) {
parser_wrapper->feedNextBuffer(new std::string(query_string));
ParseResult result = parser_wrapper->getNextStatement();
@@ -210,11 +217,8 @@ inline TypedValue executeQueryForSingleResult(const std::string &query_string,
DCHECK(query_handle->getQueryPlanMutable() != nullptr);
// Use foreman to execute the query plan.
- foreman->setQueryPlan(query_handle->getQueryPlanMutable()->getQueryPlanDAGMutable());
- foreman->reconstructQueryContextFromProto(query_handle->getQueryContextProto());
-
- foreman->start();
- foreman->join();
+ QueryExecutionUtil::ConstructAndSendAdmitRequestMessage(
+ main_thread_client_id, foreman_client_id, query_handle.get(), bus);
// Retrieve the scalar result from the result relation.
const CatalogRelation *query_result_relation = query_handle->getQueryResultRelation();
@@ -246,8 +250,10 @@ inline TypedValue executeQueryForSingleResult(const std::string &query_string,
return value;
}
-void executeAnalyze(QueryProcessor *query_processor,
- Foreman *foreman,
+void executeAnalyze(const tmb::client_id main_thread_client_id,
+ const tmb::client_id foreman_client_id,
+ MessageBus *bus,
+ QueryProcessor *query_processor,
FILE *out) {
const CatalogDatabase &database = *query_processor->getDefaultDatabase();
StorageManager *storage_manager = query_processor->getStorageManager();
@@ -273,11 +279,13 @@ void executeAnalyze(QueryProcessor *query_processor,
query_string.append(";");
TypedValue num_distinct_values =
- executeQueryForSingleResult(query_string,
+ executeQueryForSingleResult(main_thread_client_id,
+ foreman_client_id,
+ query_string,
+ bus,
storage_manager,
query_processor,
- parser_wrapper.get(),
- foreman);
+ parser_wrapper.get());
DCHECK(num_distinct_values.getTypeID() == TypeID::kLong);
mutable_relation->getStatisticsMutable()->setNumDistinctValues(
@@ -291,11 +299,13 @@ void executeAnalyze(QueryProcessor *query_processor,
query_string.append(";");
TypedValue num_tuples =
- executeQueryForSingleResult(query_string,
+ executeQueryForSingleResult(main_thread_client_id,
+ foreman_client_id,
+ query_string,
+ bus,
storage_manager,
query_processor,
- parser_wrapper.get(),
- foreman);
+ parser_wrapper.get());
DCHECK(num_tuples.getTypeID() == TypeID::kLong);
mutable_relation->getStatisticsMutable()->setNumTuples(
@@ -312,9 +322,11 @@ void executeAnalyze(QueryProcessor *query_processor,
void executeCommand(const ParseStatement &statement,
const CatalogDatabase &catalog_database,
+ const tmb::client_id main_thread_client_id,
+ const tmb::client_id foreman_client_id,
+ MessageBus *bus,
StorageManager *storage_manager,
QueryProcessor *query_processor,
- Foreman *foreman,
FILE *out) {
const ParseCommand &command = static_cast<const ParseCommand &>(statement);
const PtrVector<ParseString> *arguments = command.arguments();
@@ -328,7 +340,8 @@ void executeCommand(const ParseStatement &statement,
executeDescribeTable(arguments, catalog_database, out);
}
} else if (command_str == C::kAnalyzeCommand) {
- executeAnalyze(query_processor, foreman, out);
+ executeAnalyze(
+ main_thread_client_id, foreman_client_id, bus, query_processor, out);
} else {
THROW_SQL_ERROR_AT(command.command()) << "Invalid Command";
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/cli/CommandExecutor.hpp
----------------------------------------------------------------------
diff --git a/cli/CommandExecutor.hpp b/cli/CommandExecutor.hpp
index c819981..3435aeb 100644
--- a/cli/CommandExecutor.hpp
+++ b/cli/CommandExecutor.hpp
@@ -21,10 +21,14 @@
#include <cstdio>
#include <string>
+#include "tmb/id_typedefs.h"
+
using std::fprintf;
using std::fputc;
using std::string;
+namespace tmb { class MessageBus; }
+
namespace quickstep {
class CatalogDatabase;
@@ -53,6 +57,9 @@ constexpr char kAnalyzeCommand[] = "\\analyze";
*
* @param statement The parsed statement from the cli.
* @param catalog_database The catalog information about the current database.
+ * @param main_thread_client_id The TMB client ID of the main thread.
+ * @param foreman_client_id The TMB client ID of the Foreman thread.
+ * @param bus A pointer to the TMB.
* @param storage_manager The current StorageManager.
* @param query_processor The query processor to generate plans for SQL queries.
* @param foreman The foreman to execute query plans.
@@ -60,9 +67,11 @@ constexpr char kAnalyzeCommand[] = "\\analyze";
*/
void executeCommand(const ParseStatement &statement,
const CatalogDatabase &catalog_database,
+ const tmb::client_id main_thread_client_id,
+ const tmb::client_id foreman_client_id,
+ tmb::MessageBus *bus,
StorageManager *storage_manager,
QueryProcessor *query_processor,
- Foreman *foreman,
FILE *out);
/** @} */
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index 558d6eb..3202c96 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -53,8 +53,10 @@ typedef quickstep::LineReaderDumb LineReaderImpl;
#include "cli/PrintToScreen.hpp"
#include "parser/ParseStatement.hpp"
#include "parser/SqlParserWrapper.hpp"
+#include "query_execution/AdmitRequestMessage.hpp"
#include "query_execution/Foreman.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
+#include "query_execution/QueryExecutionUtil.hpp"
#include "query_execution/Worker.hpp"
#include "query_execution/WorkerDirectory.hpp"
#include "query_execution/WorkerMessage.hpp"
@@ -95,6 +97,7 @@ using std::string;
using std::vector;
using quickstep::Address;
+using quickstep::AdmitRequestMessage;
using quickstep::CatalogRelation;
using quickstep::DefaultsConfigurator;
using quickstep::DropRelation;
@@ -107,6 +110,7 @@ using quickstep::ParseResult;
using quickstep::ParseStatement;
using quickstep::PrintToScreen;
using quickstep::PtrVector;
+using quickstep::QueryExecutionUtil;
using quickstep::QueryHandle;
using quickstep::QueryPlan;
using quickstep::QueryProcessor;
@@ -115,9 +119,12 @@ using quickstep::TaggedMessage;
using quickstep::Worker;
using quickstep::WorkerDirectory;
using quickstep::WorkerMessage;
+using quickstep::kAdmitRequestMessage;
using quickstep::kPoisonMessage;
+using quickstep::kWorkloadCompletionMessage;
using tmb::client_id;
+using tmb::AnnotatedMessage;
namespace quickstep {
@@ -197,7 +204,9 @@ int main(int argc, char* argv[]) {
// The TMB client id for the main thread, used to kill workers at the end.
const client_id main_thread_client_id = bus.Connect();
+ bus.RegisterClientAsSender(main_thread_client_id, kAdmitRequestMessage);
bus.RegisterClientAsSender(main_thread_client_id, kPoisonMessage);
+ bus.RegisterClientAsReceiver(main_thread_client_id, kWorkloadCompletionMessage);
// Setup the paths used by StorageManager.
string fixed_storage_path(quickstep::FLAGS_storage_path);
@@ -283,12 +292,6 @@ int main(int argc, char* argv[]) {
std::chrono::duration<double>(preload_end - preload_start).count());
}
- Foreman foreman(&bus,
- query_processor->getDefaultDatabase(),
- query_processor->getStorageManager(),
- -1, /* CPU id to bind foreman. -1 is unbound. */
- num_numa_nodes_system);
-
// Get the NUMA affinities for workers.
vector<int> cpu_numa_nodes = InputParserUtil::GetNUMANodesForCPUs();
if (cpu_numa_nodes.empty()) {
@@ -323,13 +326,20 @@ int main(int argc, char* argv[]) {
worker_client_ids,
worker_numa_nodes);
- foreman.setWorkerDirectory(&worker_directory);
+ Foreman foreman(main_thread_client_id,
+ &worker_directory,
+ &bus,
+ query_processor->getDefaultDatabase(),
+ query_processor->getStorageManager(),
+ num_numa_nodes_system);
// Start the worker threads.
for (Worker &worker : workers) {
worker.start();
}
+ foreman.start();
+
LineReaderImpl line_reader("quickstep> ",
" ...> ");
std::unique_ptr<SqlParserWrapper> parser_wrapper(new SqlParserWrapper());
@@ -366,9 +376,11 @@ int main(int argc, char* argv[]) {
quickstep::cli::executeCommand(
*result.parsed_statement,
*(query_processor->getDefaultDatabase()),
+ main_thread_client_id,
+ foreman.getBusClientID(),
+ &bus,
query_processor->getStorageManager(),
query_processor.get(),
- &foreman,
stdout);
} catch (const quickstep::SqlError &sql_error) {
fprintf(stderr, "%s",
@@ -389,14 +401,20 @@ int main(int argc, char* argv[]) {
}
DCHECK(query_handle->getQueryPlanMutable() != nullptr);
- foreman.setQueryPlan(query_handle->getQueryPlanMutable()->getQueryPlanDAGMutable());
-
- foreman.reconstructQueryContextFromProto(query_handle->getQueryContextProto());
-
+ AdmitRequestMessage request_message(query_handle.get());
+ TaggedMessage admit_tagged_message(
+ &request_message, sizeof(request_message), kAdmitRequestMessage);
+
+ start = std::chrono::steady_clock::now();
+ QueryExecutionUtil::SendTMBMessage(&bus,
+ main_thread_client_id,
+ foreman.getBusClientID(),
+ std::move(admit_tagged_message));
try {
- start = std::chrono::steady_clock::now();
- foreman.start();
- foreman.join();
+ const AnnotatedMessage annotated_msg =
+ bus.Receive(main_thread_client_id, 0, true);
+ const TaggedMessage &tagged_message = annotated_msg.tagged_message;
+ DCHECK_EQ(kWorkloadCompletionMessage, tagged_message.message_type());
end = std::chrono::steady_clock::now();
const CatalogRelation *query_result_relation = query_handle->getQueryResultRelation();
@@ -440,29 +458,13 @@ int main(int argc, char* argv[]) {
}
}
- // Terminate all workers before exiting.
- // The main thread broadcasts poison message to the workers. Each worker dies
- // after receiving poison message. The order of workers' death is irrelavant.
- MessageStyle style;
- style.Broadcast(true);
- Address address;
- address.All(true);
- std::unique_ptr<WorkerMessage> poison_message(WorkerMessage::PoisonMessage());
- TaggedMessage poison_tagged_message(poison_message.get(),
- sizeof(*poison_message),
- kPoisonMessage);
-
- const tmb::MessageBus::SendStatus send_status =
- bus.Send(main_thread_client_id,
- address,
- style,
- std::move(poison_tagged_message));
- CHECK(send_status == tmb::MessageBus::SendStatus::kOK) <<
- "Broadcast message from Foreman to workers failed";
+ // Kill the foreman and workers.
+ QueryExecutionUtil::BroadcastPoisonMessage(main_thread_client_id, &bus);
for (Worker &worker : workers) {
worker.join();
}
+ foreman.join();
return 0;
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/cli/tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/cli/tests/CMakeLists.txt b/cli/tests/CMakeLists.txt
index ca37e4a..d177d6c 100644
--- a/cli/tests/CMakeLists.txt
+++ b/cli/tests/CMakeLists.txt
@@ -32,9 +32,11 @@ target_link_libraries(quickstep_cli_tests_CommandExecutorTest
quickstep_cli_PrintToScreen
quickstep_parser_ParseStatement
quickstep_parser_SqlParserWrapper
+ quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_Foreman
quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryExecutionTypedefs
+ quickstep_queryexecution_QueryExecutionUtil
quickstep_queryexecution_Worker
quickstep_queryexecution_WorkerDirectory
quickstep_queryexecution_WorkerMessage
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/cli/tests/CommandExecutorTestRunner.cpp
----------------------------------------------------------------------
diff --git a/cli/tests/CommandExecutorTestRunner.cpp b/cli/tests/CommandExecutorTestRunner.cpp
index 9cd493e..794f7e1 100644
--- a/cli/tests/CommandExecutorTestRunner.cpp
+++ b/cli/tests/CommandExecutorTestRunner.cpp
@@ -20,12 +20,15 @@
#include <cstdio>
#include <set>
#include <string>
+#include <utility>
#include "cli/CommandExecutor.hpp"
#include "cli/DropRelation.hpp"
#include "cli/PrintToScreen.hpp"
#include "parser/ParseStatement.hpp"
+#include "query_execution/AdmitRequestMessage.hpp"
#include "query_execution/Foreman.hpp"
+#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/Worker.hpp"
#include "query_optimizer/ExecutionGenerator.hpp"
#include "query_optimizer/LogicalGenerator.hpp"
@@ -41,6 +44,8 @@
#include "glog/logging.h"
+#include "tmb/tagged_message.h"
+
namespace quickstep {
class CatalogRelation;
@@ -87,9 +92,11 @@ void CommandExecutorTestRunner::runTestCase(
quickstep::cli::executeCommand(
*result.parsed_statement,
*(test_database_loader_.catalog_database()),
+ main_thread_client_id_,
+ foreman_->getBusClientID(),
+ &bus_,
test_database_loader_.storage_manager(),
nullptr,
- nullptr,
output_stream.file());
} else {
QueryHandle query_handle(optimizer_context.query_id());
@@ -100,14 +107,20 @@ void CommandExecutorTestRunner::runTestCase(
physical_generator.generatePlan(
logical_generator.generatePlan(*result.parsed_statement));
execution_generator.generatePlan(physical_plan);
- foreman_->setQueryPlan(
- query_handle.getQueryPlanMutable()->getQueryPlanDAGMutable());
-
- foreman_->reconstructQueryContextFromProto(query_handle.getQueryContextProto());
-
- foreman_->start();
- foreman_->join();
+ AdmitRequestMessage request_message(&query_handle);
+ TaggedMessage admit_tagged_message(
+ &request_message, sizeof(request_message), kAdmitRequestMessage);
+ QueryExecutionUtil::SendTMBMessage(&bus_,
+ main_thread_client_id_,
+ foreman_->getBusClientID(),
+ std::move(admit_tagged_message));
+
+ // Receive workload completion message from Foreman.
+ const AnnotatedMessage annotated_msg =
+ bus_.Receive(main_thread_client_id_, 0, true);
+ const TaggedMessage &tagged_message = annotated_msg.tagged_message;
+ DCHECK_EQ(kWorkloadCompletionMessage, tagged_message.message_type());
const CatalogRelation *query_result_relation = query_handle.getQueryResultRelation();
if (query_result_relation) {
PrintToScreen::PrintRelation(*query_result_relation,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/cli/tests/CommandExecutorTestRunner.hpp
----------------------------------------------------------------------
diff --git a/cli/tests/CommandExecutorTestRunner.hpp b/cli/tests/CommandExecutorTestRunner.hpp
index 94b1d6a..8fb5b65 100644
--- a/cli/tests/CommandExecutorTestRunner.hpp
+++ b/cli/tests/CommandExecutorTestRunner.hpp
@@ -27,6 +27,7 @@
#include "parser/SqlParserWrapper.hpp"
#include "query_execution/Foreman.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
+#include "query_execution/QueryExecutionUtil.hpp"
#include "query_execution/Worker.hpp"
#include "query_execution/WorkerDirectory.hpp"
#include "query_execution/WorkerMessage.hpp"
@@ -34,6 +35,9 @@
#include "utility/Macros.hpp"
#include "utility/textbased_test/TextBasedTestDriver.hpp"
+#include "tmb/id_typedefs.h"
+#include "tmb/message_bus.h"
+
namespace quickstep {
/**
@@ -57,9 +61,11 @@ class CommandExecutorTestRunner : public TextBasedTestRunner {
bus_.Initialize();
- foreman_.reset(new Foreman(&bus_,
- test_database_loader_.catalog_database(),
- test_database_loader_.storage_manager()));
+ main_thread_client_id_ = bus_.Connect();
+ bus_.RegisterClientAsSender(main_thread_client_id_, kAdmitRequestMessage);
+ bus_.RegisterClientAsSender(main_thread_client_id_, kPoisonMessage);
+ bus_.RegisterClientAsReceiver(main_thread_client_id_, kWorkloadCompletionMessage);
+
worker_.reset(new Worker(0, &bus_));
std::vector<client_id> worker_client_ids;
@@ -71,27 +77,20 @@ class CommandExecutorTestRunner : public TextBasedTestRunner {
workers_.reset(new WorkerDirectory(1 /* number of workers */,
worker_client_ids, numa_nodes));
- foreman_->setWorkerDirectory(workers_.get());
+ foreman_.reset(new Foreman(main_thread_client_id_,
+ workers_.get(),
+ &bus_,
+ test_database_loader_.catalog_database(),
+ test_database_loader_.storage_manager()));
+ foreman_->start();
worker_->start();
}
~CommandExecutorTestRunner() {
- std::unique_ptr<WorkerMessage> poison_message(WorkerMessage::PoisonMessage());
- TaggedMessage poison_tagged_message(poison_message.get(),
- sizeof(*poison_message),
- quickstep::kPoisonMessage);
-
- Address worker_address;
- MessageStyle single_receiver_style;
-
- worker_address.AddRecipient(worker_->getBusClientID());
- bus_.Send(foreman_->getBusClientID(),
- worker_address,
- single_receiver_style,
- std::move(poison_tagged_message));
-
+ QueryExecutionUtil::BroadcastPoisonMessage(main_thread_client_id_, &bus_);
worker_->join();
+ foreman_->join();
}
void runTestCase(const std::string &input,
@@ -102,6 +101,8 @@ class CommandExecutorTestRunner : public TextBasedTestRunner {
SqlParserWrapper sql_parser_;
optimizer::TestDatabaseLoader test_database_loader_;
+ tmb::client_id main_thread_client_id_;
+
MessageBusImpl bus_;
std::unique_ptr<Foreman> foreman_;
std::unique_ptr<Worker> worker_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/query_execution/AdmitRequestMessage.hpp
----------------------------------------------------------------------
diff --git a/query_execution/AdmitRequestMessage.hpp b/query_execution/AdmitRequestMessage.hpp
new file mode 100644
index 0000000..e33b354
--- /dev/null
+++ b/query_execution/AdmitRequestMessage.hpp
@@ -0,0 +1,73 @@
+/**
+ * Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ * University of Wisconsin\u2014Madison.
+ *
+ * Licensed 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.
+ **/
+
+#ifndef QUICKSTEP_QUERY_EXECUTION_ADMIT_REQUEST_MESSAGE_HPP_
+#define QUICKSTEP_QUERY_EXECUTION_ADMIT_REQUEST_MESSAGE_HPP_
+
+#include <vector>
+
+#include "utility/Macros.hpp"
+
+namespace quickstep {
+
+class QueryHandle;
+
+/** \addtogroup QueryExecution
+ * @{
+ */
+
+/**
+ * @brief A message requesting a query or queries to be admitted to the system.
+ **/
+class AdmitRequestMessage {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param query_handles The handles of the queries requesting to be admitted
+ * to the system.
+ **/
+ explicit AdmitRequestMessage(const std::vector<QueryHandle*> &query_handles)
+ : query_handles_(query_handles) {}
+
+ /**
+ * @brief Constructor for requesting single query admission.
+ *
+ * @param query_handle The handle of the query requesting to be admitted.
+ **/
+ explicit AdmitRequestMessage(QueryHandle *query_handle) {
+ query_handles_.push_back(query_handle);
+ }
+
+ /**
+ * @brief Get the query handles from this message.
+ **/
+ const std::vector<QueryHandle*>& getQueryHandles() const {
+ return query_handles_;
+ }
+
+ private:
+ std::vector<QueryHandle*> query_handles_;
+
+ DISALLOW_COPY_AND_ASSIGN(AdmitRequestMessage);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_QUERY_EXECUTION_ADMIT_REQUEST_MESSAGE_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 7d9d601..2f738e4 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -23,8 +23,10 @@ QS_PROTOBUF_GENERATE_CPP(queryexecution_QueryExecutionMessages_proto_srcs
if (ENABLE_DISTRIBUTED)
add_library(quickstep_queryexecution_BlockLocator BlockLocator.cpp BlockLocator.hpp)
endif()
+add_library(quickstep_queryexecution_AdmitRequestMessage ../empty_src.cpp AdmitRequestMessage.hpp)
add_library(quickstep_queryexecution_Foreman Foreman.cpp Foreman.hpp)
add_library(quickstep_queryexecution_ForemanLite ../empty_src.cpp ForemanLite.hpp)
+add_library(quickstep_queryexecution_PolicyEnforcer PolicyEnforcer.cpp PolicyEnforcer.hpp)
add_library(quickstep_queryexecution_QueryContext QueryContext.cpp QueryContext.hpp)
add_library(quickstep_queryexecution_QueryContext_proto
${queryexecution_QueryContext_proto_srcs}
@@ -43,6 +45,8 @@ add_library(quickstep_queryexecution_WorkerMessage ../empty_src.cpp WorkerMessag
add_library(quickstep_queryexecution_WorkerSelectionPolicy ../empty_src.cpp WorkerSelectionPolicy.hpp)
# Link dependencies:
+target_link_libraries(quickstep_queryexecution_AdmitRequestMessage
+ quickstep_utility_Macros)
if (ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryexecution_BlockLocator
glog
@@ -57,29 +61,17 @@ if (ENABLE_DISTRIBUTED)
tmb)
endif()
target_link_libraries(quickstep_queryexecution_Foreman
+ gflags_nothreads-static
glog
- gtest
- quickstep_catalog_CatalogDatabase
- quickstep_catalog_CatalogRelation
- quickstep_catalog_CatalogTypedefs
- quickstep_catalog_PartitionScheme
+ quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_ForemanLite
- quickstep_queryexecution_QueryContext
- quickstep_queryexecution_QueryExecutionMessages_proto
- quickstep_queryexecution_QueryExecutionState
+ quickstep_queryexecution_PolicyEnforcer
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
- quickstep_queryexecution_WorkOrdersContainer
quickstep_queryexecution_WorkerDirectory
quickstep_queryexecution_WorkerMessage
- quickstep_relationaloperators_RebuildWorkOrder
- quickstep_relationaloperators_RelationalOperator
- quickstep_relationaloperators_WorkOrder
- quickstep_storage_InsertDestination
- quickstep_storage_StorageBlock
- quickstep_storage_StorageBlockInfo
quickstep_threading_ThreadUtil
- quickstep_utility_DAG
+ quickstep_utility_EqualsAnyConstant
quickstep_utility_Macros
tmb)
target_link_libraries(quickstep_queryexecution_ForemanLite
@@ -87,6 +79,16 @@ target_link_libraries(quickstep_queryexecution_ForemanLite
quickstep_threading_Thread
quickstep_utility_Macros
tmb)
+target_link_libraries(quickstep_queryexecution_PolicyEnforcer
+ glog
+ quickstep_queryexecution_QueryExecutionMessages_proto
+ quickstep_queryexecution_QueryExecutionTypedefs
+ quickstep_queryexecution_QueryManager
+ quickstep_queryexecution_WorkerMessage
+ quickstep_queryoptimizer_QueryHandle
+ quickstep_relationaloperators_WorkOrder
+ quickstep_utility_Macros
+ tmb)
target_link_libraries(quickstep_queryexecution_QueryContext
glog
quickstep_catalog_CatalogDatabaseLite
@@ -128,7 +130,9 @@ target_link_libraries(quickstep_queryexecution_QueryExecutionTypedefs
quickstep_threading_ThreadIDBasedMap
tmb)
target_link_libraries(quickstep_queryexecution_QueryExecutionUtil
+ quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_QueryExecutionTypedefs
+ quickstep_queryexecution_WorkerMessage
quickstep_utility_Macros
tmb)
target_link_libraries(quickstep_queryexecution_QueryManager
@@ -178,8 +182,10 @@ target_link_libraries(quickstep_queryexecution_WorkerSelectionPolicy
# Module all-in-one library:
add_library(quickstep_queryexecution ../empty_src.cpp QueryExecutionModule.hpp)
target_link_libraries(quickstep_queryexecution
+ quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_Foreman
quickstep_queryexecution_ForemanLite
+ quickstep_queryexecution_PolicyEnforcer
quickstep_queryexecution_QueryContext
quickstep_queryexecution_QueryContext_proto
quickstep_queryexecution_QueryExecutionMessages_proto
@@ -223,36 +229,6 @@ if (ENABLE_DISTRIBUTED)
add_test(BlockLocator_unittest BlockLocator_unittest)
endif()
-add_executable(Foreman_unittest
- "${CMAKE_CURRENT_SOURCE_DIR}/tests/Foreman_unittest.cpp")
-target_link_libraries(Foreman_unittest
- glog
- gtest
- gtest_main
- quickstep_catalog_CatalogDatabase
- quickstep_catalog_CatalogRelation
- quickstep_catalog_CatalogTypedefs
- quickstep_queryexecution_Foreman
- quickstep_queryexecution_QueryContext
- quickstep_queryexecution_QueryContext_proto
- quickstep_queryexecution_QueryExecutionState
- quickstep_queryexecution_QueryExecutionTypedefs
- quickstep_queryexecution_WorkOrdersContainer
- quickstep_queryexecution_WorkerDirectory
- quickstep_queryexecution_WorkerMessage
- quickstep_queryoptimizer_QueryPlan
- quickstep_relationaloperators_RelationalOperator
- quickstep_relationaloperators_WorkOrder
- quickstep_storage_InsertDestination
- quickstep_storage_InsertDestination_proto
- quickstep_storage_StorageBlock
- quickstep_storage_StorageBlockInfo
- quickstep_storage_StorageManager
- quickstep_utility_DAG
- quickstep_utility_Macros
- tmb)
-add_test(Foreman_unittest Foreman_unittest)
-
add_executable(QueryManager_unittest
"${CMAKE_CURRENT_SOURCE_DIR}/tests/QueryManager_unittest.cpp")
target_link_libraries(QueryManager_unittest
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/query_execution/Foreman.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Foreman.cpp b/query_execution/Foreman.cpp
index 7705819..3609120 100644
--- a/query_execution/Foreman.cpp
+++ b/query_execution/Foreman.cpp
@@ -22,355 +22,189 @@
#include <utility>
#include <vector>
-#include "catalog/CatalogDatabase.hpp"
-#include "catalog/CatalogRelation.hpp"
-#include "catalog/CatalogTypedefs.hpp"
-#include "catalog/PartitionScheme.hpp"
-#include "query_execution/QueryContext.hpp"
-#include "query_execution/QueryExecutionMessages.pb.h"
+#include "query_execution/AdmitRequestMessage.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
#include "query_execution/WorkerDirectory.hpp"
#include "query_execution/WorkerMessage.hpp"
-#include "relational_operators/RebuildWorkOrder.hpp"
-#include "relational_operators/RelationalOperator.hpp"
-#include "relational_operators/WorkOrder.hpp"
-#include "storage/InsertDestination.hpp"
-#include "storage/StorageBlock.hpp"
-#include "storage/StorageBlockInfo.hpp"
#include "threading/ThreadUtil.hpp"
+#include "utility/EqualsAnyConstant.hpp"
#include "utility/Macros.hpp"
+#include "gflags/gflags.h"
#include "glog/logging.h"
#include "tmb/message_bus.h"
#include "tmb/tagged_message.h"
using std::move;
-using std::pair;
using std::size_t;
+using std::unique_ptr;
using std::vector;
namespace quickstep {
-void Foreman::initialize() {
+DEFINE_uint64(min_load_per_worker, 2, "The minimum load defined as the number "
+ "of pending work orders for the worker. This information is used "
+ "by the Foreman to assign work orders to worker threads");
+
+Foreman::Foreman(const tmb::client_id main_thread_client_id,
+ WorkerDirectory *worker_directory,
+ tmb::MessageBus *bus,
+ CatalogDatabaseLite *catalog_database,
+ StorageManager *storage_manager,
+ const int cpu_id,
+ const size_t num_numa_nodes)
+ : ForemanLite(bus, cpu_id),
+ main_thread_client_id_(main_thread_client_id),
+ worker_directory_(DCHECK_NOTNULL(worker_directory)),
+ catalog_database_(DCHECK_NOTNULL(catalog_database)),
+ storage_manager_(DCHECK_NOTNULL(storage_manager)) {
+ const std::vector<QueryExecutionMessageType> sender_message_types{
+ kPoisonMessage,
+ kRebuildWorkOrderMessage,
+ kWorkOrderMessage,
+ kWorkloadCompletionMessage};
+
+ for (const auto message_type : sender_message_types) {
+ bus_->RegisterClientAsSender(foreman_client_id_, message_type);
+ }
+
+ const std::vector<QueryExecutionMessageType> receiver_message_types{
+ kAdmitRequestMessage,
+ kCatalogRelationNewBlockMessage,
+ kDataPipelineMessage,
+ kPoisonMessage,
+ kRebuildWorkOrderCompleteMessage,
+ kWorkOrderFeedbackMessage,
+ kWorkOrdersAvailableMessage,
+ kWorkOrderCompleteMessage};
+
+ for (const auto message_type : receiver_message_types) {
+ bus_->RegisterClientAsReceiver(foreman_client_id_, message_type);
+ }
+
+ policy_enforcer_.reset(new PolicyEnforcer(
+ foreman_client_id_,
+ num_numa_nodes,
+ catalog_database_,
+ storage_manager_,
+ bus_));
+}
+
+void Foreman::run() {
if (cpu_id_ >= 0) {
// We can pin the foreman thread to a CPU if specified.
ThreadUtil::BindToCPU(cpu_id_);
}
- initializeState();
-
- DEBUG_ASSERT(query_dag_ != nullptr);
- const dag_node_index dag_size = query_dag_->size();
-
- // Collect all the workorders from all the relational operators in the DAG.
- for (dag_node_index index = 0; index < dag_size; ++index) {
- if (checkAllBlockingDependenciesMet(index)) {
- query_dag_->getNodePayloadMutable(index)->informAllBlockingDependenciesMet();
- processOperator(index, false);
- }
- }
-
- // Dispatch the WorkOrders generated so far.
- dispatchWorkerMessages(0, 0);
-}
-
-void Foreman::processWorkOrderCompleteMessage(const dag_node_index op_index,
- const size_t worker_thread_index) {
- query_exec_state_->decrementNumQueuedWorkOrders(op_index);
-
- // As the given worker finished executing a WorkOrder, decrement its number
- // of queued WorkOrders.
- workers_->decrementNumQueuedWorkOrders(worker_thread_index);
-
- // Check if new work orders are available and fetch them if so.
- fetchNormalWorkOrders(op_index);
-
- if (checkRebuildRequired(op_index)) {
- if (checkNormalExecutionOver(op_index)) {
- if (!checkRebuildInitiated(op_index)) {
- if (initiateRebuild(op_index)) {
- // Rebuild initiated and completed right away.
- markOperatorFinished(op_index);
- } else {
- // Rebuild under progress.
- }
- } else if (checkRebuildOver(op_index)) {
- // Rebuild was under progress and now it is over.
- markOperatorFinished(op_index);
- }
- } else {
- // Normal execution under progress for this operator.
- }
- } else if (checkOperatorExecutionOver(op_index)) {
- // Rebuild not required for this operator and its normal execution is
- // complete.
- markOperatorFinished(op_index);
- }
-
- for (const pair<dag_node_index, bool> &dependent_link :
- query_dag_->getDependents(op_index)) {
- const dag_node_index dependent_op_index = dependent_link.first;
- if (checkAllBlockingDependenciesMet(dependent_op_index)) {
- // Process the dependent operator (of the operator whose WorkOrder
- // was just executed) for which all the dependencies have been met.
- processOperator(dependent_op_index, true);
- }
- }
-
- // Dispatch the WorkerMessages to the workers. We prefer to start the search
- // for the schedulable WorkOrders beginning from 'op_index'. The first
- // candidate worker to receive the next WorkOrder is the one that sent the
- // response message to Foreman.
- dispatchWorkerMessages(worker_thread_index, op_index);
-}
-
-void Foreman::processRebuildWorkOrderCompleteMessage(const dag_node_index op_index,
- const size_t worker_thread_index) {
- query_exec_state_->decrementNumRebuildWorkOrders(op_index);
- workers_->decrementNumQueuedWorkOrders(worker_thread_index);
-
- if (checkRebuildOver(op_index)) {
- markOperatorFinished(op_index);
-
- for (const pair<dag_node_index, bool> &dependent_link :
- query_dag_->getDependents(op_index)) {
- const dag_node_index dependent_op_index = dependent_link.first;
- if (checkAllBlockingDependenciesMet(dependent_op_index)) {
- processOperator(dependent_op_index, true);
- }
- }
- }
-
- // Dispatch the WorkerMessages to the workers. We prefer to start the search
- // for the schedulable WorkOrders beginning from 'op_index'. The first
- // candidate worker to receive the next WorkOrder is the one that sent the
- // response message to Foreman.
- dispatchWorkerMessages(worker_thread_index, op_index);
-}
-
-void Foreman::processDataPipelineMessage(const dag_node_index op_index,
- const block_id block,
- const relation_id rel_id) {
- for (const dag_node_index consumer_index :
- output_consumers_[op_index]) {
- // Feed the streamed block to the consumer. Note that 'output_consumers_'
- // only contain those dependents of operator with index = op_index which are
- // eligible to receive streamed input.
- query_dag_->getNodePayloadMutable(consumer_index)->feedInputBlock(block, rel_id);
- // Because of the streamed input just fed, check if there are any new
- // WorkOrders available and if so, fetch them.
- fetchNormalWorkOrders(consumer_index);
- }
-
- // Dispatch the WorkerMessages to the workers. We prefer to start the search
- // for the schedulable WorkOrders beginning from 'op_index'. The first
- // candidate worker to receive the next WorkOrder is the one that sent the
- // response message to Foreman.
- // TODO(zuyu): Improve the data locality for the next WorkOrder.
- dispatchWorkerMessages(0, op_index);
-}
-
-void Foreman::processFeedbackMessage(const WorkOrder::FeedbackMessage &msg) {
- RelationalOperator *op =
- query_dag_->getNodePayloadMutable(msg.header().rel_op_index);
- op->receiveFeedbackMessage(msg);
-}
-
-void Foreman::run() {
- // Initialize before for Foreman eventloop.
- initialize();
// Event loop
- while (!query_exec_state_->hasQueryExecutionFinished()) {
+ for (;;) {
// Receive() causes this thread to sleep until next message is received.
- AnnotatedMessage annotated_msg = bus_->Receive(foreman_client_id_, 0, true);
+ const AnnotatedMessage annotated_msg =
+ bus_->Receive(foreman_client_id_, 0, true);
const TaggedMessage &tagged_message = annotated_msg.tagged_message;
- switch (tagged_message.message_type()) {
- case kWorkOrderCompleteMessage: {
- serialization::WorkOrderCompletionMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- processWorkOrderCompleteMessage(proto.operator_index(), proto.worker_thread_index());
- break;
- }
- case kRebuildWorkOrderCompleteMessage: {
- serialization::WorkOrderCompletionMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- processRebuildWorkOrderCompleteMessage(proto.operator_index(), proto.worker_thread_index());
+ const tmb::message_type_id message_type = tagged_message.message_type();
+ switch (message_type) {
+ case kCatalogRelationNewBlockMessage: // Fall through
+ case kDataPipelineMessage:
+ case kRebuildWorkOrderCompleteMessage:
+ case kWorkOrderCompleteMessage:
+ case kWorkOrderFeedbackMessage:
+ case kWorkOrdersAvailableMessage: {
+ policy_enforcer_->processMessage(tagged_message);
break;
}
- case kCatalogRelationNewBlockMessage: {
- serialization::CatalogRelationNewBlockMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- const block_id block = proto.block_id();
-
- CatalogRelation *relation =
- static_cast<CatalogDatabase*>(catalog_database_)->getRelationByIdMutable(proto.relation_id());
- relation->addBlock(block);
-
- if (proto.has_partition_id()) {
- relation->getPartitionSchemeMutable()->addBlockToPartition(proto.partition_id(), block);
+ case kAdmitRequestMessage: {
+ const AdmitRequestMessage *msg =
+ static_cast<const AdmitRequestMessage *>(tagged_message.message());
+ const vector<QueryHandle *> &query_handles = msg->getQueryHandles();
+
+ DCHECK(!query_handles.empty());
+ bool all_queries_admitted = true;
+ if (query_handles.size() == 1u) {
+ all_queries_admitted =
+ policy_enforcer_->admitQuery(query_handles.front());
+ } else {
+ all_queries_admitted = policy_enforcer_->admitQueries(query_handles);
+ }
+ if (!all_queries_admitted) {
+ LOG(WARNING) << "The scheduler could not admit all the queries";
+ // TODO(harshad) - Inform the main thread about the failure.
}
break;
}
- case kDataPipelineMessage: {
- // Possible message senders include InsertDestinations and some
- // operators which modify existing blocks.
- serialization::DataPipelineMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- processDataPipelineMessage(proto.operator_index(), proto.block_id(), proto.relation_id());
- break;
- }
- case kWorkOrdersAvailableMessage: {
- serialization::WorkOrdersAvailableMessage proto;
- CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- const dag_node_index op_index = proto.operator_index();
-
- // Check if new work orders are available.
- fetchNormalWorkOrders(op_index);
-
- // Dispatch the WorkerMessages to the workers. We prefer to start the search
- // for the schedulable WorkOrders beginning from 'op_index'. The first
- // candidate worker to receive the next WorkOrder is the one that sent the
- // response message to Foreman.
- // TODO(zuyu): Improve the data locality for the next WorkOrder.
- dispatchWorkerMessages(0, op_index);
- break;
- }
- case kWorkOrderFeedbackMessage: {
- WorkOrder::FeedbackMessage msg(const_cast<void *>(tagged_message.message()),
- tagged_message.message_bytes());
- processFeedbackMessage(msg);
- break;
+ case kPoisonMessage: {
+ if (policy_enforcer_->hasQueries()) {
+ LOG(WARNING) << "Foreman thread exiting while some queries are "
+ "under execution or waiting to be admitted";
+ }
+ return;
}
default:
LOG(FATAL) << "Unknown message type to Foreman";
}
- }
-
- // Clean up before exiting.
- cleanUp();
-}
-void Foreman::dispatchWorkerMessages(
- const size_t start_worker_index,
- const dag_node_index start_operator_index) {
- // Loop over all workers. Stopping criteria:
- // 1. Every worker has been assigned exactly max_msgs_per_worker_ workorders.
- // OR 2. No schedulable workorders at this time.
- size_t done_workers_count = 0;
- for (size_t curr_worker = start_worker_index;
- done_workers_count < workers_->getNumWorkers();
- curr_worker = (curr_worker + 1) % workers_->getNumWorkers()) {
- if (workers_->getNumQueuedWorkOrders(curr_worker) < max_msgs_per_worker_) {
- std::unique_ptr<WorkerMessage> msg;
- msg.reset(getNextWorkerMessage(
- start_operator_index, workers_->getNUMANode(curr_worker)));
- if (msg.get() != nullptr) {
- sendWorkerMessage(curr_worker, *msg);
- workers_->incrementNumQueuedWorkOrders(curr_worker);
- } else {
- // No schedulable workorder at this point.
- ++done_workers_count;
- }
- } else {
- // curr_worker already has been assigned max_msgs_per_worker workorders.
- ++done_workers_count;
+ if (canCollectNewMessages(message_type)) {
+ vector<unique_ptr<WorkerMessage>> new_messages;
+ policy_enforcer_->getWorkerMessages(&new_messages);
+ dispatchWorkerMessages(new_messages);
+ }
+
+ // We check again, as some queries may produce zero work orders and finish
+ // their execution.
+ if (!policy_enforcer_->hasQueries()) {
+ // Signal the main thread that there are no queries to be executed.
+ // Currently the message doesn't have any real content.
+ const int dummy_payload = 0;
+ TaggedMessage completion_tagged_message(
+ &dummy_payload, sizeof(dummy_payload), kWorkloadCompletionMessage);
+ const tmb::MessageBus::SendStatus send_status =
+ QueryExecutionUtil::SendTMBMessage(
+ bus_,
+ foreman_client_id_,
+ main_thread_client_id_,
+ move(completion_tagged_message));
+ CHECK(send_status == tmb::MessageBus::SendStatus::kOK)
+ << "Message could not be sent from Foreman with TMB client ID "
+ << foreman_client_id_ << " to main thread with TMB client ID"
+ << main_thread_client_id_;
}
}
}
-void Foreman::initializeState() {
- const dag_node_index dag_size = query_dag_->size();
-
- output_consumers_.resize(dag_size);
- blocking_dependencies_.resize(dag_size);
-
- query_exec_state_.reset(new QueryExecutionState(dag_size));
- workorders_container_.reset(new WorkOrdersContainer(dag_size, num_numa_nodes_));
-
- for (dag_node_index node_index = 0; node_index < dag_size; ++node_index) {
- const QueryContext::insert_destination_id insert_destination_index =
- query_dag_->getNodePayload(node_index).getInsertDestinationID();
- if (insert_destination_index != QueryContext::kInvalidInsertDestinationId) {
- // Rebuild is necessary whenever InsertDestination is present.
- query_exec_state_->setRebuildRequired(node_index);
- query_exec_state_->setRebuildStatus(node_index, 0, false);
- }
-
- for (const pair<dag_node_index, bool> &dependent_link :
- query_dag_->getDependents(node_index)) {
- const dag_node_index dependent_op_index = dependent_link.first;
- if (!query_dag_->getLinkMetadata(node_index, dependent_op_index)) {
- // The link is not a pipeline-breaker. Streaming of blocks is possible
- // between these two operators.
- output_consumers_[node_index].push_back(dependent_op_index);
- } else {
- // The link is a pipeline-breaker. Streaming of blocks is not possible
- // between these two operators.
- blocking_dependencies_[dependent_op_index].push_back(node_index);
- }
- }
+bool Foreman::canCollectNewMessages(const tmb::message_type_id message_type) {
+ if (QUICKSTEP_EQUALS_ANY_CONSTANT(message_type,
+ kCatalogRelationNewBlockMessage,
+ kWorkOrderFeedbackMessage)) {
+ return false;
+ } else if (worker_directory_->getLeastLoadedWorker().second <=
+ FLAGS_min_load_per_worker) {
+ // If the least loaded worker has only one pending work order, we should
+ // collect new messages and dispatch them.
+ return true;
+ } else {
+ return false;
}
}
-// TODO(harshad) : The default policy may execute remote WorkOrders for an
-// operator with a lower index even when there are local WorkOrders available for
-// an operator with higher index. We should examine if avoiding this behavior
-// has any benefits with respect to execution time and/or memory pressure.
-WorkerMessage* Foreman::getNextWorkerMessage(
- const dag_node_index start_operator_index, const int numa_node) {
- // Default policy: Operator with lowest index first.
- WorkOrder *work_order = nullptr;
- size_t num_operators_checked = 0;
- for (dag_node_index index = start_operator_index;
- num_operators_checked < query_dag_->size();
- index = (index + 1) % query_dag_->size(), ++num_operators_checked) {
- if (query_exec_state_->hasExecutionFinished(index)) {
- continue;
- }
- if (numa_node != -1) {
- // First try to get a normal WorkOrder from the specified NUMA node.
- work_order = workorders_container_->getNormalWorkOrderForNUMANode(index, numa_node);
- if (work_order != nullptr) {
- // A WorkOrder found on the given NUMA node.
- query_exec_state_->incrementNumQueuedWorkOrders(index);
- return WorkerMessage::WorkOrderMessage(work_order, index);
- } else {
- // Normal workorder not found on this node. Look for a rebuild workorder
- // on this NUMA node.
- work_order = workorders_container_->getRebuildWorkOrderForNUMANode(index, numa_node);
- if (work_order != nullptr) {
- return WorkerMessage::RebuildWorkOrderMessage(work_order, index);
- }
- }
- }
- // Either no workorder found on the given NUMA node, or numa_node is -1.
- // Try to get a normal WorkOrder from other NUMA nodes.
- work_order = workorders_container_->getNormalWorkOrder(index);
- if (work_order != nullptr) {
- query_exec_state_->incrementNumQueuedWorkOrders(index);
- return WorkerMessage::WorkOrderMessage(work_order, index);
+void Foreman::dispatchWorkerMessages(const vector<unique_ptr<WorkerMessage>> &messages) {
+ for (auto const &message : messages) {
+ DCHECK(message != nullptr);
+ int recipient_worker_thread_index = message->getRecipientHint();
+ if (recipient_worker_thread_index != -1) {
+ sendWorkerMessage(static_cast<size_t>(recipient_worker_thread_index),
+ *message);
} else {
- // Normal WorkOrder not found, look for a RebuildWorkOrder.
- work_order = workorders_container_->getRebuildWorkOrder(index);
- if (work_order != nullptr) {
- return WorkerMessage::RebuildWorkOrderMessage(work_order, index);
- }
+ sendWorkerMessage(worker_directory_->getLeastLoadedWorker().first,
+ *message);
}
}
- // No WorkOrders available right now.
- return nullptr;
}
-void Foreman::sendWorkerMessage(const std::size_t worker_thread_index,
+void Foreman::sendWorkerMessage(const size_t worker_thread_index,
const WorkerMessage &message) {
- message_type_id type;
+ tmb::message_type_id type;
if (message.getType() == WorkerMessage::WorkerMessageType::kRebuildWorkOrder) {
type = kRebuildWorkOrderMessage;
} else if (message.getType() == WorkerMessage::WorkerMessageType::kWorkOrder) {
@@ -383,152 +217,12 @@ void Foreman::sendWorkerMessage(const std::size_t worker_thread_index,
const tmb::MessageBus::SendStatus send_status =
QueryExecutionUtil::SendTMBMessage(bus_,
foreman_client_id_,
- workers_->getClientID(worker_thread_index),
+ worker_directory_->getClientID(worker_thread_index),
move(worker_tagged_message));
CHECK(send_status == tmb::MessageBus::SendStatus::kOK) <<
"Message could not be sent from Foreman with TMB client ID "
<< foreman_client_id_ << " to Foreman with TMB client ID "
- << workers_->getClientID(worker_thread_index);
-}
-
-bool Foreman::fetchNormalWorkOrders(const dag_node_index index) {
- bool generated_new_workorders = false;
- if (!query_exec_state_->hasDoneGenerationWorkOrders(index)) {
- // Do not fetch any work units until all blocking dependencies are met.
- // The releational operator is not aware of blocking dependencies for
- // uncorrelated scalar queries.
- if (!checkAllBlockingDependenciesMet(index)) {
- return false;
- }
- const size_t num_pending_workorders_before =
- workorders_container_->getNumNormalWorkOrders(index);
- const bool done_generation =
- query_dag_->getNodePayloadMutable(index)->getAllWorkOrders(workorders_container_.get(),
- query_context_.get(),
- storage_manager_,
- foreman_client_id_,
- bus_);
- if (done_generation) {
- query_exec_state_->setDoneGenerationWorkOrders(index);
- }
-
- // TODO(shoban): It would be a good check to see if operator is making
- // useful progress, i.e., the operator either generates work orders to
- // execute or still has pending work orders executing. However, this will not
- // work if Foreman polls operators without feeding data. This check can be
- // enabled, if Foreman is refactored to call getAllWorkOrders() only when
- // pending work orders are completed or new input blocks feed.
-
- generated_new_workorders =
- (num_pending_workorders_before <
- workorders_container_->getNumNormalWorkOrders(index));
- }
- return generated_new_workorders;
-}
-
-void Foreman::processOperator(const dag_node_index index,
- const bool recursively_check_dependents) {
- if (fetchNormalWorkOrders(index)) {
- // Fetched work orders. Return to wait for the generated work orders to
- // execute, and skip the execution-finished checks.
- return;
- }
-
- if (checkNormalExecutionOver(index)) {
- if (checkRebuildRequired(index)) {
- if (!checkRebuildInitiated(index)) {
- // Rebuild hasn't started, initiate it.
- if (initiateRebuild(index)) {
- // Rebuild initiated and completed right away.
- markOperatorFinished(index);
- } else {
- // Rebuild WorkOrders have been generated.
- return;
- }
- } else if (checkRebuildOver(index)) {
- // Rebuild had been initiated and it is over.
- markOperatorFinished(index);
- }
- } else {
- // Rebuild is not required and normal execution over, mark finished.
- markOperatorFinished(index);
- }
- // If we reach here, that means the operator has been marked as finished.
- if (recursively_check_dependents) {
- for (const pair<dag_node_index, bool> &dependent_link :
- query_dag_->getDependents(index)) {
- const dag_node_index dependent_op_index = dependent_link.first;
- if (checkAllBlockingDependenciesMet(dependent_op_index)) {
- processOperator(dependent_op_index, true);
- }
- }
- }
- }
-}
-
-void Foreman::markOperatorFinished(const dag_node_index index) {
- query_exec_state_->setExecutionFinished(index);
-
- RelationalOperator *op = query_dag_->getNodePayloadMutable(index);
- op->updateCatalogOnCompletion();
-
- const relation_id output_rel = op->getOutputRelationID();
- for (const pair<dag_node_index, bool> &dependent_link : query_dag_->getDependents(index)) {
- const dag_node_index dependent_op_index = dependent_link.first;
- RelationalOperator *dependent_op = query_dag_->getNodePayloadMutable(dependent_op_index);
- // Signal dependent operator that current operator is done feeding input blocks.
- if (output_rel >= 0) {
- dependent_op->doneFeedingInputBlocks(output_rel);
- }
- if (checkAllBlockingDependenciesMet(dependent_op_index)) {
- dependent_op->informAllBlockingDependenciesMet();
- }
- }
-}
-
-bool Foreman::initiateRebuild(const dag_node_index index) {
- DEBUG_ASSERT(!workorders_container_->hasRebuildWorkOrder(index));
- DEBUG_ASSERT(checkRebuildRequired(index));
- DEBUG_ASSERT(!checkRebuildInitiated(index));
-
- getRebuildWorkOrders(index, workorders_container_.get());
-
- query_exec_state_->setRebuildStatus(
- index, workorders_container_->getNumRebuildWorkOrders(index), true);
-
- return (query_exec_state_->getNumRebuildWorkOrders(index) == 0);
-}
-
-void Foreman::getRebuildWorkOrders(const dag_node_index index, WorkOrdersContainer *container) {
- const RelationalOperator &op = query_dag_->getNodePayload(index);
- const QueryContext::insert_destination_id insert_destination_index = op.getInsertDestinationID();
-
- if (insert_destination_index == QueryContext::kInvalidInsertDestinationId) {
- return;
- }
-
- vector<MutableBlockReference> partially_filled_block_refs;
-
- DCHECK(query_context_ != nullptr);
- InsertDestination *insert_destination = query_context_->getInsertDestination(insert_destination_index);
- DCHECK(insert_destination != nullptr);
-
- insert_destination->getPartiallyFilledBlocks(&partially_filled_block_refs);
-
- for (vector<MutableBlockReference>::size_type i = 0;
- i < partially_filled_block_refs.size();
- ++i) {
- // Note: The query ID used below is dummy for now, it will be replaced with
- // the true query ID when QueryManager gets used in Foreman.
- container->addRebuildWorkOrder(
- new RebuildWorkOrder(0,
- move(partially_filled_block_refs[i]),
- index,
- op.getOutputRelationID(),
- foreman_client_id_,
- bus_),
- index);
- }
+ << worker_directory_->getClientID(worker_thread_index);
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a690455e/query_execution/Foreman.hpp
----------------------------------------------------------------------
diff --git a/query_execution/Foreman.hpp b/query_execution/Foreman.hpp
index 2d6e0d3..5c4893d 100644
--- a/query_execution/Foreman.hpp
+++ b/query_execution/Foreman.hpp
@@ -22,46 +22,40 @@
#include <memory>
#include <vector>
-#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/ForemanLite.hpp"
-#include "query_execution/QueryContext.hpp"
-#include "query_execution/QueryExecutionState.hpp"
-#include "query_execution/QueryExecutionTypedefs.hpp"
-#include "query_execution/WorkOrdersContainer.hpp"
-#include "query_execution/WorkerMessage.hpp"
-#include "relational_operators/RelationalOperator.hpp"
-#include "relational_operators/WorkOrder.hpp"
-#include "storage/StorageBlockInfo.hpp"
-#include "utility/DAG.hpp"
+#include "query_execution/PolicyEnforcer.hpp"
#include "utility/Macros.hpp"
-#include "glog/logging.h"
-#include "gtest/gtest_prod.h"
+#include "gflags/gflags.h"
+#include "tmb/id_typedefs.h"
#include "tmb/message_bus.h"
namespace quickstep {
+DECLARE_uint64(min_load_per_worker);
+
class CatalogDatabaseLite;
class StorageManager;
class WorkerDirectory;
-
-namespace serialization { class QueryContext; }
+class WorkerMessage;
/** \addtogroup QueryExecution
* @{
*/
/**
- * @brief The Foreman scans the query DAG, requests each operator to produce
- * workorders. It also pipelines the intermediate output it receives to
- * the relational operators which need it.
+ * @brief The Foreman receives queries from the main thread, messages from the
+ * policy enforcer and dispatches the work to worker threads. It also
+ * receives work completion messages from workers.
**/
class Foreman final : public ForemanLite {
public:
/**
* @brief Constructor.
*
+ * @param main_thread_client_id The TMB client ID of the main thread.
+ * @param worker_directory The worker directory.
* @param bus A pointer to the TMB.
* @param catalog_database The catalog database where this query is executed.
* @param storage_manager The StorageManager to use.
@@ -71,233 +65,31 @@ class Foreman final : public ForemanLite {
* @note If cpu_id is not specified, Foreman thread can be possibly moved
* around on different CPUs by the OS.
**/
- Foreman(tmb::MessageBus *bus,
+ Foreman(const tmb::client_id main_thread_client_id,
+ WorkerDirectory *worker_directory,
+ tmb::MessageBus *bus,
CatalogDatabaseLite *catalog_database,
StorageManager *storage_manager,
const int cpu_id = -1,
- const int num_numa_nodes = 1)
- : ForemanLite(bus, cpu_id),
- catalog_database_(DCHECK_NOTNULL(catalog_database)),
- storage_manager_(DCHECK_NOTNULL(storage_manager)),
- max_msgs_per_worker_(1),
- num_numa_nodes_(num_numa_nodes) {
- bus_->RegisterClientAsSender(foreman_client_id_, kWorkOrderMessage);
- bus_->RegisterClientAsSender(foreman_client_id_, kRebuildWorkOrderMessage);
- // NOTE : Foreman thread sends poison messages in the optimizer's
- // ExecutionGeneratorTest.
- bus_->RegisterClientAsSender(foreman_client_id_, kPoisonMessage);
-
- bus_->RegisterClientAsReceiver(foreman_client_id_,
- kWorkOrderCompleteMessage);
- bus_->RegisterClientAsReceiver(foreman_client_id_,
- kRebuildWorkOrderCompleteMessage);
- bus_->RegisterClientAsReceiver(foreman_client_id_, kCatalogRelationNewBlockMessage);
- bus_->RegisterClientAsReceiver(foreman_client_id_, kDataPipelineMessage);
- bus_->RegisterClientAsReceiver(foreman_client_id_,
- kWorkOrdersAvailableMessage);
- bus_->RegisterClientAsReceiver(foreman_client_id_,
- kWorkOrderFeedbackMessage);
- }
+ const std::size_t num_numa_nodes = 1);
~Foreman() override {}
- /**
- * @brief Set the Query plan DAG for the query to be executed.
- *
- * @param query_plan_dag A pointer to the query plan DAG.
- **/
- inline void setQueryPlan(DAG<RelationalOperator, bool> *query_plan_dag) {
- query_dag_ = query_plan_dag;
- }
-
- /**
- * @brief Reconstruct the QueryContext for the query to be executed.
- *
- * @param proto The serialized QueryContext.
- **/
- inline void reconstructQueryContextFromProto(const serialization::QueryContext &proto) {
- query_context_.reset(
- new QueryContext(proto, *catalog_database_, storage_manager_, foreman_client_id_, bus_));
- }
-
- /**
- * @brief Set the WorkerDirectory pointer.
- *
- * @param workers A pointer to the WorkerDirectory.
- **/
- void setWorkerDirectory(WorkerDirectory *workers) {
- workers_ = workers;
- }
-
- /**
- * @brief Set the maximum number of messages that should be allocated to each
- * worker during a single round of WorkOrder dispatch.
- *
- * @param max_msgs_per_worker Maximum number of messages.
- **/
- void setMaxMessagesPerWorker(const std::size_t max_msgs_per_worker) {
- max_msgs_per_worker_ = max_msgs_per_worker;
- }
-
protected:
/**
- * @brief The foreman receives a DAG of relational operators, asks relational
- * operators to produce the workorders and based on the response it gets
- * pipelines the intermediate output to dependent relational operators.
- *
- * @note The workers who get the messages from the Foreman execute and
- * subsequently delete the WorkOrder contained in the message.
+ * @brief Run the event-based loop in the Foreman thread.
**/
void run() override;
private:
- typedef DAG<RelationalOperator, bool>::size_type_nodes dag_node_index;
-
- /**
- * @brief Check if all the dependencies of the node at specified index have
- * finished their execution.
- *
- * @note This function's true return value is a pre-requisite for calling
- * getRebuildWorkOrders()
- *
- * @param node_index The index of the specified node in the query DAG.
- *
- * @return True if all the dependencies have finished their execution. False
- * otherwise.
- **/
- inline bool checkAllDependenciesMet(const dag_node_index node_index) const {
- for (const dag_node_index dependency_index : query_dag_->getDependencies(node_index)) {
- // If at least one of the dependencies is not met, return false.
- if (!query_exec_state_->hasExecutionFinished(dependency_index)) {
- return false;
- }
- }
- return true;
- }
-
- /**
- * @brief Check if all the blocking dependencies of the node at specified
- * index have finished their execution.
- *
- * @note A blocking dependency is the one which is pipeline breaker. Output of
- * a dependency can't be streamed to its dependent if the link between
- * them is pipeline breaker.
- *
- * @param node_index The index of the specified node in the query DAG.
- *
- * @return True if all the blocking dependencies have finished their
- * execution. False otherwise.
- **/
- inline bool checkAllBlockingDependenciesMet(const dag_node_index node_index) const {
- for (const dag_node_index blocking_dependency_index : blocking_dependencies_[node_index]) {
- if (!query_exec_state_->hasExecutionFinished(blocking_dependency_index)) {
- return false;
- }
- }
- return true;
- }
-
/**
* @brief Dispatch schedulable WorkOrders, wrapped in WorkerMessages to the
* worker threads.
*
- * @param start_worker_index The dispatch of WorkOrders preferably begins with
- * the worker at this index.
- * @param start_operator_index The search for a schedulable WorkOrder
- * begins with the WorkOrders generated by this operator.
- **/
- void dispatchWorkerMessages(const std::size_t start_worker_index,
- const dag_node_index start_operator_index);
-
- /**
- * @brief Initialize all the local vectors and maps. If the operator has an
- * InsertDestination, pass the bus address and Foreman's TMB client ID
- * to it.
- **/
- void initializeState();
-
- /**
- * @brief Initialize the Foreman before starting the event loop. This binds
- * the Foreman thread to configured CPU, and does initial processing of
- * operator before waiting for events from Workers.
- **/
- void initialize();
-
- /**
- * @brief Process the received WorkOrder complete message.
- *
- * @param node_index The index of the specified operator node in the query DAG
- * for the completed WorkOrder.
- * @param worker_thread_index The logical index of the worker thread in
- * WorkerDirectory for the completed WorkOrder.
- **/
- void processWorkOrderCompleteMessage(const dag_node_index op_index,
- const std::size_t worker_thread_index);
-
- /**
- * @brief Process the received RebuildWorkOrder complete message.
- *
- * @param node_index The index of the specified operator node in the query DAG
- * for the completed RebuildWorkOrder.
- * @param worker_thread_index The logical index of the worker thread in
- * WorkerDirectory for the completed RebuildWorkOrder.
- **/
- void processRebuildWorkOrderCompleteMessage(const dag_node_index op_index,
- const std::size_t worker_thread_index);
-
- /**
- * @brief Process the received data pipeline message.
- *
- * @param node_index The index of the specified operator node in the query DAG
- * for the pipelining block.
- * @param block The block id.
- * @param rel_id The ID of the relation that produced 'block'.
- **/
- void processDataPipelineMessage(const dag_node_index op_index,
- const block_id block,
- const relation_id rel_id);
-
- /**
- * @brief Process the received work order feedback message and notify relational
- * operator.
- *
- * @param message Feedback message from work order.
- **/
- void processFeedbackMessage(const WorkOrder::FeedbackMessage &message);
-
- /**
- * @brief Clear some of the vectors used for a single run of a query.
- **/
- void cleanUp() {
- output_consumers_.clear();
- blocking_dependencies_.clear();
- }
-
- /**
- * @brief Process a current relational operator: Get its workorders and store
- * them in the WorkOrdersContainer for this query. If the operator can
- * be marked as done, do so.
- *
- * @param index The index of the relational operator to be processed in the
- * query plan DAG.
- * @param recursively_check_dependents If an operator is done, should we
- * call processOperator on its dependents recursively.
+ * @param messages The messages to be dispatched.
**/
- void processOperator(const dag_node_index index, const bool recursively_check_dependents);
-
- /**
- * @brief Get the next workorder to be excuted, wrapped in a WorkerMessage.
- *
- * @param start_operator_index Begin the search for the schedulable WorkOrder
- * with the operator at this index.
- * @param numa_node The next WorkOrder should preferably have its input(s)
- * from this numa_node. This is a hint and not a binding requirement.
- *
- * @return A pointer to the WorkerMessage. If there's no WorkOrder to be
- * executed, return NULL.
- **/
- WorkerMessage* getNextWorkerMessage(
- const dag_node_index start_operator_index, const int numa_node = -1);
+ void dispatchWorkerMessages(
+ const std::vector<std::unique_ptr<WorkerMessage>> &messages);
/**
* @brief Send the given message to the specified worker.
@@ -306,156 +98,24 @@ class Foreman final : public ForemanLite {
* in WorkerDirectory.
* @param message The WorkerMessage to be sent.
**/
- void sendWorkerMessage(const std::size_t worker_thread_index, const WorkerMessage &message);
+ void sendWorkerMessage(const std::size_t worker_thread_index,
+ const WorkerMessage &message);
/**
- * @brief Fetch all work orders currently available in relational operator and
- * store them internally.
- *
- * @param index The index of the relational operator to be processed in the
- * query plan DAG.
+ * @brief Check if we can collect new messages from the PolicyEnforcer.
*
- * @return Whether any work order was generated by op.
+ * @param message_type The type of the last received message.
**/
- bool fetchNormalWorkOrders(const dag_node_index index);
+ bool canCollectNewMessages(const tmb::message_type_id message_type);
- /**
- * @brief This function does the following things:
- * 1. Mark the given relational operator as "done".
- * 2. For all the dependents of this operator, check if all of their
- * blocking dependencies are met. If so inform them that the blocking
- * dependencies are met.
- * 3. Check if the given operator is done producing output. If it's
- * done, inform the dependents that they won't receive input anymore
- * from the given operator.
- *
- * @param index The index of the given relational operator in the DAG.
- **/
- void markOperatorFinished(const dag_node_index index);
+ const tmb::client_id main_thread_client_id_;
- /**
- * @brief Check if the execution of the given operator is over.
- *
- * @param index The index of the given operator in the DAG.
- *
- * @return True if the execution of the given operator is over, false
- * otherwise.
- **/
- inline bool checkOperatorExecutionOver(const dag_node_index index) const {
- if (checkRebuildRequired(index)) {
- return (checkNormalExecutionOver(index) && checkRebuildOver(index));
- } else {
- return checkNormalExecutionOver(index);
- }
- }
-
- /**
- * @brief Check if the given operator's normal execution is over.
- *
- * @note The conditions for a given operator's normal execution to get over:
- * 1. All of its normal (i.e. non rebuild) WorkOrders have finished
- * execution.
- * 2. The operator is done generating work orders.
- * 3. All of the dependencies of the given operator have been met.
- *
- * @param index The index of the given operator in the DAG.
- *
- * @return True if the normal execution of the given operator is over, false
- * otherwise.
- **/
- inline bool checkNormalExecutionOver(const dag_node_index index) const {
- return (checkAllDependenciesMet(index) &&
- !workorders_container_->hasNormalWorkOrder(index) &&
- query_exec_state_->getNumQueuedWorkOrders(index) == 0 &&
- query_exec_state_->hasDoneGenerationWorkOrders(index));
- }
-
- /**
- * @brief Check if the rebuild operation is required for a given operator.
- *
- * @param index The index of the given operator in the DAG.
- *
- * @return True if the rebuild operation is required, false otherwise.
- **/
- inline bool checkRebuildRequired(const dag_node_index index) const {
- return query_exec_state_->isRebuildRequired(index);
- }
-
- /**
- * @brief Check if the rebuild operation for a given operator is over.
- *
- * @param index The index of the given operator in the DAG.
- *
- * @return True if the rebuild operation is over, false otherwise.
- **/
- inline bool checkRebuildOver(const dag_node_index index) const {
- return query_exec_state_->hasRebuildInitiated(index) &&
- !workorders_container_->hasRebuildWorkOrder(index) &&
- (query_exec_state_->getNumRebuildWorkOrders(index) == 0);
- }
-
- /**
- * @brief Check if the rebuild operation for a given operator has been
- * initiated.
- *
- * @param index The index of the given operator in the DAG.
- *
- * @return True if the rebuild operation has been initiated, false otherwise.
- **/
- inline bool checkRebuildInitiated(const dag_node_index index) const {
- return query_exec_state_->hasRebuildInitiated(index);
- }
-
- /**
- * @brief Initiate the rebuild process for partially filled blocks generated
- * during the execution of the given operator.
- *
- * @param index The index of the given operator in the DAG.
- *
- * @return True if the rebuild is over immediately, i.e. the operator didn't
- * generate any rebuild WorkOrders, false otherwise.
- **/
- bool initiateRebuild(const dag_node_index index);
-
- /**
- * @brief Get the rebuild WorkOrders for an operator.
- *
- * @note This function should be called only once, when all the normal
- * WorkOrders generated by an operator finish their execution.
- *
- * @param index The index of the operator in the query plan DAG.
- * @param container A pointer to a WorkOrdersContainer to be used to store the
- * generated WorkOrders.
- **/
- void getRebuildWorkOrders(const dag_node_index index, WorkOrdersContainer *container);
+ WorkerDirectory *worker_directory_;
CatalogDatabaseLite *catalog_database_;
StorageManager *storage_manager_;
- DAG<RelationalOperator, bool> *query_dag_;
-
- std::unique_ptr<QueryContext> query_context_;
-
- // During a single round of WorkOrder dispatch, a Worker should be allocated
- // at most these many WorkOrders.
- std::size_t max_msgs_per_worker_;
-
- // For all nodes, store their receiving dependents.
- std::vector<std::vector<dag_node_index>> output_consumers_;
-
- // For all nodes, store their pipeline breaking dependencies (if any).
- std::vector<std::vector<dag_node_index>> blocking_dependencies_;
-
- std::unique_ptr<QueryExecutionState> query_exec_state_;
-
- std::unique_ptr<WorkOrdersContainer> workorders_container_;
-
- const int num_numa_nodes_;
-
- WorkerDirectory *workers_;
-
- friend class ForemanTest;
- FRIEND_TEST(ForemanTest, TwoNodesDAGPartiallyFilledBlocksTest);
+ std::unique_ptr<PolicyEnforcer> policy_enforcer_;
DISALLOW_COPY_AND_ASSIGN(Foreman);
};