diff --git a/doc/getstarted/quickstart_cn.rst b/doc/getstarted/quickstart_cn.rst index 51dd00f1e806e6423afe3ce53d80d53a187d2ca0..d511cead262dabafd095f68adb5ffc596a7fe596 100644 --- a/doc/getstarted/quickstart_cn.rst +++ b/doc/getstarted/quickstart_cn.rst @@ -1,6 +1,9 @@ 快速开始 ======== +快速安装 +-------- + PaddlePaddle支持使用pip快速安装，目前支持CentOS 6以上, Ubuntu 14.04以及MacOS 10.12，并安装有Python2.7。 执行下面的命令完成快速安装，版本为cpu_avx_openblas： @@ -16,6 +19,9 @@ PaddlePaddle支持使用pip快速安装，目前支持CentOS 6以上, Ubuntu 14. 更详细的安装和编译方法参考：:ref:`install_steps` 。 +快速使用 +-------- + 创建一个 housing.py 并粘贴此Python代码： .. code-block:: python diff --git a/doc/getstarted/quickstart_en.rst b/doc/getstarted/quickstart_en.rst index d1bcf82ea071e2c53760a5ccf6a5074a3ac0abd5..70f7fe0646068aa79cd72955c6848ac0250c2300 100644 --- a/doc/getstarted/quickstart_en.rst +++ b/doc/getstarted/quickstart_en.rst @@ -1,6 +1,9 @@ Quick Start ============ +Quick Install +------------- + You can use pip to install PaddlePaddle with a single command, supports CentOS 6 above, Ubuntu 14.04 above or MacOS 10.12, with Python 2.7 installed. Simply run the following command to install, the version is cpu_avx_openblas: @@ -17,6 +20,9 @@ If you need to install GPU version (cuda7.5_cudnn5_avx_openblas), run: For more details about installation and build: :ref:`install_steps` . +Quick Use +--------- + Create a new file called housing.py, and paste this Python code: diff --git a/doc/howto/cluster/index_cn.rst b/doc/howto/cluster/index_cn.rst index c68b2655b65b192814b94f0013fa92b0733b9afa..a60521b4a9646bdc6d9f1bf6da482acc989d8bf3 100644 --- a/doc/howto/cluster/index_cn.rst +++ b/doc/howto/cluster/index_cn.rst @@ -1,10 +1,22 @@ 分布式训练 ========== +本节将介绍如何使用PaddlePaddle在不同的集群框架下完成分布式训练。分布式训练架构如下图所示： + +.. image:: src/ps_cn.png + :width: 500 + +- 数据分片（Data shard): 用于训练神经网络的数据，被切分成多个部分，每个部分分别给每个trainer使用。 +- 计算节点（Trainer）: 每个trainer启动后读取切分好的一部分数据，开始神经网络的“前馈”和“后馈”计算，并和参数服务器通信。在完成一定量数据的训练后，上传计算得出的梯度（gradients），然后下载优化更新后的神经网络参数（parameters）。 +- 参数服务器（Parameter server）:每个参数服务器只保存整个神经网络所有参数的一部分。参数服务器接收从计算节点上传的梯度，并完成参数优化更新，再将更新后的参数下发到每个计算节点。 + +这样，通过计算节点和参数服务器的分布式协作，可以完成神经网络的SGD方法的训练。PaddlePaddle可以同时支持同步随机梯度下降（SGD）和异步随机梯度下降。 + +在使用同步SGD训练神经网络时，PaddlePaddle使用同步屏障（barrier），使梯度的提交和参数的更新按照顺序方式执行。在异步SGD中，则并不会等待所有trainer提交梯度才更新参数，这样极大地提高了计算的并行性：参数服务器之间不相互依赖，并行地接收梯度和更新参数，参数服务器也不会等待计算节点全部都提交梯度之后才开始下一步，计算节点之间也不会相互依赖，并行地执行模型的训练。可以看出，虽然异步SGD方式会提高参数更新并行度, 但是并不能保证参数同步更新，在任意时间某一台参数服务器上保存的参数可能比另一台要更新，与同步SGD相比，梯度会有噪声。 + .. toctree:: :maxdepth: 1 - introduction_cn.md preparations_cn.md cmd_argument_cn.md multi_cluster/index_cn.rst diff --git a/doc/howto/cluster/index_en.rst b/doc/howto/cluster/index_en.rst index af957e06cd7930ce63569a1bafdde47a1d34eb69..2640a09dcc904619bc97c9bd3f3d81a9dc307663 100644 --- a/doc/howto/cluster/index_en.rst +++ b/doc/howto/cluster/index_en.rst @@ -1,10 +1,22 @@ Distributed Training ==================== +In this section, we'll explain how to run distributed training jobs with PaddlePaddle on different types of clusters. The diagram below shows the main architecture of a distributed trainning job: + +.. image:: src/ps_en.png + :width: 500 + +- Data shard: training data will be split into multiple partitions, trainers use the partitions of the whole dataset to do the training job. +- Trainer: each trainer reads the data shard, and train the neural network. Then the trainer will upload calculated "gradients" to parameter servers, and wait for parameters to be optimized on the parameter server side. When that finishes, the trainer download optimized parameters and continues its training. +- Parameter server: every parameter server stores part of the whole neural network model data. They will do optimization calculations when gradients are uploaded from trainers, and then send updated parameters to trainers. + +PaddlePaddle can support both synchronize stochastic gradient descent (SGD) and asynchronous SGD. + +When training with synchronize SGD, PaddlePaddle uses an internal "synchronize barrier" which makes gradients update and parameter download in strict order. On the other hand, asynchronous SGD won't wait for all trainers to finish upload at a single step, this will increase the parallelism of distributed training: parameter servers do not depend on each other, they'll do parameter optimization concurrently. Parameter servers will not wait for trainers, so trainers will also do their work concurrently. But asynchronous SGD will introduce more randomness and noises in the gradient. + .. toctree:: :maxdepth: 1 - introduction_en.md preparations_en.md cmd_argument_en.md multi_cluster/index_en.rst diff --git a/doc/howto/cluster/introduction_cn.md b/doc/howto/cluster/introduction_cn.md deleted file mode 100644 index 562008a898414a6566d74d08cfeb18fb9d57582a..0000000000000000000000000000000000000000 --- a/doc/howto/cluster/introduction_cn.md +++ /dev/null @@ -1,13 +0,0 @@ -## 概述 - -本节将介绍如何使用PaddlePaddle在不同的集群框架下完成分布式训练。分布式训练架构如下图所示： - - - -- 数据分片（Data shard): 用于训练神经网络的数据，被切分成多个部分，每个部分分别给每个trainer使用。 -- 计算节点（Trainer）: 每个trainer启动后读取切分好的一部分数据，开始神经网络的“前馈”和“后馈”计算，并和参数服务器通信。在完成一定量数据的训练后，上传计算得出的梯度（gradients），然后下载优化更新后的神经网络参数（parameters）。 -- 参数服务器（Parameter server）:每个参数服务器只保存整个神经网络所有参数的一部分。参数服务器接收从计算节点上传的梯度，并完成参数优化更新，再将更新后的参数下发到每个计算节点。 - -这样，通过计算节点和参数服务器的分布式协作，可以完成神经网络的SGD方法的训练。PaddlePaddle可以同时支持同步随机梯度下降（SGD）和异步随机梯度下降。 - -在使用同步SGD训练神经网络时，PaddlePaddle使用同步屏障（barrier），使梯度的提交和参数的更新按照顺序方式执行。在异步SGD中，则并不会等待所有trainer提交梯度才更新参数，这样极大地提高了计算的并行性：参数服务器之间不相互依赖，并行地接收梯度和更新参数，参数服务器也不会等待计算节点全部都提交梯度之后才开始下一步，计算节点之间也不会相互依赖，并行地执行模型的训练。可以看出，虽然异步SGD方式会提高参数更新并行度, 但是并不能保证参数同步更新，在任意时间某一台参数服务器上保存的参数可能比另一台要更新，与同步SGD相比，梯度会有噪声。 diff --git a/doc/howto/cluster/introduction_en.md b/doc/howto/cluster/introduction_en.md deleted file mode 100644 index eb70d7cf35ab729e0da4c6a3a2e732c26905f584..0000000000000000000000000000000000000000 --- a/doc/howto/cluster/introduction_en.md +++ /dev/null @@ -1,13 +0,0 @@ -## Introduction - -In this section, we'll explain how to run distributed training jobs with PaddlePaddle on different types of clusters. The diagram below shows the main architecture of a distributed trainning job: - - - -- Data shard: training data will be split into multiple partitions, trainers use the partitions of the whole dataset to do the training job. -- Trainer: each trainer reads the data shard, and train the neural network. Then the trainer will upload calculated "gradients" to parameter servers, and wait for parameters to be optimized on the parameter server side. When that finishes, the trainer download optimized parameters and continues its training. -- Parameter server: every parameter server stores part of the whole neural network model data. They will do optimization calculations when gradients are uploaded from trainers, and then send updated parameters to trainers. - -PaddlePaddle can support both synchronize stochastic gradient descent (SGD) and asynchronous SGD. - -When training with synchronize SGD, PaddlePaddle uses an internal "synchronize barrier" which makes gradients update and parameter download in strict order. On the other hand, asynchronous SGD won't wait for all trainers to finish upload at a single step, this will increase the parallelism of distributed training: parameter servers do not depend on each other, they'll do parameter optimization concurrently. Parameter servers will not wait for trainers, so trainers will also do their work concurrently. But asynchronous SGD will introduce more randomness and noises in the gradient. diff --git a/doc/howto/cluster/src/ps_cn.png b/doc/howto/cluster/src/ps_cn.png new file mode 100644 index 0000000000000000000000000000000000000000..f9525739cc8bc6506adde642aafa0a85ae3ebebc Binary files /dev/null and b/doc/howto/cluster/src/ps_cn.png differ diff --git a/doc/howto/cluster/src/ps_en.png b/doc/howto/cluster/src/ps_en.png new file mode 100644 index 0000000000000000000000000000000000000000..6537d3d56589ca9f19a77a50a970e4b5275e6ce0 Binary files /dev/null and b/doc/howto/cluster/src/ps_en.png differ diff --git a/doc/howto/rnn/index_cn.rst b/doc/howto/rnn/index_cn.rst index 9ecab5594cff47cde4700b7ce0f58013a960a16e..bcc8c2f46eb662ec3650e829a77992224dbbb8e7 100644 --- a/doc/howto/rnn/index_cn.rst +++ b/doc/howto/rnn/index_cn.rst @@ -1,4 +1,4 @@ -RNN相关模型 +RNN模型 =========== .. toctree:: diff --git a/paddle/framework/block_desc.cc b/paddle/framework/block_desc.cc index dd2ed87252102aee6d384f37365d19305f19b281..3e344ea3790f57b0f53f36a40263dcdd326e67a9 100644 --- a/paddle/framework/block_desc.cc +++ b/paddle/framework/block_desc.cc @@ -162,9 +162,8 @@ BlockDesc::BlockDesc(const BlockDesc &other, proto::BlockDesc *desc, : prog_(prog), desc_(desc) { need_update_ = true; for (auto &op : other.ops_) { - ops_.emplace_back(new OpDesc(*op, this)); + ops_.emplace_back(new OpDesc(*op->Proto(), prog, this)); } - for (auto &it : other.vars_) { auto *var = new VarDesc(*it.second); vars_[it.first].reset(var); diff --git a/paddle/framework/channel.h b/paddle/framework/channel.h index b679387b1124e42499df158767b6c7afe1afd0c6..146f0e9e71ea9101a8f6c71e6c023178f131f967 100644 --- a/paddle/framework/channel.h +++ b/paddle/framework/channel.h @@ -1,4 +1,4 @@ -/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve. +/* Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. diff --git a/paddle/framework/channel_test.cc b/paddle/framework/channel_test.cc index df9e15e22b890347a03d6816e8549c99b010bb38..d7140dd10661c7b8582930b47872ab0b330c4d66 100644 --- a/paddle/framework/channel_test.cc +++ b/paddle/framework/channel_test.cc @@ -1,4 +1,4 @@ -/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve. +/* Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. @@ -22,6 +22,28 @@ limitations under the License. */ using paddle::framework::Channel; using paddle::framework::MakeChannel; using paddle::framework::CloseChannel; +using paddle::framework::details::Buffered; +using paddle::framework::details::UnBuffered; + +void RecevingOrderEqualToSendingOrder(Channel *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); + } + + CloseChannel(ch); + t.join(); + EXPECT_EQ(sum_send, 10U); + delete ch; +} TEST(Channel, MakeAndClose) { using paddle::framework::details::Buffered; @@ -60,13 +82,54 @@ TEST(Channel, SufficientBufferSizeDoesntBlock) { delete ch; } -TEST(Channel, SendOnClosedChannelPanics) { - const size_t buffer_size = 10; - auto ch = MakeChannel(buffer_size); - size_t i = 5; - EXPECT_EQ(ch->Send(&i), true); // should not block or panic +// This tests that a channel must return false +// on send and receive performed after closing the channel. +// Receive will only return false after close when queue is empty. +// By creating separate threads for sending and receiving, we make this +// function able to test both buffered and unbuffered channels. +void SendReceiveWithACloseChannelShouldPanic(Channel *ch) { + const size_t data = 5; + std::thread send_thread{[&]() { + size_t i = data; + EXPECT_EQ(ch->Send(&i), true); // should not block + }}; + + std::thread recv_thread{[&]() { + size_t i; + EXPECT_EQ(ch->Receive(&i), true); // should not block + EXPECT_EQ(i, data); + }}; + + send_thread.join(); + recv_thread.join(); + + // After closing send should return false. Receive should + // also return false as there is no data in queue. CloseChannel(ch); - EXPECT_EQ(ch->Send(&i), false); // should panic + send_thread = std::thread{[&]() { + size_t i = data; + EXPECT_EQ(ch->Send(&i), false); // should return false + }}; + recv_thread = std::thread{[&]() { + size_t i; + // should return false because channel is closed and queue is empty + EXPECT_EQ(ch->Receive(&i), false); + }}; + + send_thread.join(); + recv_thread.join(); +} + +TEST(Channel, SendReceiveClosedBufferedChannelPanics) { + size_t buffer_size = 10; + auto ch = MakeChannel(buffer_size); + SendReceiveWithACloseChannelShouldPanic(ch); + delete ch; +} + +TEST(Channel, SendReceiveClosedUnBufferedChannelPanics) { + auto ch = MakeChannel(0); + SendReceiveWithACloseChannelShouldPanic(ch); delete ch; } @@ -94,9 +157,7 @@ TEST(Channel, ReceiveFromBufferedChannelReturnResidualValuesTest) { for (size_t i = 0; i < buffer_size; ++i) { EXPECT_EQ(ch->Receive(&out), - false); // after receiving residual values, return zeros. - // Note: we cannot check EXPECT_EQ(out, 0), because C++ doesn't - // define zero values like Go does. + false); // receiving on closed channel should return false } delete ch; } @@ -115,7 +176,7 @@ TEST(Channel, ConcurrentSendNonConcurrentReceiveWithSufficientBufferSize) { sum += i; } }); - std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait 0.5 sec + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait 0.1 sec EXPECT_EQ(sum, 45U); CloseChannel(ch); @@ -123,31 +184,17 @@ TEST(Channel, ConcurrentSendNonConcurrentReceiveWithSufficientBufferSize) { delete ch; } -TEST(Channel, SimpleUnbufferedChannelTest) { +TEST(Channel, RecevingOrderEqualToSendingOrderWithUnBufferedChannel) { auto ch = MakeChannel(0); - 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); - } + RecevingOrderEqualToSendingOrder(ch); +} - CloseChannel(ch); - t.join(); - EXPECT_EQ(sum_send, 10U); - delete ch; +TEST(Channel, RecevingOrderEqualToSendingOrderWithBufferedChannel) { + auto ch = MakeChannel(10); + RecevingOrderEqualToSendingOrder(ch); } -// This tests that closing a buffered channel also unblocks -// any receivers waiting on the channel -TEST(Channel, BufferedChannelCloseUnblocksReceiversTest) { - auto ch = MakeChannel(1); +void ChannelCloseUnblocksReceiversTest(Channel *ch) { size_t num_threads = 5; std::thread t[num_threads]; bool thread_ended[num_threads]; @@ -158,15 +205,14 @@ TEST(Channel, BufferedChannelCloseUnblocksReceiversTest) { t[i] = std::thread( [&](bool *p) { int data; - // All reads should return false EXPECT_EQ(ch->Receive(&data), false); *p = true; }, &thread_ended[i]); } - std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait 0.1 sec - // Verify that all threads are blocked + // Verify that all the threads are blocked for (size_t i = 0; i < num_threads; i++) { EXPECT_EQ(thread_ended[i], false); } @@ -175,7 +221,7 @@ TEST(Channel, BufferedChannelCloseUnblocksReceiversTest) { // This should unblock all receivers CloseChannel(ch); - std::this_thread::sleep_for(std::chrono::milliseconds(200)); // wait + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait 0.1 sec // Verify that all threads got unblocked for (size_t i = 0; i < num_threads; i++) { @@ -183,13 +229,12 @@ TEST(Channel, BufferedChannelCloseUnblocksReceiversTest) { } for (size_t i = 0; i < num_threads; i++) t[i].join(); - delete ch; } -// This tests that closing a buffered channel also unblocks -// any senders waiting for channel to have write space -TEST(Channel, BufferedChannelCloseUnblocksSendersTest) { - auto ch = MakeChannel(1); +void ChannelCloseUnblocksSendersTest(Channel *ch) { + using paddle::framework::details::Buffered; + using paddle::framework::details::UnBuffered; + size_t num_threads = 5; std::thread t[num_threads]; bool thread_ended[num_threads]; @@ -209,34 +254,56 @@ TEST(Channel, BufferedChannelCloseUnblocksSendersTest) { } std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait - // Verify that atleast 4 threads are blocked - int ct = 0; - for (size_t i = 0; i < num_threads; i++) { - if (thread_ended[i] == false) ct++; + if (dynamic_cast *>(ch)) { + // If ch is Buffered, atleast 4 threads must be blocked. + int ct = 0; + for (size_t i = 0; i < num_threads; i++) { + if (!thread_ended[i]) ct++; + } + EXPECT_GE(ct, 4); + } else { + // If ch is UnBuffered, all the threads should be blocked. + for (size_t i = 0; i < num_threads; i++) { + EXPECT_EQ(thread_ended[i], false); + } } - // Atleast 4 threads must be blocked - EXPECT_GE(ct, 4); - // Explicitly close the thread // This should unblock all senders CloseChannel(ch); - std::this_thread::sleep_for(std::chrono::milliseconds(200)); // wait + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait // Verify that all threads got unblocked for (size_t i = 0; i < num_threads; i++) { EXPECT_EQ(thread_ended[i], true); } - // Verify that only 1 send was successful - ct = 0; - for (size_t i = 0; i < num_threads; i++) { - if (send_success[i]) ct++; + if (dynamic_cast *>(ch)) { + // Verify that only 1 send was successful + int ct = 0; + for (size_t i = 0; i < num_threads; i++) { + if (send_success[i]) ct++; + } + // Only 1 send must be successful + EXPECT_EQ(ct, 1); } - // Only 1 send must be successful - EXPECT_EQ(ct, 1); for (size_t i = 0; i < num_threads; i++) t[i].join(); +} + +// This tests that closing a buffered channel also unblocks +// any receivers waiting on the channel +TEST(Channel, BufferedChannelCloseUnblocksReceiversTest) { + auto ch = MakeChannel(1); + ChannelCloseUnblocksReceiversTest(ch); + delete ch; +} + +// This tests that closing a buffered channel also unblocks +// any senders waiting for channel to have write space +TEST(Channel, BufferedChannelCloseUnblocksSendersTest) { + auto ch = MakeChannel(1); + ChannelCloseUnblocksSendersTest(ch); delete ch; } @@ -244,40 +311,7 @@ TEST(Channel, BufferedChannelCloseUnblocksSendersTest) { // unblocks any receivers waiting for senders TEST(Channel, UnbufferedChannelCloseUnblocksReceiversTest) { auto ch = MakeChannel(0); - size_t num_threads = 5; - std::thread t[num_threads]; - bool thread_ended[num_threads]; - - // Launches threads that try to read and are blocked becausew of no writers - for (size_t i = 0; i < num_threads; i++) { - thread_ended[i] = false; - t[i] = std::thread( - [&](bool *p) { - int data; - EXPECT_EQ(ch->Receive(&data), false); - *p = true; - }, - &thread_ended[i]); - } - std::this_thread::sleep_for(std::chrono::milliseconds(500)); // wait 0.5 sec - - // Verify that all the threads are blocked - for (size_t i = 0; i < num_threads; i++) { - EXPECT_EQ(thread_ended[i], false); - } - - // Explicitly close the thread - // This should unblock all receivers - CloseChannel(ch); - - std::this_thread::sleep_for(std::chrono::milliseconds(500)); // wait 0.5 sec - - // Verify that all threads got unblocked - for (size_t i = 0; i < num_threads; i++) { - EXPECT_EQ(thread_ended[i], true); - } - - for (size_t i = 0; i < num_threads; i++) t[i].join(); + ChannelCloseUnblocksReceiversTest(ch); delete ch; } @@ -285,40 +319,7 @@ TEST(Channel, UnbufferedChannelCloseUnblocksReceiversTest) { // unblocks any senders waiting for senders TEST(Channel, UnbufferedChannelCloseUnblocksSendersTest) { auto ch = MakeChannel(0); - size_t num_threads = 5; - std::thread t[num_threads]; - bool thread_ended[num_threads]; - - // Launches threads that try to read and are blocked becausew of no writers - for (size_t i = 0; i < num_threads; i++) { - thread_ended[i] = false; - t[i] = std::thread( - [&](bool *p) { - int data = 10; - EXPECT_EQ(ch->Send(&data), false); - *p = true; - }, - &thread_ended[i]); - } - std::this_thread::sleep_for(std::chrono::milliseconds(500)); // wait 0.5 sec - - // Verify that all the threads are blocked - for (size_t i = 0; i < num_threads; i++) { - EXPECT_EQ(thread_ended[i], false); - } - - // Explicitly close the thread - // This should unblock all receivers - CloseChannel(ch); - - std::this_thread::sleep_for(std::chrono::milliseconds(500)); // wait 0.5 sec - - // Verify that all threads got unblocked - for (size_t i = 0; i < num_threads; i++) { - EXPECT_EQ(thread_ended[i], true); - } - - for (size_t i = 0; i < num_threads; i++) t[i].join(); + ChannelCloseUnblocksReceiversTest(ch); delete ch; } @@ -381,3 +382,129 @@ TEST(Channel, UnbufferedMoreReceiveLessSendTest) { EXPECT_EQ(sum_receive, 28U); delete ch; } + +// This tests that destroying a channel unblocks +// any senders waiting for channel to have write space +void ChannelDestroyUnblockSenders(Channel *ch) { + size_t num_threads = 5; + std::thread t[num_threads]; + bool thread_ended[num_threads]; + bool send_success[num_threads]; + + // Launches threads that try to write and are blocked because of no readers + for (size_t i = 0; i < num_threads; i++) { + thread_ended[i] = false; + send_success[i] = false; + t[i] = std::thread( + [&](bool *ended, bool *success) { + int data = 10; + *success = ch->Send(&data); + *ended = true; + }, + &thread_ended[i], &send_success[i]); + } + + std::this_thread::sleep_for(std::chrono::milliseconds(500)); // wait 0.5 sec + bool is_buffered_channel = false; + if (dynamic_cast *>(ch)) is_buffered_channel = true; + + if (is_buffered_channel) { + // If channel is buffered, verify that atleast 4 threads are blocked + int ct = 0; + for (size_t i = 0; i < num_threads; i++) { + if (thread_ended[i] == false) ct++; + } + // Atleast 4 threads must be blocked + EXPECT_GE(ct, 4); + } else { + // Verify that all the threads are blocked + for (size_t i = 0; i < num_threads; i++) { + EXPECT_EQ(thread_ended[i], false); + } + } + // Explicitly destroy the channel + delete ch; + std::this_thread::sleep_for(std::chrono::milliseconds(200)); // wait + + // Verify that all threads got unblocked + for (size_t i = 0; i < num_threads; i++) { + EXPECT_EQ(thread_ended[i], true); + } + + // Count number of successfuld sends + int ct = 0; + for (size_t i = 0; i < num_threads; i++) { + if (send_success[i]) ct++; + } + + if (is_buffered_channel) { + // Only 1 send must be successful + EXPECT_EQ(ct, 1); + } else { + // In unbuffered channel, no send should be successful + EXPECT_EQ(ct, 0); + } + + // Join all threads + for (size_t i = 0; i < num_threads; i++) t[i].join(); +} + +// This tests that destroying a channel also unblocks +// any receivers waiting on the channel +void ChannelDestroyUnblockReceivers(Channel *ch) { + size_t num_threads = 5; + std::thread t[num_threads]; + bool thread_ended[num_threads]; + + // Launches threads that try to read and are blocked because of no writers + for (size_t i = 0; i < num_threads; i++) { + thread_ended[i] = false; + t[i] = std::thread( + [&](bool *p) { + int data; + // All reads should return false + EXPECT_EQ(ch->Receive(&data), false); + *p = true; + }, + &thread_ended[i]); + } + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // wait + + // Verify that all threads are blocked + for (size_t i = 0; i < num_threads; i++) { + EXPECT_EQ(thread_ended[i], false); + } + // delete the channel + delete ch; + std::this_thread::sleep_for(std::chrono::milliseconds(200)); // wait + // Verify that all threads got unblocked + for (size_t i = 0; i < num_threads; i++) { + EXPECT_EQ(thread_ended[i], true); + } + + for (size_t i = 0; i < num_threads; i++) t[i].join(); +} + +TEST(Channel, BufferedChannelDestroyUnblocksReceiversTest) { + size_t buffer_size = 1; + auto ch = MakeChannel(buffer_size); + ChannelDestroyUnblockReceivers(ch); +} + +TEST(Channel, BufferedChannelDestroyUnblocksSendersTest) { + size_t buffer_size = 1; + auto ch = MakeChannel(buffer_size); + ChannelDestroyUnblockSenders(ch); +} + +// This tests that destroying an unbuffered channel also unblocks +// unblocks any receivers waiting for senders +TEST(Channel, UnbufferedChannelDestroyUnblocksReceiversTest) { + auto ch = MakeChannel(0); + ChannelDestroyUnblockReceivers(ch); +} + +TEST(Channel, UnbufferedChannelDestroyUnblocksSendersTest) { + auto ch = MakeChannel(0); + ChannelDestroyUnblockSenders(ch); +} diff --git a/paddle/framework/details/buffered_channel.h b/paddle/framework/details/buffered_channel.h index 00b63da4da7844b41168c03f55e2faa84ff44154..227a4e4811f95441158150396b5b882815fd7844 100644 --- a/paddle/framework/details/buffered_channel.h +++ b/paddle/framework/details/buffered_channel.h @@ -1,4 +1,4 @@ -/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve. +/* Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. @@ -25,6 +25,14 @@ namespace paddle { namespace framework { namespace details { +// Four of the properties of Buffered Channel: +// - A send to a full channel blocks temporarily until a receive from the +// channel or the channel is closed. +// - A receive from an empty channel blocks temporarily until a send to the +// channel or the channel is closed. +// - A send to a closed channel returns false immediately. +// - A receive from a closed channel returns false immediately. + template class Buffered : public paddle::framework::Channel { friend Channel* paddle::framework::MakeChannel(size_t); @@ -42,8 +50,11 @@ class Buffered : public paddle::framework::Channel { std::mutex mu_; std::condition_variable empty_cond_var_; std::condition_variable full_cond_var_; + std::condition_variable destructor_cond_var_; std::deque channel_; std::atomic closed_{false}; + std::atomic send_ctr{0}; + std::atomic recv_ctr{0}; Buffered(size_t cap) : cap_(cap), closed_(false) { PADDLE_ENFORCE_GT(cap, 0); @@ -58,6 +69,7 @@ bool Buffered::Send(T* item) { if (closed_) { return ret; } + send_ctr++; std::unique_lock lock(mu_); full_cond_var_.wait(lock, [this]() { return channel_.size() < cap_ || closed_; }); @@ -67,20 +79,30 @@ bool Buffered::Send(T* item) { empty_cond_var_.notify_one(); ret = true; } + send_ctr--; + destructor_cond_var_.notify_one(); return ret; } template bool Buffered::Receive(T* item) { + bool ret = false; + // Once the channel has been closed and all data has been consumed, + // just return false. Don't even try acquiring the mutex. + if (closed_ && channel_.empty()) { + return false; + } + recv_ctr++; std::unique_lock lock(mu_); empty_cond_var_.wait(lock, [this]() { return !channel_.empty() || closed_; }); - bool ret = false; if (!channel_.empty()) { *item = std::move(channel_.front()); channel_.pop_front(); full_cond_var_.notify_one(); ret = true; } + recv_ctr--; + destructor_cond_var_.notify_one(); return ret; } @@ -100,6 +122,12 @@ Buffered::~Buffered() { closed_ = true; channel_.clear(); NotifyAllParticipants(&lock); + + // The destructor must wait for all readers and writers to complete their task + // The channel has been closed, so we will not accept new readers and writers + lock.lock(); + destructor_cond_var_.wait( + lock, [this]() { return send_ctr == 0 && recv_ctr == 0; }); } template diff --git a/paddle/framework/details/cow_ptr.h b/paddle/framework/details/cow_ptr.h index 7e308ffb5a49876aa2c1833b3b7e2a2c7eb137aa..69bcea625288eba897e761a1d634f19c41dc0f79 100644 --- a/paddle/framework/details/cow_ptr.h +++ b/paddle/framework/details/cow_ptr.h @@ -1,4 +1,4 @@ -/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved. +/* Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. diff --git a/paddle/framework/details/cow_ptr_test.cc b/paddle/framework/details/cow_ptr_test.cc index 936954a2333e7e5d2a932abad641279db9ef7b9f..1f4a12bca0dcab2d146cc62cd7ce1c2d7abcddf9 100644 --- a/paddle/framework/details/cow_ptr_test.cc +++ b/paddle/framework/details/cow_ptr_test.cc @@ -1,4 +1,4 @@ -/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved. +/* Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. diff --git a/paddle/framework/details/op_registry.h b/paddle/framework/details/op_registry.h index 6d50e820b2b625f932768d2ca671d999071f1ca6..31a40bcbcb3905f01aebefe89526f3cfba8cb8c7 100644 --- a/paddle/framework/details/op_registry.h +++ b/paddle/framework/details/op_registry.h @@ -1,4 +1,4 @@ -/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve. +/* Copyright (c) 2017 PaddlePaddle Authors. All Rights Reserved. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. diff --git a/paddle/framework/details/unbuffered_channel.h b/paddle/framework/details/unbuffered_channel.h index 815cebad2d8c08aa31bb566bc6c51250870383d8..6b5c2196cb2991051c48f7da8397d2f479ca4c58 100644 --- a/paddle/framework/details/unbuffered_channel.h +++ b/paddle/framework/details/unbuffered_channel.h @@ -23,6 +23,13 @@ namespace paddle { namespace framework { namespace details { +// Four of the properties of UnBuffered Channel: +// - A send to a channel blocks temporarily until a receive from the +// channel or the channel is closed. +// - A receive from a channel blocks temporarily until a send to the +// channel or the channel is closed. +// - A send to a closed channel returns false immediately. +// - A receive from a closed channel returns false immediately. template class UnBuffered : public paddle::framework::Channel { friend Channel* paddle::framework::MakeChannel(size_t); @@ -45,9 +52,11 @@ class UnBuffered : public paddle::framework::Channel { // A transaction occurs only when both are true std::atomic reader_found_{false}, writer_found_{false}; std::condition_variable cv_channel_; - std::condition_variable_any cv_reader_, cv_writer_; + std::condition_variable_any cv_reader_, cv_writer_, cv_destructor_; T* item{nullptr}; std::atomic closed_{false}; + std::atomic send_ctr{0}; + std::atomic recv_ctr{0}; UnBuffered() : closed_(false) {} @@ -62,6 +71,7 @@ bool UnBuffered::Send(T* data) { if (closed_) { return ret; } + send_ctr++; // Prevent other writers from entering std::unique_lock writer_lock(mu_write_); writer_found_ = true; @@ -81,6 +91,8 @@ bool UnBuffered::Send(T* data) { ret = true; } writer_found_ = false; + send_ctr--; + cv_destructor_.notify_one(); return ret; } @@ -88,6 +100,12 @@ bool UnBuffered::Send(T* data) { // data that was sent by a writer is read from a reader. template bool UnBuffered::Receive(T* data) { + bool ret = false; + // If channel is closed, we don't even want any reader to enter. + // Unlike a buffered channel, an unbuffered channel does not allow + // readers to read after closing because there is no buffer to be consumed. + if (closed_) return ret; + recv_ctr++; // Prevent other readers from entering std::unique_lock read_lock{mu_read_}; reader_found_ = true; @@ -96,7 +114,6 @@ bool UnBuffered::Receive(T* data) { cv_reader_.wait(cv_lock, [this]() { return writer_found_ == true || closed_; }); cv_writer_.notify_one(); - bool ret = false; if (!closed_) { std::unique_lock lock_ch{mu_ch_}; // Reader should wait for the writer to first write its data @@ -110,6 +127,8 @@ bool UnBuffered::Receive(T* data) { cv_channel_.notify_one(); } reader_found_ = false; + recv_ctr--; + cv_destructor_.notify_one(); return ret; } @@ -135,6 +154,9 @@ UnBuffered::~UnBuffered() { item = nullptr; closed_ = true; NotifyAllParticipants(&lock); + lock.lock(); + cv_destructor_.wait(lock, + [this]() { return send_ctr == 0 && recv_ctr == 0; }); } // This function notifies all the readers, writers and diff --git a/paddle/framework/op_desc.cc b/paddle/framework/op_desc.cc index ea4028750248ec47f5094a67f736fb217216af6d..b51afe499bbc0e6b727aeeb4334f56e400ea81a5 100644 --- a/paddle/framework/op_desc.cc +++ b/paddle/framework/op_desc.cc @@ -125,11 +125,10 @@ OpDesc::OpDesc(const proto::OpDesc &desc, ProgramDesc *prog, BlockDesc *block) // restore attrs_ for (const proto::OpDesc::Attr &attr : desc_.attrs()) { std::string attr_name = attr.name(); + // The sub_block referred to by the BLOCK attr hasn't been added + // to ProgramDesc class yet, we skip setting BLOCK attr here. if (attr.type() != proto::AttrType::BLOCK) { attrs_[attr_name] = GetAttrValue(attr); - } else { - auto bid = attr.block_idx(); - attrs_[attr_name] = prog->MutableBlock(bid); } } this->block_ = block; diff --git a/paddle/framework/program_desc.cc b/paddle/framework/program_desc.cc index 15ea4035c6e6193105b621210a900e74d1466941..0e937dda4e185590648962a6d4f827eea21eb620 100644 --- a/paddle/framework/program_desc.cc +++ b/paddle/framework/program_desc.cc @@ -43,11 +43,20 @@ ProgramDesc::ProgramDesc() { ProgramDesc::ProgramDesc(const ProgramDesc &o) { desc_ = o.desc_; - for (int i = 0; i < desc_.blocks_size(); ++i) { auto *block = desc_.mutable_blocks(i); blocks_.emplace_back(new BlockDesc(*o.blocks_[i], block, this)); } + for (auto &block : blocks_) { + for (auto *op : block->AllOps()) { + for (const auto &attr : op->Proto()->attrs()) { + if (attr.type() == proto::AttrType::BLOCK) { + size_t blk_idx = attr.block_idx(); + op->SetBlockAttr(attr.name(), *this->MutableBlock(blk_idx)); + } + } + } + } } ProgramDesc::ProgramDesc(const proto::ProgramDesc &desc) { @@ -55,6 +64,16 @@ ProgramDesc::ProgramDesc(const proto::ProgramDesc &desc) { for (auto &block_desc : *desc_.mutable_blocks()) { blocks_.emplace_back(new BlockDesc(this, &block_desc)); } + for (auto &block : blocks_) { + for (auto *op : block->AllOps()) { + for (const auto &attr : op->Proto()->attrs()) { + if (attr.type() == proto::AttrType::BLOCK) { + size_t blk_idx = attr.block_idx(); + op->SetBlockAttr(attr.name(), *this->MutableBlock(blk_idx)); + } + } + } + } } ProgramDesc::ProgramDesc(const std::string &binary_str) { diff --git a/paddle/framework/prune.cc b/paddle/framework/prune.cc index bff8e0bceaca9749101b2c45edddba526d565624..ddd6b993d40f72cba919fad95318f70409c98bca 100644 --- a/paddle/framework/prune.cc +++ b/paddle/framework/prune.cc @@ -49,11 +49,28 @@ bool IsTarget(const proto::OpDesc& op_desc) { return false; } -void prune_impl(const proto::ProgramDesc& input, proto::ProgramDesc* output, - int block_id) { - // TODO(tonyyang-svail): - // - will change to use multiple blocks for RNN op and Cond Op +int GetSubBlockIndex(const proto::OpDesc& op_desc) { + for (auto& attr : op_desc.attrs()) { + if (attr.type() == proto::AttrType::BLOCK) { + PADDLE_ENFORCE(attr.has_block_idx()); + return attr.block_idx(); + } + } + return -1; +} + +bool HasSubBlock(const proto::OpDesc& op_desc) { + return GetSubBlockIndex(op_desc) > 0; +} +// block_id is the idx of the current block in the input desc +// parent_block_id is the idx of the parent of the current block +// in the output desc, -1 means the current block is global block +// dependent_vars is passed recursively from the parent block to +// the child block to help pruning +void prune_impl(const proto::ProgramDesc& input, proto::ProgramDesc* output, + int block_id, int parent_block_id, + std::set& dependent_vars) { auto& block = input.blocks(block_id); auto& ops = block.ops(); @@ -72,11 +89,9 @@ void prune_impl(const proto::ProgramDesc& input, proto::ProgramDesc* output, expect_fetch = (op_desc.type() == kFetchOpType); } - std::set dependent_vars; std::vector should_run; for (auto op_iter = ops.rbegin(); op_iter != ops.rend(); ++op_iter) { auto& op_desc = *op_iter; - if (IsTarget(op_desc) || HasDependentVar(op_desc, dependent_vars)) { // insert its input to the dependency graph for (auto& var : op_desc.inputs()) { @@ -84,7 +99,6 @@ void prune_impl(const proto::ProgramDesc& input, proto::ProgramDesc* output, dependent_vars.insert(argu); } } - should_run.push_back(true); } else { should_run.push_back(false); @@ -95,45 +109,81 @@ void prune_impl(const proto::ProgramDesc& input, proto::ProgramDesc* output, // we reverse the should_run vector std::reverse(should_run.begin(), should_run.end()); - *output = input; - auto* op_field = output->mutable_blocks(block_id)->mutable_ops(); + // copy the current block from input to output + auto* block_field = output->mutable_blocks(); + *block_field->Add() = input.blocks(block_id); + + int output_block_id = output->blocks_size() - 1; + auto* output_block = output->mutable_blocks(output_block_id); + output_block->set_idx(output_block_id); + output_block->set_parent_idx(parent_block_id); + + auto* op_field = output_block->mutable_ops(); op_field->Clear(); for (size_t i = 0; i < should_run.size(); ++i) { if (should_run[i]) { - *op_field->Add() = input.blocks(block_id).ops(i); + auto* op = op_field->Add(); + *op = input.blocks(block_id).ops(i); + if (HasSubBlock(*op)) { + // create sub_block_dependent_vars here to help prune the sub block + std::set sub_block_dependent_vars; + for (auto& var : op->inputs()) { + for (auto& argu : var.arguments()) { + sub_block_dependent_vars.insert(argu); + } + } + for (auto& var : op->outputs()) { + for (auto& argu : var.arguments()) { + sub_block_dependent_vars.insert(argu); + } + } + // GetSubBlockIndex(*op) is the idx of the sub_block in the input desc + // output_block_id is the idx of the current block in the output desc + prune_impl(input, output, GetSubBlockIndex(*op), output_block_id, + sub_block_dependent_vars); + } } } // remove the VarDescs in BlockDesc that are not referenced in // the pruned OpDescs std::unordered_map var_map; - auto* var_field = output->mutable_blocks(block_id)->mutable_vars(); + auto* var_field = output->mutable_blocks(output_block_id)->mutable_vars(); for (const auto& var : *var_field) { var_map[var.name()] = var; } - var_field->Clear(); + std::set var_names; for (const auto& op : *op_field) { - // add VarDescs of all input arguments for each OpDesc auto& input_field = op.inputs(); for (auto& input_var : input_field) { for (auto& arg : input_var.arguments()) { - *var_field->Add() = var_map[arg]; + if (var_map.count(arg) != 0) { + var_names.insert(arg); + } } } - // add VarDescs of all output arguments for each OpDesc auto& output_field = op.outputs(); for (auto& output_var : output_field) { for (auto& arg : output_var.arguments()) { - *var_field->Add() = var_map[arg]; + if (var_map.count(arg) != 0) { + var_names.insert(arg); + } } } } + + var_field->Clear(); + for (const auto& name : var_names) { + *var_field->Add() = var_map[name]; + } } // TODO(fengjiayi): Prune() could be inplaced to avoid unnecessary copies void Prune(const proto::ProgramDesc& input, proto::ProgramDesc* output) { - prune_impl(input, output, 0); + std::set dependent_vars; + output->clear_blocks(); + prune_impl(input, output, 0, -1, dependent_vars); } void inference_optimize_impl(const proto::ProgramDesc& input, diff --git a/paddle/inference/io.cc b/paddle/inference/io.cc index 1ed14b69c83a7a0fb5a55db9c179df133407440c..784e87970f77857e7f3182df904dc0133c44d6c9 100644 --- a/paddle/inference/io.cc +++ b/paddle/inference/io.cc @@ -21,6 +21,17 @@ limitations under the License. */ namespace paddle { namespace inference { +void ReadBinaryFile(const std::string& filename, std::string& contents) { + VLOG(3) << "loading model from " << filename; + std::ifstream inputfs(filename, std::ios::in | std::ios::binary); + inputfs.seekg(0, std::ios::end); + contents.clear(); + contents.resize(inputfs.tellg()); + inputfs.seekg(0, std::ios::beg); + inputfs.read(&contents[0], contents.size()); + inputfs.close(); +} + bool IsParameter(const framework::VarDesc* var, const framework::ProgramDesc& main_program) { if (var->Persistable()) { @@ -44,12 +55,15 @@ bool IsParameter(const framework::VarDesc* var, void LoadPersistables(framework::Executor& executor, framework::Scope& scope, + const framework::ProgramDesc& main_program, const std::string& dirname, - const framework::ProgramDesc& main_program) { + const std::string& param_filename) { const framework::BlockDesc& global_block = main_program.Block(0); framework::ProgramDesc* load_program = new framework::ProgramDesc(); framework::BlockDesc* load_block = load_program->MutableBlock(0); + std::vector paramlist; + for (auto* var : global_block.AllVars()) { if (IsParameter(var, main_program)) { VLOG(3) << "parameter's name: " << var->Name(); @@ -61,15 +75,33 @@ void LoadPersistables(framework::Executor& executor, new_var->SetLoDLevel(var->GetLoDLevel()); new_var->SetPersistable(true); - // append_op - framework::OpDesc* op = load_block->AppendOp(); - op->SetType("load"); - op->SetOutput("Out", {new_var->Name()}); - op->SetAttr("file_path", {dirname + "/" + new_var->Name()}); - op->CheckAttrs(); + if (!param_filename.empty()) { + paramlist.push_back(new_var->Name()); + } else { + // append_op + framework::OpDesc* op = load_block->AppendOp(); + op->SetType("load"); + op->SetOutput("Out", {new_var->Name()}); + op->SetAttr("file_path", {dirname + "/" + new_var->Name()}); + op->CheckAttrs(); + } } } + + if (!param_filename.empty()) { + // sort paramlist to have consistent ordering + std::sort(paramlist.begin(), paramlist.end()); + // append just the load_combine op + framework::OpDesc* op = load_block->AppendOp(); + op->SetType("load_combine"); + op->SetOutput("Out", paramlist); + op->SetAttr("file_path", {param_filename}); + op->CheckAttrs(); + } + executor.Run(*load_program, &scope, 0, true, true); + + VLOG(3) << "Ran loading successfully"; delete load_program; } @@ -77,20 +109,29 @@ std::unique_ptr Load(framework::Executor& executor, framework::Scope& scope, const std::string& dirname) { std::string model_filename = dirname + "/__model__"; - LOG(INFO) << "loading model from " << model_filename; - std::ifstream inputfs(model_filename, std::ios::in | std::ios::binary); std::string program_desc_str; - inputfs.seekg(0, std::ios::end); - program_desc_str.resize(inputfs.tellg()); - inputfs.seekg(0, std::ios::beg); - LOG(INFO) << "program_desc_str's size: " << program_desc_str.size(); - inputfs.read(&program_desc_str[0], program_desc_str.size()); - inputfs.close(); + ReadBinaryFile(model_filename, program_desc_str); + + std::unique_ptr main_program( + new framework::ProgramDesc(program_desc_str)); + + LoadPersistables(executor, scope, *main_program, dirname, ""); + return main_program; +} + +std::unique_ptr Load( + framework::Executor& executor, + framework::Scope& scope, + const std::string& prog_filename, + const std::string& param_filename) { + std::string model_filename = prog_filename; + std::string program_desc_str; + ReadBinaryFile(model_filename, program_desc_str); std::unique_ptr main_program( new framework::ProgramDesc(program_desc_str)); - LoadPersistables(executor, scope, dirname, *main_program); + LoadPersistables(executor, scope, *main_program, "", param_filename); return main_program; } diff --git a/paddle/inference/io.h b/paddle/inference/io.h index 962b6c4e20d30de3cc28eae1c8c5c33b3ab5f6ac..a7d7c499690620740d8627e7f5085d728d67f7c3 100644 --- a/paddle/inference/io.h +++ b/paddle/inference/io.h @@ -26,12 +26,18 @@ namespace inference { void LoadPersistables(framework::Executor& executor, framework::Scope& scope, + const framework::ProgramDesc& main_program, const std::string& dirname, - const framework::ProgramDesc& main_program); + const std::string& param_filename); std::unique_ptr Load(framework::Executor& executor, framework::Scope& scope, const std::string& dirname); +std::unique_ptr Load(framework::Executor& executor, + framework::Scope& scope, + const std::string& prog_filename, + const std::string& param_filename); + } // namespace inference } // namespace paddle diff --git a/paddle/inference/tests/book/CMakeLists.txt b/paddle/inference/tests/book/CMakeLists.txt index 63afeb18aebdf446c01cd4fdac13d238467801e4..0a96829bdd20f5dcb0c3fed501d27c27f2f73b17 100644 --- a/paddle/inference/tests/book/CMakeLists.txt +++ b/paddle/inference/tests/book/CMakeLists.txt @@ -27,3 +27,4 @@ endfunction(inference_test) inference_test(recognize_digits ARGS mlp) inference_test(image_classification ARGS vgg resnet) inference_test(label_semantic_roles) +inference_test(rnn_encoder_decoder) diff --git a/paddle/inference/tests/book/test_helper.h b/paddle/inference/tests/book/test_helper.h index 32db643fca2b026b674ea0b1ecd9aad5224e9e68..3e66ced94fe6360f0be948a6838cc37ff2f65eed 100644 --- a/paddle/inference/tests/book/test_helper.h +++ b/paddle/inference/tests/book/test_helper.h @@ -67,17 +67,28 @@ void CheckError(paddle::framework::LoDTensor& output1, EXPECT_EQ(count, 0) << "There are " << count << " different elements."; } -template +template void TestInference(const std::string& dirname, const std::vector& cpu_feeds, std::vector& cpu_fetchs) { - // 1. Define place, executor and scope + // 1. Define place, executor, scope and inference_program auto place = Place(); auto executor = paddle::framework::Executor(place); auto* scope = new paddle::framework::Scope(); + std::unique_ptr inference_program; // 2. Initialize the inference_program and load all parameters from file - auto inference_program = paddle::inference::Load(executor, *scope, dirname); + if (IsCombined) { + // Hard-coding the names for combined params case + std::string prog_filename = "__model_combined__"; + std::string param_filename = "__params_combined__"; + inference_program = paddle::inference::Load(executor, + *scope, + dirname + "/" + prog_filename, + dirname + "/" + param_filename); + } else { + inference_program = paddle::inference::Load(executor, *scope, dirname); + } // 3. Get the feed_target_names and fetch_target_names const std::vector& feed_target_names = diff --git a/paddle/inference/tests/book/test_inference_recognize_digits.cc b/paddle/inference/tests/book/test_inference_recognize_digits.cc index 48f887e6bc680087af4cce74b5c5422a4eba3726..3a48db7fe08205a3e078592651c739f77d5bf415 100644 --- a/paddle/inference/tests/book/test_inference_recognize_digits.cc +++ b/paddle/inference/tests/book/test_inference_recognize_digits.cc @@ -59,3 +59,45 @@ TEST(inference, recognize_digits) { CheckError(output1, output2); #endif } + +TEST(inference, recognize_digits_combine) { + if (FLAGS_dirname.empty()) { + LOG(FATAL) << "Usage: ./example --dirname=path/to/your/model"; + } + + LOG(INFO) << "FLAGS_dirname: " << FLAGS_dirname << std::endl; + std::string dirname = FLAGS_dirname; + + // 0. Call `paddle::framework::InitDevices()` initialize all the devices + // In unittests, this is done in paddle/testing/paddle_gtest_main.cc + + paddle::framework::LoDTensor input; + // Use normilized image pixels as input data, + // which should be in the range [-1.0, 1.0]. + SetupTensor( + input, {1, 28, 28}, static_cast(-1), static_cast(1)); + std::vector cpu_feeds; + cpu_feeds.push_back(&input); + + paddle::framework::LoDTensor output1; + std::vector cpu_fetchs1; + cpu_fetchs1.push_back(&output1); + + // Run inference on CPU + TestInference( + dirname, cpu_feeds, cpu_fetchs1); + LOG(INFO) << output1.dims(); + +#ifdef PADDLE_WITH_CUDA + paddle::framework::LoDTensor output2; + std::vector cpu_fetchs2; + cpu_fetchs2.push_back(&output2); + + // Run inference on CUDA GPU + TestInference( + dirname, cpu_feeds, cpu_fetchs2); + LOG(INFO) << output2.dims(); + + CheckError(output1, output2); +#endif +} diff --git a/paddle/inference/tests/book/test_inference_rnn_encoder_decoder.cc b/paddle/inference/tests/book/test_inference_rnn_encoder_decoder.cc new file mode 100644 index 0000000000000000000000000000000000000000..9bfc0407b7f2732a14e7ac0f319a3d39b9e641bc --- /dev/null +++ b/paddle/inference/tests/book/test_inference_rnn_encoder_decoder.cc @@ -0,0 +1,67 @@ +/* Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserve. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. */ + +#include +#include "gflags/gflags.h" +#include "test_helper.h" + +DEFINE_string(dirname, "", "Directory of the inference model."); + +TEST(inference, rnn_encoder_decoder) { + if (FLAGS_dirname.empty()) { + LOG(FATAL) << "Usage: ./example --dirname=path/to/your/model"; + } + + LOG(INFO) << "FLAGS_dirname: " << FLAGS_dirname << std::endl; + std::string dirname = FLAGS_dirname; + + // 0. Call `paddle::framework::InitDevices()` initialize all the devices + // In unittests, this is done in paddle/testing/paddle_gtest_main.cc + + paddle::framework::LoDTensor word_data, trg_word; + paddle::framework::LoD lod{{0, 4, 10}}; + + SetupLoDTensor( + word_data, lod, static_cast(0), static_cast(1)); + SetupLoDTensor( + trg_word, lod, static_cast(0), static_cast(1)); + + std::vector cpu_feeds; + cpu_feeds.push_back(&word_data); + cpu_feeds.push_back(&trg_word); + + paddle::framework::LoDTensor output1; + std::vector cpu_fetchs1; + cpu_fetchs1.push_back(&output1); + + // Run inference on CPU + TestInference( + dirname, cpu_feeds, cpu_fetchs1); + LOG(INFO) << output1.lod(); + LOG(INFO) << output1.dims(); + +#ifdef PADDLE_WITH_CUDA + paddle::framework::LoDTensor output2; + std::vector cpu_fetchs2; + cpu_fetchs2.push_back(&output2); + + // Run inference on CUDA GPU + TestInference( + dirname, cpu_feeds, cpu_fetchs2); + LOG(INFO) << output2.lod(); + LOG(INFO) << output2.dims(); + + CheckError(output1, output2); +#endif +} diff --git a/paddle/operators/compare_op.h b/paddle/operators/compare_op.h index b275fd75b3512343825170fc38565dd27f7f1c75..79b8c6f59c7ad3d77aa969f6b4f36f8050cfe823 100644 --- a/paddle/operators/compare_op.h +++ b/paddle/operators/compare_op.h @@ -62,7 +62,7 @@ class CompareOpKernel z->mutable_data(context.GetPlace()); int axis = context.Attr("axis"); ElementwiseComputeEx(context, x, y, axis, - z); + Functor(), z); } }; diff --git a/paddle/operators/elementwise_add_op.h b/paddle/operators/elementwise_add_op.h index c32288d6984f126f2374a13973541f4f663b25a4..c24f97a85092ff14e8211ca8bc4bb9b155510a2c 100644 --- a/paddle/operators/elementwise_add_op.h +++ b/paddle/operators/elementwise_add_op.h @@ -35,7 +35,8 @@ class ElementwiseAddKernel : public framework::OpKernel { auto* z = ctx.Output("Out"); z->mutable_data(ctx.GetPlace()); int axis = ctx.Attr("axis"); - ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, z); + ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, + AddFunctor(), z); } }; diff --git a/paddle/operators/elementwise_div_op.h b/paddle/operators/elementwise_div_op.h index 07ebade31ff5b3d5c89156e28ff5fa0670a9a842..dc863cc598ec6015067f166b1544a5d20223662a 100644 --- a/paddle/operators/elementwise_div_op.h +++ b/paddle/operators/elementwise_div_op.h @@ -35,7 +35,8 @@ class ElementwiseDivKernel : public framework::OpKernel { auto* z = ctx.Output("Out"); z->mutable_data(ctx.GetPlace()); int axis = ctx.Attr("axis"); - ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, z); + ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, + DivFunctor(), z); } }; diff --git a/paddle/operators/elementwise_max_op.h b/paddle/operators/elementwise_max_op.h index 717e45ab31db9b9a6629fb33e17654dbf986d8c5..67efe4e1511e054d54f91b5aa22ce28f222ed20a 100644 --- a/paddle/operators/elementwise_max_op.h +++ b/paddle/operators/elementwise_max_op.h @@ -35,7 +35,8 @@ class ElementwiseMaxKernel : public framework::OpKernel { auto* z = ctx.Output("Out"); z->mutable_data(ctx.GetPlace()); int axis = ctx.Attr("axis"); - ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, z); + ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, + MaxFunctor(), z); } }; diff --git a/paddle/operators/elementwise_min_op.h b/paddle/operators/elementwise_min_op.h index 0de9a91c52b0ab82cd62604de318ce68e56b767d..cf11759404d3342b8a1c0080fa09f6cd57e735db 100644 --- a/paddle/operators/elementwise_min_op.h +++ b/paddle/operators/elementwise_min_op.h @@ -35,7 +35,8 @@ class ElementwiseMinKernel : public framework::OpKernel { auto* z = ctx.Output("Out"); z->mutable_data(ctx.GetPlace()); int axis = ctx.Attr("axis"); - ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, z); + ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, + MinFunctor(), z); } }; diff --git a/paddle/operators/elementwise_mul_op.h b/paddle/operators/elementwise_mul_op.h index ae7a71e0244dfb8ad3e55683ac081f92bc36bea5..773125f5ca54e7b529df47a2823d56a5ad71e50d 100644 --- a/paddle/operators/elementwise_mul_op.h +++ b/paddle/operators/elementwise_mul_op.h @@ -34,7 +34,8 @@ class ElementwiseMulKernel : public framework::OpKernel { auto* z = ctx.Output("Out"); z->mutable_data(ctx.GetPlace()); int axis = ctx.Attr("axis"); - ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, z); + ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, + MulFunctor(), z); } }; diff --git a/paddle/operators/elementwise_op_function.h b/paddle/operators/elementwise_op_function.h index 213fe1f5a818873e8b666464cb112637261c598c..74abf7c4a58788eb0e53025886f10f5a43021a9e 100644 --- a/paddle/operators/elementwise_op_function.h +++ b/paddle/operators/elementwise_op_function.h @@ -365,10 +365,10 @@ template void ElementwiseComputeEx(const framework::ExecutionContext& ctx, const framework::Tensor* x, - const framework::Tensor* y, int axis, + const framework::Tensor* y, int axis, Functor func, framework::Tensor* z) { TransformFunctor functor( - x, y, z, ctx.template device_context(), Functor()); + x, y, z, ctx.template device_context(), func); auto x_dims = x->dims(); auto y_dims = y->dims(); diff --git a/paddle/operators/elementwise_pow_op.h b/paddle/operators/elementwise_pow_op.h index 874fd3f09f2afaccfbfca75799cc3448f7393b03..0c5dd031ec46ebecaabb701839c0f69c02678eb0 100644 --- a/paddle/operators/elementwise_pow_op.h +++ b/paddle/operators/elementwise_pow_op.h @@ -36,7 +36,8 @@ class ElementwisePowKernel : public framework::OpKernel { auto* z = ctx.Output("Out"); z->mutable_data(ctx.GetPlace()); int axis = ctx.Attr("axis"); - ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, z); + ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, + PowFunctor(), z); } }; diff --git a/paddle/operators/elementwise_sub_op.h b/paddle/operators/elementwise_sub_op.h index c2749a8e6ba689233dab4f3c72de10bf01f39fab..6a88c5f6b4c869f8ab5b4fa3b112ffc264be7145 100644 --- a/paddle/operators/elementwise_sub_op.h +++ b/paddle/operators/elementwise_sub_op.h @@ -34,7 +34,8 @@ class ElementwiseSubKernel : public framework::OpKernel { auto* z = ctx.Output("Out"); z->mutable_data(ctx.GetPlace()); int axis = ctx.Attr("axis"); - ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, z); + ElementwiseComputeEx, DeviceContext, T>(ctx, x, y, axis, + SubFunctor(), z); } }; diff --git a/paddle/operators/layer_norm_op.cc b/paddle/operators/layer_norm_op.cc index 1c6d2ae4d05becaeed34d66cad398cc90f9d3ece..76d5d571c31c0cdec207cd171291da1f58d29b61 100644 --- a/paddle/operators/layer_norm_op.cc +++ b/paddle/operators/layer_norm_op.cc @@ -21,13 +21,6 @@ using Tensor = framework::Tensor; using LoDTensor = framework::LoDTensor; using DataLayout = framework::DataLayout; -template -using EigenMatrixMapRowMajor = Eigen::Map< - Eigen::Matrix>; -template -using ConstEigenMatrixMapRowMajor = Eigen::Map< - const Eigen::Matrix>; - class LayerNormOp : public framework::OperatorWithKernel { public: using framework::OperatorWithKernel::OperatorWithKernel; @@ -108,7 +101,6 @@ class LayerNormOpMaker : public framework::OpProtoAndCheckerMaker { AddComment(R"DOC( Layer Normalization. - Layer Norm has been implemented as discussed in the paper: https://arxiv.org/abs/1607.06450 ... @@ -116,75 +108,6 @@ https://arxiv.org/abs/1607.06450 } }; -template -class LayerNormKernel - : public framework::OpKernel { - public: - void Compute(const framework::ExecutionContext &ctx) const override { - const float epsilon = ctx.Attr("epsilon"); - const auto *scale = ctx.Input("Scale"); - const auto *bias = ctx.Input("Bias"); - const auto *x = ctx.Input("X"); - const auto &x_dims = x->dims(); - const auto begin_norm_axis = ctx.Attr("begin_norm_axis"); - - auto *output = ctx.Output("Y"); - auto *mean = ctx.Output("Mean"); - auto *var = ctx.Output("Variance"); - output->mutable_data(ctx.GetPlace()); - mean->mutable_data(ctx.GetPlace()); - var->mutable_data(ctx.GetPlace()); - - auto matrix_dim = framework::flatten_to_2d(x_dims, begin_norm_axis); - int left = static_cast(matrix_dim[0]); - int right = static_cast(matrix_dim[1]); - - auto input_map = ConstEigenMatrixMapRowMajor(x->data(), left, right); - - auto mean_map = EigenMatrixMapRowMajor(mean->data(), left, 1); - auto var_map = EigenMatrixMapRowMajor(var->data(), left, 1); - auto output_map = EigenMatrixMapRowMajor(output->data(), left, right); - - auto squre = [](T ele) { return ele * ele; }; - auto add_epslion = [epsilon](T ele) { return ele + epsilon; }; - - mean_map = input_map.rowwise().mean(); - var_map = (input_map - mean_map.replicate(1, right)) - .unaryExpr(squre) - .rowwise() - .mean() - .unaryExpr(add_epslion); - - auto inv_std_func = [](T ele) { return std::sqrt(1 / ele); }; - // TODO(zcd): Some thinking about output_map, is it appropriate that - // `output_map` and `input_map` point to the same memory. - auto inv_std = var_map.unaryExpr(inv_std_func); - if (scale && bias) { - auto scale_map = - ConstEigenMatrixMapRowMajor(scale->data(), 1, right); - auto bias_map = ConstEigenMatrixMapRowMajor(bias->data(), 1, right); - output_map = (input_map - mean_map.replicate(1, right)) - .cwiseProduct(inv_std.replicate(1, right)) - .cwiseProduct(scale_map.replicate(left, 1)) + - bias_map.replicate(left, 1); - } else if (scale) { - auto scale_map = - ConstEigenMatrixMapRowMajor(scale->data(), 1, right); - output_map = (input_map - mean_map.replicate(1, right)) - .cwiseProduct(inv_std.replicate(1, right)) - .cwiseProduct(scale_map.replicate(left, 1)); - } else if (bias) { - auto bias_map = ConstEigenMatrixMapRowMajor(bias->data(), 1, right); - output_map = (input_map - mean_map.replicate(1, right)) - .cwiseProduct(inv_std.replicate(1, right)) + - bias_map.replicate(left, 1); - } else { - output_map = (input_map - mean_map.replicate(1, right)) - .cwiseProduct(inv_std.replicate(1, right)); - } - } -}; - class LayerNormGradOp : public framework::OperatorWithKernel { public: using framework::OperatorWithKernel::OperatorWithKernel; @@ -237,125 +160,6 @@ class LayerNormGradOp : public framework::OperatorWithKernel { } }; -template -class LayerNormGradKernel - : public framework::OpKernel { - public: - void Compute(const framework::ExecutionContext &ctx) const override { - const auto *x = ctx.Input("X"); - const auto *mean = ctx.Input("Mean"); - const auto *var = ctx.Input("Variance"); - const auto *scale = ctx.Input("Scale"); - const auto *d_y = ctx.Input(framework::GradVarName("Y")); - - const auto &x_dims = x->dims(); - - const auto begin_norm_axis = ctx.Attr("begin_norm_axis"); - auto matrix_dim = framework::flatten_to_2d(x_dims, begin_norm_axis); - int left = static_cast(matrix_dim[0]); - int right = static_cast(matrix_dim[1]); - - // init output - auto *d_x = ctx.Output(framework::GradVarName("X")); - auto *d_scale = ctx.Output(framework::GradVarName("Scale")); - auto *d_bias = ctx.Output(framework::GradVarName("Bias")); - - auto x_map = ConstEigenMatrixMapRowMajor(x->data(), left, right); - auto d_y_map = ConstEigenMatrixMapRowMajor(d_y->data(), left, right); - auto mean_map = ConstEigenMatrixMapRowMajor(mean->data(), left, 1); - auto var_map = ConstEigenMatrixMapRowMajor(var->data(), left, 1); - - if (d_bias) { - d_bias->mutable_data(ctx.GetPlace()); - auto d_bias_map = EigenMatrixMapRowMajor(d_bias->data(), 1, right); - d_bias_map = d_y_map.colwise().sum(); - } - if (d_scale) { - d_scale->mutable_data(ctx.GetPlace()); - auto d_scale_map = - EigenMatrixMapRowMajor(d_scale->data(), 1, right); - auto inv_std_func = [](T ele) { return std::sqrt(1 / ele); }; - // There are two equation to compute d_scale. One uses "Y" and the other - // does not use "Y" - d_scale_map = - ((x_map - mean_map.replicate(1, right)) - .cwiseProduct( - var_map.unaryExpr(inv_std_func).replicate(1, right)) - .cwiseProduct(d_y_map)) - .colwise() - .sum(); - } - - if (d_x) { - d_x->mutable_data(ctx.GetPlace()); - auto d_x_map = EigenMatrixMapRowMajor(d_x->data(), left, right); - auto triple_product_func = [](T ele) { return ele * ele * ele; }; - auto inv_std_func = [](T ele) { return std::sqrt(1 / ele); }; - // TODO(zcd): these code can be refined - if (d_scale) { - auto scale_map = - ConstEigenMatrixMapRowMajor(scale->data(), 1, right); - // dy_dx - auto dx_end = var_map.unaryExpr(inv_std_func) - .replicate(1, right) - .cwiseProduct(d_y_map) - .cwiseProduct(scale_map.replicate(left, 1)); - // dy_dmean_dx - auto dx_mean = (T(-1.0) / right) * - var_map.unaryExpr(inv_std_func) - .replicate(1, right) - .cwiseProduct(d_y_map) - .cwiseProduct(scale_map.replicate(left, 1)) - .rowwise() - .sum() - .replicate(1, right); - // dy_var_dx - auto dvar_end_part = (x_map - mean_map.replicate(1, right)) - .cwiseProduct(scale_map.replicate(left, 1)) - .cwiseProduct(d_y_map) - .rowwise() - .sum(); - auto dvar_end = var_map.unaryExpr(inv_std_func) - .unaryExpr(triple_product_func) - .cwiseProduct(dvar_end_part) - .replicate(1, right); - auto dx_var = - (T(-1.0) / right) * - (x_map - mean_map.replicate(1, right)).cwiseProduct(dvar_end); - - d_x_map = dx_end + dx_mean + dx_var; - } else { - // dy_dx - auto dx_end = var_map.unaryExpr(inv_std_func) - .replicate(1, right) - .cwiseProduct(d_y_map); - // dy_dmean_dx - auto dx_mean = (T(-1.0) / right) * - var_map.unaryExpr(inv_std_func) - .replicate(1, right) - .cwiseProduct(d_y_map) - .rowwise() - .sum() - .replicate(1, right); - // dy_var_dx - auto dvar_end_part = (x_map - mean_map.replicate(1, right)) - .cwiseProduct(d_y_map) - .rowwise() - .sum(); - auto dvar_end = var_map.unaryExpr(inv_std_func) - .unaryExpr(triple_product_func) - .cwiseProduct(dvar_end_part) - .replicate(1, right); - auto dx_var = - (T(-1.0) / right) * - (x_map - mean_map.replicate(1, right)).cwiseProduct(dvar_end); - - d_x_map = dx_end + dx_mean + dx_var; - } - } - } -}; - } // namespace operators } // namespace paddle @@ -363,8 +167,9 @@ namespace ops = paddle::operators; REGISTER_OP(layer_norm, ops::LayerNormOp, ops::LayerNormOpMaker, layer_norm_grad, ops::LayerNormGradOp); REGISTER_OP_CPU_KERNEL( - layer_norm, - ops::LayerNormKernel); + layer_norm, ops::LayerNormKernel, + ops::LayerNormKernel); REGISTER_OP_CPU_KERNEL( layer_norm_grad, - ops::LayerNormGradKernel); + ops::LayerNormGradKernel, + ops::LayerNormGradKernel); diff --git a/paddle/operators/layer_norm_op.cu b/paddle/operators/layer_norm_op.cu new file mode 100644 index 0000000000000000000000000000000000000000..77d13b216f0e8d6d4434742908437f1eb74818c9 --- /dev/null +++ b/paddle/operators/layer_norm_op.cu @@ -0,0 +1,25 @@ +/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. */ + +#include "paddle/operators/layer_norm_op.h" + +namespace ops = paddle::operators; +REGISTER_OP_CUDA_KERNEL( + layer_norm, + ops::LayerNormKernel, + ops::LayerNormKernel); +REGISTER_OP_CUDA_KERNEL( + layer_norm_grad, + ops::LayerNormGradKernel, + ops::LayerNormGradKernel); diff --git a/paddle/operators/layer_norm_op.h b/paddle/operators/layer_norm_op.h index bca35b91e6f52d35dee14aac9d080b52914942e3..3c436b89263758bbc0abcd1bb71cef3e1370d2a5 100644 --- a/paddle/operators/layer_norm_op.h +++ b/paddle/operators/layer_norm_op.h @@ -16,19 +16,222 @@ limitations under the License. */ #include "paddle/framework/eigen.h" #include "paddle/framework/op_registry.h" +#include "paddle/operators/elementwise_op_function.h" +#include "paddle/operators/math/math_function.h" + namespace paddle { namespace operators { +template +struct SubAndSquareFunctor { + inline HOSTDEVICE T operator()(T a, T b) const { return (a - b) * (a - b); } +}; + +template +struct DivAndSqrtFunctor { + explicit DivAndSqrtFunctor(T epsilon) { epsilon_ = epsilon; } + inline HOSTDEVICE T operator()(T a, T b) const { + return a / (sqrt(b + epsilon_)); + } + + private: + T epsilon_; +}; + +template +struct MulFunctor { + inline HOSTDEVICE T operator()(T a, T b) const { return a * b; } +}; + +template +struct AddFunctor { + inline HOSTDEVICE T operator()(T a, T b) const { return a + b; } +}; + +template +struct SubFunctor { + inline HOSTDEVICE T operator()(T a, T b) const { return a - b; } +}; + +template +struct MulInvVarFunctor { + inline HOSTDEVICE T operator()(T a, T b) const { + return a * std::sqrt(1.0 / b); + } +}; + +using Tensor = framework::Tensor; +using LoDTensor = framework::LoDTensor; +using DataLayout = framework::DataLayout; + template class LayerNormKernel : public framework::OpKernel { public: - void Compute(const framework::ExecutionContext& ctx) const override; + void Compute(const framework::ExecutionContext &ctx) const override { + const float epsilon = ctx.Attr("epsilon"); + auto *scale = ctx.Input("Scale"); + auto *bias = ctx.Input("Bias"); + auto x = *ctx.Input("X"); + + auto *y = ctx.Output("Y"); + auto *mean = ctx.Output("Mean"); + auto *var = ctx.Output("Variance"); + const auto begin_norm_axis = ctx.Attr("begin_norm_axis"); + + const auto x_dims = x.dims(); + + y->mutable_data(ctx.GetPlace()); + mean->mutable_data(ctx.GetPlace()); + var->mutable_data(ctx.GetPlace()); + + auto matrix_dim = framework::flatten_to_2d(x_dims, begin_norm_axis); + int left = static_cast(matrix_dim[0]); + int right = static_cast(matrix_dim[1]); + framework::DDim matrix_shape({left, right}); + + x.Resize(matrix_shape); + Tensor out; + out.ShareDataWith(*y); + out.Resize(matrix_shape); + + auto &dev_ctx = ctx.template device_context(); + math::RowwiseMean row_mean; + + // get mean + row_mean(dev_ctx, x, mean); + + // get variance + ElementwiseComputeEx, DeviceContext, T>( + ctx, &x, mean, /*axis*/ 0, SubAndSquareFunctor(), &out); + row_mean(dev_ctx, out, var); + + // get x_norm + ElementwiseComputeEx, DeviceContext, T>( + ctx, &x, mean, /*axis*/ 0, SubFunctor(), &out); + ElementwiseComputeEx, DeviceContext, T>( + ctx, &out, var, /*axis*/ 0, + DivAndSqrtFunctor(static_cast(epsilon)), &out); + + if (scale) { + ElementwiseComputeEx, DeviceContext, T>( + ctx, &out, scale, /*axis*/ 1, MulFunctor(), &out); + } + if (bias) { + ElementwiseComputeEx, DeviceContext, T>( + ctx, &out, bias, /*axis*/ 1, AddFunctor(), &out); + } + } }; template class LayerNormGradKernel : public framework::OpKernel { public: - void Compute(const framework::ExecutionContext& ctx) const override; + void Compute(const framework::ExecutionContext &ctx) const override { + const float epsilon = ctx.Attr("epsilon"); + auto x = *ctx.Input("X"); + auto *y = ctx.Input("Y"); + auto *mean = ctx.Input("Mean"); + auto *var = ctx.Input("Variance"); + auto *scale = ctx.Input("Scale"); + auto *bias = ctx.Input("Bias"); + auto d_y = *ctx.Input(framework::GradVarName("Y")); + const auto begin_norm_axis = ctx.Attr("begin_norm_axis"); + + // init output + auto *d_x = ctx.Output(framework::GradVarName("X")); + auto *d_scale = ctx.Output(framework::GradVarName("Scale")); + auto *d_bias = ctx.Output(framework::GradVarName("Bias")); + + const auto &x_dims = x.dims(); + auto matrix_dim = framework::flatten_to_2d(x_dims, begin_norm_axis); + int left = static_cast(matrix_dim[0]); + int right = static_cast(matrix_dim[1]); + framework::DDim matrix_shape({left, right}); + + d_y.Resize(matrix_shape); + auto &dev_ctx = ctx.template device_context(); + math::ColwiseSum colwise_sum; + + Tensor temp; + Tensor temp_norm; + if (d_scale || d_x) { + x.Resize(matrix_shape); + temp.mutable_data(matrix_shape, ctx.GetPlace()); + + if (!(bias && scale)) { + temp_norm.ShareDataWith(*y); + temp_norm.Resize(matrix_shape); + } else { + temp_norm.mutable_data(matrix_shape, ctx.GetPlace()); + // get x_norm + ElementwiseComputeEx, DeviceContext, T>( + ctx, &x, mean, /*axis*/ 0, SubFunctor(), &temp_norm); + ElementwiseComputeEx, DeviceContext, T>( + ctx, &temp_norm, var, /*axis*/ 0, + DivAndSqrtFunctor(static_cast(epsilon)), &temp_norm); + } + } + + if (d_bias) { + d_bias->mutable_data(ctx.GetPlace()); + colwise_sum(dev_ctx, d_y, d_bias); + } + if (d_scale) { + d_scale->mutable_data(ctx.GetPlace()); + ElementwiseComputeEx, DeviceContext, T>( + ctx, &temp_norm, &d_y, /*axis*/ 0, MulFunctor(), &temp); + colwise_sum(dev_ctx, temp, d_scale); + } + + if (d_x) { + framework::DDim vec_shape({left}); + d_x->mutable_data(ctx.GetPlace()); + auto dx_dim = d_x->dims(); + Tensor temp_vec; + temp_vec.mutable_data(vec_shape, ctx.GetPlace()); + + math::RowwiseMean row_mean; + + if (d_scale) { + // dy_dx + ElementwiseComputeEx, DeviceContext, T>( + ctx, &d_y, scale, /*axis*/ 1, MulFunctor(), &temp); + framework::Copy(temp, ctx.GetPlace(), ctx.device_context(), d_x); + + // dy_dmean_dx + row_mean(dev_ctx, temp, &temp_vec); + ElementwiseComputeEx, DeviceContext, T>( + ctx, d_x, &temp_vec, /*axis*/ 0, SubFunctor(), d_x); + + // dy_var_dx + ElementwiseComputeEx, DeviceContext, T>( + ctx, &temp, &temp_norm, /*axis*/ 0, MulFunctor(), &temp); + } else { + // dy_dx + framework::Copy(d_y, ctx.GetPlace(), ctx.device_context(), d_x); + + // dy_dmean_dx + row_mean(dev_ctx, d_y, &temp_vec); + ElementwiseComputeEx, DeviceContext, T>( + ctx, d_x, &temp_vec, /*axis*/ 0, SubFunctor(), d_x); + + // dy_var_dx + ElementwiseComputeEx, DeviceContext, T>( + ctx, &d_y, &temp_norm, /*axis*/ 0, MulFunctor(), &temp); + } + // dy_var_dx + row_mean(dev_ctx, temp, &temp_vec); + ElementwiseComputeEx, DeviceContext, T>( + ctx, &temp_norm, &temp_vec, /*axis*/ 0, MulFunctor(), &temp); + ElementwiseComputeEx, DeviceContext, T>( + ctx, d_x, &temp, /*axis*/ 0, SubFunctor(), d_x); + + ElementwiseComputeEx, DeviceContext, T>( + ctx, d_x, var, /*axis*/ 0, + DivAndSqrtFunctor(static_cast(epsilon)), d_x); + d_x->Resize(dx_dim); + } + } }; } // namespace operators diff --git a/paddle/operators/math/math_function.cc b/paddle/operators/math/math_function.cc index dcf4b85e1aadf88e4b1ca70ac7e8b5416fc58cd8..ce0a5f6cff873166e3308a625978ecefaed2aa29 100644 --- a/paddle/operators/math/math_function.cc +++ b/paddle/operators/math/math_function.cc @@ -331,6 +331,12 @@ template struct RowwiseAdd; template struct ColwiseSum; template struct ColwiseSum; +template struct RowwiseSum; +template struct RowwiseSum; + +template struct RowwiseMean; +template struct RowwiseMean; + } // namespace math } // namespace operators } // namespace paddle diff --git a/paddle/operators/math/math_function.cu b/paddle/operators/math/math_function.cu index d47a7f818ded61baf31e46ea3b8ae3101324111f..c0a107470a4629506fc06dabc78a4a4716be6649 100644 --- a/paddle/operators/math/math_function.cu +++ b/paddle/operators/math/math_function.cu @@ -325,6 +325,31 @@ void ColwiseSum::operator()( vector->data()); } +template struct RowwiseSum; +// template struct RowwiseSum; +// TODO(zcd): Following ColwiseSum format, need to confirm. +// The RowwiseSum failed in debug mode, +// and only failed for this case. So reimplemented it. +template <> +void RowwiseSum::operator()( + const platform::CUDADeviceContext& context, const framework::Tensor& input, + framework::Tensor* vector) { + auto in_dims = input.dims(); + auto size = input.numel() / in_dims[0]; + PADDLE_ENFORCE_EQ(vector->numel(), in_dims[0]); + framework::Tensor one; + one.mutable_data({size}, context.GetPlace()); + SetConstant set; + set(context, &one, static_cast(1.0)); + gemv( + context, true, static_cast(in_dims[1]), static_cast(in_dims[0]), + 1.0, one.data(), input.data(), 0.0, + vector->data()); +} + +template struct RowwiseMean; +template struct RowwiseMean; + } // namespace math } // namespace operators } // namespace paddle diff --git a/paddle/operators/math/math_function.h b/paddle/operators/math/math_function.h index 8cc03c2ba0facae691a0d2b8a4f2ea768cfa5491..cb14d1e57468564710640773fdabd41896c178e0 100644 --- a/paddle/operators/math/math_function.h +++ b/paddle/operators/math/math_function.h @@ -128,6 +128,18 @@ struct ColwiseSum { framework::Tensor* vec); }; +template +struct RowwiseSum { + void operator()(const DeviceContext& context, const framework::Tensor& input, + framework::Tensor* vec); +}; + +template +struct RowwiseMean { + void operator()(const DeviceContext& context, const framework::Tensor& input, + framework::Tensor* vec); +}; + } // namespace math } // namespace operators } // namespace paddle diff --git a/paddle/operators/math/math_function_impl.h b/paddle/operators/math/math_function_impl.h index de591626df28e2bc3391b609f909612411398247..af4127788af0aaeb99199f7d6e2138a449b9fe51 100644 --- a/paddle/operators/math/math_function_impl.h +++ b/paddle/operators/math/math_function_impl.h @@ -87,6 +87,88 @@ class ColwiseSum { } }; +template +void RowwiseMean::operator()(const DeviceContext& context, + const framework::Tensor& input, + framework::Tensor* out) { + auto in_dims = input.dims(); + PADDLE_ENFORCE_EQ(in_dims.size(), 2U); + PADDLE_ENFORCE_EQ(out->numel(), in_dims[0]); + + auto in = framework::EigenMatrix::From(input); + auto vec = framework::EigenVector::Flatten(*out); + + vec.device(*context.eigen_device()) = in.mean(Eigen::array({{1}})); +} +// TODO(zcd): Following ColwiseSum format, need to confirm. +// Specialize for CPU, since Eigen implement a general reduce. However, +// rowwise-sum can be easily implemented. General reduce has a huge overhead in +// CPU +template +class RowwiseMean { + public: + void operator()(const platform::CPUDeviceContext& context, + const framework::Tensor& input, framework::Tensor* out) { + auto& in_dims = input.dims(); + PADDLE_ENFORCE_EQ(in_dims.size(), 2U); + auto height = in_dims[0]; + auto size = in_dims[1]; + PADDLE_ENFORCE_EQ(out->numel(), height); + auto inv_size = 1.0 / size; + T* out_buf = out->mutable_data(out->place()); + const T* in_buf = input.data(); + + for (size_t i = 0; i < static_cast(height); ++i) { + T sum = 0; + for (size_t j = 0; j < static_cast(size); ++j) { + sum += in_buf[i * size + j]; + } + out_buf[i] = sum * inv_size; + } + } +}; + +template +void RowwiseSum::operator()(const DeviceContext& context, + const framework::Tensor& input, + framework::Tensor* out) { + auto in_dims = input.dims(); + PADDLE_ENFORCE_EQ(in_dims.size(), 2U); + PADDLE_ENFORCE_EQ(out->numel(), in_dims[0]); + + auto in = framework::EigenMatrix::From(input); + auto vec = framework::EigenVector::Flatten(*out); + + vec.device(*context.eigen_device()) = in.sum(Eigen::array({{1}})); +} +// TODO(zcd): Following ColwiseSum format, need to confirm. +// Specialize for CPU, since Eigen implement a general reduce. However, +// rowwise-sum can be easily implemented. General reduce has a huge overhead in +// CPU +template +class RowwiseSum { + public: + void operator()(const platform::CPUDeviceContext& context, + const framework::Tensor& input, framework::Tensor* out) { + auto& in_dims = input.dims(); + PADDLE_ENFORCE_EQ(in_dims.size(), 2U); + auto height = in_dims[0]; + auto size = in_dims[1]; + PADDLE_ENFORCE_EQ(out->numel(), size); + + T* out_buf = out->mutable_data(out->place()); + const T* in_buf = input.data(); + + for (size_t i = 0; i < static_cast(height); ++i) { + T sum = 0; + for (size_t j = 0; j < static_cast(size); ++j) { + sum += in_buf[i * size + j]; + } + out_buf[i] = sum; + } + } +}; + } // namespace math } // namespace operators } // namespace paddle diff --git a/python/paddle/v2/fluid/distribute_transpiler.py b/python/paddle/v2/fluid/distribute_transpiler.py index 121b407cae41fa477843b7252ebacc9053d5f7aa..c5f1d51bd718acf32d173b97ee7bb7cdeb443c63 100644 --- a/python/paddle/v2/fluid/distribute_transpiler.py +++ b/python/paddle/v2/fluid/distribute_transpiler.py @@ -300,6 +300,9 @@ class DistributeTranspiler: pass return orig_shape + def _op_input_var(self, op, varname): + pass + def _is_op_on_pserver(self, endpoint, all_ops, idx): """ Recursively check if the op need to run on current server. @@ -309,29 +312,35 @@ class DistributeTranspiler: p.name for p in self.param_grad_ep_mapping[endpoint]["params"] ] op = all_ops[idx] - if op.inputs.has_key("Param"): - if op.inputs["Param"].name in param_names: + input_names = set(op.input_names) + # TODO(typhoonzero): using Param and Grad input name to identify + # that the operator is an optimization operator, need a better way. + if "Param" in input_names: + if op.input("Param")[0] in param_names: return True else: for n in param_names: - if same_or_split_var(n, op.inputs[ - "Param"].name) and n != op.inputs["Param"].name: + if same_or_split_var(n, op.input("Param")[0]) \ + and n != op.input("Param")[0]: return True return False else: j = idx - 1 while j >= 0: prev_op = all_ops[j] - prev_output_names = [o.name for o in prev_op.outputs.values()] - prev_input_names = [o.name for o in prev_op.inputs.values()] + # prev_output_names = [o.name for o in prev_op.outputs.values()] + # prev_input_names = [o.name for o in prev_op.inputs.values()] + # NOTE(typhoonzero): consider list input/output + prev_output_names = prev_op.desc.output_arg_names() + prev_input_names = prev_op.desc.input_arg_names() found1 = False found2 = False - for _, v in op.inputs.iteritems(): - if v.name in prev_output_names: + for varname in op.desc.input_arg_names(): + if varname in prev_output_names: found1 = self._is_op_on_pserver(endpoint, all_ops, j) # later ops may produce output for prev op's next batch use. - for _, v in op.outputs.iteritems(): - if v.name in prev_input_names: + for varname in op.desc.output_arg_names(): + if varname in prev_input_names: found2 = self._is_op_on_pserver(endpoint, all_ops, j) if found1 or found2: return True @@ -342,11 +351,11 @@ class DistributeTranspiler: new_inputs = dict() # update param/grad shape first, then other inputs like # moment can use the updated shape - for key, var in opt_op.inputs.iteritems(): + for key in opt_op.input_names: if key == "Grad": grad_block = None for g in self.param_grad_ep_mapping[endpoint]["grads"]: - if same_or_split_var(g.name, var.name): + if same_or_split_var(g.name, opt_op.input(key)[0]): grad_block = g break if not grad_block: @@ -376,7 +385,7 @@ class DistributeTranspiler: # param is already created on global program param_block = None for p in self.param_grad_ep_mapping[endpoint]["params"]: - if same_or_split_var(p.name, var.name): + if same_or_split_var(p.name, opt_op.input(key)[0]): param_block = p break if not param_block: @@ -389,11 +398,12 @@ class DistributeTranspiler: new_inputs[key] = tmpvar - for key, var in opt_op.inputs.iteritems(): + for key in opt_op.input_names: if key in ["Param", "Grad"]: continue # update accumulator variable shape param_shape = new_inputs["Param"].shape + var = program.global_block().vars[opt_op.input(key)[0]] new_shape = self._get_optimizer_input_shape(opt_op.type, key, var.shape, param_shape) tmpvar = program.global_block().create_var( @@ -412,30 +422,44 @@ class DistributeTranspiler: shape=new_shape) # change output's ParamOut variable - opt_op.outputs["ParamOut"] = new_inputs["Param"] + outputs = self._get_output_map_from_op(program.global_block(), opt_op) + outputs["ParamOut"] = new_inputs["Param"] program.global_block().append_op( type=opt_op.type, inputs=new_inputs, - outputs=opt_op.outputs, + outputs=outputs, attrs=opt_op.attrs) def _append_pserver_non_opt_ops(self, program, pserver_program, opt_op): # Append the ops for parameters that do not need to be optimized/updated - for _, var in opt_op.inputs.iteritems(): - program.global_block().create_var( - name=var.name, - persistable=var.persistable, - dtype=var.dtype, - shape=var.shape) - pserver_program.global_block().create_var( - name=var.name, - persistable=var.persistable, - dtype=var.dtype, - shape=var.shape) + inputs = self._get_input_map_from_op(self.program.global_block().vars, + opt_op) + for var in inputs.itervalues(): + if type(var) == list: + varlist = var + else: + varlist = [var] + for var in varlist: + # TODO(typhoonzero): will remove below line later. + program.global_block().create_var( + name=var.name, + persistable=var.persistable, + dtype=var.dtype, + shape=var.shape) + if not pserver_program.global_block().vars.has_key(var.name): + pserver_program.global_block().create_var( + name=var.name, + persistable=var.persistable, + dtype=var.dtype, + shape=var.shape) + + outputs = self._get_output_map_from_op(self.program.global_block().vars, + opt_op) + program.global_block().append_op( type=opt_op.type, - inputs=opt_op.inputs, - outputs=opt_op.outputs, + inputs=inputs, + outputs=outputs, attrs=opt_op.attrs) def get_pserver_program(self, endpoint): @@ -472,7 +496,7 @@ class DistributeTranspiler: self.optimize_ops, idx) if not is_op_on_pserver: continue - if opt_op.inputs.has_key("Grad"): + if "Grad" in opt_op.desc.input_arg_names(): self._append_pserver_ops(optimize_sub_program, pserver_program, opt_op, endpoint) else: @@ -499,6 +523,30 @@ class DistributeTranspiler: pserver_program.sync_with_cpp() return pserver_program + def _get_input_map_from_op(self, varmap, op): + iomap = dict() + for key in op.input_names: + vars = [] + for varname in op.input(key): + vars.append(varmap[varname]) + if len(vars) == 1: + iomap[key] = vars[0] + else: + iomap[key] = vars + return iomap + + def _get_output_map_from_op(self, varmap, op): + iomap = dict() + for key in op.output_names: + vars = [] + for varname in op.output(key): + vars.append(varmap[varname]) + if len(vars) == 1: + iomap[key] = vars[0] + else: + iomap[key] = vars + return iomap + def get_startup_program(self, endpoint, pserver_program): """ Get startup program for current parameter server. @@ -529,17 +577,21 @@ class DistributeTranspiler: # 2. rename op outputs for op in orig_s_prog.global_block().ops: + new_inputs = dict() new_outputs = dict() # do not append startup op if var is not on this pserver op_on_pserver = False - for key, var in op.outputs.iteritems(): - newname, _ = _get_splited_name_and_shape(var.name) + for key in op.output_names: + newname, _ = _get_splited_name_and_shape(op.output(key)[0]) if newname: op_on_pserver = True new_outputs[key] = created_var_map[newname] - elif var.name in pserver_vars: + elif op.output(key)[0] in pserver_vars: op_on_pserver = True - new_outputs[key] = pserver_vars[var.name] + new_outputs[key] = pserver_vars[op.output(key)[0]] + + # most startup program ops have no inputs + new_inputs = self._get_input_map_from_op(pserver_vars, op) if op_on_pserver: if op.type in [ @@ -548,7 +600,7 @@ class DistributeTranspiler: op.attrs["shape"] = new_outputs["Out"].shape s_prog.global_block().append_op( type=op.type, - inputs=op.inputs, + inputs=new_inputs, outputs=new_outputs, attrs=op.attrs) return s_prog diff --git a/python/paddle/v2/fluid/framework.py b/python/paddle/v2/fluid/framework.py index a12427258e9d3142abcb84249a10dabd8e96b792..a517db68c5886fbcbe19e6981aee5bf3971352e4 100644 --- a/python/paddle/v2/fluid/framework.py +++ b/python/paddle/v2/fluid/framework.py @@ -740,6 +740,9 @@ class Block(object): raise e self.desc.remove_op(start, end + 1) + def slice_ops(self, start, end): + return list(self.ops)[start:end] + def prepend_op(self, *args, **kwargs): op_desc = self.desc.prepend_op() op = Operator(self, op_desc, *args, **kwargs) diff --git a/python/paddle/v2/fluid/io.py b/python/paddle/v2/fluid/io.py index 613dc20b6ea5533d126a73b7ec47796b3f812db5..0f43e46082a8988be4805a2b750227312ba80ff3 100644 --- a/python/paddle/v2/fluid/io.py +++ b/python/paddle/v2/fluid/io.py @@ -342,7 +342,11 @@ def save_inference_model(dirname, prepend_feed_ops(inference_program, feeded_var_names) append_fetch_ops(inference_program, fetch_var_names) - model_file_name = dirname + "/__model__" + if save_file_name == None: + model_file_name = dirname + "/__model__" + else: + model_file_name = dirname + "/__model_combined__" + with open(model_file_name, "wb") as f: f.write(inference_program.desc.serialize_to_string()) @@ -384,7 +388,11 @@ def load_inference_model(dirname, executor, load_file_name=None): if not os.path.isdir(dirname): raise ValueError("There is no directory named '%s'", dirname) - model_file_name = dirname + "/__model__" + if load_file_name == None: + model_file_name = dirname + "/__model__" + else: + model_file_name = dirname + "/__model_combined__" + with open(model_file_name, "rb") as f: program_desc_str = f.read() diff --git a/python/paddle/v2/fluid/memory_optimization_transpiler.py b/python/paddle/v2/fluid/memory_optimization_transpiler.py index 8bb8cf7b1a5ddf44427637229bdc31ac0e151e44..53e0991ee8c318e0c95018b57ad48f404ce8beae 100644 --- a/python/paddle/v2/fluid/memory_optimization_transpiler.py +++ b/python/paddle/v2/fluid/memory_optimization_transpiler.py @@ -92,14 +92,13 @@ class ControlFlowGraph(object): live_in = defaultdict(set) live_out = defaultdict(set) while True: - for i in range(self.op_size): + for i in range(self.op_size, 0, -1): live_in[i] = set(self._live_in[i]) live_out[i] = set(self._live_out[i]) - self._live_in[i] = self._uses[i] | ( - self._live_out[i] - self._defs[i]) for s in self._successors[i]: self._live_out[i] |= self._live_in[s] - + self._live_in[i] = self._uses[i] | ( + self._live_out[i] - self._defs[i]) if self._reach_fixed_point(live_in, live_out): break diff --git a/python/paddle/v2/fluid/optimizer.py b/python/paddle/v2/fluid/optimizer.py index 7844a4e2df1ce3989e48082f6472292560fbf1ee..f8a00e3a5fb4038a97a951a01c3a2f1a4488ae75 100644 --- a/python/paddle/v2/fluid/optimizer.py +++ b/python/paddle/v2/fluid/optimizer.py @@ -190,6 +190,8 @@ class Optimizer(object): # Create any accumulators program = loss.block.program with program_guard(program, startup_program): + global_block = framework.default_main_program().global_block() + start = len(global_block.ops) self.helper = LayerHelper(self.__class__.__name__) self._create_accumulators(loss.block, [p[0] for p in parameters_and_grads]) @@ -203,19 +205,14 @@ class Optimizer(object): param_and_grad) optimize_ops.append(optimize_op) - # Returned list of ops can include more ops in addition - # to optimization ops - return_ops = optimize_ops - # Get custom finish ops for subclasses # FIXME: Need to fix this once we figure out how to handle dependencies - finish_ops = self._finish_update(loss.block) - if finish_ops is not None: - return_ops += finish_ops + self._finish_update(loss.block) if self._global_step is not None: - return_ops.append(self._increment_global_step(loss.block)) - return return_ops + self._increment_global_step(loss.block) + end = len(global_block.ops) + return global_block.slice_ops(start, end) def minimize(self, loss, diff --git a/python/paddle/v2/fluid/tests/book/test_recognize_digits.py b/python/paddle/v2/fluid/tests/book/test_recognize_digits.py index d8f0ad89cd89215ac83a133bd27a53c4b904363f..6f9d85faff924c10d5179545e67f0d80ff837ce5 100644 --- a/python/paddle/v2/fluid/tests/book/test_recognize_digits.py +++ b/python/paddle/v2/fluid/tests/book/test_recognize_digits.py @@ -78,7 +78,7 @@ def conv_net(img, label): return loss_net(conv_pool_2, label) -def train(nn_type, use_cuda, parallel, save_dirname): +def train(nn_type, use_cuda, parallel, save_dirname, save_param_filename): if use_cuda and not fluid.core.is_compiled_with_cuda(): return img = fluid.layers.data(name='img', shape=[1, 28, 28], dtype='float32') @@ -143,8 +143,10 @@ def train(nn_type, use_cuda, parallel, save_dirname): avg_loss_val = numpy.array(avg_loss_set).mean() if float(acc_val) > 0.85: # test acc > 85% if save_dirname is not None: - fluid.io.save_inference_model(save_dirname, ["img"], - [prediction], exe) + fluid.io.save_inference_model( + save_dirname, ["img"], [prediction], + exe, + save_file_name=save_param_filename) return else: print( @@ -156,7 +158,7 @@ def train(nn_type, use_cuda, parallel, save_dirname): raise AssertionError("Loss of recognize digits is too large") -def infer(use_cuda, save_dirname=None): +def infer(use_cuda, save_dirname=None, param_filename=None): if save_dirname is None: return @@ -167,8 +169,8 @@ def infer(use_cuda, save_dirname=None): # the feed_target_names (the names of variables that will be feeded # data using feed operators), and the fetch_targets (variables that # we want to obtain data from using fetch operators). - [inference_program, feed_target_names, - fetch_targets] = fluid.io.load_inference_model(save_dirname, exe) + [inference_program, feed_target_names, fetch_targets + ] = fluid.io.load_inference_model(save_dirname, exe, param_filename) # The input's dimension of conv should be 4-D or 5-D. # Use normilized image pixels as input data, which should be in the range [-1.0, 1.0]. @@ -183,36 +185,45 @@ def infer(use_cuda, save_dirname=None): print("infer results: ", results[0]) -def main(use_cuda, parallel, nn_type): +def main(use_cuda, parallel, nn_type, combine): if not use_cuda and not parallel: save_dirname = "recognize_digits_" + nn_type + ".inference.model" + save_filename = None + if combine == True: + save_filename = "__params_combined__" else: save_dirname = None + save_filename = None train( nn_type=nn_type, use_cuda=use_cuda, parallel=parallel, - save_dirname=save_dirname) - infer(use_cuda=use_cuda, save_dirname=save_dirname) + save_dirname=save_dirname, + save_param_filename=save_filename) + infer( + use_cuda=use_cuda, + save_dirname=save_dirname, + param_filename=save_filename) class TestRecognizeDigits(unittest.TestCase): pass -def inject_test_method(use_cuda, parallel, nn_type): +def inject_test_method(use_cuda, parallel, nn_type, combine): def __impl__(self): prog = fluid.Program() startup_prog = fluid.Program() scope = fluid.core.Scope() with fluid.scope_guard(scope): with fluid.program_guard(prog, startup_prog): - main(use_cuda, parallel, nn_type) + main(use_cuda, parallel, nn_type, combine) - fn = 'test_{0}_{1}_{2}'.format(nn_type, 'cuda' - if use_cuda else 'cpu', 'parallel' - if parallel else 'normal') + fn = 'test_{0}_{1}_{2}_{3}'.format(nn_type, 'cuda' + if use_cuda else 'cpu', 'parallel' + if parallel else 'normal', 'combine' + if combine else 'separate') setattr(TestRecognizeDigits, fn, __impl__) @@ -221,7 +232,10 @@ def inject_all_tests(): for use_cuda in (False, True): for parallel in (False, True): for nn_type in ('mlp', 'conv'): - inject_test_method(use_cuda, parallel, nn_type) + inject_test_method(use_cuda, parallel, nn_type, True) + + # One unit-test for saving parameters as separate files + inject_test_method(False, False, 'mlp', False) inject_all_tests() diff --git a/python/paddle/v2/fluid/tests/book/test_rnn_encoder_decoder.py b/python/paddle/v2/fluid/tests/book/test_rnn_encoder_decoder.py index fdc60861760163d2ebad3b050e551929321baafd..7fe43c680ca9319682c42836986308856185a464 100644 --- a/python/paddle/v2/fluid/tests/book/test_rnn_encoder_decoder.py +++ b/python/paddle/v2/fluid/tests/book/test_rnn_encoder_decoder.py @@ -18,6 +18,10 @@ import paddle.v2.fluid as fluid import paddle.v2.fluid.core as core import paddle.v2.fluid.framework as framework import paddle.v2.fluid.layers as layers +import contextlib +import math +import sys +import unittest from paddle.v2.fluid.executor import Executor dict_size = 30000 @@ -145,7 +149,7 @@ def seq_to_seq_net(): cost = fluid.layers.cross_entropy(input=prediction, label=label) avg_cost = fluid.layers.mean(x=cost) - return avg_cost + return avg_cost, prediction def to_lodtensor(data, place): @@ -163,8 +167,16 @@ def to_lodtensor(data, place): return res -def main(): - avg_cost = seq_to_seq_net() +def create_random_lodtensor(lod, place, low, high): + data = np.random.random_integers(low, high, [lod[-1], 1]).astype("int64") + res = fluid.LoDTensor() + res.set(data, place) + res.set_lod([lod]) + return res + + +def train(use_cuda, save_dirname=None): + [avg_cost, prediction] = seq_to_seq_net() optimizer = fluid.optimizer.Adagrad(learning_rate=1e-4) optimizer.minimize(avg_cost) @@ -174,7 +186,7 @@ def main(): paddle.dataset.wmt14.train(dict_size), buf_size=1000), batch_size=batch_size) - place = core.CPUPlace() + place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() exe = Executor(place) exe.run(framework.default_startup_program()) @@ -185,6 +197,7 @@ def main(): word_data = to_lodtensor(map(lambda x: x[0], data), place) trg_word = to_lodtensor(map(lambda x: x[1], data), place) trg_word_next = to_lodtensor(map(lambda x: x[2], data), place) + outs = exe.run(framework.default_main_program(), feed={ 'source_sequence': word_data, @@ -192,13 +205,86 @@ def main(): 'label_sequence': trg_word_next }, fetch_list=[avg_cost]) + avg_cost_val = np.array(outs[0]) print('pass_id=' + str(pass_id) + ' batch=' + str(batch_id) + " avg_cost=" + str(avg_cost_val)) + if math.isnan(float(avg_cost_val[0])): + sys.exit("got NaN loss, training failed.") if batch_id > 3: - exit(0) + if save_dirname is not None: + fluid.io.save_inference_model( + save_dirname, ['source_sequence', + 'target_sequence'], [prediction], exe) + return + batch_id += 1 +def infer(use_cuda, save_dirname=None): + if save_dirname is None: + return + + place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() + exe = fluid.Executor(place) + + # Use fluid.io.load_inference_model to obtain the inference program desc, + # the feed_target_names (the names of variables that will be feeded + # data using feed operators), and the fetch_targets (variables that + # we want to obtain data from using fetch operators). + [inference_program, feed_target_names, + fetch_targets] = fluid.io.load_inference_model(save_dirname, exe) + + lod = [0, 4, 10] + word_data = create_random_lodtensor(lod, place, low=0, high=1) + trg_word = create_random_lodtensor(lod, place, low=0, high=1) + + # Construct feed as a dictionary of {feed_target_name: feed_target_data} + # and results will contain a list of data corresponding to fetch_targets. + assert feed_target_names[0] == 'source_sequence' + assert feed_target_names[1] == 'target_sequence' + results = exe.run(inference_program, + feed={ + feed_target_names[0]: word_data, + feed_target_names[1]: trg_word, + }, + fetch_list=fetch_targets, + return_numpy=False) + print(results[0].lod()) + np_data = np.array(results[0]) + print("Inference shape: ", np_data.shape) + print("Inference results: ", np_data) + + +def main(use_cuda): + if use_cuda and not fluid.core.is_compiled_with_cuda(): + return + + # Directory for saving the trained model + save_dirname = "rnn_encoder_decoder.inference.model" + + train(use_cuda, save_dirname) + infer(use_cuda, save_dirname) + + +class TestRnnEncoderDecoder(unittest.TestCase): + def test_cuda(self): + with self.scope_prog_guard(): + main(use_cuda=True) + + def test_cpu(self): + with self.scope_prog_guard(): + main(use_cuda=False) + + @contextlib.contextmanager + def scope_prog_guard(self): + prog = fluid.Program() + startup_prog = fluid.Program() + scope = fluid.core.Scope() + with fluid.scope_guard(scope): + with fluid.program_guard(prog, startup_prog): + yield + + if __name__ == '__main__': - main() + unittest.main() diff --git a/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_fit_a_line.py b/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_fit_a_line.py index 7ad5e2c594f24999e298533b6c05ba688a935f0b..045db8390cd52689a2a803c3387c90776a44ee73 100644 --- a/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_fit_a_line.py +++ b/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_fit_a_line.py @@ -15,6 +15,8 @@ import numpy as np import paddle.v2 as paddle import paddle.v2.fluid as fluid +import math +import sys # need to fix random seed and training data to compare the loss # value accurately calculated by the default and the memory optimization @@ -63,4 +65,6 @@ for pass_id in range(PASS_NUM): if avg_loss_value[0] < 10.0: exit(0) # if avg cost less than 10.0, we think our code is good. + if math.isnan(float(avg_loss_value)): + sys.exit("got NaN loss, training failed.") exit(1) diff --git a/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_image_classification_train.py b/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_image_classification_train.py index 26673afd83c48328c3f354e82bfa3725aa4805b5..9fbb36d3638bd537020247d6f762afd4ed5d402f 100644 --- a/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_image_classification_train.py +++ b/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_image_classification_train.py @@ -18,6 +18,8 @@ import sys import paddle.v2 as paddle import paddle.v2.fluid as fluid +import math +import sys # need to fix random seed and training data to compare the loss # value accurately calculated by the default and the memory optimization @@ -152,7 +154,10 @@ for pass_id in range(PASS_NUM): print("loss:" + str(loss) + " acc:" + str(acc) + " pass_acc:" + str( pass_acc)) # this model is slow, so if we can train two mini batch, we think it works properly. + if i > 2: exit(0) + if math.isnan(float(loss)): + sys.exit("got NaN loss, training failed.") i += 1 exit(1) diff --git a/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_machine_translation.py b/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_machine_translation.py index ffd53e7a78142162317a677de49c1821635a65b5..48abaa8d87563b7132c5d8962bc33283a104e67a 100644 --- a/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_machine_translation.py +++ b/python/paddle/v2/fluid/tests/book_memory_optimization/test_memopt_machine_translation.py @@ -19,6 +19,8 @@ import paddle.v2.fluid.core as core import paddle.v2.fluid.framework as framework import paddle.v2.fluid.layers as layers from paddle.v2.fluid.executor import Executor +import math +import sys dict_size = 30000 source_dict_dim = target_dict_dim = dict_size @@ -137,6 +139,8 @@ def main(): " avg_cost=" + str(avg_cost_val)) if batch_id > 2: exit(0) + if math.isnan(float(avg_cost_val)): + sys.exit("got NaN loss, training failed.") batch_id += 1 diff --git a/python/paddle/v2/fluid/tests/test_layer_norm_op.py b/python/paddle/v2/fluid/tests/test_layer_norm_op.py index 68cf8673cd46677065588f652482cd0df08b3450..4460ffaf9c46966178497419a35ef4044464ac9f 100644 --- a/python/paddle/v2/fluid/tests/test_layer_norm_op.py +++ b/python/paddle/v2/fluid/tests/test_layer_norm_op.py @@ -20,6 +20,8 @@ import paddle.v2.fluid.core as core from paddle.v2.fluid.op import Operator from paddle.v2.fluid.framework import grad_var_name +np.random.random(123) + def _reference_layer_norm_naive(x, scale, beta, epsilon, begin_norm_axis=1): x_shape = x.shape @@ -62,9 +64,9 @@ def _reference_layer_norm_grad(x, grad_y, scale, mean, var, begin_norm_axis=1): grad_x = dx_end + d_mean + d_std - grad_y.shape = x_shape - x.shape = x_shape + grad_x.shape, x.shape, grad_y.shape = x_shape, x_shape, x_shape scale.shape = scale_shape + var.shape, mean.shape = [N, ], [N, ] return grad_x, d_scale, d_bias @@ -112,10 +114,7 @@ def set_output_grad(scope, outputs, place, feed_dict=None): class TestLayerNormdOp(OpTest): def __assert_close(self, tensor, np_array, msg, atol=1e-4): - self.assertTrue( - np.allclose( - np.array(tensor).reshape(np_array.shape), np_array, atol=atol), - msg) + self.assertTrue(np.allclose(np.array(tensor), np_array, atol=atol), msg) def __assert_grad_close(self, tensor, @@ -123,7 +122,7 @@ class TestLayerNormdOp(OpTest): name, place, max_relative_error=0.02): - a = np.array(tensor).reshape(np_array.shape) + a = np.array(tensor) b = np_array abs_a = np.abs(a) abs_a[abs_a < 1e-5] = 1 @@ -151,7 +150,7 @@ class TestLayerNormdOp(OpTest): x_shape = shape D = reduce(mul, x_shape[begin_norm_axis:len(x_shape)], 1) scale_shape = [D] - np.random.random(123) + x_val = np.random.random_sample(x_shape).astype(np.float32) scale_val = np.random.random_sample(scale_shape).astype(np.float32) bias_val = np.random.random_sample(scale_shape).astype(np.float32) diff --git a/python/paddle/v2/fluid/tests/test_optimizer.py b/python/paddle/v2/fluid/tests/test_optimizer.py index 480ee7091579ba171ca957cb4d25f0034e0534c0..dc6b84dcdc04dd185d97c3cc4b9f00305a911efb 100644 --- a/python/paddle/v2/fluid/tests/test_optimizer.py +++ b/python/paddle/v2/fluid/tests/test_optimizer.py @@ -42,9 +42,9 @@ class TestOptimizer(unittest.TestCase): type="mean", inputs={"X": mul_out}, outputs={"Out": mean_out}) sgd_optimizer = optimizer.SGDOptimizer(learning_rate=0.01) opts, _ = sgd_optimizer.minimize(mean_out, init_program) - self.assertEqual(len(opts), 1) - sgd_op = opts[0] - self.assertEqual(sgd_op.type, "sgd") + self.assertEqual(len(opts), 3) + self.assertEqual([op.type for op in opts], + ["fill_constant", "elementwise_mul", "sgd"]) def test_sgd_optimizer_with_global_step(self): init_program = framework.Program() @@ -72,11 +72,10 @@ class TestOptimizer(unittest.TestCase): sgd_optimizer = optimizer.SGDOptimizer( learning_rate=learning_rate, global_step=global_step) opts, _ = sgd_optimizer.minimize(mean_out, init_program) - self.assertEqual(len(opts), 2) - sgd_op = opts[0] - self.assertEqual(sgd_op.type, "sgd") - increment_op = opts[1] - self.assertEqual(increment_op.type, "increment") + self.assertEqual(len(opts), 4) + self.assertEqual( + [op.type for op in opts], + ["fill_constant", "elementwise_mul", "sgd", "increment"]) # Check init_program init_ops = init_program.global_block().ops @@ -121,9 +120,10 @@ class TestMomentumOptimizer(unittest.TestCase): self.assertEqual(len(momentum_optimizer.get_accumulators()), 0) opts = momentum_optimizer.create_optimization_pass( params_grads, mul_out, init_program) - self.assertEqual(len(opts), 1) - sgd_op = opts[0] - self.assertEqual(sgd_op.type, "momentum") + self.assertEqual(len(opts), 3) + sgd_op = opts[-1] + self.assertEqual([op.type for op in opts], + ["fill_constant", "elementwise_mul", "momentum"]) self.assertFalse(sgd_op.attr('use_nesterov')) # Check accumulators @@ -170,9 +170,10 @@ class TestMomentumOptimizer(unittest.TestCase): self.assertEqual(len(momentum_optimizer.get_accumulators()), 0) opts = momentum_optimizer.create_optimization_pass( params_grads, mul_out, init_program) - self.assertEqual(len(opts), 1) - sgd_op = opts[0] - self.assertEqual(sgd_op.type, "momentum") + self.assertEqual(len(opts), 3) + sgd_op = opts[-1] + self.assertEqual([op.type for op in opts], + ["fill_constant", "elementwise_mul", "momentum"]) self.assertTrue(sgd_op.attr('use_nesterov')) # Check accumulators @@ -228,9 +229,9 @@ class TestAdagradOptimizer(unittest.TestCase): self.assertEqual(len(adagrad_optimizer.get_accumulators()), 0) opts = adagrad_optimizer.create_optimization_pass(params_grads, mul_out, init_program) - self.assertEqual(len(opts), 1) - adagrad_op = opts[0] - self.assertEqual(adagrad_op.type, "adagrad") + self.assertEqual(len(opts), 3) + self.assertEqual([op.type for op in opts], + ["fill_constant", "elementwise_mul", "adagrad"]) # Check accumulators accumulators = adagrad_optimizer.get_accumulators() @@ -288,9 +289,10 @@ class TestAdamOptimizer(unittest.TestCase): self.assertEqual(len(adam_optimizer.get_accumulators()), 0) opts = adam_optimizer.create_optimization_pass(params_grads, mul_out, init_program) - self.assertEqual(len(opts), 3) - adam_op = opts[0] - self.assertEqual(adam_op.type, "adam") + self.assertEqual(len(opts), 5) + self.assertEqual( + [op.type for op in opts], + ["fill_constant", "elementwise_mul", "adam", "scale", "scale"]) # Check accumulators accumulators = adam_optimizer.get_accumulators() @@ -350,9 +352,10 @@ class TestAdamaxOptimizer(unittest.TestCase): self.assertEqual(len(adamax_optimizer.get_accumulators()), 0) opts = adamax_optimizer.create_optimization_pass(params_grads, mul_out, init_program) - self.assertEqual(len(opts), 2) - adam_op = opts[0] - self.assertEqual(adam_op.type, "adamax") + self.assertEqual(len(opts), 4) + self.assertEqual( + [op.type for op in opts], + ["fill_constant", "elementwise_mul", "adamax", "scale"]) # Check accumulators accumulators = adamax_optimizer.get_accumulators() @@ -409,9 +412,10 @@ class TestDecayedAdagradOptimizer(unittest.TestCase): self.assertEqual(len(decayed_adagrad_optimizer.get_accumulators()), 0) opts = decayed_adagrad_optimizer.create_optimization_pass( params_grads, mul_out, init_program) - self.assertEqual(len(opts), 1) - decayed_adagrad_op = opts[0] - self.assertEqual(decayed_adagrad_op.type, "decayed_adagrad") + self.assertEqual(len(opts), 3) + self.assertEqual( + [op.type for op in opts], + ["fill_constant", "elementwise_mul", "decayed_adagrad"]) # Check accumulators accumulators = decayed_adagrad_optimizer.get_accumulators()