blob: 5b750156b95c0dbf656253b534b268c9dfb91819 (
plain) (
blame)
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
|
#include "count_down_latch.h"
#include "futex.h"
#include <library/cpp/yt/threading/futex.h>
#include <library/cpp/yt/assert/assert.h>
#include <cerrno>
namespace NYT::NThreading {
////////////////////////////////////////////////////////////////////////////////
TCountDownLatch::TCountDownLatch(int count)
: Count_(count)
{ }
void TCountDownLatch::CountDown()
{
#ifndef _linux_
TGuard<TMutex> guard(Mutex_);
#endif
auto previous = Count_.fetch_sub(1, std::memory_order::release);
if (previous == 1) {
#ifdef _linux_
int rv = NThreading::FutexWake(
reinterpret_cast<int*>(&Count_),
std::numeric_limits<int>::max());
YT_VERIFY(rv >= 0);
#else
ConditionVariable_.BroadCast();
#endif
}
}
void TCountDownLatch::Wait() const
{
while (true) {
#ifndef _linux_
TGuard<TMutex> guard(Mutex_);
#endif
auto count = Count_.load(std::memory_order::acquire);
if (count == 0) {
return;
}
#ifdef _linux_
int rv = NThreading::FutexWait(
const_cast<int*>(reinterpret_cast<const int*>(&Count_)),
count);
YT_VERIFY(rv >= 0 || errno == EWOULDBLOCK || errno == EINTR);
#else
ConditionVariable_.WaitI(Mutex_);
#endif
}
}
bool TCountDownLatch::TryWait() const
{
return Count_.load(std::memory_order::acquire) == 0;
}
int TCountDownLatch::GetCount() const
{
return Count_.load(std::memory_order::relaxed);
}
////////////////////////////////////////////////////////////////////////////////
} // namespace NYT::NThreading
|
