skiplist_test.cc raw
1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
4
5 #include "db/skiplist.h"
6
7 #include <atomic>
8 #include <set>
9
10 #include "leveldb/env.h"
11 #include "port/port.h"
12 #include "port/thread_annotations.h"
13 #include "util/arena.h"
14 #include "util/hash.h"
15 #include "util/random.h"
16 #include "util/testharness.h"
17
18 namespace leveldb {
19
20 typedef uint64_t Key;
21
22 struct Comparator {
23 int operator()(const Key& a, const Key& b) const {
24 if (a < b) {
25 return -1;
26 } else if (a > b) {
27 return +1;
28 } else {
29 return 0;
30 }
31 }
32 };
33
34 class SkipTest {};
35
36 TEST(SkipTest, Empty) {
37 Arena arena;
38 Comparator cmp;
39 SkipList<Key, Comparator> list(cmp, &arena);
40 ASSERT_TRUE(!list.Contains(10));
41
42 SkipList<Key, Comparator>::Iterator iter(&list);
43 ASSERT_TRUE(!iter.Valid());
44 iter.SeekToFirst();
45 ASSERT_TRUE(!iter.Valid());
46 iter.Seek(100);
47 ASSERT_TRUE(!iter.Valid());
48 iter.SeekToLast();
49 ASSERT_TRUE(!iter.Valid());
50 }
51
52 TEST(SkipTest, InsertAndLookup) {
53 const int N = 2000;
54 const int R = 5000;
55 Random rnd(1000);
56 std::set<Key> keys;
57 Arena arena;
58 Comparator cmp;
59 SkipList<Key, Comparator> list(cmp, &arena);
60 for (int i = 0; i < N; i++) {
61 Key key = rnd.Next() % R;
62 if (keys.insert(key).second) {
63 list.Insert(key);
64 }
65 }
66
67 for (int i = 0; i < R; i++) {
68 if (list.Contains(i)) {
69 ASSERT_EQ(keys.count(i), 1);
70 } else {
71 ASSERT_EQ(keys.count(i), 0);
72 }
73 }
74
75 // Simple iterator tests
76 {
77 SkipList<Key, Comparator>::Iterator iter(&list);
78 ASSERT_TRUE(!iter.Valid());
79
80 iter.Seek(0);
81 ASSERT_TRUE(iter.Valid());
82 ASSERT_EQ(*(keys.begin()), iter.key());
83
84 iter.SeekToFirst();
85 ASSERT_TRUE(iter.Valid());
86 ASSERT_EQ(*(keys.begin()), iter.key());
87
88 iter.SeekToLast();
89 ASSERT_TRUE(iter.Valid());
90 ASSERT_EQ(*(keys.rbegin()), iter.key());
91 }
92
93 // Forward iteration test
94 for (int i = 0; i < R; i++) {
95 SkipList<Key, Comparator>::Iterator iter(&list);
96 iter.Seek(i);
97
98 // Compare against model iterator
99 std::set<Key>::iterator model_iter = keys.lower_bound(i);
100 for (int j = 0; j < 3; j++) {
101 if (model_iter == keys.end()) {
102 ASSERT_TRUE(!iter.Valid());
103 break;
104 } else {
105 ASSERT_TRUE(iter.Valid());
106 ASSERT_EQ(*model_iter, iter.key());
107 ++model_iter;
108 iter.Next();
109 }
110 }
111 }
112
113 // Backward iteration test
114 {
115 SkipList<Key, Comparator>::Iterator iter(&list);
116 iter.SeekToLast();
117
118 // Compare against model iterator
119 for (std::set<Key>::reverse_iterator model_iter = keys.rbegin();
120 model_iter != keys.rend(); ++model_iter) {
121 ASSERT_TRUE(iter.Valid());
122 ASSERT_EQ(*model_iter, iter.key());
123 iter.Prev();
124 }
125 ASSERT_TRUE(!iter.Valid());
126 }
127 }
128
129 // We want to make sure that with a single writer and multiple
130 // concurrent readers (with no synchronization other than when a
131 // reader's iterator is created), the reader always observes all the
132 // data that was present in the skip list when the iterator was
133 // constructed. Because insertions are happening concurrently, we may
134 // also observe new values that were inserted since the iterator was
135 // constructed, but we should never miss any values that were present
136 // at iterator construction time.
137 //
138 // We generate multi-part keys:
139 // <key,gen,hash>
140 // where:
141 // key is in range [0..K-1]
142 // gen is a generation number for key
143 // hash is hash(key,gen)
144 //
145 // The insertion code picks a random key, sets gen to be 1 + the last
146 // generation number inserted for that key, and sets hash to Hash(key,gen).
147 //
148 // At the beginning of a read, we snapshot the last inserted
149 // generation number for each key. We then iterate, including random
150 // calls to Next() and Seek(). For every key we encounter, we
151 // check that it is either expected given the initial snapshot or has
152 // been concurrently added since the iterator started.
153 class ConcurrentTest {
154 private:
155 static const uint32_t K = 4;
156
157 static uint64_t key(Key key) { return (key >> 40); }
158 static uint64_t gen(Key key) { return (key >> 8) & 0xffffffffu; }
159 static uint64_t hash(Key key) { return key & 0xff; }
160
161 static uint64_t HashNumbers(uint64_t k, uint64_t g) {
162 uint64_t data[2] = {k, g};
163 return Hash(reinterpret_cast<char*>(data), sizeof(data), 0);
164 }
165
166 static Key MakeKey(uint64_t k, uint64_t g) {
167 static_assert(sizeof(Key) == sizeof(uint64_t), "");
168 assert(k <= K); // We sometimes pass K to seek to the end of the skiplist
169 assert(g <= 0xffffffffu);
170 return ((k << 40) | (g << 8) | (HashNumbers(k, g) & 0xff));
171 }
172
173 static bool IsValidKey(Key k) {
174 return hash(k) == (HashNumbers(key(k), gen(k)) & 0xff);
175 }
176
177 static Key RandomTarget(Random* rnd) {
178 switch (rnd->Next() % 10) {
179 case 0:
180 // Seek to beginning
181 return MakeKey(0, 0);
182 case 1:
183 // Seek to end
184 return MakeKey(K, 0);
185 default:
186 // Seek to middle
187 return MakeKey(rnd->Next() % K, 0);
188 }
189 }
190
191 // Per-key generation
192 struct State {
193 std::atomic<int> generation[K];
194 void Set(int k, int v) {
195 generation[k].store(v, std::memory_order_release);
196 }
197 int Get(int k) { return generation[k].load(std::memory_order_acquire); }
198
199 State() {
200 for (int k = 0; k < K; k++) {
201 Set(k, 0);
202 }
203 }
204 };
205
206 // Current state of the test
207 State current_;
208
209 Arena arena_;
210
211 // SkipList is not protected by mu_. We just use a single writer
212 // thread to modify it.
213 SkipList<Key, Comparator> list_;
214
215 public:
216 ConcurrentTest() : list_(Comparator(), &arena_) {}
217
218 // REQUIRES: External synchronization
219 void WriteStep(Random* rnd) {
220 const uint32_t k = rnd->Next() % K;
221 const intptr_t g = current_.Get(k) + 1;
222 const Key key = MakeKey(k, g);
223 list_.Insert(key);
224 current_.Set(k, g);
225 }
226
227 void ReadStep(Random* rnd) {
228 // Remember the initial committed state of the skiplist.
229 State initial_state;
230 for (int k = 0; k < K; k++) {
231 initial_state.Set(k, current_.Get(k));
232 }
233
234 Key pos = RandomTarget(rnd);
235 SkipList<Key, Comparator>::Iterator iter(&list_);
236 iter.Seek(pos);
237 while (true) {
238 Key current;
239 if (!iter.Valid()) {
240 current = MakeKey(K, 0);
241 } else {
242 current = iter.key();
243 ASSERT_TRUE(IsValidKey(current)) << current;
244 }
245 ASSERT_LE(pos, current) << "should not go backwards";
246
247 // Verify that everything in [pos,current) was not present in
248 // initial_state.
249 while (pos < current) {
250 ASSERT_LT(key(pos), K) << pos;
251
252 // Note that generation 0 is never inserted, so it is ok if
253 // <*,0,*> is missing.
254 ASSERT_TRUE((gen(pos) == 0) ||
255 (gen(pos) > static_cast<Key>(initial_state.Get(key(pos)))))
256 << "key: " << key(pos) << "; gen: " << gen(pos)
257 << "; initgen: " << initial_state.Get(key(pos));
258
259 // Advance to next key in the valid key space
260 if (key(pos) < key(current)) {
261 pos = MakeKey(key(pos) + 1, 0);
262 } else {
263 pos = MakeKey(key(pos), gen(pos) + 1);
264 }
265 }
266
267 if (!iter.Valid()) {
268 break;
269 }
270
271 if (rnd->Next() % 2) {
272 iter.Next();
273 pos = MakeKey(key(pos), gen(pos) + 1);
274 } else {
275 Key new_target = RandomTarget(rnd);
276 if (new_target > pos) {
277 pos = new_target;
278 iter.Seek(new_target);
279 }
280 }
281 }
282 }
283 };
284 const uint32_t ConcurrentTest::K;
285
286 // Simple test that does single-threaded testing of the ConcurrentTest
287 // scaffolding.
288 TEST(SkipTest, ConcurrentWithoutThreads) {
289 ConcurrentTest test;
290 Random rnd(test::RandomSeed());
291 for (int i = 0; i < 10000; i++) {
292 test.ReadStep(&rnd);
293 test.WriteStep(&rnd);
294 }
295 }
296
297 class TestState {
298 public:
299 ConcurrentTest t_;
300 int seed_;
301 std::atomic<bool> quit_flag_;
302
303 enum ReaderState { STARTING, RUNNING, DONE };
304
305 explicit TestState(int s)
306 : seed_(s), quit_flag_(false), state_(STARTING), state_cv_(&mu_) {}
307
308 void Wait(ReaderState s) LOCKS_EXCLUDED(mu_) {
309 mu_.Lock();
310 while (state_ != s) {
311 state_cv_.Wait();
312 }
313 mu_.Unlock();
314 }
315
316 void Change(ReaderState s) LOCKS_EXCLUDED(mu_) {
317 mu_.Lock();
318 state_ = s;
319 state_cv_.Signal();
320 mu_.Unlock();
321 }
322
323 private:
324 port::Mutex mu_;
325 ReaderState state_ GUARDED_BY(mu_);
326 port::CondVar state_cv_ GUARDED_BY(mu_);
327 };
328
329 static void ConcurrentReader(void* arg) {
330 TestState* state = reinterpret_cast<TestState*>(arg);
331 Random rnd(state->seed_);
332 int64_t reads = 0;
333 state->Change(TestState::RUNNING);
334 while (!state->quit_flag_.load(std::memory_order_acquire)) {
335 state->t_.ReadStep(&rnd);
336 ++reads;
337 }
338 state->Change(TestState::DONE);
339 }
340
341 static void RunConcurrent(int run) {
342 const int seed = test::RandomSeed() + (run * 100);
343 Random rnd(seed);
344 const int N = 1000;
345 const int kSize = 1000;
346 for (int i = 0; i < N; i++) {
347 if ((i % 100) == 0) {
348 fprintf(stderr, "Run %d of %d\n", i, N);
349 }
350 TestState state(seed + 1);
351 Env::Default()->Schedule(ConcurrentReader, &state);
352 state.Wait(TestState::RUNNING);
353 for (int i = 0; i < kSize; i++) {
354 state.t_.WriteStep(&rnd);
355 }
356 state.quit_flag_.store(true, std::memory_order_release);
357 state.Wait(TestState::DONE);
358 }
359 }
360
361 TEST(SkipTest, Concurrent1) { RunConcurrent(1); }
362 TEST(SkipTest, Concurrent2) { RunConcurrent(2); }
363 TEST(SkipTest, Concurrent3) { RunConcurrent(3); }
364 TEST(SkipTest, Concurrent4) { RunConcurrent(4); }
365 TEST(SkipTest, Concurrent5) { RunConcurrent(5); }
366
367 } // namespace leveldb
368
369 int main(int argc, char** argv) { return leveldb::test::RunAllTests(); }
370