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 <android-base/logging.h>
20
SerializedFlushToState(uint64_t start,LogMask log_mask)21 SerializedFlushToState::SerializedFlushToState(uint64_t start, LogMask log_mask)
22 : FlushToState(start, log_mask) {
23 log_id_for_each(i) {
24 if (((1 << i) & log_mask) == 0) {
25 continue;
26 }
27 logs_needed_from_next_position_[i] = true;
28 }
29 }
30
~SerializedFlushToState()31 SerializedFlushToState::~SerializedFlushToState() {
32 log_id_for_each(i) {
33 if (log_positions_[i]) {
34 log_positions_[i]->buffer_it->DecReaderRefCount();
35 }
36 }
37 }
38
CreateLogPosition(log_id_t log_id)39 void SerializedFlushToState::CreateLogPosition(log_id_t log_id) {
40 CHECK(!logs_[log_id].empty());
41 LogPosition log_position;
42 auto it = logs_[log_id].begin();
43 while (it != logs_[log_id].end() && start() > it->highest_sequence_number()) {
44 ++it;
45 }
46 if (it == logs_[log_id].end()) {
47 --it;
48 }
49 it->IncReaderRefCount();
50 log_position.buffer_it = it;
51
52 // Find the offset of the first log with sequence number >= start().
53 int read_offset = 0;
54 while (read_offset < it->write_offset()) {
55 const auto* entry = it->log_entry(read_offset);
56 if (entry->sequence() >= start()) {
57 break;
58 }
59 read_offset += entry->total_len();
60 }
61 log_position.read_offset = read_offset;
62
63 log_positions_[log_id].emplace(log_position);
64 }
65
AddMinHeapEntry(log_id_t log_id)66 void SerializedFlushToState::AddMinHeapEntry(log_id_t log_id) {
67 auto& buffer_it = log_positions_[log_id]->buffer_it;
68 auto read_offset = log_positions_[log_id]->read_offset;
69
70 // If there is another log to read in this buffer, add it to the min heap.
71 if (read_offset < buffer_it->write_offset()) {
72 auto* entry = buffer_it->log_entry(read_offset);
73 min_heap_.emplace(log_id, entry);
74 } else if (read_offset == buffer_it->write_offset()) {
75 // If there are no more logs to read in this buffer and it's the last buffer, then
76 // set logs_needed_from_next_position_ to wait until more logs get logged.
77 if (buffer_it == std::prev(logs_[log_id].end())) {
78 logs_needed_from_next_position_[log_id] = true;
79 } else {
80 // Otherwise, if there is another buffer piece, move to that and do the same check.
81 buffer_it->DecReaderRefCount();
82 ++buffer_it;
83 buffer_it->IncReaderRefCount();
84 log_positions_[log_id]->read_offset = 0;
85 if (buffer_it->write_offset() == 0) {
86 logs_needed_from_next_position_[log_id] = true;
87 } else {
88 auto* entry = buffer_it->log_entry(0);
89 min_heap_.emplace(log_id, entry);
90 }
91 }
92 } else {
93 // read_offset > buffer_it->write_offset() should never happen.
94 CHECK(false);
95 }
96 }
97
CheckForNewLogs()98 void SerializedFlushToState::CheckForNewLogs() {
99 log_id_for_each(i) {
100 if (!logs_needed_from_next_position_[i]) {
101 continue;
102 }
103 if (!log_positions_[i]) {
104 if (logs_[i].empty()) {
105 continue;
106 }
107 CreateLogPosition(i);
108 }
109 logs_needed_from_next_position_[i] = false;
110 // If it wasn't possible to insert, logs_needed_from_next_position will be set back to true.
111 AddMinHeapEntry(i);
112 }
113 }
114
PopNextUnreadLog()115 MinHeapElement SerializedFlushToState::PopNextUnreadLog() {
116 auto top = min_heap_.top();
117 min_heap_.pop();
118
119 auto* entry = top.entry;
120 auto log_id = top.log_id;
121
122 log_positions_[log_id]->read_offset += entry->total_len();
123
124 logs_needed_from_next_position_[log_id] = true;
125
126 return top;
127 }
128
Prune(log_id_t log_id,const std::list<SerializedLogChunk>::iterator & buffer_it)129 void SerializedFlushToState::Prune(log_id_t log_id,
130 const std::list<SerializedLogChunk>::iterator& buffer_it) {
131 // If we don't have a position for this log or if we're not referencing buffer_it, ignore.
132 if (!log_positions_[log_id].has_value() || log_positions_[log_id]->buffer_it != buffer_it) {
133 return;
134 }
135
136 // // Decrease the ref count since we're deleting our reference.
137 buffer_it->DecReaderRefCount();
138
139 // Delete in the reference.
140 log_positions_[log_id].reset();
141
142 // Remove the MinHeapElement referencing log_id, if it exists, but retain the others.
143 std::vector<MinHeapElement> old_elements;
144 while (!min_heap_.empty()) {
145 auto& element = min_heap_.top();
146 if (element.log_id != log_id) {
147 old_elements.emplace_back(element);
148 }
149 min_heap_.pop();
150 }
151 for (auto&& element : old_elements) {
152 min_heap_.emplace(element);
153 }
154
155 // Finally set logs_needed_from_next_position_, so CheckForNewLogs() will re-create the
156 // log_position_ object during the next read.
157 logs_needed_from_next_position_[log_id] = true;
158 }
159