1 /*
2 * Copyright (C) 2019 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 "perfetto/trace_processor/read_trace.h"
18
19 #include "perfetto/base/logging.h"
20 #include "perfetto/ext/base/file_utils.h"
21 #include "perfetto/ext/base/scoped_file.h"
22 #include "perfetto/ext/base/utils.h"
23 #include "perfetto/protozero/proto_utils.h"
24 #include "perfetto/trace_processor/trace_processor.h"
25
26 #include "src/trace_processor/forwarding_trace_parser.h"
27 #include "src/trace_processor/importers/gzip/gzip_trace_parser.h"
28 #include "src/trace_processor/importers/gzip/gzip_utils.h"
29 #include "src/trace_processor/importers/proto/proto_trace_tokenizer.h"
30 #include "src/trace_processor/util/status_macros.h"
31
32 #include "protos/perfetto/trace/trace.pbzero.h"
33 #include "protos/perfetto/trace/trace_packet.pbzero.h"
34
35 #if PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX) || \
36 PERFETTO_BUILDFLAG(PERFETTO_OS_APPLE)
37 #define PERFETTO_HAS_AIO_H() 1
38 #else
39 #define PERFETTO_HAS_AIO_H() 0
40 #endif
41
42 #if PERFETTO_HAS_AIO_H()
43 #include <aio.h>
44 #endif
45
46 namespace perfetto {
47 namespace trace_processor {
48 namespace {
49
50 // 1MB chunk size seems the best tradeoff on a MacBook Pro 2013 - i7 2.8 GHz.
51 constexpr size_t kChunkSize = 1024 * 1024;
52
ReadTraceUsingRead(TraceProcessor * tp,int fd,uint64_t * file_size,const std::function<void (uint64_t parsed_size)> & progress_callback)53 util::Status ReadTraceUsingRead(
54 TraceProcessor* tp,
55 int fd,
56 uint64_t* file_size,
57 const std::function<void(uint64_t parsed_size)>& progress_callback) {
58 // Load the trace in chunks using ordinary read().
59 for (int i = 0;; i++) {
60 if (progress_callback && i % 128 == 0)
61 progress_callback(*file_size);
62
63 std::unique_ptr<uint8_t[]> buf(new uint8_t[kChunkSize]);
64 auto rsize = base::Read(fd, buf.get(), kChunkSize);
65 if (rsize == 0)
66 break;
67
68 if (rsize < 0) {
69 return util::ErrStatus("Reading trace file failed (errno: %d, %s)", errno,
70 strerror(errno));
71 }
72
73 *file_size += static_cast<uint64_t>(rsize);
74
75 RETURN_IF_ERROR(tp->Parse(std::move(buf), static_cast<size_t>(rsize)));
76 }
77 return util::OkStatus();
78 }
79
80 class SerializingProtoTraceReader : public ChunkedTraceReader {
81 public:
SerializingProtoTraceReader(std::vector<uint8_t> * output)82 SerializingProtoTraceReader(std::vector<uint8_t>* output) : output_(output) {}
83
Parse(std::unique_ptr<uint8_t[]> data,size_t size)84 util::Status Parse(std::unique_ptr<uint8_t[]> data, size_t size) override {
85 return tokenizer_.Tokenize(
86 std::move(data), size, [this](TraceBlobView packet) {
87 uint8_t buffer[protozero::proto_utils::kMaxSimpleFieldEncodedSize];
88
89 uint8_t* pos = buffer;
90 pos = protozero::proto_utils::WriteVarInt(kTracePacketTag, pos);
91 pos = protozero::proto_utils::WriteVarInt(packet.length(), pos);
92 output_->insert(output_->end(), buffer, pos);
93
94 output_->insert(output_->end(), packet.data(),
95 packet.data() + packet.length());
96 return util::OkStatus();
97 });
98 }
99
NotifyEndOfFile()100 void NotifyEndOfFile() override {}
101
102 private:
103 static constexpr uint8_t kTracePacketTag =
104 protozero::proto_utils::MakeTagLengthDelimited(
105 protos::pbzero::Trace::kPacketFieldNumber);
106
107 ProtoTraceTokenizer tokenizer_;
108 std::vector<uint8_t>* output_;
109 };
110
111 } // namespace
112
ReadTrace(TraceProcessor * tp,const char * filename,const std::function<void (uint64_t parsed_size)> & progress_callback)113 util::Status ReadTrace(
114 TraceProcessor* tp,
115 const char* filename,
116 const std::function<void(uint64_t parsed_size)>& progress_callback) {
117 base::ScopedFile fd(base::OpenFile(filename, O_RDONLY));
118 if (!fd)
119 return util::ErrStatus("Could not open trace file (path: %s)", filename);
120
121 uint64_t file_size = 0;
122
123 #if PERFETTO_HAS_AIO_H()
124 // Load the trace in chunks using async IO. We create a simple pipeline where,
125 // at each iteration, we parse the current chunk and asynchronously start
126 // reading the next chunk.
127 struct aiocb cb {};
128 cb.aio_nbytes = kChunkSize;
129 cb.aio_fildes = *fd;
130
131 std::unique_ptr<uint8_t[]> aio_buf(new uint8_t[kChunkSize]);
132 #if defined(MEMORY_SANITIZER)
133 // Just initialize the memory to make the memory sanitizer happy as it
134 // cannot track aio calls below.
135 memset(aio_buf.get(), 0, kChunkSize);
136 #endif // defined(MEMORY_SANITIZER)
137 cb.aio_buf = aio_buf.get();
138
139 PERFETTO_CHECK(aio_read(&cb) == 0);
140 struct aiocb* aio_list[1] = {&cb};
141
142 for (int i = 0;; i++) {
143 if (progress_callback && i % 128 == 0)
144 progress_callback(file_size);
145
146 // Block waiting for the pending read to complete.
147 PERFETTO_CHECK(aio_suspend(aio_list, 1, nullptr) == 0);
148 auto rsize = aio_return(&cb);
149 if (rsize <= 0)
150 break;
151 file_size += static_cast<uint64_t>(rsize);
152
153 // Take ownership of the completed buffer and enqueue a new async read
154 // with a fresh buffer.
155 std::unique_ptr<uint8_t[]> buf(std::move(aio_buf));
156 aio_buf.reset(new uint8_t[kChunkSize]);
157 #if defined(MEMORY_SANITIZER)
158 // Just initialize the memory to make the memory sanitizer happy as it
159 // cannot track aio calls below.
160 memset(aio_buf.get(), 0, kChunkSize);
161 #endif // defined(MEMORY_SANITIZER)
162 cb.aio_buf = aio_buf.get();
163 cb.aio_offset += rsize;
164 PERFETTO_CHECK(aio_read(&cb) == 0);
165
166 // Parse the completed buffer while the async read is in-flight.
167 RETURN_IF_ERROR(tp->Parse(std::move(buf), static_cast<size_t>(rsize)));
168 }
169
170 if (file_size == 0) {
171 PERFETTO_ILOG(
172 "Failed to read any data using AIO. This is expected and not an error "
173 "on WSL. Falling back to read()");
174 RETURN_IF_ERROR(ReadTraceUsingRead(tp, *fd, &file_size, progress_callback));
175 }
176 #else // PERFETTO_HAS_AIO_H()
177 RETURN_IF_ERROR(ReadTraceUsingRead(tp, *fd, &file_size, progress_callback));
178 #endif // PERFETTO_HAS_AIO_H()
179
180 tp->NotifyEndOfFile();
181 tp->SetCurrentTraceName(filename);
182
183 if (progress_callback)
184 progress_callback(file_size);
185 return util::OkStatus();
186 }
187
DecompressTrace(const uint8_t * data,size_t size,std::vector<uint8_t> * output)188 util::Status DecompressTrace(const uint8_t* data,
189 size_t size,
190 std::vector<uint8_t>* output) {
191 TraceType type = GuessTraceType(data, size);
192 if (type != TraceType::kGzipTraceType && type != TraceType::kProtoTraceType) {
193 return util::ErrStatus(
194 "Only GZIP and proto trace types are supported by DecompressTrace");
195 }
196
197 if (type == TraceType::kGzipTraceType) {
198 std::unique_ptr<ChunkedTraceReader> reader(
199 new SerializingProtoTraceReader(output));
200 GzipTraceParser parser(std::move(reader));
201
202 RETURN_IF_ERROR(parser.ParseUnowned(data, size));
203 if (parser.needs_more_input())
204 return util::ErrStatus("Cannot decompress partial trace file");
205
206 parser.NotifyEndOfFile();
207 return util::OkStatus();
208 }
209
210 PERFETTO_CHECK(type == TraceType::kProtoTraceType);
211
212 protos::pbzero::Trace::Decoder decoder(data, size);
213 GzipDecompressor decompressor;
214 if (size > 0 && !decoder.packet()) {
215 return util::ErrStatus("Trace does not contain valid packets");
216 }
217 for (auto it = decoder.packet(); it; ++it) {
218 protos::pbzero::TracePacket::Decoder packet(*it);
219 if (!packet.has_compressed_packets()) {
220 it->SerializeAndAppendTo(output);
221 continue;
222 }
223
224 // Make sure that to reset the stream between the gzip streams.
225 auto bytes = packet.compressed_packets();
226 decompressor.Reset();
227 decompressor.SetInput(bytes.data, bytes.size);
228
229 using ResultCode = GzipDecompressor::ResultCode;
230 uint8_t out[4096];
231 for (auto ret = ResultCode::kOk; ret != ResultCode::kEof;) {
232 auto res = decompressor.Decompress(out, base::ArraySize(out));
233 ret = res.ret;
234 if (ret == ResultCode::kError || ret == ResultCode::kNoProgress ||
235 ret == ResultCode::kNeedsMoreInput) {
236 return util::ErrStatus("Failed while decompressing stream");
237 }
238 output->insert(output->end(), out, out + res.bytes_written);
239 }
240 }
241 return util::OkStatus();
242 }
243
244 } // namespace trace_processor
245 } // namespace perfetto
246