2
This file is part of TON Blockchain Library.
4
TON Blockchain Library is free software: you can redistribute it and/or modify
5
it under the terms of the GNU Lesser General Public License as published by
6
the Free Software Foundation, either version 2 of the License, or
7
(at your option) any later version.
9
TON Blockchain Library is distributed in the hope that it will be useful,
10
but WITHOUT ANY WARRANTY; without even the implied warranty of
11
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12
GNU Lesser General Public License for more details.
14
You should have received a copy of the GNU Lesser General Public License
15
along with TON Blockchain Library. If not, see <http://www.gnu.org/licenses/>.
17
Copyright 2017-2020 Telegram Systems LLP
19
#include "ChainBuffer.h"
21
#include "td/utils/buffer.h"
22
#include "td/db/utils/StreamInterface.h"
26
class ChainBuffer : public StreamWriterInterface, public StreamReaderInterface {
28
using Options = ::td::ChainBuffer::Options;
30
ChainBuffer(Options options) {
31
shared_.options_ = options;
32
reader_.io_slices_.reserve(options.max_io_slices);
33
reader_.buf_ = writer_.buf_.extract_reader();
36
// StreamReaderInterface
37
size_t reader_size() override {
38
reader_.buf_.sync_with_writer();
39
return reader_.buf_.size();
42
Slice prepare_read() override {
43
return reader_.buf_.prepare_read();
45
Span<IoSlice> prepare_readv() override {
46
reader_.io_slices_.clear();
47
auto it = reader_.buf_.clone();
48
while (!it.empty() && reader_.io_slices_.size() < reader_.io_slices_.capacity()) {
49
auto slice = it.prepare_read();
50
reader_.io_slices_.push_back(as_io_slice(slice));
51
it.confirm_read(slice.size());
53
return reader_.io_slices_;
55
void confirm_read(size_t size) override {
56
reader_.buf_.advance(size);
59
void close_reader(Status error) override {
60
CHECK(!reader_.is_closed_);
61
reader_.status_ = std::move(error);
62
reader_.is_closed_.store(true, std::memory_order_release);
64
bool is_writer_closed() const override {
65
return writer_.is_closed_.load(std::memory_order_acquire);
67
Status &writer_status() override {
68
CHECK(is_writer_closed());
69
return writer_.status_;
72
// StreamWriterInterface
73
size_t writer_size() override {
76
MutableSlice prepare_write() override {
77
return writer_.buf_.prepare_append(shared_.options_.chunk_size);
79
MutableSlice prepare_write_at_least(size_t size) override {
80
return writer_.buf_.prepare_append_at_least(size);
82
void confirm_write(size_t size) override {
83
writer_.buf_.confirm_append(size);
84
writer_.size_ += size;
86
void append(Slice data) override {
87
writer_.buf_.append(data, shared_.options_.chunk_size);
88
writer_.size_ += data.size();
90
void append(BufferSlice data) override {
91
writer_.size_ += data.size();
92
writer_.buf_.append(std::move(data));
94
void append(std::string data) override {
97
void close_writer(Status error) override {
98
CHECK(!writer_.is_closed_);
99
writer_.status_ = std::move(error);
100
writer_.is_closed_.store(true, std::memory_order_release);
102
bool is_reader_closed() const override {
103
return reader_.is_closed_.load(std::memory_order_acquire);
105
Status &reader_status() override {
106
CHECK(is_reader_closed());
107
return reader_.status_;
118
ChainBufferReader buf_;
119
std::atomic<bool> is_closed_{false};
121
std::vector<IoSlice> io_slices_;
127
ChainBufferWriter buf_;
128
std::atomic<bool> is_closed_{false};
135
std::pair<ChainBuffer::Reader, ChainBuffer::Writer> ChainBuffer::create(Options options) {
136
auto impl = std::make_shared<detail::ChainBuffer>(options);
137
return {Reader(impl), Writer(impl)};