Merge remote-tracking branch 'upstream/develop' into feature/understand_sentimental_dynrnn_2

Author: Yang Yang

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/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;
+}