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
|
#include "cru/common/io/BufferStream.h"
#include "cru/common/io/Stream.h"
namespace cru::io {
BufferStream::BufferStream(const BufferStreamOptions& options) {
block_size_ =
options.block_size <= 0 ? kDefaultBlockSize : options.block_size;
total_size_limit_ = options.total_size_limit < 0 ? kDefaultTotalSizeLimit
: options.total_size_limit;
block_count_limit_ = total_size_limit_ / block_size_;
eof_ = false;
}
BufferStream::~BufferStream() {}
bool BufferStream::CanSeek() { return false; }
Index BufferStream::Seek(Index offset, SeekOrigin origin) {
throw StreamOperationNotSupportedException(
u"BufferStream does not support seeking.");
}
bool BufferStream::CanRead() { return true; }
Index BufferStream::Read(std::byte* buffer, Index offset, Index size) {
std::unique_lock lock(mutex_);
condition_variable_.wait(lock,
[this] { return !buffer_list_.empty() || eof_; });
if (buffer_list_.empty() && eof_) {
return 0;
}
auto full = buffer_list_.size() == block_count_limit_;
Index read = 0;
while (!buffer_list_.empty()) {
auto& stream_buffer = buffer_list_.front();
auto this_read =
stream_buffer.PopFront(buffer + offset + read, size - read);
if (stream_buffer.GetUsedSize() == 0) {
buffer_list_.pop_front();
}
read += this_read;
if (read == size) {
break;
}
}
if (full && buffer_list_.size() < block_count_limit_) {
// By convention, there should be at most one producer waiting. So
// notify_one and notify_all should be the same.
condition_variable_.notify_one();
}
return read;
}
bool BufferStream::CanWrite() { return true; }
Index BufferStream::Write(const std::byte* buffer, Index offset, Index size) {
std::unique_lock lock(mutex_);
if (eof_) {
throw WriteAfterEofException(
u"Stream has been set eof. Can't write to it any more.");
}
condition_variable_.wait(lock, [this] {
return buffer_list_.size() < block_count_limit_ ||
buffer_list_.back().GetBackFree() > 0;
});
auto empty = buffer_list_.empty();
Index written = 0;
while (buffer_list_.size() != block_count_limit_) {
if (buffer_list_.back().GetBackFree() == 0) {
buffer_list_.push_back(Buffer(block_size_));
}
auto& stream_buffer = buffer_list_.back();
auto this_written =
stream_buffer.PushBack(buffer + offset + written, size - written);
written += this_written;
if (written == size) {
break;
}
}
if (empty) {
// By convention, there should be at most one consumer waiting. So
// notify_one and notify_all should be the same.
condition_variable_.notify_one();
}
return written;
}
void BufferStream::SetEof() {
std::unique_lock lock(mutex_);
eof_ = true;
if (buffer_list_.empty()) {
// By convention, there should be at most one consumer waiting. So
// notify_one and notify_all should be the same.
condition_variable_.notify_one();
}
}
} // namespace cru::io
|
