logd: add a SerializedLogBuffer suitable for compression
Initial commit for a SerializedLogBuffer. The intention here is for
the serialized data to be compressed (currently using zlib) to allow
for substantially longer logs in the same memory footprint.
Test: unit tests
Change-Id: I2528e4e1ff1cf3bc91130173a107f371f04d911a
diff --git a/logd/SerializedLogBuffer.cpp b/logd/SerializedLogBuffer.cpp
new file mode 100644
index 0000000..8bda7cf
--- /dev/null
+++ b/logd/SerializedLogBuffer.cpp
@@ -0,0 +1,351 @@
+/*
+ * 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.
+ */
+
+#include "SerializedLogBuffer.h"
+
+#include <limits>
+#include <thread>
+
+#include <android-base/logging.h>
+#include <android-base/scopeguard.h>
+
+#include "LogStatistics.h"
+#include "SerializedFlushToState.h"
+
+SerializedLogBuffer::SerializedLogBuffer(LogReaderList* reader_list, LogTags* tags,
+ LogStatistics* stats)
+ : reader_list_(reader_list), tags_(tags), stats_(stats) {
+ Init();
+}
+
+SerializedLogBuffer::~SerializedLogBuffer() {}
+
+void SerializedLogBuffer::Init() {
+ log_id_for_each(i) {
+ if (SetSize(i, __android_logger_get_buffer_size(i))) {
+ SetSize(i, LOG_BUFFER_MIN_SIZE);
+ }
+ }
+
+ // Release any sleeping reader threads to dump their current content.
+ auto reader_threads_lock = std::lock_guard{reader_list_->reader_threads_lock()};
+ for (const auto& reader_thread : reader_list_->reader_threads()) {
+ reader_thread->triggerReader_Locked();
+ }
+}
+
+bool SerializedLogBuffer::ShouldLog(log_id_t log_id, const char* msg, uint16_t len) {
+ if (log_id == LOG_ID_SECURITY) {
+ return true;
+ }
+
+ int prio = ANDROID_LOG_INFO;
+ const char* tag = nullptr;
+ size_t tag_len = 0;
+ if (IsBinary(log_id)) {
+ int32_t tag_int = MsgToTag(msg, len);
+ tag = tags_->tagToName(tag_int);
+ if (tag) {
+ tag_len = strlen(tag);
+ }
+ } else {
+ prio = *msg;
+ tag = msg + 1;
+ tag_len = strnlen(tag, len - 1);
+ }
+ return __android_log_is_loggable_len(prio, tag, tag_len, ANDROID_LOG_VERBOSE);
+}
+
+int SerializedLogBuffer::Log(log_id_t log_id, log_time realtime, uid_t uid, pid_t pid, pid_t tid,
+ const char* msg, uint16_t len) {
+ if (log_id >= LOG_ID_MAX || len == 0) {
+ return -EINVAL;
+ }
+
+ if (!ShouldLog(log_id, msg, len)) {
+ stats_->AddTotal(log_id, len);
+ return -EACCES;
+ }
+
+ auto sequence = sequence_.fetch_add(1, std::memory_order_relaxed);
+
+ auto lock = std::lock_guard{lock_};
+
+ if (logs_[log_id].empty()) {
+ logs_[log_id].push_back(SerializedLogChunk(max_size_[log_id] / 4));
+ }
+
+ auto total_len = sizeof(SerializedLogEntry) + len;
+ if (!logs_[log_id].back().CanLog(total_len)) {
+ logs_[log_id].back().FinishWriting();
+ logs_[log_id].push_back(SerializedLogChunk(max_size_[log_id] / 4));
+ }
+
+ auto entry = logs_[log_id].back().Log(sequence, realtime, uid, pid, tid, msg, len);
+ stats_->Add(entry->ToLogStatisticsElement(log_id));
+
+ MaybePrune(log_id);
+
+ reader_list_->NotifyNewLog(1 << log_id);
+ return len;
+}
+
+void SerializedLogBuffer::MaybePrune(log_id_t log_id) {
+ auto get_total_size = [](const auto& buffer) {
+ size_t total_size = 0;
+ for (const auto& chunk : buffer) {
+ total_size += chunk.PruneSize();
+ }
+ return total_size;
+ };
+ size_t total_size = get_total_size(logs_[log_id]);
+ if (total_size > max_size_[log_id]) {
+ Prune(log_id, total_size - max_size_[log_id], 0);
+ LOG(INFO) << "Pruned Logs from log_id: " << log_id << ", previous size: " << total_size
+ << " after size: " << get_total_size(logs_[log_id]);
+ }
+}
+
+// Decompresses the chunks, call LogStatistics::Subtract() on each entry, then delete the chunks and
+// the list. Note that the SerializedLogChunk objects have been removed from logs_ and their
+// references have been deleted from any SerializedFlushToState objects, so this can be safely done
+// without holding lock_. It is done in a separate thread to avoid delaying the writer thread. The
+// lambda for the thread takes ownership of the 'chunks' list and thus when the thread ends and the
+// lambda is deleted, the objects are deleted.
+void SerializedLogBuffer::DeleteLogChunks(std::list<SerializedLogChunk>&& chunks, log_id_t log_id) {
+ auto delete_thread = std::thread{[chunks = std::move(chunks), log_id, this]() mutable {
+ for (auto& chunk : chunks) {
+ chunk.IncReaderRefCount();
+ int read_offset = 0;
+ while (read_offset < chunk.write_offset()) {
+ auto* entry = chunk.log_entry(read_offset);
+ stats_->Subtract(entry->ToLogStatisticsElement(log_id));
+ read_offset += entry->total_len();
+ }
+ chunk.DecReaderRefCount(false);
+ }
+ }};
+ delete_thread.detach();
+}
+
+void SerializedLogBuffer::NotifyReadersOfPrune(
+ log_id_t log_id, const std::list<SerializedLogChunk>::iterator& chunk) {
+ for (const auto& reader_thread : reader_list_->reader_threads()) {
+ auto& state = reinterpret_cast<SerializedFlushToState&>(reader_thread->flush_to_state());
+ state.Prune(log_id, chunk);
+ }
+}
+
+bool SerializedLogBuffer::Prune(log_id_t log_id, size_t bytes_to_free, uid_t uid) {
+ // Don't prune logs that are newer than the point at which any reader threads are reading from.
+ LogReaderThread* oldest = nullptr;
+ auto reader_threads_lock = std::lock_guard{reader_list_->reader_threads_lock()};
+ for (const auto& reader_thread : reader_list_->reader_threads()) {
+ if (!reader_thread->IsWatching(log_id)) {
+ continue;
+ }
+ if (!oldest || oldest->start() > reader_thread->start() ||
+ (oldest->start() == reader_thread->start() &&
+ reader_thread->deadline().time_since_epoch().count() != 0)) {
+ oldest = reader_thread.get();
+ }
+ }
+
+ auto& log_buffer = logs_[log_id];
+
+ std::list<SerializedLogChunk> chunks_to_prune;
+ auto prune_chunks = android::base::make_scope_guard([&chunks_to_prune, log_id, this] {
+ DeleteLogChunks(std::move(chunks_to_prune), log_id);
+ });
+
+ auto it = log_buffer.begin();
+ while (it != log_buffer.end()) {
+ if (oldest != nullptr && it->highest_sequence_number() >= oldest->start()) {
+ break;
+ }
+
+ // Increment ahead of time since we're going to splice this iterator from the list.
+ auto it_to_prune = it++;
+
+ // The sequence number check ensures that all readers have read all logs in this chunk, but
+ // they may still hold a reference to the chunk to track their last read log_position.
+ // Notify them to delete the reference.
+ NotifyReadersOfPrune(log_id, it_to_prune);
+
+ if (uid != 0) {
+ // Reorder the log buffer to remove logs from the given UID. If there are no logs left
+ // in the buffer after the removal, delete it.
+ if (it_to_prune->ClearUidLogs(uid, log_id, stats_)) {
+ log_buffer.erase(it_to_prune);
+ }
+ } else {
+ size_t buffer_size = it_to_prune->PruneSize();
+ chunks_to_prune.splice(chunks_to_prune.end(), log_buffer, it_to_prune);
+ if (buffer_size >= bytes_to_free) {
+ return true;
+ }
+ bytes_to_free -= buffer_size;
+ }
+ }
+
+ // If we've deleted all buffers without bytes_to_free hitting 0, then we're called by Clear()
+ // and should return true.
+ if (it == log_buffer.end()) {
+ return true;
+ }
+
+ // Otherwise we are stuck due to a reader, so mitigate it.
+ KickReader(oldest, log_id, bytes_to_free);
+ return false;
+}
+
+// If the selected reader is blocking our pruning progress, decide on
+// what kind of mitigation is necessary to unblock the situation.
+void SerializedLogBuffer::KickReader(LogReaderThread* reader, log_id_t id, size_t bytes_to_free) {
+ if (bytes_to_free >= max_size_[id]) { // +100%
+ // A misbehaving or slow reader is dropped if we hit too much memory pressure.
+ LOG(WARNING) << "Kicking blocked reader, " << reader->name()
+ << ", from LogBuffer::kickMe()";
+ reader->release_Locked();
+ } else if (reader->deadline().time_since_epoch().count() != 0) {
+ // Allow a blocked WRAP deadline reader to trigger and start reporting the log data.
+ reader->triggerReader_Locked();
+ } else {
+ // Tell slow reader to skip entries to catch up.
+ unsigned long prune_rows = bytes_to_free / 300;
+ LOG(WARNING) << "Skipping " << prune_rows << " entries from slow reader, " << reader->name()
+ << ", from LogBuffer::kickMe()";
+ reader->triggerSkip_Locked(id, prune_rows);
+ }
+}
+
+std::unique_ptr<FlushToState> SerializedLogBuffer::CreateFlushToState(uint64_t start,
+ LogMask log_mask) {
+ return std::make_unique<SerializedFlushToState>(start, log_mask);
+}
+
+bool SerializedLogBuffer::FlushTo(
+ LogWriter* writer, FlushToState& abstract_state,
+ const std::function<FilterResult(log_id_t log_id, pid_t pid, uint64_t sequence,
+ log_time realtime)>& filter) {
+ auto lock = std::unique_lock{lock_};
+
+ auto& state = reinterpret_cast<SerializedFlushToState&>(abstract_state);
+ state.InitializeLogs(logs_);
+ state.CheckForNewLogs();
+
+ while (state.HasUnreadLogs()) {
+ MinHeapElement top = state.PopNextUnreadLog();
+ auto* entry = top.entry;
+ auto log_id = top.log_id;
+
+ if (entry->sequence() < state.start()) {
+ continue;
+ }
+ state.set_start(entry->sequence());
+
+ if (!writer->privileged() && entry->uid() != writer->uid()) {
+ continue;
+ }
+
+ if (filter) {
+ auto ret = filter(log_id, entry->pid(), entry->sequence(), entry->realtime());
+ if (ret == FilterResult::kSkip) {
+ continue;
+ }
+ if (ret == FilterResult::kStop) {
+ break;
+ }
+ }
+
+ lock.unlock();
+ // We never prune logs equal to or newer than any LogReaderThreads' `start` value, so the
+ // `entry` pointer is safe here without the lock
+ if (!entry->Flush(writer, log_id)) {
+ return false;
+ }
+ lock.lock();
+
+ // Since we released the log above, buffers that aren't in our min heap may now have had
+ // logs added, so re-check them.
+ state.CheckForNewLogs();
+ }
+
+ state.set_start(state.start() + 1);
+ return true;
+}
+
+bool SerializedLogBuffer::Clear(log_id_t id, uid_t uid) {
+ // Try three times to clear, then disconnect the readers and try one final time.
+ for (int retry = 0; retry < 3; ++retry) {
+ {
+ auto lock = std::lock_guard{lock_};
+ bool prune_success = Prune(id, ULONG_MAX, uid);
+ if (prune_success) {
+ return true;
+ }
+ }
+ sleep(1);
+ }
+ // Check if it is still busy after the sleep, we try to prune one entry, not another clear run,
+ // so we are looking for the quick side effect of the return value to tell us if we have a
+ // _blocked_ reader.
+ bool busy = false;
+ {
+ auto lock = std::lock_guard{lock_};
+ busy = !Prune(id, 1, uid);
+ }
+ // It is still busy, disconnect all readers.
+ if (busy) {
+ auto reader_threads_lock = std::lock_guard{reader_list_->reader_threads_lock()};
+ for (const auto& reader_thread : reader_list_->reader_threads()) {
+ if (reader_thread->IsWatching(id)) {
+ LOG(WARNING) << "Kicking blocked reader, " << reader_thread->name()
+ << ", from LogBuffer::clear()";
+ reader_thread->release_Locked();
+ }
+ }
+ }
+ auto lock = std::lock_guard{lock_};
+ return Prune(id, ULONG_MAX, uid);
+}
+
+unsigned long SerializedLogBuffer::GetSize(log_id_t id) {
+ auto lock = std::lock_guard{lock_};
+ size_t retval = 2 * max_size_[id] / 3; // See the comment in SetSize().
+ return retval;
+}
+
+// New SerializedLogChunk objects will be allocated according to the new size, but older one are
+// unchanged. MaybePrune() is called on the log buffer to reduce it to an appropriate size if the
+// new size is lower.
+// ChattyLogBuffer/SimpleLogBuffer don't consider the 'Overhead' of LogBufferElement or the
+// std::list<> overhead as part of the log size. SerializedLogBuffer does by its very nature, so
+// the 'size' metric is not compatible between them. Their actual memory usage is between 1.5x and
+// 2x of what they claim to use, so we conservatively set our internal size as size + size / 2.
+int SerializedLogBuffer::SetSize(log_id_t id, unsigned long size) {
+ // Reasonable limits ...
+ if (!__android_logger_valid_buffer_size(size)) {
+ return -1;
+ }
+
+ auto lock = std::lock_guard{lock_};
+ max_size_[id] = size + size / 2;
+
+ MaybePrune(id);
+
+ return 0;
+}