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
125
|
#include "sync_page_cache_file.h"
#include "record.h"
#include <util/generic/buffer.h>
#include <util/system/file.h>
#include <util/system/info.h>
#include <util/system/mutex.h>
#include <util/system/rwlock.h>
#include <util/system/align.h>
class TSyncPageCacheFileLogBackend::TImpl: public TNonCopyable {
public:
TImpl(const TString& path, size_t maxBufferSize, size_t maxPendingCacheSize)
: File_{OpenFile(path)}
, MaxBufferSize_{maxBufferSize}
, MaxPendingCacheSize_{maxPendingCacheSize}
, Buffer_{maxBufferSize}
{
ResetPtrs();
}
~TImpl() noexcept {
try {
Write();
FlushSync(GuaranteedWrittenPtr_, WrittenPtr_);
} catch (...) {
}
}
void WriteData(const TLogRecord& rec) {
TGuard guard{Lock_};
Buffer_.Append(rec.Data, rec.Len);
if (Buffer_.size() >= MaxBufferSize_) {
const i64 prevAlignedEndPtr = PageAlignedWrittenPtr_;
Write();
if (prevAlignedEndPtr < PageAlignedWrittenPtr_) {
FlushAsync(prevAlignedEndPtr, PageAlignedWrittenPtr_);
}
const i64 minPendingCacheOffset = PageAlignedWrittenPtr_ - MaxPendingCacheSize_;
if (minPendingCacheOffset > GuaranteedWrittenPtr_) {
FlushSync(GuaranteedWrittenPtr_, minPendingCacheOffset);
}
}
}
void ReopenLog() {
TGuard guard{Lock_};
Write();
FlushSync(GuaranteedWrittenPtr_, WrittenPtr_);
File_.LinkTo(OpenFile(File_.GetName()));
ResetPtrs();
}
private:
void ResetPtrs() {
WrittenPtr_ = File_.GetLength();
PageAlignedWrittenPtr_ = AlignDown(WrittenPtr_, GetPageSize());
GuaranteedWrittenPtr_ = WrittenPtr_;
}
static TFile OpenFile(const TString& path) {
return TFile{path, OpenAlways | WrOnly | ForAppend | Seq | NoReuse};
}
static i64 GetPageSize() {
static const i64 pageSize = NSystemInfo::GetPageSize();
Y_ASSUME(IsPowerOf2(pageSize));
return pageSize;
}
void Write() {
File_.Write(Buffer_.Data(), Buffer_.Size());
WrittenPtr_ += Buffer_.Size();
PageAlignedWrittenPtr_ = AlignDown(WrittenPtr_, GetPageSize());
Buffer_.Clear();
}
void FlushAsync(const i64 from, const i64 to) {
File_.FlushCache(from, to - from, /* wait = */ false);
}
void FlushSync(const i64 from, const i64 to) {
const i64 begin = AlignDown(from, GetPageSize());
const i64 end = AlignUp(to, GetPageSize());
const i64 length = end - begin;
File_.FlushCache(begin, length, /* wait = */ true);
File_.EvictCache(begin, length);
GuaranteedWrittenPtr_ = to;
}
private:
TMutex Lock_;
TFile File_;
const size_t MaxBufferSize_ = 0;
const size_t MaxPendingCacheSize_ = 0;
TBuffer Buffer_;
i64 WrittenPtr_ = 0;
i64 PageAlignedWrittenPtr_ = 0;
i64 GuaranteedWrittenPtr_ = 0;
};
TSyncPageCacheFileLogBackend::TSyncPageCacheFileLogBackend(const TString& path, size_t maxBufferSize, size_t maxPengingCacheSize)
: Impl_(MakeHolder<TImpl>(path, maxBufferSize, maxPengingCacheSize))
{}
TSyncPageCacheFileLogBackend::~TSyncPageCacheFileLogBackend() {
}
void TSyncPageCacheFileLogBackend::WriteData(const TLogRecord& rec) {
Impl_->WriteData(rec);
}
void TSyncPageCacheFileLogBackend::ReopenLog() {
Impl_->ReopenLog();
}
|
