merge baidu/develop

Author: QiJune

benchmark/cluster/vgg16/README.md

@@ -8,23 +8,24 @@
 - cpu MHz		: 2101.000
 - cache size	: 20480 KB
 
+### Blas settings
+
+Setting environment variable: `MKL_NUM_THREADS=1`.
+
 ### Single Node Single Thread
 
-- PServer Count: 10
-- Trainer Count: 20
 - Metrics: samples / sec
 
 | Batch Size | 32 | 64 | 128 | 256 |
 | -- | -- | -- | -- | -- |
 | PaddlePaddle Fluid | 15.44 | 16.32 | 16.74 | 16.79 |
 | PaddlePaddle v2 | 15.97 | 17.04 | 17.60 | 17.83 |
-| TensorFlow | - | - | - | - |
+| TensorFlow | 9.09 | 9.10 | 9.24 | 8.66 |
 
 ### Different Batch Size
 
 - PServer Count: 10
 - Trainer Count: 20
-- Per trainer CPU Core: 1
 - Metrics: samples / sec
 
 | Batch Size | 32 | 64 | 128 | 256 |

paddle/fluid/framework/block_desc.cc

@@ -16,6 +16,8 @@ limitations under the License. */
 #include "paddle/fluid/framework/operator.h"
 #include "paddle/fluid/framework/program_desc.h"
 
+#include <queue>
+
 namespace paddle {
 namespace framework {
 
@@ -64,12 +66,36 @@ VarDesc *BlockDesc::RenameVar(const std::string &old_name,
 VarDesc *BlockDesc::FindVarRecursive(const std::string &name) const {
   if (name == kEmptyVarName) return nullptr;
 
-  auto it = vars_.find(name);
-  if (it == vars_.end()) {
-    return Parent() == kNoneBlockIndex ? nullptr
-                                       : ParentBlock()->FindVarRecursive(name);
+  std::queue<const BlockDesc *> frontier;
+  std::unordered_set<const BlockDesc *> visited;
+
+  frontier.push(this);
+
+  while (!frontier.empty()) {  // BFS
+    auto cur = frontier.front();
+    frontier.pop();
+    if (visited.count(cur) != 0) {
+      continue;
+    }
+    auto var = cur->FindVar(name);
+    if (var != nullptr) {
+      return var;
+    }
+
+    auto fwd = cur->ForwardBlock();
+    auto parent = cur->ParentBlock();
+
+    if (fwd != nullptr) {
+      frontier.push(fwd);
+    }
+    if (parent != nullptr) {
+      frontier.push(parent);
+    }
+
+    visited.insert(cur);
   }
-  return it->second.get();
+
+  return nullptr;
 }
 
 VarDesc &BlockDesc::FindRecursiveOrCreateVar(const std::string &name_bytes) {
@@ -155,10 +181,7 @@ void BlockDesc::Flush() {
 }
 
 BlockDesc *BlockDesc::ParentBlock() const {
-  if (this->desc_->parent_idx() == kNoneBlockIndex) {
-    return nullptr;
-  }
-  return prog_->MutableBlock(static_cast<size_t>(this->desc_->parent_idx()));
+  return prog_->MutableBlock(static_cast<size_t>(desc_->parent_idx()));
 }
 
 proto::BlockDesc *BlockDesc::Proto() {
@@ -205,5 +228,16 @@ void BlockDesc::ClearPBVars() {
   }
 }
 
+void BlockDesc::SetForwardBlockID(int32_t forward_block_id) {
+  PADDLE_ENFORCE(!desc_->has_forward_block_idx(),
+                 "Parent block ID has been set to %d. Cannot set to %d",
+                 desc_->forward_block_idx(), forward_block_id);
+  desc_->set_forward_block_idx(forward_block_id);
+}
+
+BlockDesc *BlockDesc::ForwardBlock() const {
+  return prog_->MutableBlock(static_cast<size_t>(desc_->forward_block_idx()));
+}
+
 }  // namespace framework
 }  // namespace paddle

paddle/fluid/framework/block_desc.h

@@ -49,6 +49,8 @@ class BlockDesc {
 
   int32_t Parent() const { return desc_->parent_idx(); }
 
+  int32_t ForwardBlockID() const { return desc_->forward_block_idx(); }
+
   VarDesc *Var(const std::string &name_bytes);
 
   VarDesc *FindVar(const std::string &name_bytes) const;
@@ -75,6 +77,10 @@ class BlockDesc {
 
   BlockDesc *ParentBlock() const;
 
+  BlockDesc *ForwardBlock() const;
+
+  void SetForwardBlockID(int32_t forward_block_id);
+
   OpDesc *AppendOp();
 
   void AppendAllocatedOp(std::unique_ptr<OpDesc> &&op_desc);
@@ -93,7 +99,7 @@ class BlockDesc {
 
   proto::BlockDesc *Proto();
 
-  ProgramDesc *Program() { return this->prog_; }
+  ProgramDesc *Program() const { return this->prog_; }
 
  private:
   void ClearPBOps();

paddle/fluid/framework/channel.h

@@ -100,8 +100,7 @@ class ChannelHolder {
     virtual ~Placeholder() {}
     virtual const std::type_index Type() const = 0;
     virtual void* Ptr() const = 0;
-    virtual void Close() const = 0;
-    std::type_info type_;
+    virtual void Close() = 0;
   };
 
   template <typename T>
@@ -116,7 +115,7 @@ class ChannelHolder {
       if (channel_) channel_->Close();
     }
 
-    std::unique_ptr<Channel<T>*> channel_;
+    std::unique_ptr<Channel<T>> channel_;
     const std::type_index type_;
   };
 

paddle/fluid/framework/channel_test.cc

@@ -20,6 +20,7 @@ limitations under the License. */
 #include "gtest/gtest.h"
 
 using paddle::framework::Channel;
+using paddle::framework::ChannelHolder;
 using paddle::framework::MakeChannel;
 using paddle::framework::CloseChannel;
 using paddle::framework::details::Buffered;
@@ -508,3 +509,36 @@ TEST(Channel, UnbufferedChannelDestroyUnblocksSendersTest) {
   auto ch = MakeChannel<int>(0);
   ChannelDestroyUnblockSenders(ch);
 }
+
+void ChannelHolderSendReceive(ChannelHolder *ch) {
+  unsigned sum_send = 0;
+  std::thread t([&]() {
+    for (int i = 0; i < 5; i++) {
+      EXPECT_EQ(ch->Send(&i), true);
+      sum_send += i;
+    }
+  });
+  for (int i = 0; i < 5; i++) {
+    int recv;
+    EXPECT_EQ(ch->Receive(&recv), true);
+    EXPECT_EQ(recv, i);
+  }
+
+  ch->close();
+  t.join();
+  EXPECT_EQ(sum_send, 10U);
+}
+
+TEST(ChannelHolder, ChannelHolderBufferedSendReceiveTest) {
+  ChannelHolder *ch = new ChannelHolder();
+  ch->Reset<int>(10);
+  ChannelHolderSendReceive(ch);
+  delete ch;
+}
+
+TEST(ChannelHolder, ChannelHolderUnBufferedSendReceiveTest) {
+  ChannelHolder *ch = new ChannelHolder();
+  ch->Reset<int>(0);
+  ChannelHolderSendReceive(ch);
+  delete ch;
+}

paddle/fluid/framework/framework.proto

@@ -158,6 +158,7 @@ message BlockDesc {
   required int32 parent_idx = 2;
   repeated VarDesc vars = 3;
   repeated OpDesc ops = 4;
+  optional int32 forward_block_idx = 5 [ default = -1 ];
 }
 
 // Please refer to

paddle/fluid/framework/program_desc.h

@@ -38,7 +38,13 @@ class ProgramDesc {
 
   BlockDesc *AppendBlock(const BlockDesc &parent);
 
-  BlockDesc *MutableBlock(size_t idx) { return blocks_[idx].get(); }
+  BlockDesc *MutableBlock(size_t idx) {
+    if (idx == static_cast<size_t>(kNoneBlockIndex)) {
+      return nullptr;
+    } else {
+      return blocks_[idx].get();
+    }
+  }
 
   const BlockDesc &Block(size_t idx) const { return *blocks_[idx]; }
 

paddle/fluid/operators/while_op.cc

@@ -231,7 +231,8 @@ class WhileGradOpDescMaker : public framework::SingleGradOpDescMaker {
     while_grad->SetInput(kStepScopes, Output(kStepScopes));
 
     auto *grad_block = this->grad_block_[0];
-    auto *fwd_block = grad_block->ParentBlock();
+    auto *fwd_block = grad_block->ForwardBlock();
+    auto *parent_block = grad_block->ParentBlock();
 
     // Not all of IGs will be generated by inner gradient operators of while op.
     // Ignore IGs that is not generated by the inside block.
@@ -260,33 +261,37 @@ class WhileGradOpDescMaker : public framework::SingleGradOpDescMaker {
     for (auto &o : Output(kOutputs)) {
       block_ins.insert(o);
     }
-    std::unordered_set<std::string> extra_inputs;
+    std::unordered_set<std::string> output_grads;
     for (const auto *op : grad_block->AllOps()) {
       for (auto &input_name : op->InputArgumentNames()) {
         // If the input of Op has been recorded or is generated by the forward
         // block, do not make it as input again.
+
+        // The input is located in I/O or other op's outputs or the variable is
+        // located in grad_block's parents
         if (block_ins.find(input_name) != block_ins.end() ||
-            fwd_block->FindVar(input_name) != nullptr) {
+            (fwd_block->FindVarRecursive(input_name) != nullptr ||
+             parent_block->FindVarRecursive(input_name) != nullptr)) {
           continue;
         }
-        extra_inputs.insert(input_name);
+        output_grads.insert(input_name);
       }
       for (auto &output_name : op->OutputArgumentNames()) {
         block_ins.insert(output_name);
       }
     }
 
-    std::vector<std::string> extra_inputs_list;
-    extra_inputs_list.resize(extra_inputs.size());
-    std::copy(extra_inputs.begin(), extra_inputs.end(),
-              extra_inputs_list.begin());
-    while_grad->SetInput(framework::GradVarName(kOutputs), extra_inputs_list);
+    std::vector<std::string> output_grads_list;
+    output_grads_list.resize(output_grads.size());
+    std::copy(output_grads.begin(), output_grads.end(),
+              output_grads_list.begin());
+    while_grad->SetInput(framework::GradVarName(kOutputs), output_grads_list);
 
     while_grad->SetAttrMap(this->Attrs());
     while_grad->SetBlockAttr(kStepBlock, *grad_block);
     // record the original output gradient names, since the gradient name of
     // while operator could be renamed.
-    while_grad->SetAttr("original_output_grad", extra_inputs_list);
+    while_grad->SetAttr("original_output_grad", output_grads_list);
 
     return std::unique_ptr<framework::OpDesc>(while_grad);
   }

paddle/fluid/pybind/protobuf.cc

@@ -155,6 +155,8 @@ void BindBlockDesc(py::module &m) {
   py::class_<BlockDesc>(m, "BlockDesc", "")
       .def_property_readonly("id", &BlockDesc::ID)
       .def_property_readonly("parent", &BlockDesc::Parent)
+      .def("get_forward_block_idx", &BlockDesc::ForwardBlockID)
+      .def("set_forward_block_idx", &BlockDesc::SetForwardBlockID)
       .def("append_op", &BlockDesc::AppendOp,
            py::return_value_policy::reference)
       .def("prepend_op", &BlockDesc::PrependOp,

python/paddle/v2/fluid/backward.py

@@ -298,7 +298,8 @@ def _append_backward_ops_(block,
         # If the op has its own sub-block, deal with the sub-block first
         if op.has_attr("sub_block"):
             sub_block = program.block(op.block_attr("sub_block"))
-            grad_sub_block = program.create_block(parent_idx=sub_block.idx)
+            grad_sub_block = program.create_block()
+            grad_sub_block.set_forward_block_idx(sub_block.idx)
             cb = _callback_lookup_(op)
             if cb is not None:
                 if callbacks is None:
@@ -310,6 +311,8 @@ def _append_backward_ops_(block,
             else:
                 _append_backward_ops_(sub_block, sub_block.ops, grad_sub_block,
                                       no_grad_dict, grad_to_var, callbacks)
+
+            program.rollback()
             grad_sub_block_list.append(grad_sub_block.desc)
 
         # Getting op's corresponding grad_op