aboutsummaryrefslogtreecommitdiffstats
path: root/library/cpp/logger/sync_page_cache_file.cpp
blob: 86cf0fff8894360da2ae9ab3487d10dc16415371 (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
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
#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/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();
}