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
|
// Copyright 2025 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#ifdef UNSAFE_BUFFERS_BUILD
// TODO(crbug.com/40285824): Remove this and convert code to safer constructs.
#pragma allow_unsafe_buffers
#endif
#include "chrome/browser/actor/aggregated_journal_file_serializer.h"
#include "base/containers/span.h"
#include "base/files/file_util.h"
#include "base/task/task_traits.h"
#include "base/task/thread_pool.h"
#include "third_party/perfetto/include/perfetto/ext/tracing/core/trace_packet.h"
namespace actor {
// An object that is sequence bound to a background pool worker. This
// allows us to implement the blocking IO work on the background task.
class AggregatedJournalFileSerializer::FileWriter {
public:
FileWriter() = default;
bool Init(const base::FilePath& file_path) {
file_handle_ = std::make_unique<base::File>(
file_path, base::File::FLAG_CREATE_ALWAYS | base::File::FLAG_WRITE);
if (!file_handle_->IsValid()) {
return false;
}
return true;
}
void WritePacket(std::vector<uint8_t> message) {
perfetto::TracePacket packet;
packet.AddSlice(message.data(), message.size());
auto [preamble, preamble_size] = packet.GetProtoPreamble();
if (!file_handle_->WriteAtCurrentPosAndCheck(base::span(
reinterpret_cast<const uint8_t*>(preamble), preamble_size))) {
return;
}
for (const perfetto::Slice& slice : packet.slices()) {
if (!file_handle_->WriteAtCurrentPosAndCheck(base::span(
static_cast<const uint8_t*>(slice.start), slice.size))) {
return;
}
}
}
void Shutdown() { file_handle_.reset(); }
private:
std::unique_ptr<base::File> file_handle_;
};
AggregatedJournalFileSerializer::AggregatedJournalFileSerializer(
AggregatedJournal& journal)
: AggregatedJournalSerializer(journal) {}
AggregatedJournalFileSerializer::~AggregatedJournalFileSerializer() {
file_writer_.Reset();
}
void AggregatedJournalFileSerializer::Init(const base::FilePath& file_path,
InitResult callback) {
file_writer_ = base::SequenceBound<FileWriter>(
base::ThreadPool::CreateSequencedTaskRunnerForResource(
{base::MayBlock(), base::TaskPriority::USER_BLOCKING,
base::TaskShutdownBehavior::BLOCK_SHUTDOWN},
file_path));
file_writer_.AsyncCall(&FileWriter::Init)
.WithArgs(file_path)
.Then(base::BindOnce(&AggregatedJournalFileSerializer::InitDone,
weak_ptr_factory_.GetWeakPtr(),
std::move(callback)));
}
void AggregatedJournalFileSerializer::InitDone(InitResult callback,
bool success) {
if (success) {
InitImpl();
}
std::move(callback).Run(success);
}
void AggregatedJournalFileSerializer::Shutdown(base::OnceClosure callback) {
file_writer_.AsyncCall(&FileWriter::Shutdown).Then(std::move(callback));
}
void AggregatedJournalFileSerializer::WriteTracePacket(
std::vector<uint8_t> message) {
file_writer_.AsyncCall(&FileWriter::WritePacket).WithArgs(std::move(message));
}
} // namespace actor
|
