quickstep-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hbdeshm...@apache.org
Subject [16/19] incubator-quickstep git commit: Select query from learner in PolicyEnforcer.
Date Wed, 29 Jun 2016 21:06:45 GMT
Select query from learner in PolicyEnforcer.


Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/a70b1eca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a70b1eca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a70b1eca

Branch: refs/heads/scheduler++
Commit: a70b1ecadc5daa1ddc9a1302f6ea4874a2b91dbe
Parents: ed93b48
Author: Harshad Deshmukh <hbdeshmukh@apache.org>
Authored: Tue Jun 28 12:25:02 2016 -0500
Committer: Harshad Deshmukh <hbdeshmukh@apache.org>
Committed: Wed Jun 29 16:06:07 2016 -0500

----------------------------------------------------------------------
 query_execution/Learner.cpp                |  5 +-
 query_execution/Learner.hpp                | 31 +++++----
 query_execution/PriorityPolicyEnforcer.cpp | 86 +++++++++++++++++++++----
 query_execution/PriorityPolicyEnforcer.hpp | 19 ++++++
 4 files changed, 112 insertions(+), 29 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a70b1eca/query_execution/Learner.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.cpp b/query_execution/Learner.cpp
index 11c3735..bb24baa 100644
--- a/query_execution/Learner.cpp
+++ b/query_execution/Learner.cpp
@@ -49,7 +49,7 @@ void Learner::addCompletionFeedback(
         &workorder_completion_proto) {
   const std::size_t query_id = workorder_completion_proto.query_id();
   DCHECK(isQueryPresent(query_id));
-  const std::size_t priority_level = getQueryPriority(query_id);
+  const std::size_t priority_level = getQueryPriorityUnsafe(query_id);
   ExecutionStats *execution_stats = getExecutionStats(query_id);
   DCHECK(execution_stats != nullptr);
   execution_stats->addEntry(
@@ -106,12 +106,10 @@ void Learner::updateProbabilitiesForQueriesInPriorityLevel(
         query_id,
         1 / static_cast<float>(mean_workorders_per_query[query_id]),
         denominator);
-    // LOG(INFO) << "Added stats on query ID: " << query_id << " priority:
" << priority_level;
   } else {
     // At least one of the queries has predicted time for next work order as 0.
     // In such a case, we don't update the probabilities and continue to use
     // the older probabilities.
-    // LOG(INFO) << "Denominator is 0 QID: " << query_id << " priority:
" << priority_level;
     return;
   }
 }
@@ -248,7 +246,6 @@ void Learner::checkAndRemovePriorityLevel(const std::size_t priority_level)
{
     probabilities_of_priority_levels_->removeObject(priority_level);
     // NOTE(harshad) : Not using this cache as it gets confusing.
     // has_feedback_from_all_queries_.erase(priority_level);
-    // LOG(INFO) << "Removed priority level: " << priority_level;
     if (hasActiveQueries()) {
       if (static_cast<int>(priority_level) == highest_priority_level_) {
         // The priority level to be removed is the highest priority level.

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a70b1eca/query_execution/Learner.hpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.hpp b/query_execution/Learner.hpp
index 2c6fdef..8634842 100644
--- a/query_execution/Learner.hpp
+++ b/query_execution/Learner.hpp
@@ -70,7 +70,7 @@ class Learner {
    **/
   void removeQuery(const std::size_t query_id) {
     DCHECK(isQueryPresent(query_id));
-    const std::size_t priority_level = getQueryPriority(query_id);
+    const std::size_t priority_level = getQueryPriorityUnsafe(query_id);
     // Find the iterator to the query in execution_stats_.
     auto stats_iter_mutable = getExecutionStatsIterMutable(query_id);
     execution_stats_[priority_level].erase(stats_iter_mutable);
@@ -158,9 +158,6 @@ class Learner {
     if (hasActiveQueries()) {
       const int result = static_cast<int>(
           probabilities_of_priority_levels_->pickRandomProperty());
-      /*LOG(INFO) << "Random priority level: " << result << " has "
-                << current_probabilities_.find(result)->second->getNumObjects()
-                << " queries";*/
       return result;
     } else {
       return kInvalidPriorityLevel;
@@ -182,10 +179,8 @@ class Learner {
       DCHECK_GT(random_priority_level, 0);
       const int result = pickRandomQueryFromPriorityLevel(
           static_cast<std::size_t>(random_priority_level));
-      // LOG(INFO) << "Picked random query ID: " << result << " from priority
level " << random_priority_level;
       return result;
     } else {
-      // LOG(INFO) << "No active query right now";
       return kInvalidQueryID;
     }
   }
@@ -209,7 +204,6 @@ class Learner {
           return static_cast<int>(
               current_probabilities_.at(priority_level)->pickRandomProperty());
         }
-        // LOG(INFO) << "No queries in priority level: " << priority_level;
       } else {
         DCHECK(default_probabilities_.at(priority_level) != nullptr);
         const auto it = default_probabilities_.find(priority_level);
@@ -217,12 +211,24 @@ class Learner {
           return static_cast<int>(
               default_probabilities_.at(priority_level)->pickRandomProperty());
         }
-        // LOG(INFO) << "No queries in priority level: " << priority_level;
       }
     }
     return kInvalidQueryID;
   }
 
+  /**
+   * @brief Given a query ID, if the query exists in the learner, return its
+   *        priority, otherwise return kInvalidPriorityLevel.
+   **/
+  inline int getQueryPriority(const std::size_t query_id) const {
+    const auto it = query_id_to_priority_lookup_.find(query_id);
+    if (it != query_id_to_priority_lookup_.end()) {
+      return it->second;
+    } else {
+      return kInvalidPriorityLevel;
+    }
+  }
+
  private:
   /**
    * @brief Update the probabilities for queries in the given priority level.
@@ -324,7 +330,7 @@ class Learner {
     if (isQueryPresent(query_id)) {
       const auto stats_iter = getExecutionStatsIterMutable(query_id);
       DCHECK(stats_iter !=
-             std::end(execution_stats_[getQueryPriority(query_id)]));
+             std::end(execution_stats_[getQueryPriorityUnsafe(query_id)]));
       return stats_iter->second.get();
     }
     return nullptr;
@@ -339,7 +345,7 @@ class Learner {
   inline std::vector<
       std::pair<std::size_t, std::unique_ptr<ExecutionStats>>>::const_iterator
       getExecutionStatsIterMutable(const std::size_t query_id) {
-    const std::size_t priority_level = getQueryPriority(query_id);
+    const std::size_t priority_level = getQueryPriorityUnsafe(query_id);
     const std::vector<std::pair<std::size_t, std::unique_ptr<ExecutionStats>>>
         &stats_vector = execution_stats_[priority_level];
     // The following line uses std::find_if to reach to the desired element
@@ -356,8 +362,11 @@ class Learner {
 
   /**
    * @brief Get a query's priority level given its ID.
+   *
+   * @note This version assumes that the given query ID exists in data
+   *       structures.
    **/
-  inline const std::size_t getQueryPriority(const std::size_t query_id) const {
+  inline const std::size_t getQueryPriorityUnsafe(const std::size_t query_id) const {
     const auto it = query_id_to_priority_lookup_.find(query_id);
     DCHECK(it != query_id_to_priority_lookup_.end());
     return it->second;

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a70b1eca/query_execution/PriorityPolicyEnforcer.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.cpp b/query_execution/PriorityPolicyEnforcer.cpp
index 44ccb0a..6467367 100644
--- a/query_execution/PriorityPolicyEnforcer.cpp
+++ b/query_execution/PriorityPolicyEnforcer.cpp
@@ -53,6 +53,7 @@ bool PriorityPolicyEnforcer::admitQuery(QueryHandle *query_handle) {
           new QueryManager(foreman_client_id_, num_numa_nodes_, query_handle,
                            catalog_database_, storage_manager_, bus_));
       LOG(INFO) << "Admitted query with ID: " << query_handle->query_id();
+      priority_query_ids_[query_handle->query_priority()].emplace_back(query_id);
       learner_->addQuery(*query_handle);
       return true;
     } else {
@@ -71,6 +72,7 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
   // QueryManager, so that we need to extract message from the
   // TaggedMessage only once.
   std::size_t query_id;
+  std::size_t operator_id;
   switch (tagged_message.message_type()) {
     case kWorkOrderCompleteMessage: {
       serialization::NormalWorkOrderCompletionMessage proto;
@@ -79,6 +81,7 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
       CHECK(proto.ParseFromArray(tagged_message.message(),
                                  tagged_message.message_bytes()));
       query_id = proto.query_id();
+      operator_id = proto.operator_index();
       worker_directory_->decrementNumQueuedWorkOrders(
           proto.worker_thread_index());
       learner_->addCompletionFeedback(proto);
@@ -94,6 +97,7 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
       CHECK(proto.ParseFromArray(tagged_message.message(),
                                  tagged_message.message_bytes()));
       query_id = proto.query_id();
+      operator_id = proto.operator_index();
       worker_directory_->decrementNumQueuedWorkOrders(
           proto.worker_thread_index());
       break;
@@ -132,6 +136,7 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
   DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
   const QueryManager::QueryStatusCode return_code =
       admitted_queries_[query_id]->processMessage(tagged_message);
+  //NOTE: kQueryExecuted takes precedence over kOperatorExecuted.
   if (return_code == QueryManager::QueryStatusCode::kQueryExecuted) {
     removeQuery(query_id);
     if (!waiting_queries_.empty()) {
@@ -140,6 +145,8 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
       waiting_queries_.pop();
       admitQuery(new_query);
     }
+  } else if (return_code == QueryManager::QueryStatusCode::kOperatorExecuted) {
+    learner_->removeOperator(query_id, operator_id);
   }
 }
 
@@ -161,26 +168,25 @@ void PriorityPolicyEnforcer::getWorkerMessages(
   DCHECK_GT(per_query_share, 0u);
   std::vector<std::size_t> finished_queries_ids;
 
-  for (const auto &admitted_query_info : admitted_queries_) {
-    QueryManager *curr_query_manager = admitted_query_info.second.get();
-    DCHECK(curr_query_manager != nullptr);
-    std::size_t messages_collected_curr_query = 0;
-    while (messages_collected_curr_query < per_query_share) {
+  if (learner_->hasActiveQueries()) {
+    // Key = priority level. Value = Whether we have already checked the
+    std::unordered_map<std::size_t, bool> checked_priority_levels;
+    // While there are more priority levels to be checked ..
+    while (checked_priority_levels.size() < priority_query_ids_.size()) {
+      const int chosen_priority_level = learner_->pickRandomPriorityLevel();
+      DCHECK(chosen_priority_level != kInvalidPriorityLevel);
       WorkerMessage *next_worker_message =
-          curr_query_manager->getNextWorkerMessage(0, kAnyNUMANodeID);
+          getNextWorkerMessageFromPriorityLevel(chosen_priority_level,
+                                                &finished_queries_ids);
       if (next_worker_message != nullptr) {
-        ++messages_collected_curr_query;
         worker_messages->push_back(std::unique_ptr<WorkerMessage>(next_worker_message));
       } else {
-        // No more work ordes from the current query at this time.
-        // Check if the query's execution is over.
-        if (curr_query_manager->getQueryExecutionState().hasQueryExecutionFinished())
{
-          // If the query has been executed, remove it.
-          finished_queries_ids.push_back(admitted_query_info.first);
-        }
-        break;
+        checked_priority_levels[static_cast<std::size_t>(chosen_priority_level)] =
true;
       }
     }
+  } else {
+    LOG(INFO) << "No active queries in the learner at this point.";
+    return;
   }
   for (const std::size_t finished_qid : finished_queries_ids) {
     removeQuery(finished_qid);
@@ -194,6 +200,22 @@ void PriorityPolicyEnforcer::removeQuery(const std::size_t query_id)
{
                  << " that hasn't finished its execution";
   }
   admitted_queries_.erase(query_id);
+  // Remove the query from priority_query_ids_ structure.
+  const int query_priority = learner_->getQueryPriority(query_id);
+  DCHECK(query_priority != kInvalidPriorityLevel);
+  const std::size_t query_priority_unsigned =
+      static_cast<std::size_t>(query_priority);
+  std::vector<std::size_t> *query_ids_for_priority_level =
+      &priority_query_ids_[query_priority_unsigned];
+  query_ids_for_priority_level->erase(
+      std::remove(query_ids_for_priority_level->begin(),
+                  query_ids_for_priority_level->end(),
+                  query_id));
+  if (query_ids_for_priority_level->empty()) {
+    // No more queries for the given priority level. Remove the entry.
+    priority_query_ids_.erase(query_priority_unsigned);
+  }
+  // Remove the query from the learner.
   learner_->removeQuery(query_id);
 }
 
@@ -219,4 +241,40 @@ void PriorityPolicyEnforcer::recordTimeForWorkOrder(
       proto.execution_time_in_microseconds());
 }
 
+WorkerMessage* PriorityPolicyEnforcer::getNextWorkerMessageFromPriorityLevel(
+    const std::size_t priority_level,
+    std::vector<std::size_t> *finished_queries_ids) {
+  // Key = query ID from the given priority level, value = whether we have
+  // checked this query earlier.
+  std::unordered_map<std::size_t, bool> checked_query_ids;
+  // While there are more queries to be checked ..
+  while (checked_query_ids.size() < priority_query_ids_[priority_level].size()) {
+    const int chosen_query_id = learner_->pickRandomQueryFromPriorityLevel(priority_level);
+    if (chosen_query_id == kInvalidQueryID) {
+      // No query available at this time in this priority level.
+      return nullptr;
+    } else if (checked_query_ids.find(static_cast<std::size_t>(chosen_query_id)) !=
checked_query_ids.end()) {
+      // We have already seen this query ID, try one more time.
+      LOG(INFO) << "We have already seen this query, continue";
+      continue;
+    } else {
+      // We haven't seen this query earlier. Check if it has any schedulable
+      // WorkOrder.
+      QueryManager *chosen_query_manager = admitted_queries_[static_cast<std::size_t>(chosen_query_id)].get();
+      DCHECK(chosen_query_manager != nullptr);
+      std::unique_ptr<WorkerMessage> next_worker_message(chosen_query_manager->getNextWorkerMessage(0,
kAnyNUMANodeID));
+      if (next_worker_message != nullptr) {
+        return next_worker_message.release();
+      } else {
+        // This query doesn't have any WorkerMessage right now. Mark as checked.
+        checked_query_ids[chosen_query_id] = true;
+        if (chosen_query_manager->getQueryExecutionState().hasQueryExecutionFinished())
{
+          finished_queries_ids->emplace_back(static_cast<std::size_t>(chosen_query_id));
+        }
+      }
+    }
+  }
+  return nullptr;
+}
+
 }  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a70b1eca/query_execution/PriorityPolicyEnforcer.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.hpp b/query_execution/PriorityPolicyEnforcer.hpp
index 94cbe38..281c066 100644
--- a/query_execution/PriorityPolicyEnforcer.hpp
+++ b/query_execution/PriorityPolicyEnforcer.hpp
@@ -183,6 +183,21 @@ class PriorityPolicyEnforcer {
   void recordTimeForWorkOrder(
       const serialization::NormalWorkOrderCompletionMessage &proto);
 
+  /**
+   * @brief get a WorkerMessage from the chosen priority level.
+   *
+   * @param priority_level The priority level from which the query will be
+   *        chosen.
+   * @param finished_query_ids A vector of query IDs that have finished their
+   *        execution.
+   *
+   * @return A WorkerMessage. If no query can be chosen from this priority level,
+   *         return NULL.
+   **/
+  WorkerMessage *getNextWorkerMessageFromPriorityLevel(
+      const std::size_t priority_level,
+      std::vector<std::size_t> *finished_queries_ids);
+
   const tmb::client_id foreman_client_id_;
   const std::size_t num_numa_nodes_;
 
@@ -193,6 +208,10 @@ class PriorityPolicyEnforcer {
   tmb::MessageBus *bus_;
   const bool profile_individual_workorders_;
 
+  // Key = priority level, value = a vector of IDs of the queries belonging to
+  // the key priority level.
+  std::unordered_map<std::size_t, std::vector<std::size_t>> priority_query_ids_;
+
   // Key = query ID, value = QueryManager* for the key query.
   std::unordered_map<std::size_t, std::unique_ptr<QueryManager>> admitted_queries_;
 


Mime
View raw message