1 /*
2 * Copyright (C) 2020 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "SerializedFlushToState.h"
18
19 #include <limits>
20
21 #include <android-base/logging.h>
22
SerializedFlushToState(uint64_t start,LogMask log_mask,std::list<SerializedLogChunk> * logs)23 SerializedFlushToState::SerializedFlushToState(uint64_t start, LogMask log_mask,
24 std::list<SerializedLogChunk>* logs)
25 : FlushToState(start, log_mask), logs_(logs) {
26 log_id_for_each(i) {
27 if (((1 << i) & log_mask) == 0) {
28 continue;
29 }
30 logs_needed_from_next_position_[i] = true;
31 }
32 }
33
~SerializedFlushToState()34 SerializedFlushToState::~SerializedFlushToState() {
35 log_id_for_each(i) {
36 if (log_positions_[i]) {
37 log_positions_[i]->buffer_it->DetachReader(this);
38 }
39 }
40 }
41
CreateLogPosition(log_id_t log_id)42 void SerializedFlushToState::CreateLogPosition(log_id_t log_id) {
43 CHECK(!logs_[log_id].empty());
44 LogPosition log_position;
45 auto it = logs_[log_id].begin();
46 while (it != logs_[log_id].end() && start() > it->highest_sequence_number()) {
47 ++it;
48 }
49 if (it == logs_[log_id].end()) {
50 --it;
51 }
52 it->AttachReader(this);
53 log_position.buffer_it = it;
54
55 // Find the offset of the first log with sequence number >= start().
56 int read_offset = 0;
57 while (read_offset < it->write_offset()) {
58 const auto* entry = it->log_entry(read_offset);
59 if (entry->sequence() >= start()) {
60 break;
61 }
62 read_offset += entry->total_len();
63 }
64 log_position.read_offset = read_offset;
65
66 log_positions_[log_id].emplace(log_position);
67 }
68
UpdateLogsNeeded(log_id_t log_id)69 void SerializedFlushToState::UpdateLogsNeeded(log_id_t log_id) {
70 auto& buffer_it = log_positions_[log_id]->buffer_it;
71 auto read_offset = log_positions_[log_id]->read_offset;
72
73 // If there is another log to read in this buffer, let it be read.
74 if (read_offset < buffer_it->write_offset()) {
75 logs_needed_from_next_position_[log_id] = false;
76 } else if (read_offset == buffer_it->write_offset()) {
77 // If there are no more logs to read in this buffer and it's the last buffer, then
78 // set logs_needed_from_next_position_ to wait until more logs get logged.
79 if (buffer_it == std::prev(logs_[log_id].end())) {
80 logs_needed_from_next_position_[log_id] = true;
81 } else {
82 // Otherwise, if there is another buffer piece, move to that and do the same check.
83 buffer_it->DetachReader(this);
84 ++buffer_it;
85 buffer_it->AttachReader(this);
86 log_positions_[log_id]->read_offset = 0;
87 if (buffer_it->write_offset() == 0) {
88 logs_needed_from_next_position_[log_id] = true;
89 } else {
90 logs_needed_from_next_position_[log_id] = false;
91 }
92 }
93 } else {
94 // read_offset > buffer_it->write_offset() should never happen.
95 LOG(FATAL) << "read_offset (" << read_offset << ") > buffer_it->write_offset() ("
96 << buffer_it->write_offset() << ")";
97 }
98 }
99
CheckForNewLogs()100 void SerializedFlushToState::CheckForNewLogs() {
101 log_id_for_each(i) {
102 if (!logs_needed_from_next_position_[i]) {
103 continue;
104 }
105 if (!log_positions_[i]) {
106 if (logs_[i].empty()) {
107 continue;
108 }
109 CreateLogPosition(i);
110 }
111 UpdateLogsNeeded(i);
112 }
113 }
114
HasUnreadLogs()115 bool SerializedFlushToState::HasUnreadLogs() {
116 CheckForNewLogs();
117 log_id_for_each(i) {
118 if (log_positions_[i] && !logs_needed_from_next_position_[i]) {
119 return true;
120 }
121 }
122 return false;
123 }
124
PopNextUnreadLog()125 LogWithId SerializedFlushToState::PopNextUnreadLog() {
126 uint64_t min_sequence = std::numeric_limits<uint64_t>::max();
127 log_id_t log_id;
128 const SerializedLogEntry* entry = nullptr;
129 log_id_for_each(i) {
130 if (!log_positions_[i] || logs_needed_from_next_position_[i]) {
131 continue;
132 }
133 if (log_positions_[i]->log_entry()->sequence() < min_sequence) {
134 log_id = i;
135 entry = log_positions_[i]->log_entry();
136 min_sequence = entry->sequence();
137 }
138 }
139 CHECK_NE(nullptr, entry);
140
141 log_positions_[log_id]->read_offset += entry->total_len();
142
143 logs_needed_from_next_position_[log_id] = true;
144
145 return {log_id, entry};
146 }
147
Prune(log_id_t log_id)148 void SerializedFlushToState::Prune(log_id_t log_id) {
149 CHECK(log_positions_[log_id].has_value());
150
151 // Decrease the ref count since we're deleting our reference.
152 log_positions_[log_id]->buffer_it->DetachReader(this);
153
154 // Delete in the reference.
155 log_positions_[log_id].reset();
156
157 // Finally set logs_needed_from_next_position_, so CheckForNewLogs() will re-create the
158 // log_position_ object during the next read.
159 logs_needed_from_next_position_[log_id] = true;
160 }
161