mirror of
https://github.com/opencv/opencv.git
synced 2024-12-16 18:39:12 +08:00
6a73e5a720
GAPI: Align EXPECT/ASSERT macros * Align TEST macros * restart CI * Fix ASSERT_GT in gapi_async_test
198 lines
5.2 KiB
C++
198 lines
5.2 KiB
C++
// This file is part of OpenCV project.
|
|
// It is subject to the license terms in the LICENSE file found in the top-level directory
|
|
// of this distribution and at http://opencv.org/license.html.
|
|
//
|
|
// Copyright (C) 2019 Intel Corporation
|
|
|
|
#include "../test_precomp.hpp"
|
|
|
|
#include <unordered_set>
|
|
#include <thread>
|
|
|
|
#include "executor/conc_queue.hpp"
|
|
|
|
namespace opencv_test
|
|
{
|
|
using namespace cv::gapi;
|
|
|
|
TEST(ConcQueue, PushPop)
|
|
{
|
|
own::concurrent_bounded_queue<int> q;
|
|
for (int i = 0; i < 100; i++)
|
|
{
|
|
q.push(i);
|
|
}
|
|
|
|
for (int i = 0; i < 100; i++)
|
|
{
|
|
int x;
|
|
q.pop(x);
|
|
EXPECT_EQ(i, x);
|
|
}
|
|
}
|
|
|
|
TEST(ConcQueue, TryPop)
|
|
{
|
|
own::concurrent_bounded_queue<int> q;
|
|
int x = 0;
|
|
EXPECT_FALSE(q.try_pop(x));
|
|
|
|
q.push(1);
|
|
EXPECT_TRUE(q.try_pop(x));
|
|
EXPECT_EQ(1, x);
|
|
}
|
|
|
|
TEST(ConcQueue, Clear)
|
|
{
|
|
own::concurrent_bounded_queue<int> q;
|
|
for (int i = 0; i < 10; i++)
|
|
{
|
|
q.push(i);
|
|
}
|
|
|
|
q.clear();
|
|
int x = 0;
|
|
EXPECT_FALSE(q.try_pop(x));
|
|
}
|
|
|
|
// In this test, every writer thread produces its own range of integer
|
|
// numbers, writing those to a shared queue.
|
|
//
|
|
// Every reader thread pops elements from the queue (until -1 is
|
|
// reached) and stores those in its own associated set.
|
|
//
|
|
// Finally, the master thread waits for completion of all other
|
|
// threads and verifies that all the necessary data is
|
|
// produced/obtained.
|
|
namespace
|
|
{
|
|
using StressParam = std::tuple<int // Num writer threads
|
|
,int // Num elements per writer
|
|
,int // Num reader threads
|
|
,std::size_t>; // Queue capacity
|
|
constexpr int STOP_SIGN = -1;
|
|
constexpr int BASE = 1000;
|
|
}
|
|
struct ConcQueue_: public ::testing::TestWithParam<StressParam>
|
|
{
|
|
using Q = own::concurrent_bounded_queue<int>;
|
|
using S = std::unordered_set<int>;
|
|
|
|
static void writer(int base, int writes, Q& q)
|
|
{
|
|
for (int i = 0; i < writes; i++)
|
|
{
|
|
q.push(base + i);
|
|
}
|
|
q.push(STOP_SIGN);
|
|
}
|
|
|
|
static void reader(Q& q, S& s)
|
|
{
|
|
int x = 0;
|
|
while (true)
|
|
{
|
|
q.pop(x);
|
|
if (x == STOP_SIGN) return;
|
|
s.insert(x);
|
|
}
|
|
}
|
|
};
|
|
|
|
TEST_P(ConcQueue_, Test)
|
|
{
|
|
int num_writers = 0;
|
|
int num_writes = 0;
|
|
int num_readers = 0;
|
|
std::size_t capacity = 0u;
|
|
std::tie(num_writers, num_writes, num_readers, capacity) = GetParam();
|
|
|
|
CV_Assert(num_writers < 20);
|
|
CV_Assert(num_writes < BASE);
|
|
|
|
Q q;
|
|
if (capacity)
|
|
{
|
|
// see below (2)
|
|
CV_Assert(static_cast<int>(capacity) > (num_writers - num_readers));
|
|
q.set_capacity(capacity);
|
|
}
|
|
|
|
// Start reader threads
|
|
std::vector<S> storage(num_readers);
|
|
std::vector<std::thread> readers;
|
|
for (S& s : storage)
|
|
{
|
|
readers.emplace_back(reader, std::ref(q), std::ref(s));
|
|
}
|
|
|
|
// Start writer threads, also pre-generate reference numbers
|
|
S reference;
|
|
std::vector<std::thread> writers;
|
|
for (int w = 0; w < num_writers; w++)
|
|
{
|
|
writers.emplace_back(writer, w*BASE, num_writes, std::ref(q));
|
|
for (int r = 0; r < num_writes; r++)
|
|
{
|
|
reference.insert(w*BASE + r);
|
|
}
|
|
}
|
|
|
|
// Every writer puts a STOP_SIGN at the end,
|
|
// There are three cases:
|
|
// 1) num_writers == num_readers
|
|
// every reader should get its own STOP_SIGN from any
|
|
// of the writers
|
|
//
|
|
// 2) num_writers > num_readers
|
|
// every reader will get a STOP_SIGN but there're more
|
|
// STOP_SIGNs may be pushed to the queue - and if this
|
|
// number exceeds capacity, writers block (to a deadlock).
|
|
// The latter situation must be avoided at parameters level.
|
|
// [a] Also not every data produced by writers will be consumed
|
|
// by a reader in this case. Master thread will read the rest
|
|
//
|
|
// 3) num_readers > num_writers
|
|
// in this case, some readers will stuck and will never get
|
|
// a STOP_SIGN. Master thread will push extra STOP_SIGNs to the
|
|
// queue.
|
|
|
|
// Solution to (2a)
|
|
S remnants;
|
|
if (num_writers > num_readers)
|
|
{
|
|
int extra = num_writers - num_readers;
|
|
while (extra)
|
|
{
|
|
int x = 0;
|
|
q.pop(x);
|
|
if (x == STOP_SIGN) extra--;
|
|
else remnants.insert(x);
|
|
}
|
|
}
|
|
|
|
// Solution to (3)
|
|
if (num_readers > num_writers)
|
|
{
|
|
int extra = num_readers - num_writers;
|
|
while (extra--) q.push(STOP_SIGN);
|
|
}
|
|
|
|
// Wait for completions
|
|
for (auto &t : readers) t.join();
|
|
for (auto &t : writers) t.join();
|
|
|
|
// Accumulate and validate the result
|
|
S result(remnants.begin(), remnants.end());
|
|
for (const auto &s : storage) result.insert(s.begin(), s.end());
|
|
|
|
EXPECT_EQ(reference, result);
|
|
}
|
|
|
|
INSTANTIATE_TEST_CASE_P(ConcQueueStress, ConcQueue_,
|
|
Combine( Values(1, 2, 4, 8, 16) // writers
|
|
, Values(1, 32, 96, 256) // writes
|
|
, Values(1, 2, 10) // readers
|
|
, Values(0u, 16u, 32u))); // capacity
|
|
} // namespace opencv_test
|