Skip to content
This repository was archived by the owner on Feb 20, 2023. It is now read-only.

Commit ce5f676

Browse files
lmwnshnlmwnshn
authored
Replace dirty spinwaits with pthread cond waits in messenger test. (#1444)
1 parent f45d0ed commit ce5f676

File tree

1 file changed

+95
-18
lines changed

1 file changed

+95
-18
lines changed

‎test/messenger/messenger_test.cpp

Lines changed: 95 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#include "messenger/messenger.h"
22

3+
#include <pthread.h>
34
#include <sys/mman.h>
45
#include <unistd.h>
56

@@ -112,30 +113,56 @@ TEST_F(MessengerTests, BasicReplicationTest) {
112113
done[1] = false;
113114
done[2] = false;
114115

115-
auto spin_until_init = [init]() {
116+
auto *pmutex = static_cast<pthread_mutex_t *>(
117+
mmap(nullptr, sizeof(pthread_mutex_t), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
118+
NOISEPAGE_ASSERT(MAP_FAILED != pmutex, "mmap() failed.");
119+
pthread_mutexattr_t mutexattr;
120+
pthread_mutexattr_init(&mutexattr);
121+
pthread_mutexattr_setpshared(&mutexattr, PTHREAD_PROCESS_SHARED);
122+
pthread_mutex_init(pmutex, &mutexattr);
123+
124+
auto *pcond = static_cast<pthread_cond_t *>(
125+
mmap(nullptr, sizeof(pthread_cond_t), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
126+
NOISEPAGE_ASSERT(MAP_FAILED != pcond, "mmap() failed.");
127+
pthread_condattr_t condattr;
128+
pthread_condattr_init(&condattr);
129+
pthread_condattr_setpshared(&condattr, PTHREAD_PROCESS_SHARED);
130+
pthread_cond_init(pcond, &condattr);
131+
132+
auto sleep_until_init = [init, pmutex, pcond]() {
116133
while (!(init[0] && init[1] && init[2])) {
134+
pthread_mutex_lock(pmutex);
135+
pthread_cond_wait(pcond, pmutex);
136+
pthread_mutex_unlock(pmutex);
117137
}
118138
};
119139

120-
auto spin_until_done = [done]() {
140+
auto sleep_until_done = [done, pmutex, pcond]() {
121141
while (!(done[0] && done[1] && done[2])) {
142+
pthread_mutex_lock(pmutex);
143+
pthread_cond_wait(pcond, pmutex);
144+
pthread_mutex_unlock(pmutex);
122145
}
123146
};
124147

148+
auto wake_all = [pcond]() { pthread_cond_broadcast(pcond); };
149+
125150
VoidFn primary_fn = [=]() {
126151
auto primary = BuildDBMain(port_primary, port_messenger_primary, "primary");
127152
primary->GetNetworkLayer()->GetServer()->RunServer();
128153

129154
init[0] = true;
130-
spin_until_init();
155+
wake_all();
156+
sleep_until_init();
131157
DirtySleep();
132158

133159
while (!(done[1] && done[2])) {
134160
}
135161

136162
MESSENGER_LOG_TRACE("Primary done.");
137163
done[0] = true;
138-
spin_until_done();
164+
wake_all();
165+
sleep_until_done();
139166
MESSENGER_LOG_TRACE("Primary exit.");
140167
primary->ForceShutdown();
141168
};
@@ -145,7 +172,8 @@ TEST_F(MessengerTests, BasicReplicationTest) {
145172
replica1->GetNetworkLayer()->GetServer()->RunServer();
146173

147174
init[1] = true;
148-
spin_until_init();
175+
wake_all();
176+
sleep_until_init();
149177
DirtySleep();
150178

151179
// Set up a connection to the primary.
@@ -186,7 +214,8 @@ TEST_F(MessengerTests, BasicReplicationTest) {
186214

187215
MESSENGER_LOG_TRACE("Replica 1 done.");
188216
done[1] = true;
189-
spin_until_done();
217+
wake_all();
218+
sleep_until_done();
190219
MESSENGER_LOG_TRACE("Replica 1 exit.");
191220
replica1->ForceShutdown();
192221
};
@@ -196,7 +225,8 @@ TEST_F(MessengerTests, BasicReplicationTest) {
196225
replica2->GetNetworkLayer()->GetServer()->RunServer();
197226

198227
init[2] = true;
199-
spin_until_init();
228+
wake_all();
229+
sleep_until_init();
200230
DirtySleep();
201231

202232
// Set up a connection to the primary.
@@ -237,16 +267,21 @@ TEST_F(MessengerTests, BasicReplicationTest) {
237267

238268
MESSENGER_LOG_TRACE("Replica 2 done.");
239269
done[2] = true;
240-
spin_until_done();
270+
wake_all();
271+
sleep_until_done();
241272
MESSENGER_LOG_TRACE("Replica 2 exit.");
242273
replica2->ForceShutdown();
243274
};
244275

245276
std::vector<pid_t> pids = ForkTests({primary_fn, replica1_fn, replica2_fn});
246277

247278
// Spin until all done.
248-
while (!(done[0] && done[1] && done[2])) {
249-
}
279+
sleep_until_done();
280+
281+
pthread_mutex_destroy(pmutex);
282+
pthread_mutexattr_destroy(&mutexattr);
283+
pthread_cond_destroy(pcond);
284+
pthread_condattr_destroy(&condattr);
250285

251286
DirtySleep();
252287
{
@@ -257,6 +292,14 @@ TEST_F(MessengerTests, BasicReplicationTest) {
257292
UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(done)), 3 * sizeof(bool));
258293
NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
259294
}
295+
{
296+
UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(pmutex), sizeof(pthread_mutex_t));
297+
NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
298+
}
299+
{
300+
UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(pcond), sizeof(pthread_cond_t));
301+
NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
302+
}
260303
}
261304

262305
// NOLINTNEXTLINE
@@ -283,16 +326,34 @@ TEST_F(MessengerTests, BasicListenTest) {
283326
done[0] = false;
284327
done[1] = false;
285328

286-
auto spin_until_init = [init]() {
329+
auto *pmutex = static_cast<pthread_mutex_t *>(
330+
mmap(nullptr, sizeof(pthread_mutex_t), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
331+
NOISEPAGE_ASSERT(MAP_FAILED != pmutex, "mmap() failed.");
332+
pthread_mutexattr_t mutexattr;
333+
pthread_mutexattr_init(&mutexattr);
334+
pthread_mutexattr_setpshared(&mutexattr, PTHREAD_PROCESS_SHARED);
335+
pthread_mutex_init(pmutex, &mutexattr);
336+
337+
auto *pcond = static_cast<pthread_cond_t *>(
338+
mmap(nullptr, sizeof(pthread_cond_t), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0));
339+
NOISEPAGE_ASSERT(MAP_FAILED != pcond, "mmap() failed.");
340+
pthread_condattr_t condattr;
341+
pthread_condattr_init(&condattr);
342+
pthread_condattr_setpshared(&condattr, PTHREAD_PROCESS_SHARED);
343+
pthread_cond_init(pcond, &condattr);
344+
345+
auto sleep_until_init = [init]() {
287346
while (!(init[0] && init[1])) {
288347
}
289348
};
290349

291-
auto spin_until_done = [done]() {
350+
auto sleep_until_done = [done]() {
292351
while (!(done[0] && done[1])) {
293352
}
294353
};
295354

355+
auto wake_all = [pcond]() { pthread_cond_broadcast(pcond); };
356+
296357
VoidFn primary_fn = [=]() {
297358
auto primary = BuildDBMain(port_primary, port_messenger_primary, "primary");
298359
primary->GetNetworkLayer()->GetServer()->RunServer();
@@ -311,15 +372,17 @@ TEST_F(MessengerTests, BasicListenTest) {
311372
});
312373

313374
init[0] = true;
314-
spin_until_init();
375+
wake_all();
376+
sleep_until_init();
315377
DirtySleep();
316378

317379
while (!done[1]) {
318380
}
319381

320382
MESSENGER_LOG_TRACE("Primary done.");
321383
done[0] = true;
322-
spin_until_done();
384+
wake_all();
385+
sleep_until_done();
323386
MESSENGER_LOG_TRACE("Primary exit.");
324387
primary->ForceShutdown();
325388
};
@@ -328,7 +391,8 @@ TEST_F(MessengerTests, BasicListenTest) {
328391
auto replica1 = BuildDBMain(port_replica1, port_messenger_replica1, "replica1");
329392
replica1->GetNetworkLayer()->GetServer()->RunServer();
330393
init[1] = true;
331-
spin_until_init();
394+
wake_all();
395+
sleep_until_init();
332396
DirtySleep();
333397

334398
// Set up a connection to the primary via the listen endpoint.
@@ -363,16 +427,21 @@ TEST_F(MessengerTests, BasicListenTest) {
363427

364428
MESSENGER_LOG_TRACE("Replica 1 done.");
365429
done[1] = true;
366-
spin_until_done();
430+
wake_all();
431+
sleep_until_done();
367432
MESSENGER_LOG_TRACE("Replica 1 exit.");
368433
replica1->ForceShutdown();
369434
};
370435

371436
std::vector<pid_t> pids = ForkTests({primary_fn, replica1_fn});
372437

373438
// Spin until all done.
374-
while (!(done[0] && done[1])) {
375-
}
439+
sleep_until_done();
440+
441+
pthread_mutex_destroy(pmutex);
442+
pthread_mutexattr_destroy(&mutexattr);
443+
pthread_cond_destroy(pcond);
444+
pthread_condattr_destroy(&condattr);
376445

377446
DirtySleep();
378447
{
@@ -383,6 +452,14 @@ TEST_F(MessengerTests, BasicListenTest) {
383452
UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(const_cast<bool *>(done)), 2 * sizeof(bool));
384453
NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
385454
}
455+
{
456+
UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(pmutex), sizeof(pthread_mutex_t));
457+
NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
458+
}
459+
{
460+
UNUSED_ATTRIBUTE int munmap_retval = munmap(static_cast<void *>(pcond), sizeof(pthread_cond_t));
461+
NOISEPAGE_ASSERT(-1 != munmap_retval, "munmap() failed.");
462+
}
386463
}
387464

388465
} // namespace noisepage::messenger

0 commit comments

Comments
 (0)