b07e339b53
SerializedFlushToState::PopNextUnreadLog() was calling AddMinHeapEntry() to replenish the element that was just popped off of the heap, however AddMinHeapEntry() also manages reference counts for the buffers, and this resulting in the following scenario: PopNextUnreadLog() returns a pointer referencing log buffer #1 AddMinHeapEntry() sees that all logs from buffer #1 has been read, so it decrements the reference count The caller of PopNextUnreadLog() uses the result which references invalid memory. This calls CheckForNewLogs() within HasUnreadLogs() instead of requiring a separate call, which fixes an additional issue where continuing from the loop in SerializedLogBuffer::FlushTo() may not pick up subsequent logs in a given log buffer, since CheckForNewLogs() wouldn't be called. This was exacerbated by the above change. This adds a test to check the reference counts for this case and fixes an argument mismatch in SerializedFlushToStateTest. This adds the corpus that surfaced the issue. Bug: 159753229 Bug: 159783005 Test: these unit tests, run fuzzer without error Change-Id: Ib2636dfc14293b7e2cd00876b9def6e9dbbff4ce
103 lines
4.5 KiB
C++
103 lines
4.5 KiB
C++
/*
|
|
* Copyright (C) 2020 The Android Open Source Project
|
|
*
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
* you may not use this file except in compliance with the License.
|
|
* You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
* See the License for the specific language governing permissions and
|
|
* limitations under the License.
|
|
*/
|
|
|
|
#pragma once
|
|
|
|
#include <bitset>
|
|
#include <list>
|
|
#include <queue>
|
|
|
|
#include "LogBuffer.h"
|
|
#include "SerializedLogChunk.h"
|
|
#include "SerializedLogEntry.h"
|
|
|
|
struct LogPosition {
|
|
std::list<SerializedLogChunk>::iterator buffer_it;
|
|
int read_offset;
|
|
};
|
|
|
|
struct MinHeapElement {
|
|
MinHeapElement(log_id_t log_id, const SerializedLogEntry* entry)
|
|
: log_id(log_id), entry(entry) {}
|
|
log_id_t log_id;
|
|
const SerializedLogEntry* entry;
|
|
// The change of comparison operators is intentional, std::priority_queue uses operator<() to
|
|
// compare but creates a max heap. Since we want a min heap, we return the opposite result.
|
|
bool operator<(const MinHeapElement& rhs) const {
|
|
return entry->sequence() > rhs.entry->sequence();
|
|
}
|
|
};
|
|
|
|
// This class tracks the specific point where a FlushTo client has read through the logs. It
|
|
// directly references the std::list<> iterators from the parent SerializedLogBuffer and the offset
|
|
// into each log chunk where it has last read. All interactions with this class, except for its
|
|
// construction, must be done with SerializedLogBuffer::lock_ held. No log chunks that it
|
|
// references may be pruned, which is handled by ensuring prune does not touch any log chunk with
|
|
// highest sequence number greater or equal to start().
|
|
class SerializedFlushToState : public FlushToState {
|
|
public:
|
|
// Initializes this state object. For each log buffer set in log_mask, this sets
|
|
// logs_needed_from_next_position_.
|
|
SerializedFlushToState(uint64_t start, LogMask log_mask);
|
|
|
|
// Decrease the reference of all referenced logs. This happens when a reader is disconnected.
|
|
~SerializedFlushToState() override;
|
|
|
|
// We can't hold SerializedLogBuffer::lock_ in the constructor, so we must initialize logs here.
|
|
void InitializeLogs(std::list<SerializedLogChunk>* logs) {
|
|
if (logs_ == nullptr) logs_ = logs;
|
|
}
|
|
|
|
bool HasUnreadLogs() {
|
|
CheckForNewLogs();
|
|
return !min_heap_.empty();
|
|
}
|
|
|
|
// Pops the next unread log from the min heap and sets logs_needed_from_next_position_ to
|
|
// indicate that we're waiting for more logs from the associated log buffer.
|
|
MinHeapElement PopNextUnreadLog();
|
|
|
|
// If the parent log buffer prunes logs, the reference that this class contains may become
|
|
// invalid, so this must be called first to drop the reference to buffer_it, if any.
|
|
void Prune(log_id_t log_id, const std::list<SerializedLogChunk>::iterator& buffer_it);
|
|
|
|
private:
|
|
// If there is a log in the serialized log buffer for `log_id` at the read_offset, add it to the
|
|
// min heap for reading, otherwise set logs_needed_from_next_position_ to indicate that we're
|
|
// waiting for the next log.
|
|
void AddMinHeapEntry(log_id_t log_id);
|
|
|
|
// Create a LogPosition object for the given log_id by searching through the log chunks for the
|
|
// first chunk and then first log entry within that chunk that is greater or equal to start().
|
|
void CreateLogPosition(log_id_t log_id);
|
|
|
|
// Checks to see if any log buffers set in logs_needed_from_next_position_ have new logs and
|
|
// calls AddMinHeapEntry() if so.
|
|
void CheckForNewLogs();
|
|
|
|
std::list<SerializedLogChunk>* logs_ = nullptr;
|
|
// An optional structure that contains an iterator to the serialized log buffer and offset into
|
|
// it that this logger should handle next.
|
|
std::optional<LogPosition> log_positions_[LOG_ID_MAX];
|
|
// A bit for each log that is set if a given log_id has no logs or if this client has read all
|
|
// of its logs. In order words: `logs_[i].empty() || (buffer_it == std::prev(logs_.end) &&
|
|
// next_log_position == logs_write_position_)`. These will be re-checked in each
|
|
// loop in case new logs came in.
|
|
std::bitset<LOG_ID_MAX> logs_needed_from_next_position_ = {};
|
|
// A min heap that has up to one entry per log buffer, sorted by sequence number, of the next
|
|
// element that this reader should read.
|
|
std::priority_queue<MinHeapElement> min_heap_;
|
|
};
|