Morph MORPH
SIGN IN SIGN UP
PaddlePaddle / Paddle UNCLAIMED

PArallel Distributed Deep LEarning: Machine Learning Framework from Industrial Practice （『飞桨』核心框架，深度学习&机器学习高性能单机、分布式训练和跨平台部署）

0 0 22 C++
Normal View History Raw
Add reader blocking queue
2018-04-25 19:32:08 +08:00
// 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.
// 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 <thread> // NOLINT
use iwyu clean include (#27267) * use iwyu clean include, test=develop, test=win * compilation error, test=develop * fix compilation error2, test=develop * fix compilation error3, test=develop * fix compilation error4, test=develop * fix compilation error5, test=develop * fix compilation error6, test=develop * fix compilation error7, test=develop * fix compilation error8, test=develop * fix compilation error8, test=develop * fix compilation error10, test=develop * fix compilation error11, test=develop
2020-09-24 13:27:13 +08:00
#include "gtest/gtest.h"
Migrate reader/buffered_reader.cc to PHI [fluid_ops] (#68786) * Fix * Fix
2024-10-22 10:49:29 +08:00
#include "paddle/phi/core/operators/reader/blocking_queue.h"
Add reader blocking queue
2018-04-25 19:32:08 +08:00
using paddle::operators::reader::BlockingQueue;
TEST(BlockingQueue, CapacityTest) {
size_t cap = 10;
BlockingQueue<int> q(cap);
EXPECT_EQ(q.Cap(), cap);
}
format all files in fluid using new config (#43776)
2022-06-26 20:04:39 +08:00
void FirstInFirstOut(size_t queue_cap,
size_t elem_num,
size_t send_time_gap,
Add reader blocking queue
2018-04-25 19:32:08 +08:00
size_t receive_time_gap) {
BlockingQueue<size_t> q(queue_cap);
std::thread sender([&]() {
for (size_t i = 0; i < elem_num; ++i) {
std::this_thread::sleep_for(std::chrono::milliseconds(send_time_gap));
EXPECT_TRUE(q.Send(i));
}
q.Close();
});
size_t count = 0;
while (true) {
std::this_thread::sleep_for(std::chrono::milliseconds(receive_time_gap));
[clang-tidy] NO.10 enable `cppcoreguidelines-init-variables` (#57871) * [clang-tidy] enable cppcoreguidelines-init-variables check * fix
2023-10-11 11:53:12 +08:00
size_t elem = 0;
Add reader blocking queue
2018-04-25 19:32:08 +08:00
if (!q.Receive(&elem)) {
break;
}
EXPECT_EQ(elem, count++);
}
sender.join();
EXPECT_EQ(count, elem_num);
EXPECT_TRUE(q.IsClosed());
}
TEST(BlockingQueue, FirstInFirstOutTest) {
FirstInFirstOut(2, 5, 2, 50);
FirstInFirstOut(2, 5, 50, 2);
FirstInFirstOut(10, 3, 50, 2);
FirstInFirstOut(10, 3, 2, 50);
}
TEST(BlockingQueue, SenderBlockingTest) {
const size_t queue_cap = 2;
BlockingQueue<size_t> q(queue_cap);
size_t send_count = 0;
std::thread sender([&]() {
for (size_t i = 0; i < 5; ++i) {
if (!q.Send(i)) {
break;
}
++send_count;
}
});
fix reader_blocking_queue_test (#32505)
2021-04-25 10:47:57 +08:00
std::this_thread::sleep_for(std::chrono::milliseconds(1500));
Add reader blocking queue
2018-04-25 19:32:08 +08:00
q.Close();
sender.join();
EXPECT_EQ(send_count, queue_cap);
std::vector<size_t> res;
fix unit test error
2018-04-26 11:26:25 +08:00
while (true) {
[clang-tidy] NO.10 enable `cppcoreguidelines-init-variables` (#57871) * [clang-tidy] enable cppcoreguidelines-init-variables check * fix
2023-10-11 11:53:12 +08:00
size_t elem = 0;
fix unit test error
2018-04-26 11:27:17 +08:00
if (!q.Receive(&elem)) {
fix unit test error
2018-04-26 11:26:25 +08:00
break;
}
Add reader blocking queue
2018-04-25 19:32:08 +08:00
res.push_back(elem);
}
EXPECT_EQ(res.size(), queue_cap);
for (size_t i = 0; i < res.size(); ++i) {
EXPECT_EQ(res[i], i);
}
}
TEST(BlockingQueue, ReceiverBlockingTest) {
const size_t queue_cap = 5;
BlockingQueue<size_t> q(queue_cap);
std::vector<size_t> receive_res;
std::thread receiver([&]() {
[clang-tidy] NO.10 enable `cppcoreguidelines-init-variables` (#57871) * [clang-tidy] enable cppcoreguidelines-init-variables check * fix
2023-10-11 11:53:12 +08:00
size_t elem = 0;
Add reader blocking queue
2018-04-25 19:32:08 +08:00
while (true) {
if (!q.Receive(&elem)) {
break;
}
receive_res.push_back(elem);
}
});
std::vector<size_t> to_send{2, 1, 7};
for (auto e : to_send) {
q.Send(e);
}
q.Close();
receiver.join();
EXPECT_EQ(receive_res.size(), to_send.size());
for (size_t i = 0; i < to_send.size(); ++i) {
EXPECT_EQ(receive_res[i], to_send[i]);
}
}
void CheckIsUnorderedSame(const std::vector<std::vector<size_t>>& v1,
const std::vector<std::vector<size_t>>& v2) {
std::set<size_t> s1;
std::set<size_t> s2;
[clang-tidy] NO.21 enable performance-for-range-copy (#61712) * clangtidy 21 * fix
2024-02-21 14:59:29 +08:00
for (auto const& vec : v1) {
Add reader blocking queue
2018-04-25 19:32:08 +08:00
for (size_t elem : vec) {
s1.insert(elem);
}
}
[clang-tidy] NO.21 enable performance-for-range-copy (#61712) * clangtidy 21 * fix
2024-02-21 14:59:29 +08:00
for (auto const& vec : v2) {
Add reader blocking queue
2018-04-25 19:32:08 +08:00
for (size_t elem : vec) {
s2.insert(elem);
}
}
EXPECT_EQ(s1.size(), s2.size());
auto it1 = s1.begin();
auto it2 = s2.begin();
while (it1 != s1.end()) {
EXPECT_EQ(*it1, *it2);
++it1;
++it2;
}
}
void MultiSenderMultiReceiver(const size_t queue_cap,
const std::vector<std::vector<size_t>>& to_send,
format all files in fluid using new config (#43776)
2022-06-26 20:04:39 +08:00
size_t receiver_num,
size_t send_time_gap,
Add reader blocking queue
2018-04-25 19:32:08 +08:00
size_t receive_time_gap) {
BlockingQueue<size_t> q(queue_cap);
size_t sender_num = to_send.size();
std::vector<std::thread> senders;
for (size_t s_idx = 0; s_idx < sender_num; ++s_idx) {
[clang-tidy] enable modernize-use-emplace (#55799) * [clang-tidy] enable modernize-use-emplace * Merge branch 'develop' of https://github.com/PaddlePaddle/Paddle into modernize_use_emplace
2023-08-04 15:22:08 +08:00
senders.emplace_back([&, s_idx] {
Add reader blocking queue
2018-04-25 19:32:08 +08:00
for (size_t elem : to_send[s_idx]) {
std::this_thread::sleep_for(std::chrono::milliseconds(send_time_gap));
EXPECT_TRUE(q.Send(elem));
}
[clang-tidy] enable modernize-use-emplace (#55799) * [clang-tidy] enable modernize-use-emplace * Merge branch 'develop' of https://github.com/PaddlePaddle/Paddle into modernize_use_emplace
2023-08-04 15:22:08 +08:00
});
Add reader blocking queue
2018-04-25 19:32:08 +08:00
}
std::vector<std::thread> receivers;
std::mutex mu;
std::vector<std::vector<size_t>> res;
for (size_t r_idx = 0; r_idx < receiver_num; ++r_idx) {
[clang-tidy] enable modernize-use-emplace (#55799) * [clang-tidy] enable modernize-use-emplace * Merge branch 'develop' of https://github.com/PaddlePaddle/Paddle into modernize_use_emplace
2023-08-04 15:22:08 +08:00
receivers.emplace_back([&] {
Add reader blocking queue
2018-04-25 19:32:08 +08:00
std::vector<size_t> receiver_res;
while (true) {
std::this_thread::sleep_for(
std::chrono::milliseconds(receive_time_gap));
[clang-tidy] NO.10 enable `cppcoreguidelines-init-variables` (#57871) * [clang-tidy] enable cppcoreguidelines-init-variables check * fix
2023-10-11 11:53:12 +08:00
size_t elem = 0;
Add reader blocking queue
2018-04-25 19:32:08 +08:00
if (!q.Receive(&elem)) {
break;
}
receiver_res.push_back(elem);
}
std::lock_guard<std::mutex> lock(mu);
res.push_back(receiver_res);
[clang-tidy] enable modernize-use-emplace (#55799) * [clang-tidy] enable modernize-use-emplace * Merge branch 'develop' of https://github.com/PaddlePaddle/Paddle into modernize_use_emplace
2023-08-04 15:22:08 +08:00
});
Add reader blocking queue
2018-04-25 19:32:08 +08:00
}
for (auto& t : senders) {
t.join();
}
q.Close();
for (auto& t : receivers) {
t.join();
}
CheckIsUnorderedSame(to_send, res);
}
TEST(BlockingQueue, MultiSenderMultiReaderTest) {
std::vector<std::vector<size_t>> to_send_1{{2, 3, 4}, {9}, {0, 7, 15, 6}};
MultiSenderMultiReceiver(2, to_send_1, 2, 0, 0);
MultiSenderMultiReceiver(10, to_send_1, 2, 0, 0);
MultiSenderMultiReceiver(2, to_send_1, 20, 0, 0);
MultiSenderMultiReceiver(2, to_send_1, 2, 50, 0);
MultiSenderMultiReceiver(2, to_send_1, 2, 0, 50);
std::vector<std::vector<size_t>> to_send_2{
{2, 3, 4}, {}, {0, 7, 15, 6, 9, 32}};
MultiSenderMultiReceiver(2, to_send_2, 3, 0, 0);
MultiSenderMultiReceiver(20, to_send_2, 3, 0, 0);
MultiSenderMultiReceiver(2, to_send_2, 30, 0, 0);
MultiSenderMultiReceiver(2, to_send_2, 3, 50, 0);
MultiSenderMultiReceiver(2, to_send_2, 3, 0, 50);
}
struct MyClass {
MyClass() : val_(0) {}
explicit MyClass(int val) : val_(val) {}
MyClass(const MyClass& b) { val_ = b.val_; }
[clang-tidy] NO.42-44 enable noexcept-move-constructor,unused-but-set-variable,security.FloatLoopCounter (#61555)
2024-02-20 11:53:01 +08:00
MyClass(MyClass&& b) noexcept { val_ = b.val_; }
[clang-tidy] No. 53,54 enable cppcoreguidelines-c-copy-assignment-signature and bugprone-use-after-move (#56601)
2023-08-29 10:41:00 +08:00
MyClass& operator=(const MyClass& b) {
if (this != &b) {
val_ = b.val_;
return *this;
}
return *this;
}
Add reader blocking queue
2018-04-25 19:32:08 +08:00
int val_;
};
TEST(BlockingQueue, MyClassTest) {
BlockingQueue<MyClass> q(2);
MyClass a(200);
[clang-tidy] No. 53,54 enable cppcoreguidelines-c-copy-assignment-signature and bugprone-use-after-move (#56601)
2023-08-29 10:41:00 +08:00
q.Send(a);
Add reader blocking queue
2018-04-25 19:32:08 +08:00
MyClass b;
q.Receive(&b);
EXPECT_EQ(a.val_, b.val_);
}
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
TEST(BlockingQueue, speed_test_mode) {
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
size_t queue_size = 10;
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
BlockingQueue<size_t> q1(queue_size, false);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
for (size_t i = 0; i < queue_size; ++i) {
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
q1.Send(i);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
}
[clang-tidy] NO.10 enable `cppcoreguidelines-init-variables` (#57871) * [clang-tidy] enable cppcoreguidelines-init-variables check * fix
2023-10-11 11:53:12 +08:00
size_t b = 0;
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
for (size_t i = 0; i < queue_size; ++i) {
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
q1.Receive(&b);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
EXPECT_EQ(b, i);
}
fix some compiler warning
2018-10-17 20:47:39 +08:00
EXPECT_EQ(q1.Size(), 0UL);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
BlockingQueue<size_t> q2(queue_size, true);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
for (size_t i = 0; i < queue_size; ++i) {
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
q2.Send(i);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
}
for (size_t i = 0; i < queue_size; ++i) {
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
q2.Receive(&b);
fix typo and warning in analyzer_resnet50_test test=develop
2018-10-23 10:23:45 +08:00
EXPECT_EQ(b, 0UL);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
}
change the use of FLAGS_reader_queue_speed_test_mode test=develop
2018-10-15 17:33:12 +08:00
EXPECT_EQ(q2.Size(), queue_size);
add reader_queue_speed_test_mode_flag test
2018-10-13 19:18:27 +08:00
}
Reference in New Issue Copy Permalink