1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
#include "rwlock.h"
#include "atomic.h"
#include <library/cpp/testing/unittest/registar.h>
#include <util/thread/pool.h>
#include <util/random/random.h>
class TRWMutexTest: public TTestBase {
UNIT_TEST_SUITE(TRWMutexTest);
UNIT_TEST(TestReaders)
UNIT_TEST(TestReadersWriters)
UNIT_TEST_SUITE_END();
struct TSharedData {
TSharedData()
: writersIn(0)
, readersIn(0)
, failed(false)
{
}
TAtomic writersIn;
TAtomic readersIn;
bool failed;
TRWMutex mutex;
};
class TThreadTask: public IObjectInQueue {
public:
using PFunc = void (TThreadTask::*)(void);
TThreadTask(PFunc func, TSharedData& data, size_t id, size_t total)
: Func_(func)
, Data_(data)
, Id_(id)
, Total_(total)
{
}
void Process(void*) override {
THolder<TThreadTask> This(this);
(this->*Func_)();
}
#define FAIL_ASSERT(cond) \
if (!(cond)) { \
Data_.failed = true; \
}
void RunReaders() {
Data_.mutex.AcquireRead();
AtomicIncrement(Data_.readersIn);
usleep(100);
FAIL_ASSERT(Data_.readersIn == long(Total_));
usleep(100);
AtomicDecrement(Data_.readersIn);
Data_.mutex.ReleaseRead();
}
void RunReadersWriters() {
if (Id_ % 2 == 0) {
for (size_t i = 0; i < 10; ++i) {
Data_.mutex.AcquireRead();
AtomicIncrement(Data_.readersIn);
FAIL_ASSERT(Data_.writersIn == 0);
usleep(RandomNumber<ui32>() % 5);
AtomicDecrement(Data_.readersIn);
Data_.mutex.ReleaseRead();
}
} else {
for (size_t i = 0; i < 10; ++i) {
Data_.mutex.AcquireWrite();
AtomicIncrement(Data_.writersIn);
FAIL_ASSERT(Data_.readersIn == 0 && Data_.writersIn == 1);
usleep(RandomNumber<ui32>() % 5);
AtomicDecrement(Data_.writersIn);
Data_.mutex.ReleaseWrite();
}
}
}
#undef FAIL_ASSERT
private:
PFunc Func_;
TSharedData& Data_;
size_t Id_;
size_t Total_;
};
private:
#define RUN_CYCLE(what, count) \
Q_.Start(count); \
for (size_t i = 0; i < count; ++i) { \
UNIT_ASSERT(Q_.Add(new TThreadTask(&TThreadTask::what, Data_, i, count))); \
} \
Q_.Stop(); \
bool b = Data_.failed; \
Data_.failed = false; \
UNIT_ASSERT(!b);
void TestReaders() {
RUN_CYCLE(RunReaders, 1);
}
void TestReadersWriters() {
RUN_CYCLE(RunReadersWriters, 1);
}
#undef RUN_CYCLE
private:
TSharedData Data_;
TThreadPool Q_;
};
UNIT_TEST_SUITE_REGISTRATION(TRWMutexTest)
|
