|
1 | 1 | #include "stockpile.h" |
2 | 2 |
|
3 | | -#include "library/cpp/yt/logging/logger.h" |
| 3 | +#include <library/cpp/yt/threading/spin_lock.h> |
| 4 | + |
| 5 | +#include <library/cpp/yt/misc/global.h> |
| 6 | + |
| 7 | +#include <library/cpp/yt/memory/leaky_singleton.h> |
| 8 | + |
| 9 | +#include <library/cpp/yt/logging/logger.h> |
4 | 10 |
|
5 | 11 | #include <thread> |
6 | | -#include <mutex> |
7 | 12 |
|
8 | 13 | #include <sys/mman.h> |
9 | 14 |
|
10 | 15 | #include <util/system/thread.h> |
| 16 | + |
11 | 17 | #include <string.h> |
12 | 18 |
|
13 | 19 | namespace NYT { |
14 | 20 |
|
15 | 21 | //////////////////////////////////////////////////////////////////////////////// |
16 | 22 |
|
17 | | -static const auto Logger = NLogging::TLogger("Stockpile"); |
| 23 | +namespace { |
18 | 24 |
|
| 25 | +YT_DEFINE_GLOBAL(const NLogging::TLogger, Logger, "Stockpile"); |
19 | 26 | constexpr int MADV_STOCKPILE = 0x59410004; |
20 | 27 |
|
21 | | -//////////////////////////////////////////////////////////////////////////////// |
22 | | - |
23 | | -namespace { |
| 28 | +} // namespace |
24 | 29 |
|
25 | | -void RunWithFixedBreaks(i64 bufferSize, TDuration period) |
| 30 | +class TStockpileManagerImpl |
26 | 31 | { |
27 | | - auto returnCode = -::madvise(nullptr, bufferSize, MADV_STOCKPILE); |
28 | | - YT_LOG_DEBUG_IF(returnCode, "System call \"madvise\" failed: %v", strerror(returnCode)); |
29 | | - Sleep(period); |
30 | | -} |
| 32 | +public: |
| 33 | + static TStockpileManagerImpl* Get() |
| 34 | + { |
| 35 | + return LeakySingleton<TStockpileManagerImpl>(); |
| 36 | + } |
31 | 37 |
|
32 | | -void RunWithCappedLoad(i64 bufferSize, TDuration period) |
33 | | -{ |
34 | | - auto started = GetApproximateCpuInstant(); |
35 | | - auto returnCode = -::madvise(nullptr, bufferSize, MADV_STOCKPILE); |
36 | | - YT_LOG_DEBUG_IF(returnCode, "System call \"madvise\" failed: %v", strerror(returnCode)); |
37 | | - auto duration = CpuDurationToDuration(GetApproximateCpuInstant() - started); |
| 38 | + void Reconfigure(TStockpileOptions options) |
| 39 | + { |
| 40 | + auto guard = Guard(SpinLock_); |
| 41 | + |
| 42 | + Run_.store(false); |
| 43 | + for (const auto& thread : Threads_) { |
| 44 | + thread->join(); |
| 45 | + } |
| 46 | + |
| 47 | + Threads_.clear(); |
| 48 | + Run_.store(true); |
38 | 49 |
|
39 | | - if (duration < period) { |
40 | | - Sleep(period - duration); |
| 50 | + Options_ = options; |
| 51 | + |
| 52 | + for (int threadIndex = 0; threadIndex < Options_.ThreadCount; ++threadIndex) { |
| 53 | + Threads_.push_back(std::make_unique<std::thread>(&TStockpileManagerImpl::ThreadMain, this)); |
| 54 | + } |
41 | 55 | } |
42 | | -} |
43 | 56 |
|
44 | | -std::pair<i64, TDuration> RunWithBackoffs( |
45 | | - i64 adjustedBufferSize, |
46 | | - TDuration adjustedPeriod, |
47 | | - const TStockpileOptions& options, |
48 | | - i64 pageSize) |
49 | | -{ |
50 | | - int returnCode = -::madvise(nullptr, adjustedBufferSize, MADV_STOCKPILE); |
51 | | - YT_LOG_DEBUG_IF(returnCode, "System call \"madvise\" failed: %v", strerror(returnCode)); |
52 | | - |
53 | | - switch(returnCode) { |
54 | | - case 0: |
55 | | - Sleep(options.Period); |
56 | | - return {options.BufferSize, options.Period}; |
57 | | - |
58 | | - case ENOMEM: |
59 | | - if (adjustedBufferSize / 2 >= pageSize) { |
60 | | - // Immediately make an attempt to reclaim half as much. |
61 | | - adjustedBufferSize = adjustedBufferSize / 2; |
62 | | - } else { |
63 | | - // Unless there is not even a single reclaimable page. |
64 | | - Sleep(options.Period); |
65 | | - } |
66 | | - return {adjustedBufferSize, options.Period}; |
| 57 | +private: |
| 58 | + DECLARE_LEAKY_SINGLETON_FRIEND(); |
| 59 | + |
| 60 | + const i64 PageSize_ = sysconf(_SC_PAGESIZE); |
| 61 | + |
| 62 | + YT_DECLARE_SPIN_LOCK(NThreading::TSpinLock, SpinLock_); |
| 63 | + std::vector<std::unique_ptr<std::thread>> Threads_; |
| 64 | + TStockpileOptions Options_; |
| 65 | + std::atomic<bool> Run_ = false; |
| 66 | + |
| 67 | + void ThreadMain() |
| 68 | + { |
| 69 | + TThread::SetCurrentThreadName("Stockpile"); |
| 70 | + |
| 71 | + auto bufferSize = Options_.BufferSize; |
| 72 | + auto period = Options_.Period; |
67 | 73 |
|
68 | | - case EAGAIN: |
69 | | - case EINTR: |
70 | | - Sleep(adjustedPeriod); |
71 | | - return {options.BufferSize, adjustedPeriod + options.Period}; |
| 74 | + while (Run_.load()) { |
| 75 | + switch (Options_.Strategy) { |
| 76 | + case EStockpileStrategy::FixedBreaks: |
| 77 | + RunWithFixedBreaks(Options_.BufferSize, Options_.Period); |
| 78 | + break; |
72 | 79 |
|
73 | | - default: |
74 | | - Sleep(options.Period); |
75 | | - return {options.BufferSize, options.Period}; |
| 80 | + case EStockpileStrategy::FlooredLoad: |
| 81 | + RunWithCappedLoad(Options_.BufferSize, Options_.Period); |
| 82 | + break; |
| 83 | + |
| 84 | + case EStockpileStrategy::ProgressiveBackoff: |
| 85 | + std::tie(bufferSize, period) = RunWithBackoffs(bufferSize, period); |
| 86 | + break; |
| 87 | + |
| 88 | + default: |
| 89 | + YT_ABORT(); |
| 90 | + } |
| 91 | + } |
76 | 92 | } |
77 | | -} |
78 | 93 |
|
79 | | -} // namespace |
| 94 | + void RunWithFixedBreaks(i64 bufferSize, TDuration period) |
| 95 | + { |
| 96 | + auto returnCode = -::madvise(nullptr, bufferSize, MADV_STOCKPILE); |
| 97 | + YT_LOG_DEBUG_IF(returnCode != 0, "System call \"madvise\" failed: %v", strerror(returnCode)); |
80 | 98 |
|
81 | | -void RunStockpileThread(TStockpileOptions options, std::atomic<bool>* shouldProceed) |
82 | | -{ |
83 | | - TThread::SetCurrentThreadName("Stockpile"); |
| 99 | + Sleep(period); |
| 100 | + } |
84 | 101 |
|
85 | | - const i64 pageSize = sysconf(_SC_PAGESIZE); |
86 | | - auto bufferSize = options.BufferSize; |
87 | | - auto period = options.Period; |
| 102 | + void RunWithCappedLoad(i64 bufferSize, TDuration period) |
| 103 | + { |
| 104 | + auto started = GetApproximateCpuInstant(); |
88 | 105 |
|
89 | | - while (!shouldProceed || shouldProceed->load()) { |
90 | | - switch (options.Strategy) { |
91 | | - case EStockpileStrategy::FixedBreaks: |
92 | | - RunWithFixedBreaks(options.BufferSize, options.Period); |
93 | | - break; |
| 106 | + auto returnCode = -::madvise(nullptr, bufferSize, MADV_STOCKPILE); |
| 107 | + YT_LOG_DEBUG_IF(returnCode != 0, "System call \"madvise\" failed: %v", strerror(returnCode)); |
94 | 108 |
|
95 | | - case EStockpileStrategy::FlooredLoad: |
96 | | - RunWithCappedLoad(options.BufferSize, options.Period); |
97 | | - break; |
| 109 | + auto duration = CpuDurationToDuration(GetApproximateCpuInstant() - started); |
| 110 | + if (duration < period) { |
| 111 | + Sleep(period - duration); |
| 112 | + } |
| 113 | + } |
98 | 114 |
|
99 | | - case EStockpileStrategy::ProgressiveBackoff: |
100 | | - std::tie(bufferSize, period) = RunWithBackoffs(bufferSize, period, options, pageSize); |
101 | | - break; |
| 115 | + std::pair<i64, TDuration> RunWithBackoffs( |
| 116 | + i64 adjustedBufferSize, |
| 117 | + TDuration adjustedPeriod) |
| 118 | + { |
| 119 | + int returnCode = -::madvise(nullptr, adjustedBufferSize, MADV_STOCKPILE); |
| 120 | + YT_LOG_DEBUG_IF(returnCode != 0, "System call \"madvise\" failed: %v", strerror(returnCode)); |
| 121 | + |
| 122 | + switch(returnCode) { |
| 123 | + case 0: |
| 124 | + Sleep(Options_.Period); |
| 125 | + return {Options_.BufferSize, Options_.Period}; |
| 126 | + |
| 127 | + case ENOMEM: |
| 128 | + if (adjustedBufferSize / 2 >= PageSize_) { |
| 129 | + // Immediately make an attempt to reclaim half as much. |
| 130 | + adjustedBufferSize = adjustedBufferSize / 2; |
| 131 | + } else { |
| 132 | + // Unless there is not even a single reclaimable page. |
| 133 | + Sleep(Options_.Period); |
| 134 | + } |
| 135 | + return {adjustedBufferSize, Options_.Period}; |
| 136 | + |
| 137 | + case EAGAIN: |
| 138 | + case EINTR: |
| 139 | + Sleep(adjustedPeriod); |
| 140 | + return {Options_.BufferSize, adjustedPeriod + Options_.Period}; |
102 | 141 |
|
103 | 142 | default: |
104 | | - YT_ABORT(); |
| 143 | + Sleep(Options_.Period); |
| 144 | + return {Options_.BufferSize, Options_.Period}; |
105 | 145 | } |
106 | 146 | } |
107 | | -} |
| 147 | +}; |
| 148 | + |
| 149 | +//////////////////////////////////////////////////////////////////////////////// |
108 | 150 |
|
109 | | -void RunDetachedStockpileThreads(TStockpileOptions options) |
| 151 | +void TStockpileManager::Reconfigure(TStockpileOptions options) |
110 | 152 | { |
111 | | - static std::once_flag OnceFlag; |
112 | | - std::call_once(OnceFlag, [options = std::move(options)] { |
113 | | - for (int i = 0; i < options.ThreadCount; ++i) { |
114 | | - std::thread(RunStockpileThread, options, nullptr).detach(); |
115 | | - } |
116 | | - }); |
| 153 | + TStockpileManagerImpl::Get()->Reconfigure(std::move(options)); |
117 | 154 | } |
118 | 155 |
|
119 | 156 | //////////////////////////////////////////////////////////////////////////////// |
|
0 commit comments