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/21 04:10:57 UTC
[01/10] incubator-quickstep git commit: Refactored
printWorkOrderProfilingResults in Foreman. [Forced Update!]
Repository: incubator-quickstep
Updated Branches:
refs/heads/multi-shiftboss-test 4d24659e2 -> f4ad9383d (forced update)
Refactored printWorkOrderProfilingResults in Foreman.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/7fb96f3d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/7fb96f3d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/7fb96f3d
Branch: refs/heads/multi-shiftboss-test
Commit: 7fb96f3da140f4631060132b58ded2fd38191167
Parents: 76e37ca
Author: Zuyu Zhang <zu...@apache.org>
Authored: Fri Nov 18 22:07:27 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 19:41:16 2016 -0800
----------------------------------------------------------------------
query_execution/CMakeLists.txt | 3 +++
query_execution/ForemanBase.hpp | 33 +++++++++++++++++++++++++++++
query_execution/ForemanDistributed.cpp | 11 ++++++----
query_execution/ForemanDistributed.hpp | 14 +-----------
query_execution/ForemanSingleNode.cpp | 17 ++++++++-------
query_execution/ForemanSingleNode.hpp | 29 +------------------------
query_execution/PolicyEnforcerBase.hpp | 18 ++++++++--------
7 files changed, 63 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7fb96f3d/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 0f6c282..cf9d5b0 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -86,6 +86,7 @@ if (ENABLE_DISTRIBUTED)
endif(ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryexecution_ForemanBase
glog
+ quickstep_queryexecution_PolicyEnforcerBase
quickstep_threading_Thread
quickstep_utility_Macros
tmb)
@@ -98,6 +99,7 @@ if (ENABLE_DISTRIBUTED)
quickstep_catalog_Catalog_proto
quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_ForemanBase
+ quickstep_queryexecution_PolicyEnforcerBase
quickstep_queryexecution_PolicyEnforcerDistributed
quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionTypedefs
@@ -113,6 +115,7 @@ target_link_libraries(quickstep_queryexecution_ForemanSingleNode
glog
quickstep_queryexecution_AdmitRequestMessage
quickstep_queryexecution_ForemanBase
+ quickstep_queryexecution_PolicyEnforcerBase
quickstep_queryexecution_PolicyEnforcerSingleNode
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7fb96f3d/query_execution/ForemanBase.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanBase.hpp b/query_execution/ForemanBase.hpp
index 951f34d..ee6c7ce 100644
--- a/query_execution/ForemanBase.hpp
+++ b/query_execution/ForemanBase.hpp
@@ -20,6 +20,11 @@
#ifndef QUICKSTEP_QUERY_EXECUTION_FOREMAN_BASE_HPP_
#define QUICKSTEP_QUERY_EXECUTION_FOREMAN_BASE_HPP_
+#include <cstdio>
+#include <memory>
+#include <vector>
+
+#include "query_execution/PolicyEnforcerBase.hpp"
#include "threading/Thread.hpp"
#include "utility/Macros.hpp"
@@ -30,6 +35,8 @@
namespace quickstep {
+struct WorkOrderTimeEntry;
+
/** \addtogroup QueryExecution
* @{
*/
@@ -59,6 +66,30 @@ class ForemanBase : public Thread {
~ForemanBase() override {}
/**
+ * @brief Print the results of profiling individual work orders for a given
+ * query.
+ *
+ * TODO(harshad) - Add the name of the operator to the output.
+ *
+ * @param query_id The ID of the query for which the results are to be printed.
+ * @param out The file stream.
+ **/
+ virtual void printWorkOrderProfilingResults(const std::size_t query_id,
+ std::FILE *out) const = 0;
+
+ /**
+ * @brief Get the results of profiling individual work orders for a given
+ * query.
+ *
+ * @param query_id The ID of the query for which the results are to be printed.
+ * @return A vector of records, each being a single profiling entry.
+ **/
+ const std::vector<WorkOrderTimeEntry>& getWorkOrderProfilingResults(
+ const std::size_t query_id) const {
+ return policy_enforcer_->getProfilingResults(query_id);
+ }
+
+ /**
* @brief Get the TMB client ID of Foreman thread.
*
* @return TMB client ID of foreman thread.
@@ -77,6 +108,8 @@ class ForemanBase : public Thread {
// The ID of the CPU that the Foreman thread can optionally be pinned to.
const int cpu_id_;
+ std::unique_ptr<PolicyEnforcerBase> policy_enforcer_;
+
private:
DISALLOW_COPY_AND_ASSIGN(ForemanBase);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7fb96f3d/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 56b319b..aa4db17 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -28,6 +28,7 @@
#include "catalog/CatalogRelation.hpp"
#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/AdmitRequestMessage.hpp"
+#include "query_execution/PolicyEnforcerBase.hpp"
#include "query_execution/PolicyEnforcerDistributed.hpp"
#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/QueryExecutionTypedefs.hpp"
@@ -98,12 +99,12 @@ ForemanDistributed::ForemanDistributed(
bus_->RegisterClientAsReceiver(foreman_client_id_, message_type);
}
- policy_enforcer_.reset(new PolicyEnforcerDistributed(
+ policy_enforcer_ = std::make_unique<PolicyEnforcerDistributed>(
foreman_client_id_,
catalog_database_,
&shiftboss_directory_,
bus_,
- profile_individual_workorders));
+ profile_individual_workorders);
}
void ForemanDistributed::run() {
@@ -180,7 +181,8 @@ void ForemanDistributed::run() {
}
case kInitiateRebuildResponseMessage: {
// A unique case in the distributed version.
- policy_enforcer_->processInitiateRebuildResponseMessage(tagged_message);
+ static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->
+ processInitiateRebuildResponseMessage(tagged_message);
break;
}
case kSaveQueryResultResponseMessage: {
@@ -228,7 +230,8 @@ void ForemanDistributed::run() {
if (canCollectNewMessages(message_type)) {
vector<unique_ptr<S::WorkOrderMessage>> new_messages;
- policy_enforcer_->getWorkOrderProtoMessages(&new_messages);
+ static_cast<PolicyEnforcerDistributed*>(policy_enforcer_.get())->
+ getWorkOrderProtoMessages(&new_messages);
dispatchWorkOrderMessages(new_messages);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7fb96f3d/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index b42795c..7f2d2f9 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -24,7 +24,6 @@
#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/ForemanBase.hpp"
-#include "query_execution/PolicyEnforcerDistributed.hpp"
#include "query_execution/ShiftbossDirectory.hpp"
#include "utility/Macros.hpp"
@@ -68,17 +67,8 @@ class ForemanDistributed final : public ForemanBase {
~ForemanDistributed() override {}
- /**
- * @brief Print the results of profiling individual work orders for a given
- * query.
- *
- * TODO(harshad) - Add the name of the operator to the output.
- *
- * @param query_id The ID of the query for which the results are to be printed.
- * @param out The file stream.
- **/
void printWorkOrderProfilingResults(const std::size_t query_id,
- std::FILE *out) const;
+ std::FILE *out) const override;
protected:
void run() override;
@@ -120,8 +110,6 @@ class ForemanDistributed final : public ForemanBase {
CatalogDatabaseLite *catalog_database_;
- std::unique_ptr<PolicyEnforcerDistributed> policy_enforcer_;
-
// From a query id to a set of Shiftbosses that save query result.
std::unordered_map<std::size_t, std::unordered_set<std::size_t>> query_result_saved_shiftbosses_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7fb96f3d/query_execution/ForemanSingleNode.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanSingleNode.cpp b/query_execution/ForemanSingleNode.cpp
index 4661c37..dfdfb71 100644
--- a/query_execution/ForemanSingleNode.cpp
+++ b/query_execution/ForemanSingleNode.cpp
@@ -26,6 +26,7 @@
#include <vector>
#include "query_execution/AdmitRequestMessage.hpp"
+#include "query_execution/PolicyEnforcerBase.hpp"
#include "query_execution/PolicyEnforcerSingleNode.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
@@ -92,14 +93,14 @@ ForemanSingleNode::ForemanSingleNode(
bus_->RegisterClientAsReceiver(foreman_client_id_, message_type);
}
- policy_enforcer_.reset(new PolicyEnforcerSingleNode(
+ policy_enforcer_ = std::make_unique<PolicyEnforcerSingleNode>(
foreman_client_id_,
num_numa_nodes,
catalog_database_,
storage_manager_,
worker_directory_,
bus_,
- profile_individual_workorders));
+ profile_individual_workorders);
}
void ForemanSingleNode::run() {
@@ -157,7 +158,8 @@ void ForemanSingleNode::run() {
if (canCollectNewMessages(message_type)) {
vector<unique_ptr<WorkerMessage>> new_messages;
- policy_enforcer_->getWorkerMessages(&new_messages);
+ static_cast<PolicyEnforcerSingleNode*>(policy_enforcer_.get())->
+ getWorkerMessages(&new_messages);
dispatchWorkerMessages(new_messages);
}
@@ -233,13 +235,12 @@ void ForemanSingleNode::sendWorkerMessage(const size_t worker_thread_index,
CHECK(send_status == tmb::MessageBus::SendStatus::kOK);
}
-const std::vector<WorkOrderTimeEntry>& ForemanSingleNode
- ::getWorkOrderProfilingResults(const std::size_t query_id) const {
- return policy_enforcer_->getProfilingResults(query_id);
-}
-
void ForemanSingleNode::printWorkOrderProfilingResults(const std::size_t query_id,
std::FILE *out) const {
+ // TODO(harshad) - Add the CPU core ID of the operator to the output. This
+ // will require modifying the WorkerDirectory to remember worker affinities.
+ // Until then, the users can refer to the worker_affinities provided to the
+ // cli to infer the CPU core ID where a given worker is pinned.
const std::vector<WorkOrderTimeEntry> &recorded_times =
policy_enforcer_->getProfilingResults(query_id);
fputs("Query ID,Worker ID,NUMA Socket,Operator ID,Time (microseconds)\n", out);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7fb96f3d/query_execution/ForemanSingleNode.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanSingleNode.hpp b/query_execution/ForemanSingleNode.hpp
index 5a368aa..4cc7a63 100644
--- a/query_execution/ForemanSingleNode.hpp
+++ b/query_execution/ForemanSingleNode.hpp
@@ -26,7 +26,6 @@
#include <vector>
#include "query_execution/ForemanBase.hpp"
-#include "query_execution/PolicyEnforcerSingleNode.hpp"
#include "utility/Macros.hpp"
#include "tmb/id_typedefs.h"
@@ -78,32 +77,8 @@ class ForemanSingleNode final : public ForemanBase {
~ForemanSingleNode() override {}
-
- /**
- * @brief Get the results of profiling individual work orders for a given
- * query.
- *
- * @param query_id The ID of the query for which the results are to be printed.
- * @return A vector of records, each being a single profiling entry.
- **/
- const std::vector<WorkOrderTimeEntry>& getWorkOrderProfilingResults(
- const std::size_t query_id) const;
-
- /**
- * @brief Print the results of profiling individual work orders for a given
- * query.
- *
- * TODO(harshad) - Add the name of the operator to the output.
- * TODO(harshad) - Add the CPU core ID of the operator to the output. This
- * will require modifying the WorkerDirectory to remember worker affinities.
- * Until then, the users can refer to the worker_affinities provided to the
- * cli to infer the CPU core ID where a given worker is pinned.
- *
- * @param query_id The ID of the query for which the results are to be printed.
- * @param out The file stream.
- **/
void printWorkOrderProfilingResults(const std::size_t query_id,
- std::FILE *out) const;
+ std::FILE *out) const override;
protected:
void run() override;
@@ -142,8 +117,6 @@ class ForemanSingleNode final : public ForemanBase {
CatalogDatabaseLite *catalog_database_;
StorageManager *storage_manager_;
- std::unique_ptr<PolicyEnforcerSingleNode> policy_enforcer_;
-
DISALLOW_COPY_AND_ASSIGN(ForemanSingleNode);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/7fb96f3d/query_execution/PolicyEnforcerBase.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerBase.hpp b/query_execution/PolicyEnforcerBase.hpp
index 414367e..aa070b8 100644
--- a/query_execution/PolicyEnforcerBase.hpp
+++ b/query_execution/PolicyEnforcerBase.hpp
@@ -146,6 +146,15 @@ class PolicyEnforcerBase {
return workorder_time_recorder_.at(query_id);
}
+ /**
+ * @brief Admit a query to the system.
+ *
+ * @param query_handle The QueryHandle for the new query.
+ *
+ * @return Whether the query was admitted to the system.
+ **/
+ virtual bool admitQuery(QueryHandle *query_handle) = 0;
+
protected:
static constexpr std::size_t kMaxConcurrentQueries = 1;
@@ -185,15 +194,6 @@ class PolicyEnforcerBase {
private:
/**
- * @brief Admit a query to the system.
- *
- * @param query_handle The QueryHandle for the new query.
- *
- * @return Whether the query was admitted to the system.
- **/
- virtual bool admitQuery(QueryHandle *query_handle) = 0;
-
- /**
* @brief Decrement the number of queued workorders for the given worker by 1.
*
* @param proto The completion message proto received after the WorkOrder
[03/10] incubator-quickstep git commit: Moved a CLI flag into a new
file.
Posted by zu...@apache.org.
Moved a CLI flag into a new file.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/f7d15436
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/f7d15436
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/f7d15436
Branch: refs/heads/multi-shiftboss-test
Commit: f7d1543671a3d3b3e52d2fe0c8170b64796aa9d6
Parents: e7c664b
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sat Nov 19 00:18:55 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 19:56:10 2016 -0800
----------------------------------------------------------------------
CMakeLists.txt | 2 ++
cli/CMakeLists.txt | 5 +++++
cli/Flags.cpp | 43 +++++++++++++++++++++++++++++++++++++++++++
cli/Flags.hpp | 41 +++++++++++++++++++++++++++++++++++++++++
cli/QuickstepCli.cpp | 23 +++++++++--------------
5 files changed, 100 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f7d15436/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 6191de0..cd53967 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -752,6 +752,7 @@ target_link_libraries(quickstep_cli_shell
quickstep_cli_CommandExecutor
quickstep_cli_DefaultsConfigurator
quickstep_cli_DropRelation
+ quickstep_cli_Flags
quickstep_cli_InputParserUtil
quickstep_cli_LineReader
quickstep_cli_PrintToScreen
@@ -770,6 +771,7 @@ target_link_libraries(quickstep_cli_shell
quickstep_queryoptimizer_QueryProcessor
quickstep_storage_PreloaderThread
quickstep_storage_StorageConstants
+ quickstep_storage_StorageManager
quickstep_threading_ThreadIDBasedMap
quickstep_utility_ExecutionDAGVisualizer
quickstep_utility_Macros
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f7d15436/cli/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/cli/CMakeLists.txt b/cli/CMakeLists.txt
index b86821a..9928f3f 100644
--- a/cli/CMakeLists.txt
+++ b/cli/CMakeLists.txt
@@ -54,6 +54,7 @@ target_link_libraries(quickstep_cli_DropRelation
quickstep_storage_StorageBlockInfo
quickstep_storage_StorageManager
quickstep_utility_Macros)
+add_library(quickstep_cli_Flags Flags.cpp Flags.hpp)
if(USE_LINENOISE)
add_library(quickstep_cli_LineReader
@@ -110,6 +111,9 @@ if(QUICKSTEP_HAVE_LIBNUMA)
target_link_libraries(quickstep_cli_DefaultsConfigurator
${LIBNUMA_LIBRARY})
endif()
+target_link_libraries(quickstep_cli_Flags
+ quickstep_storage_StorageConstants
+ ${GFLAGS_LIB_NAME})
target_link_libraries(quickstep_cli_InputParserUtil
glog
quickstep_utility_Macros
@@ -139,6 +143,7 @@ target_link_libraries(quickstep_cli
quickstep_cli_CommandExecutor
quickstep_cli_DefaultsConfigurator
quickstep_cli_DropRelation
+ quickstep_cli_Flags
quickstep_cli_InputParserUtil
quickstep_cli_LineReader
quickstep_cli_PrintToScreen)
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f7d15436/cli/Flags.cpp
----------------------------------------------------------------------
diff --git a/cli/Flags.cpp b/cli/Flags.cpp
new file mode 100644
index 0000000..384ef71
--- /dev/null
+++ b/cli/Flags.cpp
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#include "cli/Flags.hpp"
+
+#include <string>
+
+#include "storage/StorageConstants.hpp"
+
+#include "gflags/gflags.h"
+
+namespace quickstep {
+
+static bool ValidateStoragePath(const char *flagname,
+ const std::string &value) {
+ if (!value.empty() && value.back() != kPathSeparator) {
+ FLAGS_storage_path.push_back(kPathSeparator);
+ }
+
+ return true;
+}
+DEFINE_string(storage_path, kDefaultStoragePath,
+ "Filesystem path to store the Quickstep database.");
+static const volatile bool storage_path_dummy
+ = gflags::RegisterFlagValidator(&FLAGS_storage_path, &ValidateStoragePath);
+
+} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f7d15436/cli/Flags.hpp
----------------------------------------------------------------------
diff --git a/cli/Flags.hpp b/cli/Flags.hpp
new file mode 100644
index 0000000..a623448
--- /dev/null
+++ b/cli/Flags.hpp
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_CLI_FLAGS_HPP_
+#define QUICKSTEP_CLI_FLAGS_HPP_
+
+#include "gflags/gflags_declare.h"
+
+namespace quickstep {
+
+/** \addtogroup CLI
+ * @{
+ */
+
+/**
+ * @brief A collection of common flags shared by Quickstep CLIs in both the
+ * single-node and the distributed version.
+ **/
+DECLARE_string(storage_path);
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_CLI_FLAGS_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/f7d15436/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index a4e55d7..33a9ff4 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -54,6 +54,7 @@ typedef quickstep::LineReaderDumb LineReaderImpl;
#endif
#include "cli/DefaultsConfigurator.hpp"
+#include "cli/Flags.hpp"
#include "cli/InputParserUtil.hpp"
#include "cli/PrintToScreen.hpp"
#include "parser/ParseStatement.hpp"
@@ -76,6 +77,7 @@ typedef quickstep::LineReaderDumb LineReaderImpl;
#include "storage/PreloaderThread.hpp"
#include "storage/StorageConstants.hpp"
+#include "storage/StorageManager.hpp"
#include "threading/ThreadIDBasedMap.hpp"
#include "utility/ExecutionDAGVisualizer.hpp"
#include "utility/Macros.hpp"
@@ -107,6 +109,7 @@ using quickstep::AdmitRequestMessage;
using quickstep::CatalogRelation;
using quickstep::DefaultsConfigurator;
using quickstep::DropRelation;
+using quickstep::FLAGS_storage_path;
using quickstep::ForemanSingleNode;
using quickstep::InputParserUtil;
using quickstep::MessageBusImpl;
@@ -142,8 +145,6 @@ DEFINE_bool(preload_buffer_pool, false,
"If true, pre-load all known blocks into buffer pool before "
"accepting queries (should also set --buffer_pool_slots to be "
"large enough to accomodate the entire database).");
-DEFINE_string(storage_path, kDefaultStoragePath,
- "Filesystem path to store the Quickstep database.");
DEFINE_string(worker_affinities, "",
"A comma-separated list of CPU IDs to pin worker threads to "
"(leaving this empty will cause all worker threads to inherit "
@@ -231,15 +232,9 @@ int main(int argc, char* argv[]) {
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);
- if (!fixed_storage_path.empty()
- && (fixed_storage_path.back() != quickstep::kPathSeparator)) {
- fixed_storage_path.push_back(quickstep::kPathSeparator);
- }
- quickstep::StorageManager storage_manager(fixed_storage_path);
+ quickstep::StorageManager storage_manager(FLAGS_storage_path);
- string catalog_path(fixed_storage_path);
+ string catalog_path(FLAGS_storage_path);
catalog_path.append(kCatalogFilename);
if (quickstep::FLAGS_initialize_db) { // Initialize the database
// TODO(jmp): Refactor the code in this file!
@@ -249,14 +244,14 @@ int main(int argc, char* argv[]) {
// TODO(jmp): At some point, likely in C++-17, we will just have the
// filesystem path, and we can clean this up
#ifdef QUICKSTEP_OS_WINDOWS
- std::filesystem::create_directories(fixed_storage_path);
- LOG(FATAL) << "Failed when attempting to create the directory: " << fixed_storage_path << "\n";
+ std::filesystem::create_directories(FLAGS_storage_path);
+ LOG(FATAL) << "Failed when attempting to create the directory: " << FLAGS_storage_path << "\n";
LOG(FATAL) << "Check if the directory already exists. If so, delete it or move it before initializing \n";
#else
{
- string path_name = "mkdir " + fixed_storage_path;
+ string path_name = "mkdir " + FLAGS_storage_path;
if (std::system(path_name.c_str())) {
- LOG(FATAL) << "Failed when attempting to create the directory: " << fixed_storage_path << "\n";
+ LOG(FATAL) << "Failed when attempting to create the directory: " << FLAGS_storage_path << "\n";
}
}
#endif
[04/10] incubator-quickstep git commit: Added an util to get block
domain from BlockLocator.
Posted by zu...@apache.org.
Added an util to get block domain from BlockLocator.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/56555117
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/56555117
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/56555117
Branch: refs/heads/multi-shiftboss-test
Commit: 565551173e797c65651eef0408c74b55e27a5796
Parents: f7d1543
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 20 15:17:46 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 19:58:14 2016 -0800
----------------------------------------------------------------------
query_execution/BlockLocatorUtil.cpp | 93 ++++++++++++++++++++
query_execution/BlockLocatorUtil.hpp | 59 +++++++++++++
query_execution/CMakeLists.txt | 9 ++
query_execution/tests/BlockLocator_unittest.cpp | 42 +--------
query_optimizer/tests/CMakeLists.txt | 4 +-
.../DistributedExecutionGeneratorTestRunner.cpp | 54 +++---------
.../DistributedExecutionGeneratorTestRunner.hpp | 4 +-
storage/CMakeLists.txt | 4 +-
storage/tests/DataExchange_unittest.cpp | 50 ++---------
9 files changed, 187 insertions(+), 132 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/query_execution/BlockLocatorUtil.cpp
----------------------------------------------------------------------
diff --git a/query_execution/BlockLocatorUtil.cpp b/query_execution/BlockLocatorUtil.cpp
new file mode 100644
index 0000000..d2d1e96
--- /dev/null
+++ b/query_execution/BlockLocatorUtil.cpp
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#include "query_execution/BlockLocatorUtil.hpp"
+
+#include <cstdlib>
+#include <string>
+#include <utility>
+
+#include "query_execution/QueryExecutionMessages.pb.h"
+#include "query_execution/QueryExecutionTypedefs.hpp"
+#include "storage/StorageBlockInfo.hpp"
+
+#include "glog/logging.h"
+
+#include "tmb/address.h"
+#include "tmb/id_typedefs.h"
+#include "tmb/message_bus.h"
+#include "tmb/message_style.h"
+#include "tmb/tagged_message.h"
+
+using tmb::TaggedMessage;
+using tmb::MessageBus;
+using tmb::client_id;
+
+namespace quickstep {
+namespace block_locator {
+
+namespace S = ::quickstep::serialization;
+
+block_id_domain getBlockDomain(const std::string &network_address,
+ const client_id cli_id,
+ client_id *locator_client_id,
+ MessageBus *bus) {
+ tmb::Address address;
+ address.All(true);
+ // NOTE(zuyu): The singleton BlockLocator would need only one copy of the message.
+ tmb::MessageStyle style;
+
+ S::BlockDomainRegistrationMessage proto;
+ proto.set_domain_network_address(network_address);
+
+ const int proto_length = proto.ByteSize();
+ char *proto_bytes = static_cast<char*>(std::malloc(proto_length));
+ CHECK(proto.SerializeToArray(proto_bytes, proto_length));
+
+ TaggedMessage message(static_cast<const void*>(proto_bytes),
+ proto_length,
+ kBlockDomainRegistrationMessage);
+ std::free(proto_bytes);
+
+ DLOG(INFO) << "Client (id '" << cli_id
+ << "') broadcasts BlockDomainRegistrationMessage (typed '" << kBlockDomainRegistrationMessage
+ << "') to BlockLocator.";
+
+ CHECK(MessageBus::SendStatus::kOK ==
+ bus->Send(cli_id, address, style, std::move(message)));
+
+ const tmb::AnnotatedMessage annotated_message(bus->Receive(cli_id, 0, true));
+ const TaggedMessage &tagged_message = annotated_message.tagged_message;
+ CHECK_EQ(kBlockDomainRegistrationResponseMessage, tagged_message.message_type());
+
+ *locator_client_id = annotated_message.sender;
+
+ DLOG(INFO) << "Client (id '" << cli_id
+ << "') received BlockDomainRegistrationResponseMessage (typed '"
+ << kBlockDomainRegistrationResponseMessage
+ << "') from BlockLocator (id '" << *locator_client_id << "').";
+
+ S::BlockDomainMessage response_proto;
+ CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
+
+ return static_cast<block_id_domain>(response_proto.block_domain());
+}
+
+} // namespace block_locator
+} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/query_execution/BlockLocatorUtil.hpp
----------------------------------------------------------------------
diff --git a/query_execution/BlockLocatorUtil.hpp b/query_execution/BlockLocatorUtil.hpp
new file mode 100644
index 0000000..74f65e4
--- /dev/null
+++ b/query_execution/BlockLocatorUtil.hpp
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_QUERY_EXECUTION_BLOCK_LOCATOR_UTIL_HPP_
+#define QUICKSTEP_QUERY_EXECUTION_BLOCK_LOCATOR_UTIL_HPP_
+
+#include <string>
+
+#include "storage/StorageBlockInfo.hpp"
+
+#include "tmb/id_typedefs.h"
+
+namespace tmb { class MessageBus; }
+
+namespace quickstep {
+namespace block_locator {
+
+/** \addtogroup QueryExecution
+ * @{
+ */
+
+/**
+ * @brief Broadcast to find BlockLocator to get a block domain for
+ * StorageManager with the given network address.
+ *
+ * @param network_address The network address of the StorageManager.
+ * @param cli_id The client ID of the block domain requester.
+ * @param locator_client_id The client ID of BlockLocator to set.
+ * @param bus A pointer to the TMB.
+ *
+ * @return The requested block domain.
+ **/
+block_id_domain getBlockDomain(const std::string &network_address,
+ const tmb::client_id cli_id,
+ tmb::client_id *locator_client_id,
+ tmb::MessageBus *bus);
+
+/** @} */
+
+} // namespace block_locator
+} // namespace quickstep
+
+#endif // QUICKSTEP_QUERY_EXECUTION_BLOCK_LOCATOR_UTIL_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index eec0029..719d9f3 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -31,6 +31,7 @@ endif()
add_library(quickstep_queryexecution_AdmitRequestMessage ../empty_src.cpp AdmitRequestMessage.hpp)
if (ENABLE_DISTRIBUTED)
add_library(quickstep_queryexecution_BlockLocator BlockLocator.cpp BlockLocator.hpp)
+ add_library(quickstep_queryexecution_BlockLocatorUtil BlockLocatorUtil.cpp BlockLocatorUtil.hpp)
endif(ENABLE_DISTRIBUTED)
add_library(quickstep_queryexecution_ForemanBase ../empty_src.cpp ForemanBase.hpp)
if (ENABLE_DISTRIBUTED)
@@ -83,6 +84,12 @@ if (ENABLE_DISTRIBUTED)
quickstep_threading_ThreadUtil
quickstep_utility_Macros
tmb)
+ target_link_libraries(quickstep_queryexecution_BlockLocatorUtil
+ glog
+ quickstep_queryexecution_QueryExecutionMessages_proto
+ quickstep_queryexecution_QueryExecutionTypedefs
+ quickstep_storage_StorageBlockInfo
+ tmb)
endif(ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryexecution_ForemanBase
glog
@@ -345,6 +352,7 @@ target_link_libraries(quickstep_queryexecution
if (ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryexecution
quickstep_queryexecution_BlockLocator
+ quickstep_queryexecution_BlockLocatorUtil
quickstep_queryexecution_ForemanDistributed
quickstep_queryexecution_PolicyEnforcerDistributed
quickstep_queryexecution_QueryManagerDistributed
@@ -363,6 +371,7 @@ if (ENABLE_DISTRIBUTED)
quickstep_catalog_CatalogAttribute
quickstep_catalog_CatalogRelation
quickstep_queryexecution_BlockLocator
+ quickstep_queryexecution_BlockLocatorUtil
quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/query_execution/tests/BlockLocator_unittest.cpp
----------------------------------------------------------------------
diff --git a/query_execution/tests/BlockLocator_unittest.cpp b/query_execution/tests/BlockLocator_unittest.cpp
index 465f2a3..32437c3 100644
--- a/query_execution/tests/BlockLocator_unittest.cpp
+++ b/query_execution/tests/BlockLocator_unittest.cpp
@@ -26,6 +26,7 @@
#include "catalog/CatalogAttribute.hpp"
#include "catalog/CatalogRelation.hpp"
#include "query_execution/BlockLocator.hpp"
+#include "query_execution/BlockLocatorUtil.hpp"
#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
@@ -71,7 +72,6 @@ class BlockLocatorTest : public ::testing::Test {
bus_.Initialize();
locator_.reset(new BlockLocator(&bus_));
- locator_client_id_ = locator_->getBusClientID();
locator_->start();
worker_client_id_ = bus_.Connect();
@@ -84,7 +84,9 @@ class BlockLocatorTest : public ::testing::Test {
bus_.RegisterClientAsSender(worker_client_id_, kPoisonMessage);
- block_domain_ = getBlockDomain(kDomainNetworkAddress);
+ block_domain_ =
+ block_locator::getBlockDomain(kDomainNetworkAddress, worker_client_id_, &locator_client_id_, &bus_);
+ DCHECK_EQ(locator_->getBusClientID(), locator_client_id_);
storage_manager_.reset(
new StorageManager(kStoragePath, block_domain_, locator_client_id_, &bus_));
@@ -168,42 +170,6 @@ class BlockLocatorTest : public ::testing::Test {
unique_ptr<StorageManager> storage_manager_;
private:
- block_id_domain getBlockDomain(const string &network_address) {
- serialization::BlockDomainRegistrationMessage proto;
- proto.set_domain_network_address(network_address);
-
- const int proto_length = proto.ByteSize();
- char *proto_bytes = static_cast<char*>(malloc(proto_length));
- CHECK(proto.SerializeToArray(proto_bytes, proto_length));
-
- TaggedMessage message(static_cast<const void*>(proto_bytes),
- proto_length,
- kBlockDomainRegistrationMessage);
- free(proto_bytes);
-
- LOG(INFO) << "Worker (id '" << worker_client_id_
- << "') sent BlockDomainRegistrationMessage (typed '" << kBlockDomainRegistrationMessage
- << "') to BlockLocator (id '" << locator_client_id_ << "')";
-
- CHECK(MessageBus::SendStatus::kOK ==
- QueryExecutionUtil::SendTMBMessage(&bus_,
- worker_client_id_,
- locator_client_id_,
- move(message)));
-
- const AnnotatedMessage annotated_message(bus_.Receive(worker_client_id_, 0, true));
- const TaggedMessage &tagged_message = annotated_message.tagged_message;
- EXPECT_EQ(locator_client_id_, annotated_message.sender);
- EXPECT_EQ(kBlockDomainRegistrationResponseMessage, tagged_message.message_type());
- LOG(INFO) << "Worker (id '" << worker_client_id_
- << "') received BlockDomainRegistrationResponseMessage from BlockLocator";
-
- serialization::BlockDomainMessage response_proto;
- CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- return static_cast<block_id_domain>(response_proto.block_domain());
- }
-
MessageBusImpl bus_;
unique_ptr<BlockLocator> locator_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/query_optimizer/tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/CMakeLists.txt b/query_optimizer/tests/CMakeLists.txt
index 9c764e4..b987322 100644
--- a/query_optimizer/tests/CMakeLists.txt
+++ b/query_optimizer/tests/CMakeLists.txt
@@ -115,8 +115,8 @@ if (ENABLE_DISTRIBUTED)
quickstep_parser_ParseStatement
quickstep_parser_SqlParserWrapper
quickstep_queryexecution_BlockLocator
+ quickstep_queryexecution_BlockLocatorUtil
quickstep_queryexecution_ForemanDistributed
- quickstep_queryexecution_QueryExecutionMessages_proto
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
quickstep_queryexecution_Shiftboss
@@ -127,7 +127,7 @@ if (ENABLE_DISTRIBUTED)
quickstep_queryoptimizer_QueryHandle
quickstep_queryoptimizer_tests_TestDatabaseLoader
quickstep_storage_DataExchangerAsync
- quickstep_storage_StorageBlockInfo
+ quickstep_storage_StorageManager
quickstep_utility_Macros
quickstep_utility_MemStream
quickstep_utility_SqlError
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.cpp b/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.cpp
index 0403e77..2351dcd 100644
--- a/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.cpp
+++ b/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.cpp
@@ -32,14 +32,14 @@
#include "cli/PrintToScreen.hpp"
#include "parser/ParseStatement.hpp"
#include "query_execution/BlockLocator.hpp"
+#include "query_execution/BlockLocatorUtil.hpp"
#include "query_execution/ForemanDistributed.hpp"
-#include "query_execution/QueryExecutionMessages.pb.h"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
#include "query_optimizer/OptimizerContext.hpp"
#include "query_optimizer/QueryHandle.hpp"
#include "storage/DataExchangerAsync.hpp"
-#include "storage/StorageBlockInfo.hpp"
+#include "storage/StorageManager.hpp"
#include "utility/MemStream.hpp"
#include "utility/SqlError.hpp"
@@ -81,14 +81,15 @@ DistributedExecutionGeneratorTestRunner::DistributedExecutionGeneratorTestRunner
bus_.RegisterClientAsReceiver(cli_id_, kBlockDomainRegistrationResponseMessage);
block_locator_ = make_unique<BlockLocator>(&bus_);
- locator_client_id_ = block_locator_->getBusClientID();
block_locator_->start();
test_database_loader_ = make_unique<TestDatabaseLoader>(
storage_path,
- getBlockDomain(test_database_loader_data_exchanger_.network_address()),
+ block_locator::getBlockDomain(
+ test_database_loader_data_exchanger_.network_address(), cli_id_, &locator_client_id_, &bus_),
locator_client_id_,
&bus_);
+ DCHECK_EQ(block_locator_->getBusClientID(), locator_client_id_);
test_database_loader_data_exchanger_.set_storage_manager(test_database_loader_->storage_manager());
test_database_loader_data_exchanger_.start();
@@ -111,7 +112,11 @@ DistributedExecutionGeneratorTestRunner::DistributedExecutionGeneratorTestRunner
make_unique<WorkerDirectory>(worker_client_ids.size(), worker_client_ids, numa_nodes));
auto storage_manager = make_unique<StorageManager>(
- storage_path, getBlockDomain(data_exchangers_[i].network_address()), locator_client_id_, &bus_);
+ storage_path,
+ block_locator::getBlockDomain(
+ data_exchangers_[i].network_address(), cli_id_, &locator_client_id_, &bus_),
+ locator_client_id_, &bus_);
+ DCHECK_EQ(block_locator_->getBusClientID(), locator_client_id_);
data_exchangers_[i].set_storage_manager(storage_manager.get());
shiftbosses_.push_back(
@@ -193,44 +198,5 @@ void DistributedExecutionGeneratorTestRunner::runTestCase(
}
}
-block_id_domain DistributedExecutionGeneratorTestRunner::getBlockDomain(
- const string &network_address) {
- serialization::BlockDomainRegistrationMessage proto;
- proto.set_domain_network_address(network_address);
-
- const int proto_length = proto.ByteSize();
- char *proto_bytes = static_cast<char*>(malloc(proto_length));
- CHECK(proto.SerializeToArray(proto_bytes, proto_length));
-
- TaggedMessage message(static_cast<const void*>(proto_bytes),
- proto_length,
- kBlockDomainRegistrationMessage);
- free(proto_bytes);
-
- DLOG(INFO) << "Client (id '" << cli_id_
- << "') sent BlockDomainRegistrationMessage (typed '" << kBlockDomainRegistrationMessage
- << "') to BlockLocator (id '" << locator_client_id_ << "')";
-
- CHECK(MessageBus::SendStatus::kOK ==
- QueryExecutionUtil::SendTMBMessage(&bus_,
- cli_id_,
- locator_client_id_,
- move(message)));
-
- const tmb::AnnotatedMessage annotated_message(bus_.Receive(cli_id_, 0, true));
- const TaggedMessage &tagged_message = annotated_message.tagged_message;
- CHECK_EQ(locator_client_id_, annotated_message.sender);
- CHECK_EQ(kBlockDomainRegistrationResponseMessage, tagged_message.message_type());
- DLOG(INFO) << "Client (id '" << cli_id_
- << "') received BlockDomainRegistrationResponseMessage (typed '"
- << kBlockDomainRegistrationResponseMessage
- << "') from BlockLocator";
-
- serialization::BlockDomainMessage response_proto;
- CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- return static_cast<block_id_domain>(response_proto.block_domain());
-}
-
} // namespace optimizer
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp b/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
index d2b13e4..63e320d 100644
--- a/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
+++ b/query_optimizer/tests/DistributedExecutionGeneratorTestRunner.hpp
@@ -38,7 +38,7 @@
#include "query_optimizer/Optimizer.hpp"
#include "query_optimizer/tests/TestDatabaseLoader.hpp"
#include "storage/DataExchangerAsync.hpp"
-#include "storage/StorageBlockInfo.hpp"
+#include "storage/StorageManager.hpp"
#include "utility/Macros.hpp"
#include "utility/textbased_test/TextBasedTestRunner.hpp"
@@ -115,8 +115,6 @@ class DistributedExecutionGeneratorTestRunner : public TextBasedTestRunner {
std::string *output) override;
private:
- block_id_domain getBlockDomain(const std::string &network_address);
-
std::size_t query_id_;
SqlParserWrapper sql_parser_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/storage/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/storage/CMakeLists.txt b/storage/CMakeLists.txt
index be60662..559d86d 100644
--- a/storage/CMakeLists.txt
+++ b/storage/CMakeLists.txt
@@ -862,7 +862,7 @@ target_link_libraries(quickstep_storage_PartitionedHashTablePool
quickstep_storage_FastHashTableFactory
quickstep_storage_HashTableBase
quickstep_utility_Macros
- quickstep_utility_StringUtil)
+ quickstep_utility_StringUtil)
target_link_libraries(quickstep_storage_PreloaderThread
glog
quickstep_catalog_CatalogDatabase
@@ -1502,7 +1502,7 @@ if (ENABLE_DISTRIBUTED)
quickstep_catalog_CatalogRelation
quickstep_catalog_CatalogTypedefs
quickstep_queryexecution_BlockLocator
- quickstep_queryexecution_QueryExecutionMessages_proto
+ quickstep_queryexecution_BlockLocatorUtil
quickstep_queryexecution_QueryExecutionTypedefs
quickstep_queryexecution_QueryExecutionUtil
quickstep_storage_CountedReference
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/56555117/storage/tests/DataExchange_unittest.cpp
----------------------------------------------------------------------
diff --git a/storage/tests/DataExchange_unittest.cpp b/storage/tests/DataExchange_unittest.cpp
index 9c75150..ac39728 100644
--- a/storage/tests/DataExchange_unittest.cpp
+++ b/storage/tests/DataExchange_unittest.cpp
@@ -27,7 +27,7 @@
#include "catalog/CatalogRelation.hpp"
#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/BlockLocator.hpp"
-#include "query_execution/QueryExecutionMessages.pb.h"
+#include "query_execution/BlockLocatorUtil.hpp"
#include "query_execution/QueryExecutionTypedefs.hpp"
#include "query_execution/QueryExecutionUtil.hpp"
#include "storage/CountedReference.hpp"
@@ -77,7 +77,6 @@ class DataExchangeTest : public ::testing::Test {
bus_.Initialize();
locator_.reset(new BlockLocator(&bus_));
- locator_client_id_ = locator_->getBusClientID();
locator_->start();
worker_client_id_ = bus_.Connect();
@@ -88,18 +87,22 @@ class DataExchangeTest : public ::testing::Test {
storage_manager_expected_.reset(new StorageManager(
kStoragePath,
- getBlockDomain(data_exchanger_expected_.network_address()),
+ block_locator::getBlockDomain(
+ data_exchanger_expected_.network_address(), worker_client_id_, &locator_client_id_, &bus_),
locator_client_id_,
&bus_));
+ DCHECK_EQ(locator_->getBusClientID(), locator_client_id_);
data_exchanger_expected_.set_storage_manager(storage_manager_expected_.get());
data_exchanger_expected_.start();
storage_manager_checked_.reset(new StorageManager(
kStoragePath,
- getBlockDomain(kCheckedDomainNetworkAddress),
+ block_locator::getBlockDomain(
+ kCheckedDomainNetworkAddress, worker_client_id_, &locator_client_id_, &bus_),
locator_client_id_,
&bus_));
+ DCHECK_EQ(locator_->getBusClientID(), locator_client_id_);
}
virtual void TearDown() {
@@ -123,45 +126,6 @@ class DataExchangeTest : public ::testing::Test {
unique_ptr<StorageManager> storage_manager_expected_, storage_manager_checked_;
private:
- block_id_domain getBlockDomain(const string &network_address) {
- serialization::BlockDomainRegistrationMessage proto;
- proto.set_domain_network_address(network_address);
-
- const int proto_length = proto.ByteSize();
- char *proto_bytes = static_cast<char*>(malloc(proto_length));
- CHECK(proto.SerializeToArray(proto_bytes, proto_length));
-
- TaggedMessage message(static_cast<const void*>(proto_bytes),
- proto_length,
- kBlockDomainRegistrationMessage);
- free(proto_bytes);
-
- LOG(INFO) << "Worker (id '" << worker_client_id_
- << "') sent BlockDomainRegistrationMessage (typed '" << kBlockDomainRegistrationMessage
- << "') to BlockLocator (id '" << locator_client_id_ << "')";
-
- CHECK(MessageBus::SendStatus::kOK ==
- QueryExecutionUtil::SendTMBMessage(&bus_,
- worker_client_id_,
- locator_client_id_,
- move(message)));
-
- const tmb::AnnotatedMessage annotated_message(bus_.Receive(worker_client_id_, 0, true));
- const TaggedMessage &tagged_message = annotated_message.tagged_message;
- EXPECT_EQ(locator_client_id_, annotated_message.sender);
- EXPECT_EQ(kBlockDomainRegistrationResponseMessage, tagged_message.message_type());
-
- LOG(INFO) << "Worker (id '" << worker_client_id_
- << "') received BlockDomainRegistrationResponseMessage (typed '"
- << kBlockDomainRegistrationResponseMessage
- << "') from BlockLocator";
-
- serialization::BlockDomainMessage response_proto;
- CHECK(response_proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));
-
- return static_cast<block_id_domain>(response_proto.block_domain());
- }
-
MessageBusImpl bus_;
unique_ptr<BlockLocator> locator_;
[02/10] incubator-quickstep git commit: Refactor regarding the
profile flag.
Posted by zu...@apache.org.
Refactor regarding the profile flag.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/e7c664ba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/e7c664ba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/e7c664ba
Branch: refs/heads/multi-shiftboss-test
Commit: e7c664baa216fad0c7d04f08cad8398d6d9db2bd
Parents: 7fb96f3
Author: Zuyu Zhang <zu...@apache.org>
Authored: Fri Nov 18 17:38:17 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 19:51:25 2016 -0800
----------------------------------------------------------------------
cli/QuickstepCli.cpp | 13 ++++---------
query_execution/CMakeLists.txt | 3 ++-
query_execution/ForemanDistributed.cpp | 6 ++----
query_execution/ForemanDistributed.hpp | 5 +----
query_execution/ForemanSingleNode.cpp | 6 ++----
query_execution/ForemanSingleNode.hpp | 5 +----
query_execution/PolicyEnforcerBase.cpp | 15 +++++++++++++++
query_execution/PolicyEnforcerBase.hpp | 6 +-----
query_execution/PolicyEnforcerDistributed.hpp | 5 ++---
query_execution/PolicyEnforcerSingleNode.hpp | 5 ++---
10 files changed, 32 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/cli/QuickstepCli.cpp
----------------------------------------------------------------------
diff --git a/cli/QuickstepCli.cpp b/cli/QuickstepCli.cpp
index 0e8468e..a4e55d7 100644
--- a/cli/QuickstepCli.cpp
+++ b/cli/QuickstepCli.cpp
@@ -133,9 +133,6 @@ using tmb::client_id;
namespace quickstep {
-DEFINE_bool(profile_and_report_workorder_perf, false,
- "If true, Quickstep will record the exceution time of all the individual "
- "normal work orders and report it at the end of query execution.");
DEFINE_int32(num_workers, 0, "Number of worker threads. If this value is "
"specified and is greater than 0, then this "
"user-supplied value is used. Else (i.e. the"
@@ -181,10 +178,9 @@ DEFINE_string(profile_file_name, "",
// To put things in perspective, the first run is, in my experiments, about 5-10
// times more expensive than the average run. That means the query needs to be
// run at least a hundred times to make the impact of the first run small (< 5 %).
-DEFINE_bool(visualize_execution_dag, false,
- "If true, visualize the execution plan DAG into a graph in DOT "
- "format (DOT is a plain text graph description language) which is "
- "then printed via stderr.");
+
+DECLARE_bool(profile_and_report_workorder_perf);
+DECLARE_bool(visualize_execution_dag);
} // namespace quickstep
@@ -361,8 +357,7 @@ int main(int argc, char* argv[]) {
query_processor->getDefaultDatabase(),
&storage_manager,
-1, // Don't pin the Foreman thread.
- num_numa_nodes_system,
- quickstep::FLAGS_profile_and_report_workorder_perf || quickstep::FLAGS_visualize_execution_dag);
+ num_numa_nodes_system);
// Start the worker threads.
for (Worker &worker : workers) {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index cf9d5b0..eec0029 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -138,7 +138,8 @@ target_link_libraries(quickstep_queryexecution_PolicyEnforcerBase
quickstep_relationaloperators_WorkOrder
quickstep_storage_StorageBlockInfo
quickstep_utility_Macros
- tmb)
+ tmb
+ ${GFLAGS_LIB_NAME})
if (ENABLE_DISTRIBUTED)
target_link_libraries(quickstep_queryexecution_PolicyEnforcerDistributed
glog
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index aa4db17..d619657 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -64,8 +64,7 @@ class QueryHandle;
ForemanDistributed::ForemanDistributed(
MessageBus *bus,
CatalogDatabaseLite *catalog_database,
- const int cpu_id,
- const bool profile_individual_workorders)
+ const int cpu_id)
: ForemanBase(bus, cpu_id),
catalog_database_(DCHECK_NOTNULL(catalog_database)) {
const std::vector<QueryExecutionMessageType> sender_message_types{
@@ -103,8 +102,7 @@ ForemanDistributed::ForemanDistributed(
foreman_client_id_,
catalog_database_,
&shiftboss_directory_,
- bus_,
- profile_individual_workorders);
+ bus_);
}
void ForemanDistributed::run() {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index 7f2d2f9..ccdd0ae 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -54,16 +54,13 @@ class ForemanDistributed final : public ForemanBase {
* @param bus A pointer to the TMB.
* @param catalog_database The catalog database where this query is executed.
* @param cpu_id The ID of the CPU to which the Foreman thread can be pinned.
- * @param profile_individual_workorders Whether every workorder's execution
- * be profiled or not.
*
* @note If cpu_id is not specified, Foreman thread can be possibly moved
* around on different CPUs by the OS.
**/
ForemanDistributed(tmb::MessageBus *bus,
CatalogDatabaseLite *catalog_database,
- const int cpu_id = -1,
- const bool profile_individual_workorders = false);
+ const int cpu_id = -1);
~ForemanDistributed() override {}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/ForemanSingleNode.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanSingleNode.cpp b/query_execution/ForemanSingleNode.cpp
index dfdfb71..02799c7 100644
--- a/query_execution/ForemanSingleNode.cpp
+++ b/query_execution/ForemanSingleNode.cpp
@@ -63,8 +63,7 @@ ForemanSingleNode::ForemanSingleNode(
CatalogDatabaseLite *catalog_database,
StorageManager *storage_manager,
const int cpu_id,
- const size_t num_numa_nodes,
- const bool profile_individual_workorders)
+ const size_t num_numa_nodes)
: ForemanBase(bus, cpu_id),
main_thread_client_id_(main_thread_client_id),
worker_directory_(DCHECK_NOTNULL(worker_directory)),
@@ -99,8 +98,7 @@ ForemanSingleNode::ForemanSingleNode(
catalog_database_,
storage_manager_,
worker_directory_,
- bus_,
- profile_individual_workorders);
+ bus_);
}
void ForemanSingleNode::run() {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/ForemanSingleNode.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanSingleNode.hpp b/query_execution/ForemanSingleNode.hpp
index 4cc7a63..d2db51b 100644
--- a/query_execution/ForemanSingleNode.hpp
+++ b/query_execution/ForemanSingleNode.hpp
@@ -60,8 +60,6 @@ class ForemanSingleNode final : public ForemanBase {
* @param storage_manager The StorageManager to use.
* @param cpu_id The ID of the CPU to which the Foreman thread can be pinned.
* @param num_numa_nodes The number of NUMA nodes in the system.
- * @param profile_individual_workorders Whether every workorder's execution
- * be profiled or not.
*
* @note If cpu_id is not specified, Foreman thread can be possibly moved
* around on different CPUs by the OS.
@@ -72,8 +70,7 @@ class ForemanSingleNode final : public ForemanBase {
CatalogDatabaseLite *catalog_database,
StorageManager *storage_manager,
const int cpu_id = -1,
- const std::size_t num_numa_nodes = 1,
- const bool profile_individual_workorders = false);
+ const std::size_t num_numa_nodes = 1);
~ForemanSingleNode() override {}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/PolicyEnforcerBase.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerBase.cpp b/query_execution/PolicyEnforcerBase.cpp
index 6e922a8..b799d5f 100644
--- a/query_execution/PolicyEnforcerBase.cpp
+++ b/query_execution/PolicyEnforcerBase.cpp
@@ -35,10 +35,25 @@
#include "relational_operators/WorkOrder.hpp"
#include "storage/StorageBlockInfo.hpp"
+#include "gflags/gflags.h"
#include "glog/logging.h"
namespace quickstep {
+DEFINE_bool(profile_and_report_workorder_perf, false,
+ "If true, Quickstep will record the exceution time of all the individual "
+ "normal work orders and report it at the end of query execution.");
+
+DEFINE_bool(visualize_execution_dag, false,
+ "If true, visualize the execution plan DAG into a graph in DOT "
+ "format (DOT is a plain text graph description language) which is "
+ "then printed via stderr.");
+
+PolicyEnforcerBase::PolicyEnforcerBase(CatalogDatabaseLite *catalog_database)
+ : catalog_database_(catalog_database),
+ profile_individual_workorders_(FLAGS_profile_and_report_workorder_perf || FLAGS_visualize_execution_dag) {
+}
+
void PolicyEnforcerBase::processMessage(const TaggedMessage &tagged_message) {
std::size_t query_id;
QueryManagerBase::dag_node_index op_index;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/PolicyEnforcerBase.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerBase.hpp b/query_execution/PolicyEnforcerBase.hpp
index aa070b8..baf9c68 100644
--- a/query_execution/PolicyEnforcerBase.hpp
+++ b/query_execution/PolicyEnforcerBase.hpp
@@ -54,12 +54,8 @@ class PolicyEnforcerBase {
* @brief Constructor.
*
* @param catalog_database The CatalogDatabase used.
- * @param profile_individual_workorders If true, profile each normal work order.
**/
- PolicyEnforcerBase(CatalogDatabaseLite *catalog_database,
- const bool profile_individual_workorders)
- : catalog_database_(catalog_database),
- profile_individual_workorders_(profile_individual_workorders) {}
+ explicit PolicyEnforcerBase(CatalogDatabaseLite *catalog_database);
/**
* @brief Virtual Destructor.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/PolicyEnforcerDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerDistributed.hpp b/query_execution/PolicyEnforcerDistributed.hpp
index 0bf249c..146e9af 100644
--- a/query_execution/PolicyEnforcerDistributed.hpp
+++ b/query_execution/PolicyEnforcerDistributed.hpp
@@ -59,9 +59,8 @@ class PolicyEnforcerDistributed final : public PolicyEnforcerBase {
PolicyEnforcerDistributed(const tmb::client_id foreman_client_id,
CatalogDatabaseLite *catalog_database,
ShiftbossDirectory *shiftboss_directory,
- tmb::MessageBus *bus,
- const bool profile_individual_workorders = false)
- : PolicyEnforcerBase(catalog_database, profile_individual_workorders),
+ tmb::MessageBus *bus)
+ : PolicyEnforcerBase(catalog_database),
foreman_client_id_(foreman_client_id),
shiftboss_directory_(shiftboss_directory),
bus_(bus) {}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/e7c664ba/query_execution/PolicyEnforcerSingleNode.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PolicyEnforcerSingleNode.hpp b/query_execution/PolicyEnforcerSingleNode.hpp
index 16c7a0d..f87d670 100644
--- a/query_execution/PolicyEnforcerSingleNode.hpp
+++ b/query_execution/PolicyEnforcerSingleNode.hpp
@@ -64,9 +64,8 @@ class PolicyEnforcerSingleNode final : public PolicyEnforcerBase {
CatalogDatabaseLite *catalog_database,
StorageManager *storage_manager,
WorkerDirectory *worker_directory,
- tmb::MessageBus *bus,
- const bool profile_individual_workorders = false)
- : PolicyEnforcerBase(catalog_database, profile_individual_workorders),
+ tmb::MessageBus *bus)
+ : PolicyEnforcerBase(catalog_database),
foreman_client_id_(foreman_client_id),
num_numa_nodes_(num_numa_nodes),
storage_manager_(storage_manager),
[10/10] 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/86d40d53
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/86d40d53
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/86d40d53
Branch: refs/heads/multi-shiftboss-test
Commit: 86d40d5321fd0e4f0ea789d34e9f8054982cac4d
Parents: c886f7a
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sat Nov 12 22:31:42 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 20:10:49 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/86d40d53/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/86d40d53/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/86d40d53/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/86d40d53/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/86d40d53/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
[08/10] incubator-quickstep git commit: A better debug log for worker.
Posted by zu...@apache.org.
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/f4ad9383
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/f4ad9383
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/f4ad9383
Branch: refs/heads/multi-shiftboss-test
Commit: f4ad9383d1c00125521c395757d0badf6a21f2fe
Parents: 8f5e71f
Author: Zuyu Zhang <zu...@apache.org>
Authored: Thu Nov 17 15:10:03 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 20:10:49 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/f4ad9383/query_execution/Worker.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Worker.cpp b/query_execution/Worker.cpp
index 31eb964..ec6c503 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"
+#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(
[06/10] incubator-quickstep git commit: Used the network util in
DataExchanger.
Posted by zu...@apache.org.
Used the network util in DataExchanger.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/c886f7af
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/c886f7af
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/c886f7af
Branch: refs/heads/multi-shiftboss-test
Commit: c886f7af5835a8577f516b2c7a39692d18d5b9b2
Parents: 8f1b0e5
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 13 15:55:31 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 20:08:27 2016 -0800
----------------------------------------------------------------------
storage/CMakeLists.txt | 1 +
storage/DataExchangerAsync.cpp | 9 +++++----
storage/DataExchangerAsync.hpp | 2 --
3 files changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c886f7af/storage/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/storage/CMakeLists.txt b/storage/CMakeLists.txt
index 559d86d..61a8a99 100644
--- a/storage/CMakeLists.txt
+++ b/storage/CMakeLists.txt
@@ -625,6 +625,7 @@ if (ENABLE_DISTRIBUTED)
quickstep_storage_StorageManager
quickstep_threading_Thread
quickstep_utility_Macros
+ quickstep_utility_NetworkUtil
${GRPCPLUSPLUS_LIBRARIES})
endif()
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c886f7af/storage/DataExchangerAsync.cpp
----------------------------------------------------------------------
diff --git a/storage/DataExchangerAsync.cpp b/storage/DataExchangerAsync.cpp
index 1d2f7db..d349430 100644
--- a/storage/DataExchangerAsync.cpp
+++ b/storage/DataExchangerAsync.cpp
@@ -28,6 +28,7 @@
#include "storage/DataExchange.grpc.pb.h"
#include "storage/DataExchange.pb.h"
#include "storage/StorageManager.hpp"
+#include "utility/NetworkUtil.hpp"
#include "glog/logging.h"
@@ -127,18 +128,18 @@ void CallContext::Proceed() {
} // namespace
-const char *DataExchangerAsync::kLocalNetworkAddress = "0.0.0.0:";
-
DataExchangerAsync::DataExchangerAsync() {
+ const std::string ipv4_address(GetIpv4Address() + ':');
+
grpc::ServerBuilder builder;
- builder.AddListeningPort(kLocalNetworkAddress, grpc::InsecureServerCredentials(), &port_);
+ builder.AddListeningPort(ipv4_address, grpc::InsecureServerCredentials(), &port_);
builder.RegisterService(&service_);
queue_ = builder.AddCompletionQueue();
server_ = builder.BuildAndStart();
DCHECK_GT(port_, 0);
- server_address_ = kLocalNetworkAddress + std::to_string(port_);
+ server_address_ = ipv4_address + std::to_string(port_);
LOG(INFO) << "DataExchangerAsync Service listening on " << server_address_;
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c886f7af/storage/DataExchangerAsync.hpp
----------------------------------------------------------------------
diff --git a/storage/DataExchangerAsync.hpp b/storage/DataExchangerAsync.hpp
index b777108..4504c08 100644
--- a/storage/DataExchangerAsync.hpp
+++ b/storage/DataExchangerAsync.hpp
@@ -79,8 +79,6 @@ class DataExchangerAsync final : public Thread {
void run() override;
private:
- static const char *kLocalNetworkAddress;
-
DataExchange::AsyncService service_;
int port_ = -1;
[07/10] 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/a62129fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a62129fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a62129fe
Branch: refs/heads/multi-shiftboss-test
Commit: a62129fe0dfdfd976d8c2985c1bb1ebba18b0471
Parents: 86d40d5
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 13 20:20:56 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 20:10:49 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/a62129fe/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/a62129fe/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 7dccce4..900a71f 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 unique_ptr<S::WorkOrderMessage> &message,
+ const size_t next_shiftboss_index_to_schedule,
+ size_t *shiftboss_index_for_hash_join) {
+ const S::WorkOrder &work_order_proto = message->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;
+ }
+
+ policy_enforcer_->getShiftbossIndexForHashJoin(
+ message->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);
+ size_t shiftboss_index_for_hash_join;
+ if (isHashJoinRelatedWorkOrder(message, shiftboss_index, &shiftboss_index_for_hash_join)) {
+ sendWorkOrderMessage(shiftboss_index_for_hash_join, *message);
+ 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, *message);
+ 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/a62129fe/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index ccdd0ae..55e747f 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 std::unique_ptr<serialization::WorkOrderMessage> &message,
+ 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/a62129fe/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/a62129fe/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/a62129fe/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);
};
[09/10] incubator-quickstep git commit: Scheduled FinalizeAggr /
DestroyAggr WorkOrder on the same Shiftboss of AggrWorkOrder.
Posted by zu...@apache.org.
Scheduled FinalizeAggr / DestroyAggr WorkOrder on the same Shiftboss of AggrWorkOrder.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/8f5e71f9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/8f5e71f9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/8f5e71f9
Branch: refs/heads/multi-shiftboss-test
Commit: 8f5e71f94de468084a4256a9ed3a39b4c79638ce
Parents: a62129f
Author: Zuyu Zhang <zu...@apache.org>
Authored: Thu Nov 17 12:01:26 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 20:10:49 2016 -0800
----------------------------------------------------------------------
query_execution/ForemanDistributed.cpp | 71 +++++++++++++++++-----
query_execution/ForemanDistributed.hpp | 9 ++-
query_execution/PolicyEnforcerDistributed.cpp | 12 ++++
query_execution/PolicyEnforcerDistributed.hpp | 18 ++++++
query_execution/QueryManagerDistributed.hpp | 23 +++++++
5 files changed, 116 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8f5e71f9/query_execution/ForemanDistributed.cpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.cpp b/query_execution/ForemanDistributed.cpp
index 900a71f..7ca246d 100644
--- a/query_execution/ForemanDistributed.cpp
+++ b/query_execution/ForemanDistributed.cpp
@@ -243,28 +243,55 @@ bool ForemanDistributed::canCollectNewMessages(const tmb::message_type_id messag
kWorkOrderFeedbackMessage);
}
-bool ForemanDistributed::isHashJoinRelatedWorkOrder(const unique_ptr<S::WorkOrderMessage> &message,
+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;
+ }
+
+ policy_enforcer_->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) {
- const S::WorkOrder &work_order_proto = message->work_order();
+ 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);
+ 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;
}
policy_enforcer_->getShiftbossIndexForHashJoin(
- message->query_id(), join_hash_table_index, next_shiftboss_index_to_schedule,
+ proto.query_id(), join_hash_table_index, next_shiftboss_index_to_schedule,
shiftboss_index_for_hash_join);
return true;
@@ -275,14 +302,11 @@ void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::Wo
size_t shiftboss_index = 0u;
for (const auto &message : messages) {
DCHECK(message != nullptr);
- size_t shiftboss_index_for_hash_join;
- if (isHashJoinRelatedWorkOrder(message, shiftboss_index, &shiftboss_index_for_hash_join)) {
- sendWorkOrderMessage(shiftboss_index_for_hash_join, *message);
- 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(*message, shiftboss_index, &shiftboss_index_for_particular_work_order_type)) {
+ dispatchWorkOrderMessagesHelper(*message, shiftboss_index_for_particular_work_order_type, &shiftboss_index);
+ } else if (isHashJoinRelatedWorkOrder(*message, shiftboss_index, &shiftboss_index_for_particular_work_order_type)) {
+ dispatchWorkOrderMessagesHelper(*message, shiftboss_index_for_particular_work_order_type, &shiftboss_index);
} else {
sendWorkOrderMessage(shiftboss_index, *message);
shiftboss_directory_.incrementNumQueuedWorkOrders(shiftboss_index);
@@ -293,6 +317,21 @@ void ForemanDistributed::dispatchWorkOrderMessages(const vector<unique_ptr<S::Wo
}
}
+void ForemanDistributed::dispatchWorkOrderMessagesHelper(const S::WorkOrderMessage &proto,
+ const size_t shiftboss_index_for_particular_work_order_type,
+ size_t *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) % shiftboss_directory_.size();
+ } else {
+ // NOTE(zuyu): This is not the exact round-robin scheduling, as in this case,
+ // <shiftboss_index_for_particular_work_order_type> will be scheduled one
+ // more WorkOrder.
+ }
+}
+
void ForemanDistributed::sendWorkOrderMessage(const size_t shiftboss_index,
const S::WorkOrderMessage &proto) {
const size_t proto_length = proto.ByteSize();
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8f5e71f9/query_execution/ForemanDistributed.hpp
----------------------------------------------------------------------
diff --git a/query_execution/ForemanDistributed.hpp b/query_execution/ForemanDistributed.hpp
index 55e747f..f6d8597 100644
--- a/query_execution/ForemanDistributed.hpp
+++ b/query_execution/ForemanDistributed.hpp
@@ -71,7 +71,11 @@ class ForemanDistributed final : public ForemanBase {
void run() override;
private:
- bool isHashJoinRelatedWorkOrder(const std::unique_ptr<serialization::WorkOrderMessage> &message,
+ 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);
@@ -84,6 +88,9 @@ class ForemanDistributed final : public ForemanBase {
void dispatchWorkOrderMessages(
const std::vector<std::unique_ptr<serialization::WorkOrderMessage>> &messages);
+ void dispatchWorkOrderMessagesHelper(const serialization::WorkOrderMessage &proto,
+ const std::size_t shiftboss_index_for_particular_work_order_type,
+ std::size_t *shiftboss_index);
/**
* @brief Send the given message to the specified worker.
*
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8f5e71f9/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/8f5e71f9/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/8f5e71f9/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_;
[05/10] incubator-quickstep git commit: Added an util method to get
the ip address.
Posted by zu...@apache.org.
Added an util method to get the ip address.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/8f1b0e56
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/8f1b0e56
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/8f1b0e56
Branch: refs/heads/multi-shiftboss-test
Commit: 8f1b0e56947b49f544bb6e0f5a4b425e8735ccf9
Parents: 5655511
Author: Zuyu Zhang <zu...@apache.org>
Authored: Sun Nov 13 15:13:09 2016 -0800
Committer: Zuyu Zhang <zu...@apache.org>
Committed: Sun Nov 20 20:04:21 2016 -0800
----------------------------------------------------------------------
utility/CMakeLists.txt | 20 ++++++++
utility/NetworkUtil.cpp | 78 +++++++++++++++++++++++++++++
utility/NetworkUtil.hpp | 42 ++++++++++++++++
utility/tests/NetworkUtil_unittest.cpp | 44 ++++++++++++++++
4 files changed, 184 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8f1b0e56/utility/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/utility/CMakeLists.txt b/utility/CMakeLists.txt
index e9be2ec..6a656fb 100644
--- a/utility/CMakeLists.txt
+++ b/utility/CMakeLists.txt
@@ -15,6 +15,12 @@
# specific language governing permissions and limitations
# under the License.
+if (BUILD_SHARED_LIBS)
+ set(GFLAGS_LIB_NAME gflags_nothreads-shared)
+else()
+ set(GFLAGS_LIB_NAME gflags_nothreads-static)
+endif()
+
include(CheckCXXSourceCompiles)
# Look for GCC-style builtins (also in clang, ICC, and possibly others) that
@@ -180,6 +186,7 @@ add_library(quickstep_utility_Glob Glob.cpp Glob.hpp)
add_library(quickstep_utility_HashPair ../empty_src.cpp HashPair.hpp)
add_library(quickstep_utility_Macros ../empty_src.cpp Macros.hpp)
add_library(quickstep_utility_MemStream ../empty_src.cpp MemStream.hpp)
+add_library(quickstep_utility_NetworkUtil NetworkUtil.cpp NetworkUtil.hpp)
add_library(quickstep_utility_PlanVisualizer PlanVisualizer.cpp PlanVisualizer.hpp)
add_library(quickstep_utility_PrimeNumber PrimeNumber.cpp PrimeNumber.hpp)
add_library(quickstep_utility_PtrList ../empty_src.cpp PtrList.hpp)
@@ -257,6 +264,9 @@ target_link_libraries(quickstep_utility_Glob
target_link_libraries(quickstep_utility_MemStream
glog
quickstep_utility_Macros)
+target_link_libraries(quickstep_utility_NetworkUtil
+ glog
+ ${GFLAGS_LIB_NAME})
target_link_libraries(quickstep_utility_PrimeNumber
glog)
target_link_libraries(quickstep_utility_PlanVisualizer
@@ -339,6 +349,7 @@ target_link_libraries(quickstep_utility
quickstep_utility_HashPair
quickstep_utility_Macros
quickstep_utility_MemStream
+ quickstep_utility_NetworkUtil
quickstep_utility_PlanVisualizer
quickstep_utility_PrimeNumber
quickstep_utility_PtrList
@@ -404,6 +415,15 @@ target_link_libraries(EqualsAnyConstant_unittest
${LIBS})
add_test(EqualsAnyConstant_unittest EqualsAnyConstant_unittest)
+add_executable(NetworkUtil_unittest "${CMAKE_CURRENT_SOURCE_DIR}/tests/NetworkUtil_unittest.cpp")
+target_link_libraries(NetworkUtil_unittest
+ gtest
+ gtest_main
+ quickstep_utility_NetworkUtil
+ ${GFLAGS_LIB_NAME}
+ ${LIBS})
+add_test(NetworkUtil_unittest NetworkUtil_unittest)
+
add_executable(PrimeNumber_unittest "${CMAKE_CURRENT_SOURCE_DIR}/tests/PrimeNumber_unittest.cpp")
target_link_libraries(PrimeNumber_unittest
gtest
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8f1b0e56/utility/NetworkUtil.cpp
----------------------------------------------------------------------
diff --git a/utility/NetworkUtil.cpp b/utility/NetworkUtil.cpp
new file mode 100644
index 0000000..f665b62
--- /dev/null
+++ b/utility/NetworkUtil.cpp
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#include "utility/NetworkUtil.hpp"
+
+#include "utility/UtilityConfig.h" // For QUICKSTEP_HAVE_WINDOWS.
+
+#ifndef QUICKSTEP_HAVE_WINDOWS
+#include <arpa/inet.h>
+#include <net/if.h> // Need to include before <ifaddrs.h> for NetBSD.
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <ifaddrs.h>
+#include <netdb.h>
+
+#include <cstring>
+#endif // QUICKSTEP_HAVE_WINDOWS
+
+#include <string>
+
+#include "gflags/gflags.h"
+#include "glog/logging.h"
+
+namespace quickstep {
+
+DEFINE_bool(use_ethernet_ip, false,
+ "Use the Ethernet ip address, instead of the loopback.");
+
+std::string GetIpv4Address() {
+#ifndef QUICKSTEP_HAVE_WINDOWS
+ struct ifaddrs *ifaddrs;
+ CHECK_EQ(0, getifaddrs(&ifaddrs));
+
+ char ip_address[NI_MAXHOST] = { '\0' };
+
+ for (struct ifaddrs *ifa = ifaddrs; ifa; ifa = ifa->ifa_next) {
+ if (ifa->ifa_addr == nullptr ||
+ ifa->ifa_addr->sa_family != AF_INET ||
+ (ifa->ifa_flags & IFF_UP) == 0) {
+ continue;
+ }
+
+ if (FLAGS_use_ethernet_ip && !std::strncmp(ifa->ifa_name, "lo", 2)) {
+ // NOTE: On Linux it starts with 'eth', while on Mac OS X, 'en'.
+ continue;
+ }
+
+ struct sockaddr_in *s4 = (struct sockaddr_in *)(ifa->ifa_addr);
+ CHECK(inet_ntop(AF_INET, reinterpret_cast<void *>(&(s4->sin_addr)), ip_address, sizeof(ip_address)) != nullptr);
+ break;
+ }
+
+ freeifaddrs(ifaddrs);
+
+ return ip_address;
+#else
+ CHECK(!FLAGS_use_ethernet_ip);
+ return kLocalIpv4Address;
+#endif // QUICKSTEP_HAVE_WINDOWS
+}
+
+} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8f1b0e56/utility/NetworkUtil.hpp
----------------------------------------------------------------------
diff --git a/utility/NetworkUtil.hpp b/utility/NetworkUtil.hpp
new file mode 100644
index 0000000..b825fd8
--- /dev/null
+++ b/utility/NetworkUtil.hpp
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_UTILITY_NETWORK_UTIL_HPP_
+#define QUICKSTEP_UTILITY_NETWORK_UTIL_HPP_
+
+#include <string>
+
+namespace quickstep {
+
+/** \addtogroup Utility
+ * @{
+ */
+
+constexpr char kLocalIpv4Address[] = "127.0.0.1";
+
+/**
+ * @brief Get the IPv4 network address in the text format, i.e., "127.0.0.1".
+ */
+extern std::string GetIpv4Address();
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_UTILITY_NETWORK_UTIL_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8f1b0e56/utility/tests/NetworkUtil_unittest.cpp
----------------------------------------------------------------------
diff --git a/utility/tests/NetworkUtil_unittest.cpp b/utility/tests/NetworkUtil_unittest.cpp
new file mode 100644
index 0000000..0013d7c
--- /dev/null
+++ b/utility/tests/NetworkUtil_unittest.cpp
@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#include <string>
+
+#include "utility/NetworkUtil.hpp"
+
+#include "gflags/gflags_declare.h"
+#include "gtest/gtest.h"
+
+using std::string;
+
+namespace quickstep {
+
+DECLARE_bool(use_ethernet_ip);
+
+TEST(NetworkUtilTest, LoopbackTest) {
+ string lo_ip_address = GetIpv4Address();
+ EXPECT_STREQ(kLocalIpv4Address, lo_ip_address.c_str());
+}
+
+TEST(NetworkUtilTest, EthernetTest) {
+ FLAGS_use_ethernet_ip = true;
+ string ethernet_ip_address = GetIpv4Address();
+ EXPECT_STRNE(kLocalIpv4Address, ethernet_ip_address.c_str());
+}
+
+} // namespace quickstep