From 98dda08a8535d6faa1442dd6452d7ce5d035712d Mon Sep 17 00:00:00 2001
From: dongdaxiang <dongdaxiang@baidu.com>
Date: Fri, 29 Mar 2019 00:04:57 +0800
Subject: [PATCH] fix pull sparse slow problem test=develop

---
 paddle/fluid/framework/async_executor.cc      |  3 +++
 paddle/fluid/framework/downpour_worker.cc     | 11 +++++++++++
 paddle/fluid/framework/fleet/fleet_wrapper.cc | 18 ++++++++++++++----
 paddle/fluid/framework/fleet/fleet_wrapper.h  |  2 +-
 paddle/fluid/platform/lodtensor_printer.cc    |  2 +-
 5 files changed, 30 insertions(+), 6 deletions(-)

diff --git a/paddle/fluid/framework/async_executor.cc b/paddle/fluid/framework/async_executor.cc
index b13eefba2e..89153d82d0 100644
--- a/paddle/fluid/framework/async_executor.cc
+++ b/paddle/fluid/framework/async_executor.cc
@@ -153,11 +153,14 @@ void AsyncExecutor::RunFromFile(const ProgramDesc& main_program,
   for (auto& th : threads) {
     th.join();
   }
+  // TODO(guru4elephant): we don't need this
+  /*
 #ifdef PADDLE_WITH_PSLIB
   if (mode == "mpi") {
     _pull_dense_thread->stop();
   }
 #endif
+  */
   VLOG(3) << "start to run from files in async_executor";
   VLOG(3) << "Drop current scope kids";
   root_scope_->DropKids();
diff --git a/paddle/fluid/framework/downpour_worker.cc b/paddle/fluid/framework/downpour_worker.cc
index dab3113fbc..4ca7842fa2 100644
--- a/paddle/fluid/framework/downpour_worker.cc
+++ b/paddle/fluid/framework/downpour_worker.cc
@@ -210,6 +210,7 @@ void DownpourWorker::TrainFilesWithProfiler() {
       timeline.Pause();
       pull_sparse_time += timeline.ElapsedSec();
       total_time += timeline.ElapsedSec();
+      timeline.Start();
       CollectLabelInfo(i);
       timeline.Pause();
       collect_label_time += timeline.ElapsedSec();
@@ -336,6 +337,16 @@ void DownpourWorker::TrainFilesWithProfiler() {
         }
         fprintf(stderr, "mean read time: %fs\n", read_time / batch_cnt);
         fprintf(stderr, "IO percent: %f\n", read_time / total_time * 100);
+        fprintf(stderr, "pull sparse time percent: %f\n",
+                pull_sparse_time / total_time * 100);
+        fprintf(stderr, "collect label time percent: %f\n",
+                collect_label_time / total_time * 100);
+        fprintf(stderr, "fill sparse time percent: %f\n",
+                fill_sparse_time / total_time * 100);
+        fprintf(stderr, "push sparse time percent: %f\n",
+                push_sparse_time / total_time * 100);
+        fprintf(stderr, "push dense time percent: %f\n",
+                push_dense_time / total_time * 100);
         fprintf(stderr, "%6.2f instances/s\n", total_inst / total_time);
       }
     }
diff --git a/paddle/fluid/framework/fleet/fleet_wrapper.cc b/paddle/fluid/framework/fleet/fleet_wrapper.cc
index 6af8ba9518..72fd1a9cf1 100644
--- a/paddle/fluid/framework/fleet/fleet_wrapper.cc
+++ b/paddle/fluid/framework/fleet/fleet_wrapper.cc
@@ -142,6 +142,7 @@ void FleetWrapper::PullSparseVarsSync(
       }
       fea_keys->push_back(static_cast<uint64_t>(ids[i]));
     }
+    /*
     fea_values->resize(fea_keys->size() + 1);
     for (auto& t : *fea_values) {
       t.resize(fea_value_dim);
@@ -150,10 +151,19 @@ void FleetWrapper::PullSparseVarsSync(
     for (auto& t : *fea_values) {
       pull_result_ptr.push_back(t.data());
     }
-    auto status = pslib_ptr_->_worker_ptr->pull_sparse(
-        pull_result_ptr.data(), table_id, fea_keys->data(), fea_keys->size());
-    pull_sparse_status.push_back(std::move(status));
+    */
   }
+  fea_values->resize(fea_keys->size() + 1);
+  for (auto& t : *fea_values) {
+    t.resize(fea_value_dim);
+  }
+  std::vector<float*> pull_result_ptr;
+  for (auto& t : *fea_values) {
+    pull_result_ptr.push_back(t.data());
+  }
+  auto status = pslib_ptr_->_worker_ptr->pull_sparse(
+      pull_result_ptr.data(), table_id, fea_keys->data(), fea_keys->size());
+  pull_sparse_status.push_back(std::move(status));
   for (auto& t : pull_sparse_status) {
     t.wait();
     auto status = t.get();
@@ -207,7 +217,7 @@ void FleetWrapper::PullDenseVarsSync(
 }
 
 void FleetWrapper::PushDenseParamSync(
-    const ProgramDesc& program, const uint64_t table_id,
+    const Scope& scope, const uint64_t table_id,
     const std::vector<std::string>& var_names) {
 #ifdef PADDLE_WITH_PSLIB
   auto place = platform::CPUPlace();
diff --git a/paddle/fluid/framework/fleet/fleet_wrapper.h b/paddle/fluid/framework/fleet/fleet_wrapper.h
index 40ed3c5511..07eb670cbe 100644
--- a/paddle/fluid/framework/fleet/fleet_wrapper.h
+++ b/paddle/fluid/framework/fleet/fleet_wrapper.h
@@ -73,7 +73,7 @@ class FleetWrapper {
       const std::vector<std::string>& var_names,
       std::vector<::std::future<int32_t>>* pull_dense_status);
 
-  void PushDenseParamSync(const ProgramDesc& program, const uint64_t table_id,
+  void PushDenseParamSync(const Scope& scope, const uint64_t table_id,
                           const std::vector<std::string>& var_names);
 
   // Push dense variables to server in async mode
diff --git a/paddle/fluid/platform/lodtensor_printer.cc b/paddle/fluid/platform/lodtensor_printer.cc
index fb8e761f1a..a5aa1a4148 100644
--- a/paddle/fluid/platform/lodtensor_printer.cc
+++ b/paddle/fluid/platform/lodtensor_printer.cc
@@ -41,7 +41,7 @@ void print_lod_tensor(const std::string& var_name,
 void PrintVar(framework::Scope* scope, const std::string& var_name,
               const std::string& print_info) {
   framework::Variable* var = scope->FindVar(var_name);
-  if (tensor == nullptr) {
+  if (var == nullptr) {
     VLOG(1) << "Variable Name " << var_name << " does not exist in your scope";
     return;
   }
-- 
GitLab