changed ETFeeder::readNextWindow() to reduce extra resolveDeps.

Author: changhai0109

et_feeder/et_feeder.cpp

@@ -1,16 +1,16 @@
 #include "et_feeder/et_feeder.h"
+#include <cassert>
 
 using namespace std;
 using namespace Chakra;
 
 ETFeeder::ETFeeder(string filename)
-  : trace_(filename), window_size_(4096 * 256), et_complete_(false) {
+    : trace_(filename), window_size_(4096 * 256), et_complete_(false) {
   readGlobalMetadata();
   readNextWindow();
 }
 
-ETFeeder::~ETFeeder() {
-}
+ETFeeder::~ETFeeder() {}
 
 void ETFeeder::addNode(shared_ptr<ETFeederNode> node) {
   dep_graph_[node->getChakraNode()->id()] = node;
@@ -19,8 +19,7 @@ void ETFeeder::addNode(shared_ptr<ETFeederNode> node) {
 void ETFeeder::removeNode(uint64_t node_id) {
   dep_graph_.erase(node_id);
 
-  if (!et_complete_
-      && (dep_free_node_queue_.size() < window_size_)) {
+  if (!et_complete_ && (dep_free_node_queue_.size() < window_size_)) {
     readNextWindow();
   }
 }
@@ -47,16 +46,18 @@ void ETFeeder::pushBackIssuableNode(uint64_t node_id) {
 }
 
 shared_ptr<ETFeederNode> ETFeeder::lookupNode(uint64_t node_id) {
-  return dep_graph_[node_id];
+  auto node = dep_graph_.find(node_id);
+  assert(node != dep_graph_.end());
+  return node->second;
 }
 
 void ETFeeder::freeChildrenNodes(uint64_t node_id) {
   shared_ptr<ETFeederNode> node = dep_graph_[node_id];
-  for (auto child: node->getChildren()) {
+  for (auto child : node->getChildren()) {
     auto child_chakra = child->getChakraNode();
     for (auto it = child_chakra->mutable_data_deps()->begin();
-        it != child_chakra->mutable_data_deps()->end();
-        ++it) {
+         it != child_chakra->mutable_data_deps()->end();
+         ++it) {
       if (*it == node_id) {
         child_chakra->mutable_data_deps()->erase(it);
         break;
@@ -70,12 +71,14 @@ void ETFeeder::freeChildrenNodes(uint64_t node_id) {
 }
 
 void ETFeeder::readGlobalMetadata() {
-  shared_ptr<ChakraProtoMsg::GlobalMetadata> pkt_msg = make_shared<ChakraProtoMsg::GlobalMetadata>();
+  shared_ptr<ChakraProtoMsg::GlobalMetadata> pkt_msg =
+      make_shared<ChakraProtoMsg::GlobalMetadata>();
   trace_.read(*pkt_msg);
 }
 
 shared_ptr<ETFeederNode> ETFeeder::readNode() {
-  shared_ptr<ChakraProtoMsg::Node> pkt_msg = make_shared<ChakraProtoMsg::Node>();
+  shared_ptr<ChakraProtoMsg::Node> pkt_msg =
+      make_shared<ChakraProtoMsg::Node>();
   if (!trace_.read(*pkt_msg)) {
     return nullptr;
   }
@@ -101,11 +104,12 @@ shared_ptr<ETFeederNode> ETFeeder::readNode() {
 
 void ETFeeder::resolveDep() {
   for (auto it = dep_unresolved_node_set_.begin();
-      it != dep_unresolved_node_set_.end();) {
+       it != dep_unresolved_node_set_.end();) {
     shared_ptr<ETFeederNode> node = *it;
-    vector<uint64_t> dep_unresolved_parent_ids = node->getDepUnresolvedParentIDs();
+    vector<uint64_t> dep_unresolved_parent_ids =
+        node->getDepUnresolvedParentIDs();
     for (auto inner_it = dep_unresolved_parent_ids.begin();
-        inner_it != dep_unresolved_parent_ids.end();) {
+         inner_it != dep_unresolved_parent_ids.end();) {
       auto parent_node = dep_graph_.find(*inner_it);
       if (parent_node != dep_graph_.end()) {
         parent_node->second->addChild(node);
@@ -126,24 +130,27 @@ void ETFeeder::resolveDep() {
 void ETFeeder::readNextWindow() {
   uint32_t num_read = 0;
   do {
-    shared_ptr<ETFeederNode> new_node = readNode();
-    if (new_node == nullptr) {
-      et_complete_ = true;
-      break;
+    if (this->et_complete_) {
+      // graph read finished, but still nodes unresolved
+      // which means the graph is broken
+      assert(false);
+    }
+    for (uint32_t num_read = 0; num_read < this->window_size_; num_read++) {
+      std::shared_ptr<ETFeederNode> new_node = readNode();
+      if (new_node == nullptr) {
+        et_complete_ = true;
+        break;
+      }
+      addNode(new_node);
     }
-
-    addNode(new_node);
-    ++num_read;
-
     resolveDep();
-  } while ((num_read < window_size_)
-      || (dep_unresolved_node_set_.size() != 0));
+  } while (dep_unresolved_node_set_.size() != 0);
 
-  for (auto node_id_node: dep_graph_) {
+  for (auto node_id_node : dep_graph_) {
     uint64_t node_id = node_id_node.first;
     shared_ptr<ETFeederNode> node = node_id_node.second;
-    if ((dep_free_node_id_set_.count(node_id) == 0)
-        && (node->getChakraNode()->data_deps().size() == 0)) {
+    if ((dep_free_node_id_set_.count(node_id) == 0) &&
+        (node->getChakraNode()->data_deps().size() == 0)) {
       dep_free_node_id_set_.emplace(node_id);
       dep_free_node_queue_.emplace(node);
     }