Google Git
Sign in
flutter/third_party/perfetto/refs/heads/canary/./src/tracing/service/tracing_service_impl.cc
blob: 7495c705aca293f14946d5e0e84d1ad1f1af07fc [file] [edit]
/*
* Copyright (C) 2017 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 "src/tracing/service/tracing_service_impl.h"
#include <fcntl.h>
#include <limits.h>
#include <string.h>
#include <algorithm>
#include <array>
#include <atomic>
#include <cinttypes>
#include <cstdint>
#include <cstdlib>
#include <functional>
#include <limits>
#include <map>
#include <memory>
#include <optional>
#include <set>
#include <string>
#include <tuple>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>
#if !PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) && \
!PERFETTO_BUILDFLAG(PERFETTO_OS_NACL)
#include <sys/uio.h>
#include <sys/utsname.h>
#include <unistd.h>
#endif
#if PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID) && \
PERFETTO_BUILDFLAG(PERFETTO_ANDROID_BUILD)
#include "src/android_internal/lazy_library_loader.h" // nogncheck
#include "src/android_internal/tracing_service_proxy.h" // nogncheck
#endif
#if PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID) || \
PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX) || \
PERFETTO_BUILDFLAG(PERFETTO_OS_FREEBSD) || \
PERFETTO_BUILDFLAG(PERFETTO_OS_APPLE)
#define PERFETTO_HAS_CHMOD
#include <sys/stat.h>
#endif
#include "perfetto/base/build_config.h"
#include "perfetto/base/compiler.h"
#include "perfetto/base/logging.h"
#include "perfetto/base/status.h"
#include "perfetto/base/task_runner.h"
#include "perfetto/ext/base/android_utils.h"
#include "perfetto/ext/base/clock_snapshots.h"
#include "perfetto/ext/base/file_utils.h"
#include "perfetto/ext/base/flags.h"
#include "perfetto/ext/base/flat_hash_map.h"
#include "perfetto/ext/base/fnv_hash.h"
#include "perfetto/ext/base/metatrace.h"
#include "perfetto/ext/base/periodic_task.h"
#include "perfetto/ext/base/regex.h"
#include "perfetto/ext/base/scoped_file.h"
#include "perfetto/ext/base/scoped_sched_boost.h"
#include "perfetto/ext/base/string_utils.h" // IWYU pragma: keep
#include "perfetto/ext/base/sys_types.h"
#include "perfetto/ext/base/thread_checker.h"
#include "perfetto/ext/base/utils.h"
#include "perfetto/ext/base/uuid.h"
#include "perfetto/ext/base/version.h"
#include "perfetto/ext/base/watchdog.h"
#include "perfetto/ext/tracing/core/basic_types.h"
#include "perfetto/ext/tracing/core/client_identity.h"
#include "perfetto/ext/tracing/core/consumer.h"
#include "perfetto/ext/tracing/core/priority_boost_config.h"
#include "perfetto/ext/tracing/core/producer.h"
#include "perfetto/ext/tracing/core/shared_memory.h"
#include "perfetto/ext/tracing/core/shared_memory_abi.h"
#include "perfetto/ext/tracing/core/slice.h"
#include "perfetto/ext/tracing/core/trace_packet.h"
#include "perfetto/ext/tracing/core/trace_writer.h"
#include "perfetto/ext/tracing/core/tracing_service.h"
#include "perfetto/protozero/scattered_heap_buffer.h"
#include "perfetto/protozero/static_buffer.h"
#include "perfetto/tracing/core/flush_flags.h"
#include "perfetto/tracing/core/forward_decls.h"
#include "perfetto/tracing/core/trace_config.h"
#include "protos/perfetto/config/trace_config.gen.h"
#include "src/android_stats/perfetto_atoms.h"
#include "src/android_stats/statsd_logging_helper.h"
#include "src/protovm/vm.h"
#include "src/protozero/filtering/message_filter.h"
#include "src/protozero/filtering/message_filter_config.h"
#include "src/protozero/filtering/string_filter.h"
#include "src/tracing/core/shared_memory_arbiter_impl.h"
#include "src/tracing/service/clock.h"
#include "src/tracing/service/dependencies.h"
#include "src/tracing/service/packet_stream_validator.h"
#include "src/tracing/service/random.h"
#include "src/tracing/service/trace_buffer.h"
#include "src/tracing/service/trace_buffer_v1.h"
#include "src/tracing/service/trace_buffer_v2.h"
#include "src/tracing/service/tracing_service_endpoints_impl.h"
#include "src/tracing/service/tracing_service_session.h"
#include "src/tracing/service/tracing_service_structs.h"
#include "protos/perfetto/common/builtin_clock.gen.h"
#include "protos/perfetto/common/builtin_clock.pbzero.h"
#include "protos/perfetto/common/system_info.pbzero.h"
#include "protos/perfetto/common/trace_stats.pbzero.h" // IWYU pragma: keep
#include "protos/perfetto/config/protovm/protovm_config.gen.h"
#include "protos/perfetto/config/trace_config.pbzero.h"
#include "protos/perfetto/protovm/vm_program.gen.h"
#include "protos/perfetto/trace/clock_snapshot.pbzero.h"
#include "protos/perfetto/trace/extension_descriptor.pbzero.h"
#include "protos/perfetto/trace/perfetto/trace_provenance.pbzero.h"
#include "protos/perfetto/trace/perfetto/tracing_service_event.pbzero.h"
#include "protos/perfetto/trace/remote_clock_sync.pbzero.h"
#include "protos/perfetto/trace/trace_packet.pbzero.h"
#include "protos/perfetto/trace/trace_uuid.pbzero.h"
#include "protos/perfetto/trace/trigger.pbzero.h"
// General note: this class must assume that Producers are malicious and will
// try to crash / exploit this class. We can trust pointers because they come
// from the IPC layer, but we should never assume that that the producer calls
// come in the right order or their arguments are sane / within bounds.
// This is a macro because we want the call-site line number for the ELOG.
#define PERFETTO_SVC_ERR(...) \
(PERFETTO_ELOG(__VA_ARGS__), ::perfetto::base::ErrStatus(__VA_ARGS__))
namespace perfetto {
// static
std::unique_ptr<TracingService> TracingService::CreateInstance(
std::unique_ptr<SharedMemory::Factory> shm_factory,
base::TaskRunner* task_runner,
InitOpts init_opts) {
tracing_service::Dependencies deps;
deps.clock = std::make_unique<tracing_service::ClockImpl>();
uint32_t seed = static_cast<uint32_t>(deps.clock->GetWallTimeMs().count());
deps.random = std::make_unique<tracing_service::RandomImpl>(seed);
return std::unique_ptr<TracingService>(
new tracing_service::TracingServiceImpl(
std::move(shm_factory), task_runner, std::move(deps), init_opts));
}
namespace tracing_service {
namespace {
constexpr int kMaxBuffersPerConsumer = 128;
constexpr uint32_t kDefaultSnapshotsIntervalMs = 10 * 1000;
constexpr int kDefaultWriteIntoFilePeriodMs = 5000;
constexpr int kMinWriteIntoFilePeriodMs = 100;
constexpr uint32_t kAllDataSourceStartedTimeout = 20000;
constexpr int kMaxConcurrentTracingSessions = 15;
constexpr int kMaxConcurrentTracingSessionsPerUid = 5;
#if PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
constexpr int kMaxConcurrentTracingSessionsForStatsdUid = 10;
#endif
constexpr int64_t kMinSecondsBetweenTracesGuardrail = 5 * 60;
constexpr uint32_t kMillisPerHour = 3600000;
constexpr uint32_t kMillisPerDay = kMillisPerHour * 24;
constexpr uint32_t kMaxTracingDurationMillis = 7 * 24 * kMillisPerHour;
// These apply only if enable_extra_guardrails is true.
constexpr uint32_t kGuardrailsMaxTracingBufferSizeKb = 128 * 1024;
constexpr uint32_t kGuardrailsMaxTracingDurationMillis = 24 * kMillisPerHour;
constexpr size_t kMaxLifecycleEventsListedDataSources = 32;
constexpr uint32_t kTracePacketSystemInfoFieldId = 45;
#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN) || PERFETTO_BUILDFLAG(PERFETTO_OS_NACL)
struct iovec {
void* iov_base; // Address
size_t iov_len; // Block size
};
// Simple implementation of writev. Note that this does not give the atomicity
// guarantees of a real writev, but we don't depend on these (we aren't writing
// to the same file from another thread).
ssize_t writev(int fd, const struct iovec* iov, int iovcnt) {
ssize_t total_size = 0;
for (int i = 0; i < iovcnt; ++i) {
ssize_t current_size = base::WriteAll(fd, iov[i].iov_base, iov[i].iov_len);
if (current_size != static_cast<ssize_t>(iov[i].iov_len))
return -1;
total_size += current_size;
}
return total_size;
}
#define IOV_MAX 1024 // Linux compatible limit.
#elif PERFETTO_BUILDFLAG(PERFETTO_OS_QNX)
#define IOV_MAX 1024 // Linux compatible limit.
#endif
void SerializeAndAppendPacket(std::vector<TracePacket>* packets,
const std::vector<uint8_t>& packet) {
Slice slice = Slice::Allocate(packet.size());
memcpy(slice.own_data(), packet.data(), packet.size());
packets->emplace_back();
packets->back().AddSlice(std::move(slice));
}
std::tuple<size_t /*shm_size*/, size_t /*page_size*/> EnsureValidShmSizes(
size_t shm_size,
size_t page_size) {
// Theoretically the max page size supported by the ABI is 64KB.
// However, the current implementation of TraceBuffer (the non-shared
// userspace buffer where the service copies data) supports at most
// 32K. Setting 64K "works" from the producer<>consumer viewpoint
// but then causes the data to be discarded when copying it into
// TraceBuffer.
constexpr size_t kMaxPageSize = 32 * 1024;
static_assert(kMaxPageSize <= SharedMemoryABI::kMaxPageSize, "");
if (page_size == 0)
page_size = TracingServiceImpl::kDefaultShmPageSize;
if (shm_size == 0)
shm_size = TracingServiceImpl::kDefaultShmSize;
page_size = std::min<size_t>(page_size, kMaxPageSize);
shm_size = std::min<size_t>(shm_size, TracingServiceImpl::kMaxShmSize);
// The tracing page size has to be multiple of 4K. On some systems (e.g. Mac
// on Arm64) the system page size can be larger (e.g., 16K). That doesn't
// matter here, because the tracing page size is just a logical partitioning
// and does not have any dependencies on kernel mm syscalls (read: it's fine
// to have trace page sizes of 4K on a system where the kernel page size is
// 16K).
bool page_size_is_valid = page_size >= SharedMemoryABI::kMinPageSize;
page_size_is_valid &= page_size % SharedMemoryABI::kMinPageSize == 0;
// Only allow power of two numbers of pages, i.e. 1, 2, 4, 8 pages.
size_t num_pages = page_size / SharedMemoryABI::kMinPageSize;
page_size_is_valid &= (num_pages & (num_pages - 1)) == 0;
if (!page_size_is_valid || shm_size < page_size ||
shm_size % page_size != 0) {
return std::make_tuple(TracingServiceImpl::kDefaultShmSize,
TracingServiceImpl::kDefaultShmPageSize);
}
return std::make_tuple(shm_size, page_size);
}
bool NameMatchesFilter(const std::string& name,
const std::vector<std::string>& name_filter,
const std::vector<std::string>& name_regex_filter) {
if (std::find(name_filter.begin(), name_filter.end(), name) !=
name_filter.end()) {
return true;
}
return std::any_of(
name_regex_filter.begin(), name_regex_filter.end(),
[&](const std::string& pattern) {
return base::Regex::CreateOrCheck(pattern).FullMatch(name);
});
}
// Used when TraceConfig.write_into_file == true and output_path is not empty.
base::ScopedFile CreateTraceFile(const std::string& path, bool overwrite) {
#if PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID) && \
PERFETTO_BUILDFLAG(PERFETTO_ANDROID_BUILD)
// This is NOT trying to preserve any security property, SELinux does that.
// It just improves the actionability of the error when people try to save the
// trace in a location that is not SELinux-allowed (a generic "permission
// denied" vs "don't put it here, put it there").
// These are the only SELinux approved dir for trace files that are created
// directly by traced.
static const char* kTraceDirBasePath = "/data/misc/perfetto-traces/";
if (!base::StartsWith(path, kTraceDirBasePath)) {
PERFETTO_ELOG("Invalid output_path %s. On Android it must be within %s.",
path.c_str(), kTraceDirBasePath);
return base::ScopedFile();
}
#endif
// O_CREAT | O_EXCL will fail if the file exists already.
const int flags = O_RDWR | O_CREAT | (overwrite ? O_TRUNC : O_EXCL);
auto fd = base::OpenFile(path, flags, 0600);
if (fd) {
#if defined(PERFETTO_HAS_CHMOD)
// Passing 0644 directly above won't work because of umask.
PERFETTO_CHECK(fchmod(*fd, 0644) == 0);
#endif
} else {
PERFETTO_PLOG("Failed to create %s", path.c_str());
}
return fd;
}
bool ShouldLogEvent(const TraceConfig& cfg) {
switch (cfg.statsd_logging()) {
case TraceConfig::STATSD_LOGGING_ENABLED:
return true;
case TraceConfig::STATSD_LOGGING_DISABLED:
return false;
case TraceConfig::STATSD_LOGGING_UNSPECIFIED:
break;
}
// For backward compatibility with older versions of perfetto_cmd.
return cfg.enable_extra_guardrails();
}
// Appends `data` (which has `size` bytes), to `*packet`. Splits the data in
// slices no larger than `max_slice_size`.
void AppendOwnedSlicesToPacket(std::unique_ptr<uint8_t[]> data,
size_t size,
size_t max_slice_size,
perfetto::TracePacket* packet) {
if (size <= max_slice_size) {
packet->AddSlice(Slice::TakeOwnership(std::move(data), size));
return;
}
uint8_t* src_ptr = data.get();
for (size_t size_left = size; size_left > 0;) {
const size_t slice_size = std::min(size_left, max_slice_size);
Slice slice = Slice::Allocate(slice_size);
memcpy(slice.own_data(), src_ptr, slice_size);
packet->AddSlice(std::move(slice));
src_ptr += slice_size;
size_left -= slice_size;
}
}
// Shmem emulation is only for relay (remote-host) producers whose SMB is copied
// over IPC. An in-process producer always has a real shared SMB, so it must use
// kDefault even when it carries a non-default machine id.
SharedMemoryABI::ShmemMode GetShmemMode(const ClientIdentity& client_identity,
bool in_process) {
return (client_identity.machine_id() == kDefaultMachineID || in_process)
? SharedMemoryABI::ShmemMode::kDefault
: SharedMemoryABI::ShmemMode::kShmemEmulation;
}
} // namespace
TracingServiceImpl::TracingServiceImpl(
std::unique_ptr<SharedMemory::Factory> shm_factory,
base::TaskRunner* task_runner,
tracing_service::Dependencies deps,
InitOpts init_opts)
: clock_(std::move(deps.clock)),
random_(std::move(deps.random)),
init_opts_(init_opts),
shm_factory_(std::move(shm_factory)),
uid_(base::GetCurrentUserId()),
buffer_ids_(kMaxTraceBufferID),
weak_runner_(task_runner) {
PERFETTO_DCHECK(task_runner);
}
TracingServiceImpl::~TracingServiceImpl() {
// TODO(fmayer): handle teardown of all Producer.
}
std::unique_ptr<TracingService::ProducerEndpoint>
TracingServiceImpl::ConnectProducer(Producer* producer,
const ClientIdentity& client_identity,
const std::string& producer_name,
size_t shared_memory_size_hint_bytes,
bool in_process,
ProducerSMBScrapingMode smb_scraping_mode,
size_t shared_memory_page_size_hint_bytes,
std::unique_ptr<SharedMemory> shm,
const std::string& sdk_version,
const std::string& machine_name) {
PERFETTO_DCHECK_THREAD(thread_checker_);
auto uid = client_identity.uid();
if (lockdown_mode_ && uid != base::GetCurrentUserId()) {
PERFETTO_DLOG("Lockdown mode. Rejecting producer with UID %lu",
static_cast<unsigned long>(uid));
return nullptr;
}
if (producers_.size() >= kMaxProducerID) {
PERFETTO_DFATAL("Too many producers.");
return nullptr;
}
const ProducerID id = GetNextProducerID();
PERFETTO_DLOG("Producer %" PRIu16 " connected, uid=%d", id,
static_cast<int>(uid));
bool smb_scraping_enabled = smb_scraping_enabled_;
switch (smb_scraping_mode) {
case ProducerSMBScrapingMode::kDefault:
break;
case ProducerSMBScrapingMode::kEnabled:
smb_scraping_enabled = true;
break;
case ProducerSMBScrapingMode::kDisabled:
smb_scraping_enabled = false;
break;
}
std::unique_ptr<ProducerEndpointImpl> endpoint(new ProducerEndpointImpl(
id, client_identity, this, weak_runner_.task_runner(), producer,
producer_name, machine_name, sdk_version, in_process,
smb_scraping_enabled));
auto it_and_inserted = producers_.emplace(id, endpoint.get());
PERFETTO_DCHECK(it_and_inserted.second);
// Remember an in-process producer's machine so the service can attribute its
// own packets to it (see SetServiceTracePacketHeader), leaving a
// single-machine in-process trace with no separate host machine. Relayed
// producers connect with in_process=false and carry a remote machine id; they
// must not redirect the host service's own packets, so only an in-process
// producer is adopted here. A default machine id is fine to store; the stamp
// decision is made at emit time.
if (in_process)
local_machine_id_ = client_identity.machine_id();
endpoint->shmem_size_hint_bytes_ = shared_memory_size_hint_bytes;
endpoint->shmem_page_size_hint_bytes_ = shared_memory_page_size_hint_bytes;
// Producer::OnConnect() should run before Producer::OnTracingSetup(). The
// latter may be posted by SetupSharedMemory() below, so post OnConnect() now.
endpoint->weak_runner_.PostTask(
[endpoint = endpoint.get()] { endpoint->producer_->OnConnect(); });
if (shm) {
// The producer supplied an SMB. This is used only by Chrome; in the most
// common cases the SMB is created by the service and passed via
// OnTracingSetup(). Verify that it is correctly sized before we attempt to
// use it. The transport layer has to verify the integrity of the SMB (e.g.
// ensure that the producer can't resize if after the fact).
size_t shm_size, page_size;
std::tie(shm_size, page_size) =
EnsureValidShmSizes(shm->size(), endpoint->shmem_page_size_hint_bytes_);
if (shm_size == shm->size() &&
page_size == endpoint->shmem_page_size_hint_bytes_) {
PERFETTO_DLOG(
"Adopting producer-provided SMB of %zu kB for producer \"%s\"",
shm_size / 1024, endpoint->name_.c_str());
auto shmem_mode = GetShmemMode(client_identity, in_process);
endpoint->SetupSharedMemory(std::move(shm), page_size,
/*provided_by_producer=*/true, shmem_mode);
} else {
PERFETTO_LOG(
"Discarding incorrectly sized producer-provided SMB for producer "
"\"%s\", falling back to service-provided SMB. Requested sizes: %zu "
"B total, %zu B page size; suggested corrected sizes: %zu B total, "
"%zu B page size",
endpoint->name_.c_str(), shm->size(),
endpoint->shmem_page_size_hint_bytes_, shm_size, page_size);
shm.reset();
}
}
return std::unique_ptr<ProducerEndpoint>(std::move(endpoint));
}
void TracingServiceImpl::DisconnectProducer(ProducerID id) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Producer %" PRIu16 " disconnected", id);
PERFETTO_DCHECK(producers_.count(id));
if (auto* producer = GetProducer(id)) {
// Scrape remaining chunks for this producer to ensure we don't lose data.
for (auto& session_id_and_session : tracing_sessions_) {
ScrapeSharedMemoryBuffers(&session_id_and_session.second, producer);
}
// Fire a disconnect trigger so pre-configured sessions can capture
// diagnostics when the host traced_probes crashes. Skip producers
// relayed from another machine (e.g. a VM): they share the same
// producer name but their disconnects are expected on VM teardown.
if constexpr (PERFETTO_FLAGS(
TRIGGER_PERFETTO_ON_TRACED_PROBES_DISCONNECT)) {
if (producer->name_ == "perfetto.traced_probes" &&
producer->client_identity().machine_id() == kDefaultMachineID) {
PERFETTO_ELOG("traced_probes disconnected, firing disconnect trigger");
ActivateTriggers(id, {"perfetto.traced_probes.disconnect"});
}
}
}
for (auto it = data_sources_.begin(); it != data_sources_.end();) {
auto next = it;
next++;
if (it->second.producer_id == id)
UnregisterDataSource(id, it->second.descriptor.name());
it = next;
}
producers_.erase(id);
UpdateMemoryGuardrail();
}
ProducerEndpointImpl* TracingServiceImpl::GetProducer(ProducerID id) const {
PERFETTO_DCHECK_THREAD(thread_checker_);
auto it = producers_.find(id);
if (it == producers_.end())
return nullptr;
return it->second;
}
std::unique_ptr<TracingService::ConsumerEndpoint>
TracingServiceImpl::ConnectConsumer(Consumer* consumer, uid_t uid) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Consumer %p connected from UID %" PRIu64,
reinterpret_cast<void*>(consumer), static_cast<uint64_t>(uid));
std::unique_ptr<ConsumerEndpointImpl> endpoint(new ConsumerEndpointImpl(
this, weak_runner_.task_runner(), consumer, uid));
// Consumer might go away before we're able to send the connect notification,
// if that is the case just bail out.
auto weak_ptr = endpoint->weak_ptr_factory_.GetWeakPtr();
weak_runner_.task_runner()->PostTask([weak_ptr = std::move(weak_ptr)] {
if (weak_ptr)
weak_ptr->consumer_->OnConnect();
});
return std::unique_ptr<ConsumerEndpoint>(std::move(endpoint));
}
void TracingServiceImpl::DisconnectConsumer(ConsumerEndpointImpl* consumer) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Consumer %p disconnected", reinterpret_cast<void*>(consumer));
if (consumer->tracing_session_id_)
FreeBuffers(consumer->tracing_session_id_); // Will also DisableTracing().
// At this point no more pointers to |consumer| should be around.
PERFETTO_DCHECK(!std::any_of(
tracing_sessions_.begin(), tracing_sessions_.end(),
[consumer](const std::pair<const TracingSessionID, TracingSession>& kv) {
return kv.second.consumer_maybe_null == consumer;
}));
}
bool TracingServiceImpl::DetachConsumer(ConsumerEndpointImpl* consumer,
const std::string& key) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Consumer %p detached", reinterpret_cast<void*>(consumer));
TracingSessionID tsid = consumer->tracing_session_id_;
TracingSession* tracing_session;
if (!tsid || !(tracing_session = GetTracingSession(tsid)))
return false;
if (GetDetachedSession(consumer->uid_, key)) {
PERFETTO_ELOG("Another session has been detached with the same key \"%s\"",
key.c_str());
return false;
}
PERFETTO_DCHECK(tracing_session->consumer_maybe_null == consumer);
tracing_session->consumer_maybe_null = nullptr;
tracing_session->detach_key = key;
consumer->tracing_session_id_ = 0;
return true;
}
std::unique_ptr<TracingService::RelayEndpoint>
TracingServiceImpl::ConnectRelayClient(RelayClientID relay_client_id) {
PERFETTO_DCHECK_THREAD(thread_checker_);
auto endpoint = std::make_unique<RelayEndpointImpl>(relay_client_id, this);
relay_clients_[relay_client_id] = endpoint.get();
return std::move(endpoint);
}
void TracingServiceImpl::DisconnectRelayClient(RelayClientID relay_client_id) {
PERFETTO_DCHECK_THREAD(thread_checker_);
if (relay_clients_.find(relay_client_id) == relay_clients_.end())
return;
relay_clients_.erase(relay_client_id);
}
bool TracingServiceImpl::AttachConsumer(ConsumerEndpointImpl* consumer,
const std::string& key) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Consumer %p attaching to session %s",
reinterpret_cast<void*>(consumer), key.c_str());
if (consumer->tracing_session_id_) {
PERFETTO_ELOG(
"Cannot reattach consumer to session %s"
" while it already attached tracing session ID %" PRIu64,
key.c_str(), consumer->tracing_session_id_);
return false;
}
auto* tracing_session = GetDetachedSession(consumer->uid_, key);
if (!tracing_session) {
PERFETTO_ELOG(
"Failed to attach consumer, session '%s' not found for uid %d",
key.c_str(), static_cast<int>(consumer->uid_));
return false;
}
consumer->tracing_session_id_ = tracing_session->id;
tracing_session->consumer_maybe_null = consumer;
tracing_session->detach_key.clear();
return true;
}
base::Status TracingServiceImpl::EnableTracing(ConsumerEndpointImpl* consumer,
const TraceConfig& cfg,
base::ScopedFile fd) {
PERFETTO_DCHECK_THREAD(thread_checker_);
// If the producer is specifying a UUID, respect that (at least for the first
// snapshot). Otherwise generate a new UUID.
base::Uuid uuid(cfg.trace_uuid_lsb(), cfg.trace_uuid_msb());
if (!uuid)
uuid = base::Uuidv4();
PERFETTO_DLOG("Enabling tracing for consumer %p, UUID: %s",
reinterpret_cast<void*>(consumer),
uuid.ToPrettyString().c_str());
MaybeLogUploadEvent(cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracing);
if (cfg.lockdown_mode() == TraceConfig::LOCKDOWN_SET)
lockdown_mode_ = true;
if (cfg.lockdown_mode() == TraceConfig::LOCKDOWN_CLEAR)
lockdown_mode_ = false;
// Scope |tracing_session| to this block to prevent accidental use of a null
// pointer later in this function.
{
TracingSession* tracing_session =
GetTracingSession(consumer->tracing_session_id_);
if (tracing_session) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingExistingTraceSession);
return PERFETTO_SVC_ERR(
"A Consumer is trying to EnableTracing() but another tracing "
"session is already active (forgot a call to FreeBuffers() ?)");
}
}
const uint32_t max_duration_ms = cfg.enable_extra_guardrails()
? kGuardrailsMaxTracingDurationMillis
: kMaxTracingDurationMillis;
if (cfg.duration_ms() > max_duration_ms) {
MaybeLogUploadEvent(cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingTooLongTrace);
return PERFETTO_SVC_ERR("Requested too long trace (%" PRIu32
"ms > %" PRIu32 " ms)",
cfg.duration_ms(), max_duration_ms);
}
const bool has_trigger_config =
GetTriggerMode(cfg) != TraceConfig::TriggerConfig::UNSPECIFIED;
if (has_trigger_config &&
(cfg.trigger_config().trigger_timeout_ms() == 0 ||
cfg.trigger_config().trigger_timeout_ms() > max_duration_ms)) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingInvalidTriggerTimeout);
return PERFETTO_SVC_ERR(
"Traces with START_TRACING triggers must provide a positive "
"trigger_timeout_ms < 7 days (received %" PRIu32 "ms)",
cfg.trigger_config().trigger_timeout_ms());
}
// This check has been introduced in May 2023 after finding b/274931668.
if (static_cast<int>(cfg.trigger_config().trigger_mode()) >
TraceConfig::TriggerConfig::TriggerMode_MAX) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingInvalidTriggerMode);
return PERFETTO_SVC_ERR(
"The trace config specified an invalid trigger_mode");
}
if (cfg.trigger_config().use_clone_snapshot_if_available() &&
cfg.trigger_config().trigger_mode() !=
TraceConfig::TriggerConfig::STOP_TRACING) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingInvalidTriggerMode);
return PERFETTO_SVC_ERR(
"trigger_mode must be STOP_TRACING when "
"use_clone_snapshot_if_available=true");
}
if (has_trigger_config && cfg.duration_ms() != 0) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingDurationWithTrigger);
return PERFETTO_SVC_ERR(
"duration_ms was set, this must not be set for traces with triggers.");
}
for (char c : cfg.bugreport_filename()) {
if (!((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
(c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.')) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingInvalidBrFilename);
return PERFETTO_SVC_ERR(
"bugreport_filename contains invalid chars. Use [a-zA-Z0-9-_.]+");
}
}
if ((GetTriggerMode(cfg) == TraceConfig::TriggerConfig::STOP_TRACING ||
GetTriggerMode(cfg) == TraceConfig::TriggerConfig::CLONE_SNAPSHOT) &&
cfg.write_into_file()) {
// We don't support this usecase because there are subtle assumptions which
// break around TracingServiceEvents and windowed sorting (i.e. if we don't
// drain the events in ReadBuffersIntoFile because we are waiting for
// STOP_TRACING, we can end up queueing up a lot of TracingServiceEvents and
// emitting them wildy out of order breaking windowed sorting in trace
// processor).
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingStopTracingWriteIntoFile);
return PERFETTO_SVC_ERR(
"Specifying trigger mode STOP_TRACING/CLONE_SNAPSHOT and "
"write_into_file together is unsupported");
}
std::unordered_set<std::string> triggers;
for (const auto& trigger : cfg.trigger_config().triggers()) {
if (!triggers.insert(trigger.name()).second) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingDuplicateTriggerName);
return PERFETTO_SVC_ERR("Duplicate trigger name: %s",
trigger.name().c_str());
}
}
if (cfg.enable_extra_guardrails()) {
if (cfg.deferred_start()) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingInvalidDeferredStart);
return PERFETTO_SVC_ERR(
"deferred_start=true is not supported in unsupervised traces");
}
uint64_t buf_size_sum = 0;
for (const auto& buf : cfg.buffers()) {
if (buf.size_kb() % 4 != 0) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingInvalidBufferSize);
return PERFETTO_SVC_ERR(
"buffers.size_kb must be a multiple of 4, got %" PRIu32,
buf.size_kb());
}
buf_size_sum += buf.size_kb();
}
uint32_t max_tracing_buffer_size_kb =
std::max(kGuardrailsMaxTracingBufferSizeKb,
cfg.guardrail_overrides().max_tracing_buffer_size_kb());
if (buf_size_sum > max_tracing_buffer_size_kb) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingBufferSizeTooLarge);
return PERFETTO_SVC_ERR("Requested too large trace buffer (%" PRIu64
"kB > %" PRIu32 " kB)",
buf_size_sum, max_tracing_buffer_size_kb);
}
}
const size_t num_buffers = static_cast<size_t>(cfg.buffers_size());
if (num_buffers > kMaxBuffersPerConsumer) {
MaybeLogUploadEvent(cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingTooManyBuffers);
return PERFETTO_SVC_ERR("Too many buffers configured (%zu)", num_buffers);
}
// Build a map of buffer names to indices for named buffer resolution.
// Also validate that buffer names are unique within the session.
base::FlatHashMap<std::string, size_t> buffer_name_to_index;
for (size_t i = 0; i < num_buffers; ++i) {
const auto& buf = cfg.buffers()[i];
if (!buf.name().empty()) {
size_t* existing_idx = buffer_name_to_index.Find(buf.name());
if (existing_idx) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingDuplicateBufferName);
return PERFETTO_SVC_ERR(
"Duplicate buffer name \"%s\" at index %zu (first occurrence at "
"index %zu)",
buf.name().c_str(), i, *existing_idx);
}
buffer_name_to_index.Insert(buf.name(), i);
}
}
// Check that the config specifies all buffers for its data sources. This
// is also checked in SetupDataSource, but it is simpler to return a proper
// error to the consumer from here (and there will be less state to undo).
for (const TraceConfig::DataSource& cfg_data_source : cfg.data_sources()) {
const auto& ds_config = cfg_data_source.config();
// Resolve target buffer: if target_buffer_name is set, look it up.
size_t target_buffer = ds_config.target_buffer();
if (!ds_config.target_buffer_name().empty()) {
size_t* resolved_index_ptr =
buffer_name_to_index.Find(ds_config.target_buffer_name());
if (!resolved_index_ptr) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingOobTargetBuffer);
return PERFETTO_SVC_ERR(
"Data source \"%s\" specified target_buffer_name \"%s\" which does "
"not match any buffer",
ds_config.name().c_str(), ds_config.target_buffer_name().c_str());
}
size_t resolved_index = *resolved_index_ptr;
// If both target_buffer and target_buffer_name are specified, they must
// resolve to the same index. This allows configs to work with both old
// and new versions of the tracing service.
// We have to be lax on target_buffer = 0, because we shouldn't try to
// distinguish between a missing target_buffer and = 0, as per
// proto best practices.
if (target_buffer > 0 && target_buffer != resolved_index) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingOobTargetBuffer);
return PERFETTO_SVC_ERR(
"Data source \"%s\" specified both target_buffer=%zu and "
"target_buffer_name=\"%s\" (index %zu) but they don't match",
ds_config.name().c_str(), target_buffer,
ds_config.target_buffer_name().c_str(), resolved_index);
}
target_buffer = resolved_index;
}
if (target_buffer >= num_buffers) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingOobTargetBuffer);
return PERFETTO_SVC_ERR(
"Data source \"%s\" specified an out of bounds target_buffer (%zu >= "
"%zu)",
ds_config.name().c_str(), target_buffer, num_buffers);
}
}
uint32_t current_exclusive_prio = 0;
for (const auto& [_, session] : tracing_sessions_) {
current_exclusive_prio =
std::max(current_exclusive_prio, session.config.exclusive_prio());
}
// If an exclusive session is active, any new session must have a strictly
// higher priority.
if (current_exclusive_prio > 0 &&
current_exclusive_prio >= cfg.exclusive_prio()) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingExclusiveSessionRejected);
return PERFETTO_SVC_ERR(
"An exclusive session with priority %u >= requested priority %u is "
"already active.",
current_exclusive_prio, cfg.exclusive_prio());
}
if (cfg.exclusive_prio() > 0) { // Exclusive mode.
#if PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
if (consumer->uid_ != AID_ROOT && consumer->uid_ != AID_SHELL) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingExclusiveSessionNotAllowed);
return PERFETTO_SVC_ERR(
"On android, exclusive mode can only be requested by root or shell.");
}
#endif
// Abort all existing sessions.
const std::string abort_reason =
"Aborted due to user requested higher-priority (" +
std::to_string(cfg.exclusive_prio()) + ") exclusive session.";
for (auto it = tracing_sessions_.begin(); it != tracing_sessions_.end();) {
auto next_it = it;
++next_it;
const auto session_consumer = it->second.consumer_maybe_null;
// FreeBuffers() will complete the teardown of the TracingSession and also
// remove it from tracing_sessions_.
FreeBuffers(it->first, abort_reason);
// Disassociate the consumer from the obsolete tracing session.
if (session_consumer) {
session_consumer->tracing_session_id_ = 0;
}
it = next_it;
}
}
if (!cfg.unique_session_name().empty()) {
const std::string& name = cfg.unique_session_name();
for (auto& kv : tracing_sessions_) {
if (kv.second.state == TracingSession::CLONED_READ_ONLY)
continue; // Don't consider cloned sessions in uniqueness checks.
if (kv.second.config.unique_session_name() == name) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingDuplicateSessionName);
static const char fmt[] =
"A trace with this unique session name (%s) already exists";
// This happens frequently, don't make it an "E"LOG.
PERFETTO_LOG(fmt, name.c_str());
return base::ErrStatus(fmt, name.c_str());
}
}
}
if (!cfg.session_semaphores().empty()) {
struct SemaphoreSessionsState {
uint64_t smallest_max_other_session_count =
std::numeric_limits<uint64_t>::max();
uint64_t session_count = 0;
};
// For each semaphore, compute the number of active sessions and the
// MIN(limit).
std::unordered_map<std::string, SemaphoreSessionsState>
sem_to_sessions_state;
for (const auto& id_and_session : tracing_sessions_) {
const auto& session = id_and_session.second;
if (session.state == TracingSession::CLONED_READ_ONLY ||
session.state == TracingSession::DISABLED) {
// Don't consider cloned or disabled sessions in checks.
continue;
}
for (const auto& sem : session.config.session_semaphores()) {
auto& sessions_state = sem_to_sessions_state[sem.name()];
sessions_state.smallest_max_other_session_count =
std::min(sessions_state.smallest_max_other_session_count,
sem.max_other_session_count());
sessions_state.session_count++;
}
}
// Check if any of the semaphores declared by the config clashes with any of
// the currently active semaphores.
for (const auto& semaphore : cfg.session_semaphores()) {
auto it = sem_to_sessions_state.find(semaphore.name());
if (it == sem_to_sessions_state.end()) {
continue;
}
uint64_t max_other_session_count =
std::min(semaphore.max_other_session_count(),
it->second.smallest_max_other_session_count);
if (it->second.session_count > max_other_session_count) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::
kTracedEnableTracingFailedSessionSemaphoreCheck);
return PERFETTO_SVC_ERR(
"Semaphore \"%s\" exceeds maximum allowed other session count "
"(%" PRIu64 " > min(%" PRIu64 ", %" PRIu64 "))",
semaphore.name().c_str(), it->second.session_count,
semaphore.max_other_session_count(),
it->second.smallest_max_other_session_count);
}
}
}
if (cfg.enable_extra_guardrails()) {
// unique_session_name can be empty
const std::string& name = cfg.unique_session_name();
int64_t now_s = clock_->GetBootTimeS().count();
// Remove any entries where the time limit has passed so this map doesn't
// grow indefinitely:
std::map<std::string, int64_t>& sessions = session_to_last_trace_s_;
for (auto it = sessions.cbegin(); it != sessions.cend();) {
if (now_s - it->second > kMinSecondsBetweenTracesGuardrail) {
it = sessions.erase(it);
} else {
++it;
}
}
int64_t& previous_s = session_to_last_trace_s_[name];
if (previous_s == 0) {
previous_s = now_s;
} else {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingSessionNameTooRecent);
return PERFETTO_SVC_ERR(
"A trace with unique session name \"%s\" began less than %" PRId64
"s ago (%" PRId64 "s)",
name.c_str(), kMinSecondsBetweenTracesGuardrail, now_s - previous_s);
}
}
const int sessions_for_uid = static_cast<int>(std::count_if(
tracing_sessions_.begin(), tracing_sessions_.end(),
[consumer](const decltype(tracing_sessions_)::value_type& s) {
return s.second.consumer_uid == consumer->uid_;
}));
int per_uid_limit = kMaxConcurrentTracingSessionsPerUid;
#if PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
if (consumer->uid_ == AID_STATSD) {
per_uid_limit = kMaxConcurrentTracingSessionsForStatsdUid;
}
#endif
if (sessions_for_uid >= per_uid_limit) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingTooManySessionsForUid);
return PERFETTO_SVC_ERR(
"Too many concurrent tracing sesions (%d) for uid %d limit is %d",
sessions_for_uid, static_cast<int>(consumer->uid_), per_uid_limit);
}
// TODO(primiano): This is a workaround to prevent that a producer gets stuck
// in a state where it stalls by design by having more TraceWriterImpl
// instances than free pages in the buffer. This is really a bug in
// trace_probes and the way it handles stalls in the shmem buffer.
if (tracing_sessions_.size() >= kMaxConcurrentTracingSessions) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnableTracingTooManyConcurrentSessions);
return PERFETTO_SVC_ERR("Too many concurrent tracing sesions (%zu)",
tracing_sessions_.size());
}
// If the trace config provides a filter bytecode, setup the filter now.
// If the filter loading fails, abort the tracing session rather than running
// unfiltered.
std::unique_ptr<protozero::MessageFilter> trace_filter;
if (cfg.has_trace_filter()) {
trace_filter.reset(new protozero::MessageFilter());
auto filter_status = protozero::LoadMessageFilterConfig(cfg.trace_filter(),
trace_filter.get());
if (!filter_status.ok()) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingInvalidFilter);
return PERFETTO_SVC_ERR("%s", filter_status.c_message());
}
// The filter is created using perfetto.protos.Trace as root message
// (because that makes it possible to play around with the `proto_filter`
// tool on actual traces). Here in the service, however, we deal with
// perfetto.protos.TracePacket(s), which are one level down (Trace.packet).
// The IPC client (or the write_into_filte logic in here) are responsible
// for pre-pending the packet preamble (See GetProtoPreamble() calls), but
// the preamble is not there at ReadBuffer time. Hence we change the root of
// the filtering to start at the Trace.packet level.
if (!trace_filter->SetFilterRoot({TracePacket::kPacketFieldNumber})) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnableTracingInvalidFilter);
return PERFETTO_SVC_ERR("Failed to set filter root.");
}
}
std::optional<base::ScopedSchedBoost> priority_boost;
if (cfg.has_priority_boost()) {
auto sched_policy = CreateSchedPolicyFromConfig(cfg.priority_boost());
if (!sched_policy.ok()) {
MaybeLogUploadEvent(
cfg, uuid,
PerfettoStatsdAtom::kTracedEnablePriorityBoostInvalidConfig);
return PERFETTO_SVC_ERR("Invalid priority boost config: %s",
sched_policy.status().c_message());
}
auto boost = base::ScopedSchedBoost::Boost(sched_policy.value());
if (!boost.ok()) {
MaybeLogUploadEvent(
cfg, uuid, PerfettoStatsdAtom::kTracedEnablePriorityBoostOtherError);
return PERFETTO_SVC_ERR("Failed to boost priority: %s",
boost.status().c_message());
}
priority_boost = std::move(*boost);
}
const TracingSessionID tsid = ++last_tracing_session_id_;
TracingSession* tracing_session =
&tracing_sessions_
.emplace(std::piecewise_construct, std::forward_as_tuple(tsid),
std::forward_as_tuple(tsid, consumer, cfg,
weak_runner_.task_runner()))
.first->second;
tracing_session->trace_uuid = uuid;
if (trace_filter)
tracing_session->trace_filter = std::move(trace_filter);
if (priority_boost)
tracing_session->priority_boost = std::move(priority_boost);
if (cfg.flush_period_ms() > 0) {
tracing_session->flush_strategy = TracingSession::FlushStrategy::kPeriodic;
tracing_session->periodic_flush_ms = cfg.flush_period_ms();
} else {
tracing_session->flush_strategy = TracingSession::FlushStrategy::kDisabled;
}
if (cfg.write_into_file()) {
if (!fd ^ !cfg.output_path().empty()) {
MaybeLogUploadEvent(
tracing_session->config, uuid,
PerfettoStatsdAtom::kTracedEnableTracingInvalidFdOutputFile);
tracing_sessions_.erase(tsid);
return PERFETTO_SVC_ERR(
"When write_into_file==true either a FD needs to be passed or "
"output_path must be populated (but not both)");
}
if (!cfg.output_path().empty()) {
fd = CreateTraceFile(cfg.output_path(), /*overwrite=*/false);
if (!fd) {
MaybeLogUploadEvent(
tracing_session->config, uuid,
PerfettoStatsdAtom::kTracedEnableTracingFailedToCreateFile);
tracing_sessions_.erase(tsid);
return PERFETTO_SVC_ERR("Failed to create the trace file %s",
cfg.output_path().c_str());
}
}
tracing_session->write_into_file = std::move(fd);
uint32_t write_period_ms = cfg.file_write_period_ms();
if (write_period_ms == 0)
write_period_ms = kDefaultWriteIntoFilePeriodMs;
if (write_period_ms < kMinWriteIntoFilePeriodMs)
write_period_ms = kMinWriteIntoFilePeriodMs;
auto flush_mode = cfg.write_flush_mode();
if (flush_mode == TraceConfig::WRITE_FLUSH_AUTO ||
flush_mode == TraceConfig::WRITE_FLUSH_UNSPECIFIED) {
if (write_period_ms <=
static_cast<uint32_t>(kDefaultWriteIntoFilePeriodMs)) {
tracing_session->flush_strategy =
TracingSession::FlushStrategy::kPeriodic;
tracing_session->periodic_flush_ms =
static_cast<uint32_t>(kDefaultWriteIntoFilePeriodMs);
} else {
tracing_session->flush_strategy =
TracingSession::FlushStrategy::kOnWrite;
}
if (cfg.flush_period_ms() > 0) {
PERFETTO_LOG(
"Warning: flush_period_ms is ignored because write_flush_mode is "
"in AUTO mode. Set write_flush_mode to WRITE_FLUSH_DISABLED to "
"use flush_period_ms.");
}
} else if (flush_mode == TraceConfig::WRITE_FLUSH_ENABLED) {
tracing_session->flush_strategy = TracingSession::FlushStrategy::kOnWrite;
if (cfg.flush_period_ms() > 0) {
PERFETTO_LOG(
"Warning: flush_period_ms is ignored because write_flush_mode is "
"in WRITE_FLUSH_ENABLED mode. Set write_flush_mode to "
"WRITE_FLUSH_DISABLED to use flush_period_ms.");
}
}
tracing_session->write_period_ms = write_period_ms;
tracing_session->max_file_size_bytes = cfg.max_file_size_bytes();
tracing_session->bytes_written_into_file = 0;
tracing_session->fflush_post_write =
cfg.fflush_post_write() == TraceConfig::FFLUSH_ENABLED;
}
if (cfg.compression_type() == TraceConfig::COMPRESSION_TYPE_DEFLATE) {
if (init_opts_.compressor_fn) {
tracing_session->compress_deflate = true;
} else {
PERFETTO_LOG(
"COMPRESSION_TYPE_DEFLATE is not supported in the current build "
"configuration. Skipping compression");
}
}
// Initialize the log buffers.
bool did_allocate_all_buffers = true;
bool invalid_buffer_config = false;
// Allocate the trace buffers. Also create a map to translate a consumer
// relative index (TraceConfig.DataSourceConfig.target_buffer) into the
// corresponding BufferID, which is a global ID namespace for the service and
// all producers.
size_t total_buf_size_kb = 0;
tracing_session->buffers_index.reserve(num_buffers);
for (size_t i = 0; i < num_buffers; i++) {
const TraceConfig::BufferConfig& buffer_cfg = cfg.buffers()[i];
BufferID global_id = buffer_ids_.Allocate();
if (!global_id) {
did_allocate_all_buffers = false; // We ran out of IDs.
break;
}
tracing_session->buffers_index.push_back(global_id);
// TraceBuffer size is limited to 32-bit.
const uint32_t buf_size_kb = buffer_cfg.size_kb();
const uint64_t buf_size_bytes = buf_size_kb * static_cast<uint64_t>(1024);
const size_t buf_size = static_cast<size_t>(buf_size_bytes);
if (buf_size_bytes == 0 ||
buf_size_bytes > std::numeric_limits<uint32_t>::max() ||
buf_size != buf_size_bytes) {
invalid_buffer_config = true;
did_allocate_all_buffers = false;
break;
}
total_buf_size_kb += buf_size_kb;
TraceBuffer::OverwritePolicy policy =
buffer_cfg.fill_policy() == TraceConfig::BufferConfig::DISCARD
? TraceBuffer::kDiscard
: TraceBuffer::kOverwrite;
std::unique_ptr<TraceBuffer> new_buffer;
switch (buffer_cfg.experimental_mode()) {
case TraceConfig::BufferConfig::TRACE_BUFFER_V2:
new_buffer = TraceBufferV2::Create(buf_size, policy);
break;
case TraceConfig::BufferConfig::MODE_UNSPECIFIED:
new_buffer = TraceBufferV1::Create(buf_size, policy);
break;
}
auto it_and_inserted = buffers_.emplace(global_id, std::move(new_buffer));
PERFETTO_DCHECK(it_and_inserted.second); // buffers_.count(global_id) == 0.
std::unique_ptr<TraceBuffer>& trace_buffer = it_and_inserted.first->second;
if (!trace_buffer) {
did_allocate_all_buffers = false;
break;
}
}
// This can happen if either:
// - All the kMaxTraceBufferID slots are taken.
// - OOM, or, more realistically, we exhausted virtual memory.
// - The buffer size in the config is invalid.
// In any case, free all the previously allocated buffers and abort.
if (!did_allocate_all_buffers) {
for (BufferID global_id : tracing_session->buffers_index) {
buffer_ids_.Free(global_id);
buffers_.erase(global_id);
}
MaybeLogUploadEvent(tracing_session->config, uuid,
PerfettoStatsdAtom::kTracedEnableTracingOom);
tracing_sessions_.erase(tsid);
if (invalid_buffer_config) {
return PERFETTO_SVC_ERR(
"Failed to allocate tracing buffers: Invalid buffer sizes");
}
return PERFETTO_SVC_ERR(
"Failed to allocate tracing buffers: OOM or too many buffers");
}
UpdateMemoryGuardrail();
consumer->tracing_session_id_ = tsid;
// Setup the data sources on the producers without starting them.
for (const TraceConfig::DataSource& cfg_data_source : cfg.data_sources()) {
// Scan all the registered data sources with a matching name.
auto range = data_sources_.equal_range(cfg_data_source.config().name());
for (auto it = range.first; it != range.second; it++) {
TraceConfig::ProducerConfig producer_config;
for (const auto& config : cfg.producers()) {
if (GetProducer(it->second.producer_id)->name_ ==
config.producer_name()) {
producer_config = config;
break;
}
}
SetupDataSource(cfg_data_source, producer_config, it->second,
tracing_session);
}
}
bool has_start_trigger = false;
switch (GetTriggerMode(cfg)) {
case TraceConfig::TriggerConfig::UNSPECIFIED:
// no triggers are specified so this isn't a trace that is using triggers.
PERFETTO_DCHECK(!has_trigger_config);
break;
case TraceConfig::TriggerConfig::START_TRACING:
// For traces which use START_TRACE triggers we need to ensure that the
// tracing session will be cleaned up when it times out.
has_start_trigger = true;
weak_runner_.PostDelayedTask(
[tsid, this]() { OnStartTriggersTimeout(tsid); },
cfg.trigger_config().trigger_timeout_ms());
break;
case TraceConfig::TriggerConfig::STOP_TRACING:
case TraceConfig::TriggerConfig::CLONE_SNAPSHOT:
// Update the tracing_session's duration_ms to ensure that if no trigger
// is received the session will end and be cleaned up equal to the
// timeout.
//
// TODO(nuskos): Refactor this so that rather then modifying the config we
// have a field we look at on the tracing_session.
tracing_session->config.set_duration_ms(
cfg.trigger_config().trigger_timeout_ms());
break;
// The case of unknown modes (coming from future versions of the service)
// is handled few lines above (search for TriggerMode_MAX).
}
tracing_session->state = TracingSession::CONFIGURED;
PERFETTO_LOG(
"Configured tracing session %" PRIu64
", #sources:%zu, duration:%u ms%s, #buffers:%d, total "
"buffer size:%zu KB, total sessions:%zu, uid:%u session name: \"%s\"",
tsid, cfg.data_sources().size(), tracing_session->config.duration_ms(),
tracing_session->config.prefer_suspend_clock_for_duration()
? " (suspend_clock)"
: "",
cfg.buffers_size(), total_buf_size_kb, tracing_sessions_.size(),
static_cast<unsigned int>(consumer->uid_),
cfg.unique_session_name().c_str());
// Start the data sources, unless this is a case of early setup + fast
// triggering, either through TraceConfig.deferred_start or
// TraceConfig.trigger_config(). If both are specified which ever one occurs
// first will initiate the trace.
if (!cfg.deferred_start() && !has_start_trigger)
StartTracing(tsid);
return base::OkStatus();
}
void TracingServiceImpl::ChangeTraceConfig(ConsumerEndpointImpl* consumer,
const TraceConfig& updated_cfg) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session =
GetTracingSession(consumer->tracing_session_id_);
PERFETTO_DCHECK(tracing_session);
if ((tracing_session->state != TracingSession::STARTED) &&
(tracing_session->state != TracingSession::CONFIGURED)) {
PERFETTO_ELOG(
"ChangeTraceConfig() was called for a tracing session which isn't "
"running.");
return;
}
// We only support updating producer_name_{,regex}_filter (and pass-through
// configs) for now; null out any changeable fields and make sure the rest are
// identical.
TraceConfig new_config_copy(updated_cfg);
for (auto& ds_cfg : *new_config_copy.mutable_data_sources()) {
ds_cfg.clear_producer_name_filter();
ds_cfg.clear_producer_name_regex_filter();
}
TraceConfig current_config_copy(tracing_session->config);
for (auto& ds_cfg : *current_config_copy.mutable_data_sources()) {
ds_cfg.clear_producer_name_filter();
ds_cfg.clear_producer_name_regex_filter();
}
if (new_config_copy != current_config_copy) {
PERFETTO_LOG(
"ChangeTraceConfig() was called with a config containing unsupported "
"changes; only adding to the producer_name_{,regex}_filter is "
"currently supported and will have an effect.");
}
for (TraceConfig::DataSource& cfg_data_source :
*tracing_session->config.mutable_data_sources()) {
// Find the updated producer_filter in the new config.
std::vector<std::string> new_producer_name_filter;
std::vector<std::string> new_producer_name_regex_filter;
bool found_data_source = false;
for (const auto& it : updated_cfg.data_sources()) {
if (cfg_data_source.config().name() == it.config().name()) {
new_producer_name_filter = it.producer_name_filter();
new_producer_name_regex_filter = it.producer_name_regex_filter();
found_data_source = true;
break;
}
}
// Bail out if data source not present in the new config.
if (!found_data_source) {
PERFETTO_ELOG(
"ChangeTraceConfig() called without a current data source also "
"present in the new config: %s",
cfg_data_source.config().name().c_str());
continue;
}
// TODO(oysteine): Just replacing the filter means that if
// there are any filter entries which were present in the original config,
// but removed from the config passed to ChangeTraceConfig, any matching
// producers will keep producing but newly added producers after this
// point will never start.
*cfg_data_source.mutable_producer_name_filter() = new_producer_name_filter;
*cfg_data_source.mutable_producer_name_regex_filter() =
new_producer_name_regex_filter;
// Get the list of producers that are already set up.
std::unordered_set<uint16_t> set_up_producers;
auto& ds_instances = tracing_session->data_source_instances;
for (auto instance_it = ds_instances.begin();
instance_it != ds_instances.end(); ++instance_it) {
set_up_producers.insert(instance_it->first);
}
// Scan all the registered data sources with a matching name.
auto range = data_sources_.equal_range(cfg_data_source.config().name());
for (auto it = range.first; it != range.second; it++) {
ProducerEndpointImpl* producer = GetProducer(it->second.producer_id);
PERFETTO_DCHECK(producer);
// Check if the producer name of this data source is present
// in the name filters. We currently only support new filters, not
// removing old ones.
if ((!new_producer_name_filter.empty() ||
!new_producer_name_regex_filter.empty()) &&
!NameMatchesFilter(producer->name_, new_producer_name_filter,
new_producer_name_regex_filter)) {
continue;
}
// If this producer is already set up, we assume that all datasources
// in it started already.
if (set_up_producers.count(it->second.producer_id))
continue;
// If it wasn't previously setup, set it up now.
// (The per-producer config is optional).
TraceConfig::ProducerConfig producer_config;
for (const auto& config : tracing_session->config.producers()) {
if (producer->name_ == config.producer_name()) {
producer_config = config;
break;
}
}
DataSourceInstance* ds_inst = SetupDataSource(
cfg_data_source, producer_config, it->second, tracing_session);
if (ds_inst && tracing_session->state == TracingSession::STARTED)
StartDataSourceInstance(producer, tracing_session, ds_inst);
}
}
}
uint32_t TracingServiceImpl::DelayToNextWritePeriodMs(
const TracingSession& session) {
PERFETTO_DCHECK(session.write_period_ms > 0);
return session.write_period_ms -
static_cast<uint32_t>(clock_->GetWallTimeMs().count() %
session.write_period_ms);
}
void TracingServiceImpl::StartTracing(TracingSessionID tsid) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session) {
PERFETTO_ELOG("StartTracing() failed, invalid session ID %" PRIu64, tsid);
return;
}
MaybeLogUploadEvent(tracing_session->config, tracing_session->trace_uuid,
PerfettoStatsdAtom::kTracedStartTracing);
if (tracing_session->state != TracingSession::CONFIGURED) {
MaybeLogUploadEvent(
tracing_session->config, tracing_session->trace_uuid,
PerfettoStatsdAtom::kTracedStartTracingInvalidSessionState);
PERFETTO_ELOG("StartTracing() failed, invalid session state: %d",
tracing_session->state);
return;
}
tracing_session->state = TracingSession::STARTED;
// We store the start of trace snapshot separately as it's important to make
// sure we can interpret all the data in the trace and storing it in the ring
// buffer means it could be overwritten by a later snapshot.
if (!tracing_session->config.builtin_data_sources()
.disable_clock_snapshotting()) {
SnapshotClocks(&tracing_session->initial_clock_snapshot);
}
// We don't snapshot the clocks here because we just did this above.
SnapshotLifecycleEvent(
tracing_session,
protos::pbzero::TracingServiceEvent::kTracingStartedFieldNumber,
false /* snapshot_clocks */);
// Periodically snapshot clocks, stats, sync markers while the trace is
// active. The snapshots are emitted on the future ReadBuffers() calls, which
// means that:
// (a) If we're streaming to a file (or to a consumer) while tracing, we
// write snapshots periodically into the trace.
// (b) If ReadBuffers() is only called after tracing ends, we emit the latest
// snapshot into the trace. For clock snapshots, we keep track of the
// snapshot recorded at the beginning of the session
// (initial_clock_snapshot above), as well as the most recent sampled
// snapshots that showed significant new drift between different clocks.
// The latter clock snapshots are sampled periodically and at lifecycle
// events.
base::PeriodicTask::Args snapshot_task_args;
snapshot_task_args.start_first_task_immediately = true;
snapshot_task_args.use_suspend_aware_timer =
tracing_session->config.builtin_data_sources()
.prefer_suspend_clock_for_snapshot();
snapshot_task_args.task = [this, tsid] { PeriodicSnapshotTask(tsid); };
snapshot_task_args.period_ms =
tracing_session->config.builtin_data_sources().snapshot_interval_ms();
if (!snapshot_task_args.period_ms)
snapshot_task_args.period_ms = kDefaultSnapshotsIntervalMs;
tracing_session->snapshot_periodic_task.Start(snapshot_task_args);
// Trigger delayed task if the trace is time limited.
const uint32_t trace_duration_ms = tracing_session->config.duration_ms();
if (trace_duration_ms > 0) {
auto stop_task =
std::bind(&TracingServiceImpl::StopOnDurationMsExpiry, this, tsid);
if (tracing_session->config.prefer_suspend_clock_for_duration()) {
base::PeriodicTask::Args stop_args;
stop_args.use_suspend_aware_timer = true;
stop_args.period_ms = trace_duration_ms;
stop_args.one_shot = true;
stop_args.task = std::move(stop_task);
tracing_session->timed_stop_task.Start(stop_args);
} else {
weak_runner_.PostDelayedTask(std::move(stop_task), trace_duration_ms);
}
} // if (trace_duration_ms > 0).
// Start the periodic drain tasks if we should to save the trace into a file.
if (tracing_session->config.write_into_file()) {
bool async_flush_buffers_before_read =
tracing_session->flush_strategy ==
TracingSession::FlushStrategy::kOnWrite;
weak_runner_.PostDelayedTask(
[this, tsid, async_flush_buffers_before_read] {
ReadBuffersIntoFile(tsid, async_flush_buffers_before_read);
},
DelayToNextWritePeriodMs(*tracing_session));
}
// Start the periodic flush tasks if the config specified a flush period.
if (tracing_session->flush_strategy ==
TracingSession::FlushStrategy::kPeriodic) {
PeriodicFlushTask(tsid, /*post_next_only=*/true);
}
// Start the periodic incremental state clear tasks if the config specified a
// period.
if (tracing_session->config.incremental_state_config().clear_period_ms()) {
PeriodicClearIncrementalStateTask(tsid, /*post_next_only=*/true);
}
for (auto& [prod_id, data_source] : tracing_session->data_source_instances) {
ProducerEndpointImpl* producer = GetProducer(prod_id);
if (!producer) {
PERFETTO_DFATAL("Producer does not exist.");
continue;
}
StartDataSourceInstance(producer, tracing_session, &data_source);
}
MaybeNotifyAllDataSourcesStarted(tracing_session);
// `did_notify_all_data_source_started` is only set if a consumer is
// connected.
if (tracing_session->consumer_maybe_null) {
weak_runner_.PostDelayedTask(
[this, tsid] { OnAllDataSourceStartedTimeout(tsid); },
kAllDataSourceStartedTimeout);
}
}
void TracingServiceImpl::StopOnDurationMsExpiry(TracingSessionID tsid) {
auto* tracing_session_ptr = GetTracingSession(tsid);
if (!tracing_session_ptr)
return;
// If this trace was using STOP_TRACING triggers and we've seen
// one, then the trigger overrides the normal timeout. In this
// case we just return and let the other task clean up this trace.
if (GetTriggerMode(tracing_session_ptr->config) ==
TraceConfig::TriggerConfig::STOP_TRACING &&
!tracing_session_ptr->received_triggers.empty())
return;
// In all other cases (START_TRACING or no triggers) we flush
// after |trace_duration_ms| unconditionally.
FlushAndDisableTracing(tsid);
}
void TracingServiceImpl::StartDataSourceInstance(
ProducerEndpointImpl* producer,
TracingSession* tracing_session,
DataSourceInstance* instance) {
PERFETTO_DCHECK(instance->state == DataSourceInstance::CONFIGURED);
bool start_immediately = !instance->will_notify_on_start;
if (producer->IsAndroidProcessFrozen()) {
PERFETTO_DLOG(
"skipping waiting of data source \"%s\" on producer \"%s\" (pid=%d) "
"because it is frozen",
instance->data_source_name.c_str(), producer->name_.c_str(),
static_cast<int>(producer->pid()));
start_immediately = true;
}
if (!start_immediately) {
instance->state = DataSourceInstance::STARTING;
} else {
instance->state = DataSourceInstance::STARTED;
}
if (tracing_session->consumer_maybe_null) {
tracing_session->consumer_maybe_null->OnDataSourceInstanceStateChange(
*producer, *instance);
}
producer->StartDataSource(instance->instance_id, instance->config);
// If all data sources are started, notify the consumer.
if (instance->state == DataSourceInstance::STARTED)
MaybeNotifyAllDataSourcesStarted(tracing_session);
}
// DisableTracing just stops the data sources but doesn't free up any buffer.
// This is to allow the consumer to freeze the buffers (by stopping the trace)
// and then drain the buffers. The actual teardown of the TracingSession happens
// in FreeBuffers().
void TracingServiceImpl::DisableTracing(TracingSessionID tsid,
bool disable_immediately,
const std::string& error) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session) {
// Can happen if the consumer calls this before EnableTracing() or after
// FreeBuffers().
PERFETTO_DLOG("DisableTracing() failed, invalid session ID %" PRIu64, tsid);
return;
}
switch (tracing_session->state) {
// Spurious call to DisableTracing() while already disabled, nothing to do.
case TracingSession::DISABLED:
PERFETTO_DCHECK(tracing_session->AllDataSourceInstancesStopped());
return;
case TracingSession::CLONED_READ_ONLY:
return;
// This is either:
// A) The case of a graceful DisableTracing() call followed by a call to
// FreeBuffers(), iff |disable_immediately| == true. In this case we want
// to forcefully transition in the disabled state without waiting for the
// outstanding acks because the buffers are going to be destroyed soon.
// B) A spurious call, iff |disable_immediately| == false, in which case
// there is nothing to do.
case TracingSession::DISABLING_WAITING_STOP_ACKS:
PERFETTO_DCHECK(!tracing_session->AllDataSourceInstancesStopped());
if (disable_immediately)
DisableTracingNotifyConsumerAndFlushFile(tracing_session, error);
return;
// Continues below.
case TracingSession::CONFIGURED:
// If the session didn't even start there is no need to orchestrate a
// graceful stop of data sources.
disable_immediately = true;
break;
// This is the nominal case, continues below.
case TracingSession::STARTED:
// Log the disable tracing event only when the session was actually
// started. This avoids double-logging in scenarios where DisableTracing
// is called multiple times for the same session. A common case is with
// traces that have a timeout (e.g. using `trigger_timeout_ms`):
// 1. The service's timer expires and it calls `DisableTracing`
// internally.
// 2. The service notifies the consumer (e.g. `perfetto_cmd`) that the
// trace has ended.
// 3. The consumer, as part of its cleanup, calls `FreeBuffers()`.
// 4. `FreeBuffers()` on the service-side calls `DisableTracing()` again
// as a safeguard.
// By logging only when transitioning from the `STARTED` state, we ensure
// we only log the effective disable event.
MaybeLogUploadEvent(tracing_session->config, tracing_session->trace_uuid,
PerfettoStatsdAtom::kTracedDisableTracing);
break;
}
for (auto& data_source_inst : tracing_session->data_source_instances) {
const ProducerID producer_id = data_source_inst.first;
DataSourceInstance& instance = data_source_inst.second;
ProducerEndpointImpl* producer = GetProducer(producer_id);
PERFETTO_DCHECK(producer);
PERFETTO_DCHECK(instance.state == DataSourceInstance::CONFIGURED ||
instance.state == DataSourceInstance::STARTING ||
instance.state == DataSourceInstance::STARTED);
StopDataSourceInstance(producer, tracing_session, &instance,
disable_immediately);
}
// If the periodic task is running, we can stop the periodic snapshot timer
// here instead of waiting until FreeBuffers to prevent useless snapshots
// which won't be read.
tracing_session->snapshot_periodic_task.Reset();
// Either this request is flagged with |disable_immediately| or there are no
// data sources that are requesting a final handshake. In both cases just mark
// the session as disabled immediately, notify the consumer and flush the
// trace file (if used).
if (tracing_session->AllDataSourceInstancesStopped())
return DisableTracingNotifyConsumerAndFlushFile(tracing_session, error);
tracing_session->state = TracingSession::DISABLING_WAITING_STOP_ACKS;
weak_runner_.PostDelayedTask([this, tsid] { OnDisableTracingTimeout(tsid); },
tracing_session->data_source_stop_timeout_ms());
// Deliberately NOT removing the session from |tracing_session_|, it's still
// needed to call ReadBuffers(). FreeBuffers() will erase() the session.
}
void TracingServiceImpl::NotifyDataSourceStarted(
ProducerID producer_id,
DataSourceInstanceID instance_id) {
PERFETTO_DCHECK_THREAD(thread_checker_);
for (auto& kv : tracing_sessions_) {
TracingSession& tracing_session = kv.second;
DataSourceInstance* instance =
tracing_session.GetDataSourceInstance(producer_id, instance_id);
if (!instance)
continue;
// If the tracing session was already stopped, ignore this notification.
if (tracing_session.state != TracingSession::STARTED)
continue;
if (instance->state != DataSourceInstance::STARTING) {
PERFETTO_ELOG("Started data source instance in incorrect state: %d",
instance->state);
continue;
}
instance->state = DataSourceInstance::STARTED;
ProducerEndpointImpl* producer = GetProducer(producer_id);
PERFETTO_DCHECK(producer);
if (tracing_session.consumer_maybe_null) {
tracing_session.consumer_maybe_null->OnDataSourceInstanceStateChange(
*producer, *instance);
}
// If all data sources are started, notify the consumer.
MaybeNotifyAllDataSourcesStarted(&tracing_session);
} // for (tracing_session)
}
void TracingServiceImpl::OnAllDataSourceStartedTimeout(TracingSessionID tsid) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
// It would be possible to check for `AllDataSourceInstancesStarted()` here,
// but it doesn't make much sense, because a data source can be registered
// after the session has started. Therefore this is tied to
// `did_notify_all_data_source_started`: if that notification happened, do not
// record slow data sources.
if (!tracing_session || !tracing_session->consumer_maybe_null ||
tracing_session->did_notify_all_data_source_started) {
return;
}
int64_t timestamp = clock_->GetBootTimeNs().count();
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
packet->set_timestamp(static_cast<uint64_t>(timestamp));
SetServiceTracePacketHeader(packet.get());
size_t i = 0;
protos::pbzero::TracingServiceEvent::DataSources* slow_data_sources =
packet->set_service_event()->set_slow_starting_data_sources();
for (const auto& [producer_id, ds_instance] :
tracing_session->data_source_instances) {
if (ds_instance.state == DataSourceInstance::STARTED) {
continue;
}
ProducerEndpointImpl* producer = GetProducer(producer_id);
if (!producer) {
continue;
}
if (++i > kMaxLifecycleEventsListedDataSources) {
break;
}
auto* ds = slow_data_sources->add_data_source();
ds->set_producer_name(producer->name_);
ds->set_data_source_name(ds_instance.data_source_name);
PERFETTO_LOG(
"Data source failed to start within 20s data_source=\"%s\", "
"producer=\"%s\", tsid=%" PRIu64,
ds_instance.data_source_name.c_str(), producer->name_.c_str(), tsid);
}
tracing_session->slow_start_event = TracingSession::ArbitraryLifecycleEvent{
timestamp, packet.SerializeAsArray()};
}
void TracingServiceImpl::MaybeNotifyAllDataSourcesStarted(
TracingSession* tracing_session) {
if (!tracing_session->consumer_maybe_null)
return;
if (!tracing_session->AllDataSourceInstancesStarted())
return;
// In some rare cases, we can get in this state more than once. Consider the
// following scenario: 3 data sources are registered -> trace starts ->
// all 3 data sources ack -> OnAllDataSourcesStarted() is called.
// Imagine now that a 4th data source registers while the trace is ongoing.
// This would hit the AllDataSourceInstancesStarted() condition again.
// In this case, however, we don't want to re-notify the consumer again.
// That would be unexpected (even if, perhaps, technically correct) and
// trigger bugs in the consumer.
if (tracing_session->did_notify_all_data_source_started)
return;
PERFETTO_DLOG("All data sources started");
SnapshotLifecycleEvent(
tracing_session,
protos::pbzero::TracingServiceEvent::kAllDataSourcesStartedFieldNumber,
true /* snapshot_clocks */);
tracing_session->did_notify_all_data_source_started = true;
tracing_session->consumer_maybe_null->OnAllDataSourcesStarted();
}
void TracingServiceImpl::NotifyDataSourceStopped(
ProducerID producer_id,
DataSourceInstanceID instance_id) {
PERFETTO_DCHECK_THREAD(thread_checker_);
for (auto& kv : tracing_sessions_) {
TracingSession& tracing_session = kv.second;
DataSourceInstance* instance =
tracing_session.GetDataSourceInstance(producer_id, instance_id);
if (!instance)
continue;
if (instance->state != DataSourceInstance::STOPPING) {
PERFETTO_ELOG("Stopped data source instance in incorrect state: %d",
instance->state);
continue;
}
instance->state = DataSourceInstance::STOPPED;
ProducerEndpointImpl* producer = GetProducer(producer_id);
PERFETTO_DCHECK(producer);
if (tracing_session.consumer_maybe_null) {
tracing_session.consumer_maybe_null->OnDataSourceInstanceStateChange(
*producer, *instance);
}
if (!tracing_session.AllDataSourceInstancesStopped())
continue;
if (tracing_session.state != TracingSession::DISABLING_WAITING_STOP_ACKS)
continue;
// All data sources acked the termination.
DisableTracingNotifyConsumerAndFlushFile(&tracing_session);
} // for (tracing_session)
}
void TracingServiceImpl::ActivateTriggers(
ProducerID producer_id,
const std::vector<std::string>& triggers) {
PERFETTO_DCHECK_THREAD(thread_checker_);
auto* producer = GetProducer(producer_id);
PERFETTO_DCHECK(producer);
int64_t now_ns = clock_->GetBootTimeNs().count();
for (const auto& trigger_name : triggers) {
PERFETTO_DLOG("Received ActivateTriggers request for \"%s\"",
trigger_name.c_str());
android_stats::MaybeLogTriggerEvent(PerfettoTriggerAtom::kTracedTrigger,
/* trace_uuid_lsb */ 0, trigger_name);
base::FnvHasher hash;
hash.Update(trigger_name.c_str(), trigger_name.size());
std::string triggered_session_name;
base::Uuid triggered_session_uuid;
TracingSessionID triggered_session_id = 0;
auto trigger_mode = TraceConfig::TriggerConfig::UNSPECIFIED;
uint64_t trigger_name_hash = hash.digest();
size_t count_in_window =
PurgeExpiredAndCountTriggerInWindow(now_ns, trigger_name_hash);
bool trigger_matched = false;
bool trigger_activated = false;
for (auto& id_and_tracing_session : tracing_sessions_) {
auto& tracing_session = id_and_tracing_session.second;
TracingSessionID tsid = id_and_tracing_session.first;
auto iter = std::find_if(
tracing_session.config.trigger_config().triggers().begin(),
tracing_session.config.trigger_config().triggers().end(),
[&trigger_name](const TraceConfig::TriggerConfig::Trigger& trigger) {
return trigger.name() == trigger_name;
});
if (iter == tracing_session.config.trigger_config().triggers().end())
continue;
if (tracing_session.state == TracingSession::CLONED_READ_ONLY)
continue;
// If this trigger requires a certain producer to have sent it
// (non-empty producer_name()) ensure the producer who sent this trigger
// matches.
if (!iter->producer_name_regex().empty()) {
auto re = base::Regex::CreateOrCheck(iter->producer_name_regex());
if (!re.FullMatch(producer->name_)) {
continue;
}
}
// Use a random number between 0 and 1 to check if we should allow this
// trigger through or not.
double trigger_rnd = random_->GetValue();
PERFETTO_DCHECK(trigger_rnd >= 0 && trigger_rnd < 1);
if (trigger_rnd < iter->skip_probability()) {
MaybeLogTriggerEvent(tracing_session.config, tracing_session.trace_uuid,
PerfettoTriggerAtom::kTracedLimitProbability,
trigger_name);
continue;
}
// If we already triggered more times than the limit, silently ignore
// this trigger.
if (iter->max_per_24_h() > 0 && count_in_window >= iter->max_per_24_h()) {
MaybeLogTriggerEvent(tracing_session.config, tracing_session.trace_uuid,
PerfettoTriggerAtom::kTracedLimitMaxPer24h,
trigger_name);
continue;
}
trigger_matched = true;
triggered_session_id = tracing_session.id;
triggered_session_name = tracing_session.config.unique_session_name();
triggered_session_uuid.set_lsb_msb(tracing_session.trace_uuid.lsb(),
tracing_session.trace_uuid.msb());
trigger_mode = GetTriggerMode(tracing_session.config);
const bool triggers_already_received =
!tracing_session.received_triggers.empty();
const TriggerInfo trigger = {static_cast<uint64_t>(now_ns), iter->name(),
producer->name_, producer->uid(),
iter->stop_delay_ms()};
MaybeSnapshotClocksIntoRingBuffer(&tracing_session);
tracing_session.received_triggers.push_back(trigger);
switch (trigger_mode) {
case TraceConfig::TriggerConfig::START_TRACING:
// If the session has already been triggered and moved past
// CONFIGURED then we don't need to repeat StartTracing. This would
// work fine (StartTracing would return false) but would add error
// logs.
if (tracing_session.state != TracingSession::CONFIGURED)
break;
trigger_activated = true;
MaybeLogUploadEvent(
tracing_session.config, tracing_session.trace_uuid,
PerfettoStatsdAtom::kTracedTriggerStartTracing, iter->name());
// We override the trace duration to be the trigger's requested
// value, this ensures that the trace will end after this amount
// of time has passed.
tracing_session.config.set_duration_ms(iter->stop_delay_ms());
StartTracing(tsid);
break;
case TraceConfig::TriggerConfig::STOP_TRACING:
// Only stop the trace once to avoid confusing log messages. I.E.
// when we've already hit the first trigger we've already Posted the
// task to FlushAndDisable. So all future triggers will just break
// out.
if (triggers_already_received)
break;
trigger_activated = true;
MaybeLogUploadEvent(
tracing_session.config, tracing_session.trace_uuid,
PerfettoStatsdAtom::kTracedTriggerStopTracing, iter->name());
// Now that we've seen a trigger we need to stop, flush, and disable
// this session after the configured |stop_delay_ms|.
weak_runner_.PostDelayedTask(
[this, tsid] {
// Skip entirely the flush if the trace session doesn't exist
// anymore. This is to prevent misleading error messages to be
// logged.
if (GetTracingSession(tsid))
FlushAndDisableTracing(tsid);
},
// If this trigger is zero this will immediately executable and
// will happen shortly.
iter->stop_delay_ms());
break;
case TraceConfig::TriggerConfig::CLONE_SNAPSHOT:
trigger_activated = true;
MaybeLogUploadEvent(
tracing_session.config, tracing_session.trace_uuid,
PerfettoStatsdAtom::kTracedTriggerCloneSnapshot, iter->name());
weak_runner_.PostDelayedTask(
[this, tsid, trigger] {
auto* tsess = GetTracingSession(tsid);
if (!tsess || !tsess->consumer_maybe_null)
return;
tsess->consumer_maybe_null->NotifyCloneSnapshotTrigger(trigger);
},
iter->stop_delay_ms());
break;
case TraceConfig::TriggerConfig::UNSPECIFIED:
PERFETTO_ELOG("Trigger activated but trigger mode unspecified.");
break;
}
} // for (.. : tracing_sessions_)
if (trigger_matched) {
trigger_history_.emplace_back(TriggerHistory{now_ns, trigger_name_hash});
}
if (trigger_activated) {
// Log only the trigger that actually caused a trace stop/start, don't log
// the follow-up ones, even if they matched.
PERFETTO_LOG(
"Trace trigger activated: trigger_name=\"%s\" trigger_mode=%d "
"trace_name=\"%s\" trace_uuid=\"%s\" tsid=%" PRIu64,
trigger_name.c_str(), trigger_mode, triggered_session_name.c_str(),
triggered_session_uuid.ToPrettyString().c_str(),
triggered_session_id);
}
} // for (trigger_name : triggers)
}
// Always invoked TraceConfig.data_source_stop_timeout_ms (by default
// kDataSourceStopTimeoutMs) after DisableTracing(). In nominal conditions all
// data sources should have acked the stop and this will early out.
void TracingServiceImpl::OnDisableTracingTimeout(TracingSessionID tsid) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session ||
tracing_session->state != TracingSession::DISABLING_WAITING_STOP_ACKS) {
return; // Tracing session was successfully disabled.
}
PERFETTO_ILOG("Timeout while waiting for ACKs for tracing session %" PRIu64,
tsid);
PERFETTO_DCHECK(!tracing_session->AllDataSourceInstancesStopped());
DisableTracingNotifyConsumerAndFlushFile(tracing_session);
}
void TracingServiceImpl::DisableTracingNotifyConsumerAndFlushFile(
TracingSession* tracing_session,
const std::string& error) {
PERFETTO_DCHECK(tracing_session->state != TracingSession::DISABLED);
for (auto& inst_kv : tracing_session->data_source_instances) {
if (inst_kv.second.state == DataSourceInstance::STOPPED)
continue;
inst_kv.second.state = DataSourceInstance::STOPPED;
ProducerEndpointImpl* producer = GetProducer(inst_kv.first);
PERFETTO_DCHECK(producer);
if (tracing_session->consumer_maybe_null) {
tracing_session->consumer_maybe_null->OnDataSourceInstanceStateChange(
*producer, inst_kv.second);
}
}
tracing_session->state = TracingSession::DISABLED;
// Scrape any remaining chunks that weren't flushed by the producers.
for (auto& producer_id_and_producer : producers_)
ScrapeSharedMemoryBuffers(tracing_session, producer_id_and_producer.second);
SnapshotLifecycleEvent(
tracing_session,
protos::pbzero::TracingServiceEvent::kTracingDisabledFieldNumber,
true /* snapshot_clocks */);
if (tracing_session->write_into_file) {
tracing_session->write_period_ms = 0;
// Buffers are scraped, no need to flush before reading into file.
ReadBuffersIntoFile(tracing_session->id,
/* async_flush_buffers_before_read = */ false);
}
MaybeLogUploadEvent(tracing_session->config, tracing_session->trace_uuid,
PerfettoStatsdAtom::kTracedNotifyTracingDisabled);
if (tracing_session->consumer_maybe_null)
tracing_session->consumer_maybe_null->NotifyOnTracingDisabled(error);
}
void TracingServiceImpl::Flush(TracingSessionID tsid,
uint32_t timeout_ms,
ConsumerEndpoint::FlushCallback callback,
FlushFlags flush_flags) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session) {
PERFETTO_DLOG("Flush() failed, invalid session ID %" PRIu64, tsid);
return;
}
SnapshotLifecycleEvent(
tracing_session,
protos::pbzero::TracingServiceEvent::kFlushStartedFieldNumber,
false /* snapshot_clocks */);
std::map<ProducerID, std::vector<DataSourceInstanceID>> data_source_instances;
for (const auto& [producer_id, ds_inst] :
tracing_session->data_source_instances) {
if (!ds_inst.no_flush) {
data_source_instances[producer_id].push_back(ds_inst.instance_id);
continue;
}
ProducerEndpointImpl* producer = GetProducer(producer_id);
if (producer->smb_scraping_enabled_ && producer->IsShmemEmulated()) {
// Remote producers (connected via traced_relay) don't share buffer memory
// with the tracing service, therefore we NEED to trigger a flush request
// so that the producer can self-scrape their emulated SMB. Trigger the
// flush by inserting an empty vector if the producer not in the map
// already. See RFC-0010.
data_source_instances.try_emplace(producer_id);
}
}
FlushDataSourceInstances(tracing_session, timeout_ms, data_source_instances,
std::move(callback), flush_flags);
}
void TracingServiceImpl::FlushDataSourceInstances(
TracingSession* tracing_session,
uint32_t timeout_ms,
const std::map<ProducerID, std::vector<DataSourceInstanceID>>&
data_source_instances,
ConsumerEndpoint::FlushCallback callback,
FlushFlags flush_flags) {
PERFETTO_DCHECK_THREAD(thread_checker_);
if (!timeout_ms)
timeout_ms = tracing_session->flush_timeout_ms();
if (tracing_session->pending_flushes.size() > 1000) {
PERFETTO_ELOG("Too many flushes (%zu) pending for the tracing session",
tracing_session->pending_flushes.size());
callback(false);
return;
}
if (tracing_session->state != TracingSession::STARTED) {
PERFETTO_LOG("Flush() called, but tracing has not been started");
callback(false);
return;
}
tracing_session->last_flush_events.clear();
++tracing_session->flushes_requested;
FlushRequestID flush_request_id = ++last_flush_request_id_;
PendingFlush& pending_flush =
tracing_session->pending_flushes
.emplace_hint(tracing_session->pending_flushes.end(),
flush_request_id, PendingFlush(std::move(callback)))
->second;
// Send a flush request to each producer involved in the tracing session. In
// order to issue a flush request we have to build a map of all data source
// instance ids enabled for each producer.
for (const auto& [producer_id, data_sources] : data_source_instances) {
ProducerEndpointImpl* producer = GetProducer(producer_id);
producer->Flush(flush_request_id, data_sources, flush_flags);
if (!producer->IsAndroidProcessFrozen()) {
pending_flush.producers.insert(producer_id);
} else {
PERFETTO_DLOG(
"skipping waiting flush for on producer \"%s\" (pid=%" PRIu32
") because it is frozen",
producer->name_.c_str(), static_cast<uint32_t>(producer->pid()));
}
}
// If there are no producers to flush (realistically this happens only in
// some tests) fire OnFlushTimeout() straight away, without waiting.
if (data_source_instances.empty())
timeout_ms = 0;
weak_runner_.PostDelayedTask(
[this, tsid = tracing_session->id, flush_request_id, flush_flags] {
OnFlushTimeout(tsid, flush_request_id, flush_flags);
},
timeout_ms);
}
void TracingServiceImpl::NotifyFlushDoneForProducer(
ProducerID producer_id,
FlushRequestID flush_request_id) {
for (auto& kv : tracing_sessions_) {
// Remove all pending flushes <= |flush_request_id| for |producer_id|.
auto& pending_flushes = kv.second.pending_flushes;
auto end_it = pending_flushes.upper_bound(flush_request_id);
for (auto it = pending_flushes.begin(); it != end_it;) {
PendingFlush& pending_flush = it->second;
pending_flush.producers.erase(producer_id);
if (pending_flush.producers.empty()) {
TracingSessionID tsid = kv.first;
auto callback = std::move(pending_flush.callback);
weak_runner_.PostTask([this, tsid, callback = std::move(callback)]() {
CompleteFlush(tsid, std::move(callback),
/*success=*/true);
});
it = pending_flushes.erase(it);
} else {
it++;
}
} // for (pending_flushes)
} // for (tracing_session)
}
void TracingServiceImpl::OnFlushTimeout(TracingSessionID tsid,
FlushRequestID flush_request_id,
FlushFlags flush_flags) {
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session)
return;
auto it = tracing_session->pending_flushes.find(flush_request_id);
if (it == tracing_session->pending_flushes.end())
return; // Nominal case: flush was completed and acked on time.
PendingFlush& pending_flush = it->second;
// If there were no producers to flush, consider it a success.
bool success = pending_flush.producers.empty();
auto callback = std::move(pending_flush.callback);
// If flush failed and this is a "final" flush, log which data sources were
// slow.
if ((flush_flags.reason() == FlushFlags::Reason::kTraceClone ||
flush_flags.reason() == FlushFlags::Reason::kTraceStop) &&
!success) {
int64_t timestamp = clock_->GetBootTimeNs().count();
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
packet->set_timestamp(static_cast<uint64_t>(timestamp));
SetServiceTracePacketHeader(packet.get());
size_t i = 0;
protos::pbzero::TracingServiceEvent::DataSources* event =
packet->set_service_event()->set_last_flush_slow_data_sources();
for (const auto& producer_id : pending_flush.producers) {
ProducerEndpointImpl* producer = GetProducer(producer_id);
if (!producer) {
continue;
}
if (++i > kMaxLifecycleEventsListedDataSources) {
break;
}
auto ds_id_range =
tracing_session->data_source_instances.equal_range(producer_id);
for (auto ds_it = ds_id_range.first; ds_it != ds_id_range.second;
ds_it++) {
auto* ds = event->add_data_source();
ds->set_producer_name(producer->name_);
ds->set_data_source_name(ds_it->second.data_source_name);
if (++i > kMaxLifecycleEventsListedDataSources) {
break;
}
}
}
tracing_session->last_flush_events.push_back(
{timestamp, packet.SerializeAsArray()});
}
tracing_session->pending_flushes.erase(it);
CompleteFlush(tsid, std::move(callback), success);
}
void TracingServiceImpl::CompleteFlush(TracingSessionID tsid,
ConsumerEndpoint::FlushCallback callback,
bool success) {
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session) {
callback(false);
return;
}
// Producers may not have been able to flush all their data, even if they
// indicated flush completion. If possible, also collect uncommitted chunks
// to make sure we have everything they wrote so far.
for (auto& producer_id_and_producer : producers_) {
ScrapeSharedMemoryBuffers(tracing_session, producer_id_and_producer.second);
}
SnapshotLifecycleEvent(
tracing_session,
protos::pbzero::TracingServiceEvent::kAllDataSourcesFlushedFieldNumber,
true /* snapshot_clocks */);
tracing_session->flushes_succeeded += success ? 1 : 0;
tracing_session->flushes_failed += success ? 0 : 1;
callback(success);
}
void TracingServiceImpl::ScrapeSharedMemoryBuffers(
TracingSession* tracing_session,
ProducerEndpointImpl* producer) {
if (!producer->smb_scraping_enabled_ || producer->IsShmemEmulated())
return;
// Can't copy chunks if we don't know about any trace writers.
if (producer->writers_.empty())
return;
// Performance optimization: On flush or session disconnect, this method is
// called for each producer. If the producer doesn't participate in the
// session, there's no need to scrape its chunks right now. We can tell if a
// producer participates in the session by checking if the producer is allowed
// to write into the session's log buffers.
const auto& session_buffers = tracing_session->buffers_index;
bool producer_in_session =
std::any_of(session_buffers.begin(), session_buffers.end(),
[producer](BufferID buffer_id) {
return producer->allowed_target_buffers_.count(buffer_id);
});
if (!producer_in_session)
return;
PERFETTO_DLOG("Scraping SMB for producer %" PRIu16, producer->id_);
// Find and copy any uncommitted chunks from the SMB.
//
// In nominal conditions, the page header bitmap of the used SMB pages should
// never change because the service is the only one who is supposed to modify
// used pages (to make them free again).
//
// However, the code here needs to deal with the case of a malicious producer
// altering the SMB in unpredictable ways. Thankfully the SMB size is
// immutable, so a chunk will always point to some valid memory, even if the
// producer alters the intended layout and chunk header concurrently.
// Ultimately a malicious producer altering the SMB's chunk header bitamp
// while we are iterating in this function is not any different from the case
// of a malicious producer asking to commit a chunk made of random data,
// which is something this class has to deal with regardless.
//
// The only legitimate mutations that can happen from sane producers,
// concurrently to this function, are:
// A. free pages being partitioned,
// B. free chunks being migrated to kChunkBeingWritten,
// C. kChunkBeingWritten chunks being migrated to kChunkCompleted.
SharedMemoryArbiterImpl::ForEachScrapableChunk(
&producer->shmem_abi_,
[&](SharedMemoryABI::Chunk* chunk, bool chunk_complete,
uint16_t packet_count, uint8_t packet_flags) {
WriterID writer_id = chunk->writer_id();
std::optional<BufferID> target_buffer_id =
producer->buffer_id_for_writer(writer_id);
// We can only scrape this chunk if we know which log buffer to copy it
// into.
if (!target_buffer_id)
return;
// Skip chunks that don't belong to the requested tracing session.
bool target_buffer_belongs_to_session =
std::find(session_buffers.begin(), session_buffers.end(),
*target_buffer_id) != session_buffers.end();
if (!target_buffer_belongs_to_session)
return;
uint32_t chunk_id =
chunk->header()->chunk_id.load(std::memory_order_relaxed);
CopyProducerPageIntoLogBuffer(
producer->id_, producer->client_identity_, writer_id, chunk_id,
*target_buffer_id, packet_count, packet_flags, chunk_complete,
chunk->payload_begin(), chunk->payload_size());
});
}
void TracingServiceImpl::FlushAndDisableTracing(TracingSessionID tsid) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Triggering final flush for %" PRIu64, tsid);
Flush(
tsid, 0,
[this, tsid](bool success) {
// This was a DLOG up to Jun 2021 (v16, Android S).
PERFETTO_LOG("FlushAndDisableTracing(%" PRIu64 ") done, success=%d",
tsid, success);
TracingSession* session = GetTracingSession(tsid);
if (!session) {
return;
}
session->final_flush_outcome = success
? TraceStats::FINAL_FLUSH_SUCCEEDED
: TraceStats::FINAL_FLUSH_FAILED;
if (session->consumer_maybe_null) {
// If the consumer is still attached, just disable the session but
// give it a chance to read the contents.
DisableTracing(tsid);
} else {
// If the consumer detached, destroy the session. If the consumer did
// start the session in long-tracing mode, the service will have saved
// the contents to the passed file. If not, the contents will be
// destroyed.
FreeBuffers(tsid);
}
},
FlushFlags(FlushFlags::Initiator::kTraced,
FlushFlags::Reason::kTraceStop));
}
void TracingServiceImpl::PeriodicFlushTask(TracingSessionID tsid,
bool post_next_only) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session || tracing_session->state != TracingSession::STARTED)
return;
PERFETTO_CHECK(tracing_session->flush_strategy ==
TracingSession::FlushStrategy::kPeriodic);
uint32_t flush_period_ms = tracing_session->periodic_flush_ms;
weak_runner_.PostDelayedTask(
[this, tsid] { PeriodicFlushTask(tsid, /*post_next_only=*/false); },
flush_period_ms - static_cast<uint32_t>(clock_->GetWallTimeMs().count() %
flush_period_ms));
if (post_next_only)
return;
PERFETTO_DLOG("Triggering periodic flush for trace session %" PRIu64, tsid);
Flush(
tsid, 0,
[](bool success) {
if (!success)
PERFETTO_ELOG("Periodic flush timed out");
},
FlushFlags(FlushFlags::Initiator::kTraced,
FlushFlags::Reason::kPeriodic));
}
void TracingServiceImpl::PeriodicClearIncrementalStateTask(
TracingSessionID tsid,
bool post_next_only) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session || tracing_session->state != TracingSession::STARTED)
return;
uint32_t clear_period_ms =
tracing_session->config.incremental_state_config().clear_period_ms();
weak_runner_.PostDelayedTask(
[this, tsid] {
PeriodicClearIncrementalStateTask(tsid, /*post_next_only=*/false);
},
clear_period_ms - static_cast<uint32_t>(clock_->GetWallTimeMs().count() %
clear_period_ms));
if (post_next_only)
return;
PERFETTO_DLOG(
"Performing periodic incremental state clear for trace session %" PRIu64,
tsid);
// Queue the IPCs to producers with active data sources that opted in.
std::map<ProducerID, std::vector<DataSourceInstanceID>> clear_map;
for (const auto& kv : tracing_session->data_source_instances) {
ProducerID producer_id = kv.first;
const DataSourceInstance& data_source = kv.second;
if (data_source.handles_incremental_state_clear) {
clear_map[producer_id].push_back(data_source.instance_id);
}
}
for (const auto& kv : clear_map) {
ProducerID producer_id = kv.first;
const std::vector<DataSourceInstanceID>& data_sources = kv.second;
ProducerEndpointImpl* producer = GetProducer(producer_id);
if (!producer) {
PERFETTO_DFATAL("Producer does not exist.");
continue;
}
producer->ClearIncrementalState(data_sources);
}
}
bool TracingServiceImpl::ReadBuffersIntoConsumer(
TracingSessionID tsid,
ConsumerEndpointImpl* consumer) {
PERFETTO_DCHECK(consumer);
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session) {
PERFETTO_DLOG(
"Cannot ReadBuffersIntoConsumer(): no tracing session is active");
return false;
}
if (tracing_session->write_into_file) {
// If the consumer enabled tracing and asked to save the contents into the
// passed file makes little sense to also try to read the buffers over IPC,
// as that would just steal data from the periodic draining task.
PERFETTO_ELOG("Consumer trying to read from write_into_file session.");
return false;
}
if (IsWaitingForTrigger(tracing_session))
return false;
// This is a rough threshold to determine how much to read from the buffer in
// each task. This is to avoid executing a single huge sending task for too
// long and risk to hit the watchdog. This is *not* an upper bound: we just
// stop accumulating new packets and PostTask *after* we cross this threshold.
// This constant essentially balances the PostTask and IPC overhead vs the
// responsiveness of the service. An extremely small value will cause one IPC
// and one PostTask for each slice but will keep the service extremely
// responsive. An extremely large value will batch the send for the full
// buffer in one large task, will hit the blocking send() once the socket
// buffers are full and hang the service for a bit (until the consumer
// catches up).
static constexpr size_t kApproxBytesPerTask = 32768;
bool has_more;
std::vector<TracePacket> packets =
ReadBuffers(tracing_session, kApproxBytesPerTask, &has_more);
if (has_more) {
auto weak_consumer = consumer->weak_ptr_factory_.GetWeakPtr();
weak_runner_.PostTask(
[this, weak_consumer = std::move(weak_consumer), tsid] {
if (!weak_consumer)
return;
ReadBuffersIntoConsumer(tsid, weak_consumer.get());
});
}
// Keep this as tail call, just in case the consumer re-enters.
consumer->consumer_->OnTraceData(std::move(packets), has_more);
return true;
}
bool TracingServiceImpl::ReadBuffersIntoFile(
TracingSessionID tsid,
bool async_flush_buffers_before_read) {
PERFETTO_DCHECK_THREAD(thread_checker_);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session) {
// This will be hit systematically from the PostDelayedTask. Avoid logging,
// it would be just spam.
return false;
}
// This can happen if the file is closed by a previous task because it reaches
// |max_file_size_bytes|.
if (!tracing_session->write_into_file)
return false;
if (IsWaitingForTrigger(tracing_session))
return false;
auto do_read_buffers_into_file_fn =
[this, tsid](bool async_flush_buffers_before_read) {
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session)
return;
// ReadBuffers() can allocate memory internally, for filtering. By
// limiting the data that ReadBuffers() reads to kWriteIntoChunksSize
// per iteration, we limit the amount of memory used on each iteration.
//
// It would be tempting to split this into multiple tasks like in
// ReadBuffersIntoConsumer, but that's not currently possible.
// ReadBuffersIntoFile has to read the whole available data before
// returning, to support the disable_immediately=true code paths.
bool has_more = true;
bool stop_writing_into_file = false;
do {
std::vector<TracePacket> packets =
ReadBuffers(tracing_session, kWriteIntoFileChunkSize, &has_more);
stop_writing_into_file =
WriteIntoFile(tracing_session, std::move(packets));
} while (has_more && !stop_writing_into_file);
if (stop_writing_into_file || tracing_session->write_period_ms == 0) {
// Ensure all data was written to the file before we close it.
base::FlushFile(tracing_session->write_into_file.get());
tracing_session->write_into_file.reset();
tracing_session->write_period_ms = 0;
if (tracing_session->state == TracingSession::STARTED)
DisableTracing(tsid);
return;
}
if (tracing_session->fflush_post_write) {
// Ensure all data was written to the file.
base::FlushFile(tracing_session->write_into_file.get());
}
weak_runner_.PostDelayedTask(
[this, tsid, async_flush_buffers_before_read] {
ReadBuffersIntoFile(tsid, async_flush_buffers_before_read);
},
DelayToNextWritePeriodMs(*tracing_session));
};
if (async_flush_buffers_before_read) {
Flush(
tsid, 0,
[do_read_buffers_into_file_fn](bool success) {
if (!success)
PERFETTO_ELOG("ReadBuffersIntoFile flush timed out");
do_read_buffers_into_file_fn(
/* async_flush_buffers_before_read= */ true);
},
FlushFlags(FlushFlags::Initiator::kTraced,
FlushFlags::Reason::kPeriodic));
} else {
do_read_buffers_into_file_fn(/* async_flush_buffers_before_read= */ false);
}
return true;
}
bool TracingServiceImpl::IsWaitingForTrigger(TracingSession* tracing_session) {
// Ignore the logic below for cloned tracing sessions. In this case we
// actually want to read the (cloned) trace buffers even if no trigger was
// hit.
if (tracing_session->state == TracingSession::CLONED_READ_ONLY) {
return false;
}
// When a tracing session is waiting for a trigger, it is considered empty. If
// a tracing session finishes and moves into DISABLED without ever receiving a
// trigger, the trace should never return any data. This includes the
// synthetic packets like TraceConfig and Clock snapshots. So we bail out
// early and let the consumer know there is no data.
if (!tracing_session->config.trigger_config().triggers().empty() &&
tracing_session->received_triggers.empty()) {
PERFETTO_DLOG(
"ReadBuffers(): tracing session has not received a trigger yet.");
return true;
}
// Traces with CLONE_SNAPSHOT triggers are a special case of the above. They
// can be read only via a CloneSession() request. This is to keep the
// behavior consistent with the STOP_TRACING+triggers case and avoid periodic
// finalizations and uploads of the main CLONE_SNAPSHOT triggers.
if (GetTriggerMode(tracing_session->config) ==
TraceConfig::TriggerConfig::CLONE_SNAPSHOT) {
PERFETTO_DLOG(
"ReadBuffers(): skipping because the tracing session has "
"CLONE_SNAPSHOT triggers defined");
return true;
}
return false;
}
std::vector<TracePacket> TracingServiceImpl::ReadBuffers(
TracingSession* tracing_session,
size_t threshold,
bool* has_more) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DCHECK(tracing_session);
*has_more = false;
std::vector<TracePacket> packets;
packets.reserve(1024); // Just an educated guess to avoid trivial expansions.
if (!tracing_session->initial_clock_snapshot.empty()) {
EmitClockSnapshot(tracing_session,
std::move(tracing_session->initial_clock_snapshot),
&packets);
}
for (auto& snapshot : tracing_session->clock_snapshot_ring_buffer) {
PERFETTO_DCHECK(!snapshot.empty());
EmitClockSnapshot(tracing_session, std::move(snapshot), &packets);
}
tracing_session->clock_snapshot_ring_buffer.clear();
if (tracing_session->should_emit_sync_marker) {
EmitSyncMarker(&packets);
tracing_session->should_emit_sync_marker = false;
}
if (!tracing_session->config.builtin_data_sources().disable_trace_config()) {
MaybeEmitTraceConfig(tracing_session, &packets);
MaybeEmitCloneTrigger(tracing_session, &packets);
MaybeEmitReceivedTriggers(tracing_session, &packets);
}
if (!tracing_session->did_emit_initial_packets) {
EmitUuid(tracing_session, &packets);
if (!tracing_session->config.builtin_data_sources()
.disable_extension_descriptors()) {
EmitExtensionDescriptors(tracing_session, &packets);
}
EmitTraceProvenance(tracing_session, &packets);
if (!tracing_session->config.builtin_data_sources().disable_system_info()) {
EmitSystemInfo(&packets);
if (!relay_clients_.empty())
MaybeEmitRemoteSystemInfo(&packets);
}
}
tracing_session->did_emit_initial_packets = true;
// Note that in the proto comment, we guarantee that the tracing_started
// lifecycle event will be emitted before any data packets so make sure to
// keep this before reading the tracing buffers.
if (!tracing_session->config.builtin_data_sources().disable_service_events())
EmitLifecycleEvents(tracing_session, &packets);
// In a multi-machine tracing session, emit clock synchronization messages for
// remote machines.
if (!tracing_session->config.builtin_data_sources()
.disable_clock_snapshotting() &&
!relay_clients_.empty()) {
MaybeEmitRemoteClockSync(tracing_session, &packets);
}
MaybeEmitProtoVmInstances(tracing_session, &packets);
size_t packets_bytes = 0; // SUM(slice.size() for each slice in |packets|).
// Add up size for packets added by the Maybe* calls above.
for (const TracePacket& packet : packets) {
packets_bytes += packet.size();
}
bool did_hit_threshold = false;
for (size_t buf_idx = 0;
buf_idx < tracing_session->num_buffers() && !did_hit_threshold;
buf_idx++) {
auto tbuf_iter = buffers_.find(tracing_session->buffers_index[buf_idx]);
if (tbuf_iter == buffers_.end()) {
PERFETTO_DFATAL("Buffer not found.");
continue;
}
TraceBuffer& tbuf = *tbuf_iter->second;
tbuf.BeginRead();
while (!did_hit_threshold) {
TracePacket packet;
TraceBuffer::PacketSequenceProperties sequence_properties{};
uint32_t previous_packet_dropped;
if (!tbuf.ReadNextTracePacket(&packet, &sequence_properties,
&previous_packet_dropped)) {
break;
}
packet.set_buffer_index_for_stats(static_cast<uint32_t>(buf_idx));
PERFETTO_DCHECK(sequence_properties.producer_id_trusted != 0);
PERFETTO_DCHECK(sequence_properties.writer_id != 0);
PERFETTO_DCHECK(sequence_properties.client_identity_trusted.has_uid());
// Not checking sequence_properties.client_identity_trusted.has_pid():
// it is false if the platform doesn't support it.
PERFETTO_DCHECK(packet.size() > 0);
if (!PacketStreamValidator::Validate(packet.slices())) {
tracing_session->invalid_packets++;
PERFETTO_DLOG("Dropping invalid packet");
continue;
}
// Append a slice with the trusted field data. This can't be spoofed
// because above we validated that the existing slices don't contain any
// trusted fields. For added safety we append instead of prepending
// because according to protobuf semantics, if the same field is
// encountered multiple times the last instance takes priority. Note that
// truncated packets are also rejected, so the producer can't give us a
// partial packet (e.g., a truncated string) which only becomes valid when
// the trusted data is appended here.
Slice slice = Slice::Allocate(32);
protozero::StaticBuffered<protos::pbzero::TracePacket> trusted_packet(
slice.own_data(), slice.size);
const auto& client_identity_trusted =
sequence_properties.client_identity_trusted;
// Producer data, not a service packet: keeps the producer's own sequence
// and machine id, so it can't use SetServiceTracePacketHeader.
trusted_packet->set_trusted_uid(
static_cast<int32_t>(client_identity_trusted.uid()));
trusted_packet->set_trusted_packet_sequence_id(
tracing_session->GetPacketSequenceID(
client_identity_trusted.machine_id(),
sequence_properties.producer_id_trusted,
sequence_properties.writer_id));
if (client_identity_trusted.has_pid()) {
// Not supported on all platforms.
trusted_packet->set_trusted_pid(
static_cast<int32_t>(client_identity_trusted.pid()));
}
if (client_identity_trusted.has_non_default_machine_id()) {
trusted_packet->set_machine_id(client_identity_trusted.machine_id());
}
if (previous_packet_dropped)
trusted_packet->set_previous_packet_dropped(previous_packet_dropped);
slice.size = trusted_packet.Finalize();
packet.AddSlice(std::move(slice));
// Append the packet (inclusive of the trusted uid) to |packets|.
packets_bytes += packet.size();
did_hit_threshold = packets_bytes >= threshold;
packets.emplace_back(std::move(packet));
} // for(packets...)
} // for(buffers...)
*has_more = did_hit_threshold;
// Only emit the "read complete" lifetime event when there is no more trace
// data available to read. These events are used as safe points to limit
// sorting in trace processor: the code shouldn't emit the event unless the
// buffers are empty.
if (!*has_more && !tracing_session->config.builtin_data_sources()
.disable_service_events()) {
// We don't bother snapshotting clocks here because we wouldn't be able to
// emit it and we shouldn't have significant drift from the last snapshot in
// any case.
SnapshotLifecycleEvent(tracing_session,
protos::pbzero::TracingServiceEvent::
kReadTracingBuffersCompletedFieldNumber,
false /* snapshot_clocks */);
EmitLifecycleEvents(tracing_session, &packets);
}
// Only emit the stats when there is no more trace data is available to read.
// That way, any problems that occur while reading from the buffers are
// reflected in the emitted stats. This is particularly important for use
// cases where ReadBuffers is only ever called after the tracing session is
// stopped.
if (!*has_more && tracing_session->should_emit_stats) {
EmitStats(tracing_session, &packets);
tracing_session->should_emit_stats = false;
}
MaybeFilterPackets(tracing_session, &packets);
MaybeCompressPackets(tracing_session, &packets);
if (!*has_more) {
// We've observed some extremely high memory usage by scudo after
// MaybeFilterPackets in the past. The original bug (b/195145848) is fixed
// now, but this code asks scudo to release memory just in case.
base::MaybeReleaseAllocatorMemToOS();
}
return packets;
}
void TracingServiceImpl::MaybeFilterPackets(TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
// If the tracing session specified a filter, run all packets through the
// filter and replace them with the filter results.
// The process below mantains the cardinality of input packets. Even if an
// entire packet is filtered out, we emit a zero-sized TracePacket proto. That
// makes debugging and reasoning about the trace stats easier.
// This place swaps the contents of each |packets| entry in place.
if (!tracing_session->trace_filter) {
return;
}
protozero::MessageFilter& trace_filter = *tracing_session->trace_filter;
// The filter root should be reset from protos.Trace to protos.TracePacket
// by the earlier call to SetFilterRoot() in EnableTracing().
PERFETTO_DCHECK(trace_filter.config().root_msg_index() != 0);
std::vector<protozero::MessageFilter::InputSlice> filter_input;
auto start = clock_->GetWallTimeNs();
for (TracePacket& packet : *packets) {
const auto& packet_slices = packet.slices();
const size_t input_packet_size = packet.size();
filter_input.clear();
filter_input.resize(packet_slices.size());
++tracing_session->filter_input_packets;
tracing_session->filter_input_bytes += input_packet_size;
for (size_t i = 0; i < packet_slices.size(); ++i)
filter_input[i] = {packet_slices[i].start, packet_slices[i].size};
auto filtered_packet = trace_filter.FilterMessageFragments(
&filter_input[0], filter_input.size());
// Replace the packet in-place with the filtered one (unless failed).
std::optional<uint32_t> maybe_buffer_idx = packet.buffer_index_for_stats();
packet = TracePacket();
if (filtered_packet.error) {
++tracing_session->filter_errors;
PERFETTO_DLOG("Trace packet filtering failed @ packet %" PRIu64,
tracing_session->filter_input_packets);
continue;
}
tracing_session->filter_output_bytes += filtered_packet.size;
if (maybe_buffer_idx.has_value()) {
// Keep the per-buffer stats updated. Also propagate the
// buffer_index_for_stats in the output packet to allow accounting by
// other parts of the ReadBuffer pipeline.
uint32_t buffer_idx = maybe_buffer_idx.value();
packet.set_buffer_index_for_stats(buffer_idx);
auto& vec = tracing_session->filter_bytes_discarded_per_buffer;
if (static_cast<size_t>(buffer_idx) >= vec.size())
vec.resize(buffer_idx + 1);
PERFETTO_DCHECK(input_packet_size >= filtered_packet.size);
size_t bytes_filtered_out = input_packet_size - filtered_packet.size;
vec[buffer_idx] += bytes_filtered_out;
}
AppendOwnedSlicesToPacket(std::move(filtered_packet.data),
filtered_packet.size, kMaxTracePacketSliceSize,
&packet);
}
auto end = clock_->GetWallTimeNs();
tracing_session->filter_time_taken_ns +=
static_cast<uint64_t>((end - start).count());
}
void TracingServiceImpl::MaybeCompressPackets(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
if (!tracing_session->compress_deflate) {
return;
}
init_opts_.compressor_fn(packets);
}
bool TracingServiceImpl::WriteIntoFile(TracingSession* tracing_session,
std::vector<TracePacket> packets) {
if (!tracing_session->write_into_file) {
return false;
}
const uint64_t max_size = tracing_session->max_file_size_bytes
? tracing_session->max_file_size_bytes
: std::numeric_limits<size_t>::max();
size_t total_slices = 0;
for (const TracePacket& packet : packets) {
total_slices += packet.slices().size();
}
// When writing into a file, the file should look like a root trace.proto
// message. Each packet should be prepended with a proto preamble stating
// its field id (within trace.proto) and size. Hence the addition below.
const size_t max_iovecs = total_slices + packets.size();
size_t num_iovecs = 0;
bool stop_writing_into_file = false;
std::unique_ptr<struct iovec[]> iovecs(new struct iovec[max_iovecs]);
size_t num_iovecs_at_last_packet = 0;
uint64_t bytes_about_to_be_written = 0;
for (TracePacket& packet : packets) {
std::tie(iovecs[num_iovecs].iov_base, iovecs[num_iovecs].iov_len) =
packet.GetProtoPreamble();
bytes_about_to_be_written += iovecs[num_iovecs].iov_len;
num_iovecs++;
for (const Slice& slice : packet.slices()) {
// writev() doesn't change the passed pointer. However, struct iovec
// take a non-const ptr because it's the same struct used by readv().
// Hence the const_cast here.
char* start = static_cast<char*>(const_cast<void*>(slice.start));
bytes_about_to_be_written += slice.size;
iovecs[num_iovecs++] = {start, slice.size};
}
if (tracing_session->bytes_written_into_file + bytes_about_to_be_written >=
max_size) {
stop_writing_into_file = true;
num_iovecs = num_iovecs_at_last_packet;
break;
}
num_iovecs_at_last_packet = num_iovecs;
}
PERFETTO_DCHECK(num_iovecs <= max_iovecs);
int fd = *tracing_session->write_into_file;
uint64_t total_wr_size = 0;
// writev() can take at most IOV_MAX entries per call. Batch them.
constexpr size_t kIOVMax = IOV_MAX;
for (size_t i = 0; i < num_iovecs; i += kIOVMax) {
int iov_batch_size = static_cast<int>(std::min(num_iovecs - i, kIOVMax));
ssize_t wr_size = PERFETTO_EINTR(writev(fd, &iovecs[i], iov_batch_size));
if (wr_size <= 0) {
PERFETTO_PLOG("writev() failed");
stop_writing_into_file = true;
break;
}
total_wr_size += static_cast<size_t>(wr_size);
}
tracing_session->bytes_written_into_file += total_wr_size;
PERFETTO_DLOG("Draining into file, written: %" PRIu64 " KB, stop: %d",
(total_wr_size + 1023) / 1024, stop_writing_into_file);
return stop_writing_into_file;
}
void TracingServiceImpl::FreeBuffers(TracingSessionID tsid,
const std::string& error) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Freeing buffers for session %" PRIu64, tsid);
TracingSession* tracing_session = GetTracingSession(tsid);
if (!tracing_session) {
PERFETTO_DLOG("FreeBuffers() failed, invalid session ID %" PRIu64, tsid);
return; // TODO(primiano): signal failure?
}
DisableTracing(tsid, /*disable_immediately=*/true, error);
PERFETTO_DCHECK(tracing_session->AllDataSourceInstancesStopped());
tracing_session->data_source_instances.clear();
for (auto& producer_entry : producers_) {
ProducerEndpointImpl* producer = producer_entry.second;
producer->OnFreeBuffers(tracing_session->buffers_index);
}
for (BufferID buffer_id : tracing_session->buffers_index) {
buffer_ids_.Free(buffer_id);
PERFETTO_DCHECK(buffers_.count(buffer_id) == 1);
buffers_.erase(buffer_id);
}
bool notify_traceur =
tracing_session->config.notify_traceur() &&
tracing_session->state != TracingSession::CLONED_READ_ONLY;
bool is_long_trace =
(tracing_session->config.write_into_file() &&
tracing_session->config.file_write_period_ms() < kMillisPerDay);
auto pending_clones = std::move(tracing_session->pending_clones);
tracing_sessions_.erase(tsid);
tracing_session = nullptr;
UpdateMemoryGuardrail();
for (const auto& id_to_clone_op : pending_clones) {
const PendingClone& clone_op = id_to_clone_op.second;
if (clone_op.weak_consumer) {
weak_runner_.task_runner()->PostTask(
[weak_consumer = clone_op.weak_consumer] {
if (weak_consumer) {
weak_consumer->consumer_->OnSessionCloned(
{false, "Original session ended", {}, false});
}
});
}
}
PERFETTO_LOG("Tracing session %" PRIu64 " ended, total sessions:%zu", tsid,
tracing_sessions_.size());
#if PERFETTO_BUILDFLAG(PERFETTO_ANDROID_BUILD) && \
PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
if (notify_traceur && is_long_trace) {
PERFETTO_LAZY_LOAD(android_internal::NotifyTraceSessionEnded, notify_fn);
if (!notify_fn || !notify_fn(/*session_stolen=*/false))
PERFETTO_ELOG("Failed to notify Traceur long tracing has ended");
}
#else
base::ignore_result(notify_traceur);
base::ignore_result(is_long_trace);
#endif
}
void TracingServiceImpl::MaybeSetUpProtoVm(
const DataSourceConfig& ds_config,
const RegisteredDataSource& ds_instance,
BufferID buffer_id) {
if (!ds_config.has_protovm_config() &&
!ds_instance.descriptor.has_protovm_program()) {
return; // Data source has no ProtoVM
}
// TODO(keanmariotti): report the errors below in a trace packet as well, so
// that we can surface them on the UI.
if (!ds_instance.descriptor.has_protovm_program()) {
PERFETTO_ELOG(
"ProtoVM config for data source %s specifies a ProtoVM, but the data "
"source instance (ProducerID: %d) doesn't specify a ProtoVM program",
ds_config.name().c_str(), ds_instance.producer_id);
return;
}
if (!ds_config.has_protovm_config()) {
PERFETTO_ELOG(
"Received ProtoVM program from data souce instance (ProducerID: %d, "
"Data source name: %s), but the data source config doesn't specify a "
"ProtoVM.",
ds_instance.producer_id, ds_config.name().c_str());
return;
}
if (!ds_config.protovm_config().has_memory_limit_kb()) {
PERFETTO_ELOG(
"ProtoVM config for data source %s doesn't specify a memory limit",
ds_config.name().c_str());
return;
}
auto it = buffers_.find(buffer_id);
PERFETTO_CHECK(it != buffers_.cend());
if (it->second->buf_type() != TraceBuffer::BufType::kV2) {
PERFETTO_ELOG(
"Data source %s is configured to run a ProtoVM, but the corresponding "
"buffer is not of type V2",
ds_config.name().c_str());
return;
}
auto* buffer = static_cast<TraceBufferV2*>(it->second.get());
buffer->MaybeSetUpProtoVm(
ds_config.name(),
ds_instance.descriptor.protovm_program().SerializeAsString(),
ds_config.protovm_config().memory_limit_kb(), ds_instance.producer_id);
}
void TracingServiceImpl::RegisterDataSource(ProducerID producer_id,
const DataSourceDescriptor& desc) {
PERFETTO_DCHECK_THREAD(thread_checker_);
if (desc.name().empty()) {
PERFETTO_DLOG("Received RegisterDataSource() with empty name");
return;
}
ProducerEndpointImpl* producer = GetProducer(producer_id);
if (!producer) {
PERFETTO_DFATAL("Producer not found.");
return;
}
// Check that the producer doesn't register two data sources with the same ID.
// Note that we tolerate |id| == 0 because until Android T / v22 the |id|
// field didn't exist.
for (const auto& kv : data_sources_) {
if (desc.id() && kv.second.producer_id == producer_id &&
kv.second.descriptor.id() == desc.id()) {
PERFETTO_ELOG(
"Failed to register data source \"%s\". A data source with the same "
"id %" PRIu64 " (name=\"%s\") is already registered for producer %d",
desc.name().c_str(), desc.id(), kv.second.descriptor.name().c_str(),
producer_id);
return;
}
}
PERFETTO_DLOG("Producer %" PRIu16 " registered data source \"%s\"",
producer_id, desc.name().c_str());
auto reg_ds = data_sources_.emplace(desc.name(),
RegisteredDataSource{producer_id, desc});
// If there are existing tracing sessions, we need to check if the new
// data source is enabled by any of them.
for (auto& iter : tracing_sessions_) {
TracingSession& tracing_session = iter.second;
if (tracing_session.state != TracingSession::STARTED &&
tracing_session.state != TracingSession::CONFIGURED) {
continue;
}
TraceConfig::ProducerConfig producer_config;
for (const auto& config : tracing_session.config.producers()) {
if (producer->name_ == config.producer_name()) {
producer_config = config;
break;
}
}
for (const TraceConfig::DataSource& cfg_data_source :
tracing_session.config.data_sources()) {
if (cfg_data_source.config().name() != desc.name())
continue;
DataSourceInstance* ds_inst = SetupDataSource(
cfg_data_source, producer_config, reg_ds->second, &tracing_session);
if (ds_inst && tracing_session.state == TracingSession::STARTED)
StartDataSourceInstance(producer, &tracing_session, ds_inst);
}
} // for(iter : tracing_sessions_)
}
void TracingServiceImpl::UpdateDataSource(
ProducerID producer_id,
const DataSourceDescriptor& new_desc) {
if (new_desc.id() == 0) {
PERFETTO_ELOG("UpdateDataSource() must have a non-zero id");
return;
}
// If this producer has already registered a matching descriptor name and id,
// just update the descriptor.
RegisteredDataSource* data_source = nullptr;
auto range = data_sources_.equal_range(new_desc.name());
for (auto it = range.first; it != range.second; ++it) {
if (it->second.producer_id == producer_id &&
it->second.descriptor.id() == new_desc.id()) {
data_source = &it->second;
break;
}
}
if (!data_source) {
PERFETTO_ELOG(
"UpdateDataSource() failed, could not find an existing data source "
"with name=\"%s\" id=%" PRIu64,
new_desc.name().c_str(), new_desc.id());
return;
}
data_source->descriptor = new_desc;
}
void TracingServiceImpl::StopDataSourceInstance(ProducerEndpointImpl* producer,
TracingSession* tracing_session,
DataSourceInstance* instance,
bool disable_immediately) {
const DataSourceInstanceID ds_inst_id = instance->instance_id;
if (producer->IsAndroidProcessFrozen()) {
PERFETTO_DLOG(
"skipping waiting of data source \"%s\" on producer \"%s\" (pid=%d) "
"because it is frozen",
instance->data_source_name.c_str(), producer->name_.c_str(),
static_cast<int>(producer->pid()));
disable_immediately = true;
}
if (instance->will_notify_on_stop && !disable_immediately) {
instance->state = DataSourceInstance::STOPPING;
} else {
instance->state = DataSourceInstance::STOPPED;
}
if (tracing_session->consumer_maybe_null) {
tracing_session->consumer_maybe_null->OnDataSourceInstanceStateChange(
*producer, *instance);
}
producer->StopDataSource(ds_inst_id);
}
void TracingServiceImpl::UnregisterDataSource(ProducerID producer_id,
const std::string& name) {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_DLOG("Producer %" PRIu16 " unregistered data source \"%s\"",
producer_id, name.c_str());
PERFETTO_CHECK(producer_id);
ProducerEndpointImpl* producer = GetProducer(producer_id);
PERFETTO_DCHECK(producer);
for (auto& kv : tracing_sessions_) {
auto& ds_instances = kv.second.data_source_instances;
bool removed = false;
for (auto it = ds_instances.begin(); it != ds_instances.end();) {
if (it->first == producer_id && it->second.data_source_name == name) {
DataSourceInstanceID ds_inst_id = it->second.instance_id;
if (it->second.state != DataSourceInstance::STOPPED) {
if (it->second.state != DataSourceInstance::STOPPING) {
StopDataSourceInstance(producer, &kv.second, &it->second,
/* disable_immediately = */ false);
}
// Mark the instance as stopped immediately, since we are
// unregistering it below.
//
// The StopDataSourceInstance above might have set the state to
// STOPPING so this condition isn't an else.
if (it->second.state == DataSourceInstance::STOPPING)
NotifyDataSourceStopped(producer_id, ds_inst_id);
}
it = ds_instances.erase(it);
removed = true;
} else {
++it;
}
} // for (data_source_instances)
if (removed)
MaybeNotifyAllDataSourcesStarted(&kv.second);
} // for (tracing_session)
for (auto it = data_sources_.begin(); it != data_sources_.end(); ++it) {
if (it->second.producer_id == producer_id &&
it->second.descriptor.name() == name) {
data_sources_.erase(it);
return;
}
}
PERFETTO_DFATAL(
"Tried to unregister a non-existent data source \"%s\" for "
"producer %" PRIu16,
name.c_str(), producer_id);
}
bool TracingServiceImpl::IsInitiatorPrivileged(
const TracingSession& tracing_session) {
#if PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
if (tracing_session.consumer_uid == AID_STATSD &&
tracing_session.config.statsd_metadata().triggering_config_uid() !=
AID_SHELL &&
tracing_session.config.statsd_metadata().triggering_config_uid() !=
AID_ROOT) {
// StatsD can be triggered either by shell, root or an app that has DUMP and
// USAGE_STATS permission. When triggered by shell or root, we do not want
// to consider the trace a trusted system trace, as it was initiated by the
// user. Otherwise, it has to come from an app with DUMP and
// PACKAGE_USAGE_STATS, which has to be preinstalled and trusted by the
// system.
// Check for shell / root: https://bit.ly/3b7oZNi
// Check for DUMP or PACKAGE_USAGE_STATS: https://bit.ly/3ep0NrR
return true;
}
if (tracing_session.consumer_uid == 1000 /* AID_SYSTEM */) {
// AID_SYSTEM is considered a privileged initiator so that system_server can
// profile apps that are not profileable by shell. Other AID_SYSTEM
// processes are not allowed by SELinux to connect to the consumer socket or
// to exec perfetto.
return true;
}
#else
base::ignore_result(tracing_session);
#endif
return false;
}
DataSourceInstance* TracingServiceImpl::SetupDataSource(
const TraceConfig::DataSource& cfg_data_source,
const TraceConfig::ProducerConfig& producer_config,
const RegisteredDataSource& data_source,
TracingSession* tracing_session) {
PERFETTO_DCHECK_THREAD(thread_checker_);
ProducerEndpointImpl* producer = GetProducer(data_source.producer_id);
PERFETTO_DCHECK(producer);
// An existing producer that is not ftrace could have registered itself as
// ftrace, we must not enable it in that case.
if (lockdown_mode_ && producer->uid() != uid_) {
PERFETTO_DLOG("Lockdown mode: not enabling producer %hu", producer->id_);
return nullptr;
}
// Check machine name filter for multi-machine (traced_relay) tracing.
bool is_host_machine =
producer->client_identity().machine_id() == kDefaultMachineID;
if (cfg_data_source.machine_name_filter_size()) {
// Explicit set of machines to match.
const auto& cfg_names = cfg_data_source.machine_name_filter();
bool filter_match =
NameMatchesFilter(producer->machine_name_, cfg_names, {});
// Special case: "host" also always matches the host machine.
bool host_match =
is_host_machine && NameMatchesFilter("host", cfg_names, {});
if (!filter_match && !host_match) {
PERFETTO_DLOG("Data source: %s is filtered out for machine: %s",
cfg_data_source.config().name().c_str(),
producer->machine_name_.c_str());
return nullptr;
}
} else if (!tracing_session->config.trace_all_machines() &&
!is_host_machine) {
// Default matching behaviour starting from perfetto v54: match only host,
// unless the config sets a top level flag.
PERFETTO_DLOG("Data source: %s is filtered out for remote machine: %s",
cfg_data_source.config().name().c_str(),
producer->machine_name_.c_str());
return nullptr;
}
// Check producer name filter.
if (cfg_data_source.producer_name_filter_size() ||
cfg_data_source.producer_name_regex_filter_size()) {
if (!NameMatchesFilter(producer->name_,
cfg_data_source.producer_name_filter(),
cfg_data_source.producer_name_regex_filter())) {
PERFETTO_DLOG("Data source: %s is filtered out for producer: %s",
cfg_data_source.config().name().c_str(),
producer->name_.c_str());
return nullptr;
}
}
// Resolve the target buffer index. If target_buffer_name is specified, look
// up the index from the buffer names in the config.
uint32_t relative_buffer_id = cfg_data_source.config().target_buffer();
const auto& ds_cfg = cfg_data_source.config();
if (!ds_cfg.target_buffer_name().empty()) {
// Look up the buffer index by name from the trace config.
for (size_t i = 0; i < tracing_session->num_buffers(); ++i) {
const auto& buf = tracing_session->config.buffers()[i];
if (buf.name() == ds_cfg.target_buffer_name()) {
relative_buffer_id = static_cast<uint32_t>(i);
break;
}
}
}
if (relative_buffer_id >= tracing_session->num_buffers()) {
PERFETTO_LOG(
"The TraceConfig for DataSource %s specified a target_buffer out of "
"bound (%u). Skipping it.",
cfg_data_source.config().name().c_str(), relative_buffer_id);
return nullptr;
}
// Create a copy of the DataSourceConfig specified in the trace config. This
// will be passed to the producer after translating the |target_buffer| id.
// The |target_buffer| parameter passed by the consumer in the trace config is
// relative to the buffers declared in the same trace config. This has to be
// translated to the global BufferID before passing it to the producers, which
// don't know anything about tracing sessions and consumers.
DataSourceInstanceID inst_id = ++last_data_source_instance_id_;
auto insert_iter = tracing_session->data_source_instances.emplace(
std::piecewise_construct, //
std::forward_as_tuple(producer->id_),
std::forward_as_tuple(
inst_id,
cfg_data_source.config(), // Deliberate copy.
data_source.descriptor.name(),
data_source.descriptor.will_notify_on_start(),
data_source.descriptor.will_notify_on_stop(),
data_source.descriptor.handles_incremental_state_clear(),
data_source.descriptor.no_flush()));
DataSourceInstance* ds_instance = &insert_iter->second;
// New data source instance starts out in CONFIGURED state.
if (tracing_session->consumer_maybe_null) {
tracing_session->consumer_maybe_null->OnDataSourceInstanceStateChange(
*producer, *ds_instance);
}
DataSourceConfig& ds_config = ds_instance->config;
ds_config.set_trace_duration_ms(tracing_session->config.duration_ms());
// Rationale for `if (prefer) set_prefer(true)`, rather than `set(prefer)`:
// ComputeStartupConfigHash() in tracing_muxer_impl.cc compares hashes of the
// DataSourceConfig and expects to know (and clear) the fields generated by
// the tracing service. Unconditionally adding a new field breaks backward
// compatibility of startup tracing with older SDKs, because the serialization
// also propagates unkonwn fields, breaking the hash matching check.
if (tracing_session->config.prefer_suspend_clock_for_duration())
ds_config.set_prefer_suspend_clock_for_duration(true);
ds_config.set_stop_timeout_ms(tracing_session->data_source_stop_timeout_ms());
ds_config.set_enable_extra_guardrails(
tracing_session->config.enable_extra_guardrails());
if (IsInitiatorPrivileged(*tracing_session)) {
ds_config.set_session_initiator(
DataSourceConfig::SESSION_INITIATOR_TRUSTED_SYSTEM);
} else {
// Unset in case the consumer set it.
// We need to be able to trust this field.
ds_config.set_session_initiator(
DataSourceConfig::SESSION_INITIATOR_UNSPECIFIED);
}
ds_config.set_tracing_session_id(tracing_session->id);
BufferID global_id = tracing_session->buffers_index[relative_buffer_id];
PERFETTO_DCHECK(global_id);
ds_config.set_target_buffer(global_id);
MaybeSetUpProtoVm(ds_config, data_source, global_id);
PERFETTO_DLOG("Setting up data source %s with target buffer %" PRIu16,
ds_config.name().c_str(), global_id);
if (!producer->shared_memory()) {
// Determine the SMB page size. Must be an integer multiple of 4k.
// As for the SMB size below, the decision tree is as follows:
// 1. Give priority to what is defined in the trace config.
// 2. If unset give priority to the hint passed by the producer.
// 3. Keep within bounds and ensure it's a multiple of 4k.
size_t page_size = producer_config.page_size_kb() * 1024;
if (page_size == 0)
page_size = producer->shmem_page_size_hint_bytes_;
// Determine the SMB size. Must be an integer multiple of the SMB page size.
// The decision tree is as follows:
// 1. Give priority to what defined in the trace config.
// 2. If unset give priority to the hint passed by the producer.
// 3. Keep within bounds and ensure it's a multiple of the page size.
size_t shm_size = producer_config.shm_size_kb() * 1024;
if (shm_size == 0)
shm_size = producer->shmem_size_hint_bytes_;
auto valid_sizes = EnsureValidShmSizes(shm_size, page_size);
if (valid_sizes != std::tie(shm_size, page_size)) {
PERFETTO_DLOG(
"Invalid configured SMB sizes: shm_size %zu page_size %zu. Falling "
"back to shm_size %zu page_size %zu.",
shm_size, page_size, std::get<0>(valid_sizes),
std::get<1>(valid_sizes));
}
std::tie(shm_size, page_size) = valid_sizes;
// TODO(primiano): right now Create() will suicide in case of OOM if the
// mmap fails. We should instead gracefully fail the request and tell the
// client to go away.
PERFETTO_DLOG("Creating SMB of %zu KB for producer \"%s\"", shm_size / 1024,
producer->name_.c_str());
// In the case the producer is using shmem emulation, because we use
// MMAP to allocate the memory and we never write to it, the shared
// memory is mapped to the zero page, essentially costing zero
// physical memory.
auto shared_memory = shm_factory_->CreateSharedMemory(shm_size);
auto shmem_mode =
GetShmemMode(producer->client_identity(), producer->in_process_);
producer->SetupSharedMemory(std::move(shared_memory), page_size,
/*provided_by_producer=*/false, shmem_mode);
}
producer->SetupDataSource(inst_id, ds_config);
return ds_instance;
}
// Note: all the fields % *_trusted ones are untrusted, as in, the Producer
// might be lying / returning garbage contents. |src| and |size| can be trusted
// in terms of being a valid pointer, but not the contents.
void TracingServiceImpl::CopyProducerPageIntoLogBuffer(
ProducerID producer_id_trusted,
const ClientIdentity& client_identity_trusted,
WriterID writer_id,
ChunkID chunk_id,
BufferID buffer_id,
uint16_t num_fragments,
uint8_t chunk_flags,
bool chunk_complete,
const uint8_t* src,
size_t size) {
PERFETTO_DCHECK_THREAD(thread_checker_);
ProducerEndpointImpl* producer = GetProducer(producer_id_trusted);
if (!producer) {
PERFETTO_DFATAL("Producer not found.");
chunks_discarded_++;
return;
}
TraceBuffer* buf = GetBufferByID(buffer_id);
if (!buf) {
PERFETTO_DLOG("Could not find target buffer %" PRIu16
" for producer %" PRIu16,
buffer_id, producer_id_trusted);
chunks_discarded_++;
return;
}
// Verify that the producer is actually allowed to write into the target
// buffer specified in the request. This prevents a malicious producer from
// injecting data into a log buffer that belongs to a tracing session the
// producer is not part of.
if (!producer->is_allowed_target_buffer(buffer_id)) {
PERFETTO_ELOG("Producer %" PRIu16
" tried to write into forbidden target buffer %" PRIu16,
producer_id_trusted, buffer_id);
PERFETTO_DFATAL("Forbidden target buffer");
chunks_discarded_++;
return;
}
// If the writer was registered by the producer, it should only write into the
// buffer it was registered with.
std::optional<BufferID> associated_buffer =
producer->buffer_id_for_writer(writer_id);
if (associated_buffer && *associated_buffer != buffer_id) {
PERFETTO_ELOG("Writer %" PRIu16 " of producer %" PRIu16
" was registered to write into target buffer %" PRIu16
", but tried to write into buffer %" PRIu16,
writer_id, producer_id_trusted, *associated_buffer,
buffer_id);
PERFETTO_DFATAL("Wrong target buffer");
chunks_discarded_++;
return;
}
buf->CopyChunkUntrusted(producer_id_trusted, client_identity_trusted,
writer_id, chunk_id, num_fragments, chunk_flags,
chunk_complete, src, size);
}
void TracingServiceImpl::ApplyChunkPatches(
ProducerID producer_id_trusted,
const std::vector<CommitDataRequest::ChunkToPatch>& chunks_to_patch) {
PERFETTO_DCHECK_THREAD(thread_checker_);
for (const auto& chunk : chunks_to_patch) {
const ChunkID chunk_id = static_cast<ChunkID>(chunk.chunk_id());
const WriterID writer_id = static_cast<WriterID>(chunk.writer_id());
TraceBuffer* buf =
GetBufferByID(static_cast<BufferID>(chunk.target_buffer()));
static_assert(std::numeric_limits<ChunkID>::max() == kMaxChunkID,
"Add a '|| chunk_id > kMaxChunkID' below if this fails");
if (!writer_id || writer_id > kMaxWriterID || !buf) {
// This can genuinely happen when the trace is stopped. The producers
// might see the stop signal with some delay and try to keep sending
// patches left soon after.
PERFETTO_DLOG(
"Received invalid chunks_to_patch request from Producer: %" PRIu16
", BufferID: %" PRIu32 " ChunkdID: %" PRIu32 " WriterID: %" PRIu16,
producer_id_trusted, chunk.target_buffer(), chunk_id, writer_id);
patches_discarded_ += static_cast<uint64_t>(chunk.patches_size());
continue;
}
// Note, there's no need to validate that the producer is allowed to write
// to the specified buffer ID (or that it's the correct buffer ID for a
// registered TraceWriter). That's because TraceBuffer uses the producer ID
// and writer ID to look up the chunk to patch. If the producer specifies an
// incorrect buffer, this lookup will fail and TraceBuffer will ignore the
// patches. Because the producer ID is trusted, there's also no way for a
// malicious producer to patch another producer's data.
// Speculate on the fact that there are going to be a limited amount of
// patches per request, so we can allocate the |patches| array on the stack.
std::array<TraceBuffer::Patch, 1024> patches; // Uninitialized.
if (chunk.patches().size() > patches.size()) {
PERFETTO_ELOG("Too many patches (%zu) batched in the same request",
patches.size());
PERFETTO_DFATAL("Too many patches");
patches_discarded_ += static_cast<uint64_t>(chunk.patches_size());
continue;
}
size_t i = 0;
for (const auto& patch : chunk.patches()) {
const std::string& patch_data = patch.data();
if (patch_data.size() != patches[i].data.size()) {
PERFETTO_ELOG("Received patch from producer: %" PRIu16
" of unexpected size %zu",
producer_id_trusted, patch_data.size());
patches_discarded_++;
continue;
}
patches[i].offset_untrusted = patch.offset();
memcpy(&patches[i].data[0], patch_data.data(), patches[i].data.size());
i++;
}
buf->TryPatchChunkContents(producer_id_trusted, writer_id, chunk_id,
&patches[0], i, chunk.has_more_patches());
}
}
TracingSession* TracingServiceImpl::GetDetachedSession(uid_t uid,
const std::string& key) {
PERFETTO_DCHECK_THREAD(thread_checker_);
for (auto& kv : tracing_sessions_) {
TracingSession* session = &kv.second;
if (session->consumer_uid == uid && session->detach_key == key) {
PERFETTO_DCHECK(session->consumer_maybe_null == nullptr);
return session;
}
}
return nullptr;
}
TracingSession* TracingServiceImpl::GetTracingSession(TracingSessionID tsid) {
PERFETTO_DCHECK_THREAD(thread_checker_);
auto it = tsid ? tracing_sessions_.find(tsid) : tracing_sessions_.end();
if (it == tracing_sessions_.end())
return nullptr;
return &it->second;
}
TracingSession* TracingServiceImpl::GetTracingSessionByUniqueName(
const std::string& unique_session_name) {
PERFETTO_DCHECK_THREAD(thread_checker_);
if (unique_session_name.empty()) {
return nullptr;
}
for (auto& session_id_and_session : tracing_sessions_) {
TracingSession& session = session_id_and_session.second;
if (session.state == TracingSession::CLONED_READ_ONLY) {
continue;
}
if (session.config.unique_session_name() == unique_session_name) {
return &session;
}
}
return nullptr;
}
TracingSession* TracingServiceImpl::FindTracingSessionWithMaxBugreportScore() {
TracingSession* max_session = nullptr;
for (auto& session_id_and_session : tracing_sessions_) {
auto& session = session_id_and_session.second;
const int32_t score = session.config.bugreport_score();
// Exclude sessions with 0 (or below) score. By default tracing sessions
// should NOT be eligible to be attached to bugreports.
if (score <= 0 || session.state != TracingSession::STARTED)
continue;
if (!max_session || score > max_session->config.bugreport_score())
max_session = &session;
}
return max_session;
}
ProducerID TracingServiceImpl::GetNextProducerID() {
PERFETTO_DCHECK_THREAD(thread_checker_);
PERFETTO_CHECK(producers_.size() < kMaxProducerID);
do {
++last_producer_id_;
} while (producers_.count(last_producer_id_) || last_producer_id_ == 0);
PERFETTO_DCHECK(last_producer_id_ > 0 && last_producer_id_ <= kMaxProducerID);
return last_producer_id_;
}
TraceBuffer* TracingServiceImpl::GetBufferByID(BufferID buffer_id) {
auto buf_iter = buffers_.find(buffer_id);
if (buf_iter == buffers_.end())
return nullptr;
return buf_iter->second.get();
}
void TracingServiceImpl::OnStartTriggersTimeout(TracingSessionID tsid) {
// Skip entirely the flush if the trace session doesn't exist anymore.
// This is to prevent misleading error messages to be logged.
//
// if the trace has started from the trigger we rely on
// the |stop_delay_ms| from the trigger so don't flush and
// disable if we've moved beyond a CONFIGURED state
auto* tracing_session_ptr = GetTracingSession(tsid);
if (tracing_session_ptr &&
tracing_session_ptr->state == TracingSession::CONFIGURED) {
PERFETTO_DLOG("Disabling TracingSession %" PRIu64
" since no triggers activated.",
tsid);
// No data should be returned from ReadBuffers() regardless of if we
// call FreeBuffers() or DisableTracing(). This is because in
// STOP_TRACING we need this promise in either case, and using
// DisableTracing() allows a graceful shutdown. Consumers can follow
// their normal path and check the buffers through ReadBuffers() and
// the code won't hang because the tracing session will still be
// alive just disabled.
DisableTracing(tsid);
}
}
void TracingServiceImpl::UpdateMemoryGuardrail() {
#if PERFETTO_BUILDFLAG(PERFETTO_WATCHDOG)
uint64_t total_buffer_bytes = 0;
// Sum up all the shared memory buffers.
for (const auto& id_to_producer : producers_) {
if (id_to_producer.second->shared_memory())
total_buffer_bytes += id_to_producer.second->shared_memory()->size();
}
// Sum up all the trace buffers.
for (const auto& id_to_buffer : buffers_) {
total_buffer_bytes += id_to_buffer.second->GetMemoryUsageBytes();
}
// Sum up all the cloned traced buffers.
for (const auto& id_to_ts : tracing_sessions_) {
const TracingSession& ts = id_to_ts.second;
for (const auto& id_to_clone_op : ts.pending_clones) {
const PendingClone& clone_op = id_to_clone_op.second;
for (const std::unique_ptr<TraceBuffer>& buf : clone_op.buffers) {
if (buf) {
total_buffer_bytes += buf->GetMemoryUsageBytes();
}
}
}
}
// Set the guard rail to 32MB + the sum of all the buffers over a 30 second
// interval.
uint64_t guardrail = base::kWatchdogDefaultMemorySlack + total_buffer_bytes;
base::Watchdog::GetInstance()->SetMemoryLimit(guardrail, 30 * 1000);
#endif
}
void TracingServiceImpl::PeriodicSnapshotTask(TracingSessionID tsid) {
auto* tracing_session = GetTracingSession(tsid);
if (!tracing_session)
return;
if (tracing_session->state != TracingSession::STARTED)
return;
tracing_session->should_emit_sync_marker = true;
tracing_session->should_emit_stats = true;
MaybeSnapshotClocksIntoRingBuffer(tracing_session);
}
void TracingServiceImpl::SnapshotLifecycleEvent(TracingSession* tracing_session,
uint32_t field_id,
bool snapshot_clocks) {
// field_id should be an id of a field in TracingServiceEvent.
auto& lifecycle_events = tracing_session->lifecycle_events;
auto event_it =
std::find_if(lifecycle_events.begin(), lifecycle_events.end(),
[field_id](const TracingSession::LifecycleEvent& event) {
return event.field_id == field_id;
});
TracingSession::LifecycleEvent* event;
if (event_it == lifecycle_events.end()) {
lifecycle_events.emplace_back(field_id);
event = &lifecycle_events.back();
} else {
event = &*event_it;
}
// Snapshot the clocks before capturing the timestamp for the event so we can
// use this snapshot to resolve the event timestamp if necessary.
if (snapshot_clocks)
MaybeSnapshotClocksIntoRingBuffer(tracing_session);
// Erase before emplacing to prevent a unncessary doubling of memory if
// not needed.
if (event->timestamps.size() >= event->max_size) {
event->timestamps.erase_front(1 + event->timestamps.size() -
event->max_size);
}
event->timestamps.emplace_back(clock_->GetBootTimeNs().count());
}
void TracingServiceImpl::SetSingleLifecycleEvent(
TracingSession* tracing_session,
uint32_t field_id,
int64_t boot_timestamp_ns) {
// field_id should be an id of a field in TracingServiceEvent.
auto& lifecycle_events = tracing_session->lifecycle_events;
auto event_it =
std::find_if(lifecycle_events.begin(), lifecycle_events.end(),
[field_id](const TracingSession::LifecycleEvent& event) {
return event.field_id == field_id;
});
TracingSession::LifecycleEvent* event;
if (event_it == lifecycle_events.end()) {
lifecycle_events.emplace_back(field_id);
event = &lifecycle_events.back();
} else {
event = &*event_it;
}
event->timestamps.clear();
event->timestamps.emplace_back(boot_timestamp_ns);
}
void TracingServiceImpl::MaybeSnapshotClocksIntoRingBuffer(
TracingSession* tracing_session) {
if (tracing_session->config.builtin_data_sources()
.disable_clock_snapshotting()) {
return;
}
// We are making an explicit copy of the latest snapshot (if it exists)
// because SnapshotClocks reads this data and computes the drift based on its
// content. If the clock drift is high enough, it will update the contents of
// |snapshot| and return true. Otherwise, it will return false.
TracingSession::ClockSnapshotData snapshot =
tracing_session->clock_snapshot_ring_buffer.empty()
? TracingSession::ClockSnapshotData()
: tracing_session->clock_snapshot_ring_buffer.back();
bool did_update = SnapshotClocks(&snapshot);
if (did_update) {
// This means clocks drifted enough since last snapshot. See the comment
// in SnapshotClocks.
auto* snapshot_buffer = &tracing_session->clock_snapshot_ring_buffer;
// Erase before emplacing to prevent a unncessary doubling of memory if
// not needed.
static constexpr uint32_t kClockSnapshotRingBufferSize = 16;
if (snapshot_buffer->size() >= kClockSnapshotRingBufferSize) {
snapshot_buffer->erase_front(1 + snapshot_buffer->size() -
kClockSnapshotRingBufferSize);
}
snapshot_buffer->emplace_back(std::move(snapshot));
}
}
// Returns true when the data in |snapshot_data| is updated with the new state
// of the clocks and false otherwise.
bool TracingServiceImpl::SnapshotClocks(
TracingSession::ClockSnapshotData* snapshot_data) {
// Minimum drift that justifies replacing a prior clock snapshot that hasn't
// been emitted into the trace yet (see comment below).
static constexpr int64_t kSignificantDriftNs = 10 * 1000 * 1000; // 10 ms
TracingSession::ClockSnapshotData new_snapshot_data =
base::CaptureClockSnapshots();
// If we're about to update a session's latest clock snapshot that hasn't been
// emitted into the trace yet, check whether the clocks have drifted enough to
// warrant overriding the current snapshot values. The older snapshot would be
// valid for a larger part of the currently buffered trace data because the
// clock sync protocol in trace processor uses the latest clock <= timestamp
// to translate times (see https://perfetto.dev/docs/concepts/clock-sync), so
// we try to keep it if we can.
if (!snapshot_data->empty()) {
PERFETTO_DCHECK(snapshot_data->size() == new_snapshot_data.size());
PERFETTO_DCHECK((*snapshot_data)[0].clock_id ==
protos::gen::BUILTIN_CLOCK_BOOTTIME);
bool update_snapshot = false;
uint64_t old_boot_ns = (*snapshot_data)[0].timestamp;
uint64_t new_boot_ns = new_snapshot_data[0].timestamp;
int64_t boot_diff =
static_cast<int64_t>(new_boot_ns) - static_cast<int64_t>(old_boot_ns);
for (size_t i = 1; i < snapshot_data->size(); i++) {
uint64_t old_ns = (*snapshot_data)[i].timestamp;
uint64_t new_ns = new_snapshot_data[i].timestamp;
int64_t diff =
static_cast<int64_t>(new_ns) - static_cast<int64_t>(old_ns);
// Compare the boottime delta against the delta of this clock.
if (std::abs(boot_diff - diff) >= kSignificantDriftNs) {
update_snapshot = true;
break;
}
}
if (!update_snapshot)
return false;
snapshot_data->clear();
}
*snapshot_data = std::move(new_snapshot_data);
return true;
}
void TracingServiceImpl::SetServiceTracePacketHeader(
protos::pbzero::TracePacket* tp) {
tp->set_trusted_uid(static_cast<int32_t>(uid_));
tp->set_trusted_packet_sequence_id(kServicePacketSequenceID);
// When a local machine was adopted (an in-process producer with a non-default
// machine id; see ConnectProducer), attribute the service's own packets to it
// so the trace has no separate host machine. Host and relay sessions leave
// local_machine_id_ at the default and keep these packets on the host
// machine.
if (local_machine_id_ != kDefaultMachineID)
tp->set_machine_id(local_machine_id_);
}
void TracingServiceImpl::EmitClockSnapshot(
TracingSession* tracing_session,
TracingSession::ClockSnapshotData snapshot_data,
std::vector<TracePacket>* packets) {
PERFETTO_DCHECK(!tracing_session->config.builtin_data_sources()
.disable_clock_snapshotting());
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
auto* snapshot = packet->set_clock_snapshot();
protos::gen::BuiltinClock trace_clock =
tracing_session->config.builtin_data_sources().primary_trace_clock();
if (!trace_clock)
trace_clock = protos::gen::BUILTIN_CLOCK_BOOTTIME;
snapshot->set_primary_trace_clock(
static_cast<protos::pbzero::BuiltinClock>(trace_clock));
for (auto& clock_id_and_ts : snapshot_data) {
auto* c = snapshot->add_clocks();
c->set_clock_id(clock_id_and_ts.clock_id);
c->set_timestamp(clock_id_and_ts.timestamp);
}
SetServiceTracePacketHeader(packet.get());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
void TracingServiceImpl::EmitSyncMarker(std::vector<TracePacket>* packets) {
// The sync marks are used to tokenize large traces efficiently.
// See description in trace_packet.proto.
if (sync_marker_packet_size_ == 0) {
// The marker ABI expects that the marker is written after the uid.
// Protozero guarantees that fields are written in the same order of the
// calls. The ResynchronizeTraceStreamUsingSyncMarker test verifies the ABI.
protozero::StaticBuffered<protos::pbzero::TracePacket> packet(
&sync_marker_packet_[0], sizeof(sync_marker_packet_));
// Can't use SetServiceTracePacketHeader: fixed ABI (marker written last,
// after uid) and cached/reused across machines, so it must not gain a
// machine_id field. It's a stream-resync marker; host is fine.
packet->set_trusted_uid(static_cast<int32_t>(uid_));
packet->set_trusted_packet_sequence_id(kServicePacketSequenceID);
// Keep this last.
packet->set_synchronization_marker(kSyncMarker, sizeof(kSyncMarker));
sync_marker_packet_size_ = packet.Finalize();
}
packets->emplace_back();
packets->back().AddSlice(&sync_marker_packet_[0], sync_marker_packet_size_);
}
void TracingServiceImpl::EmitStats(TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
SetServiceTracePacketHeader(packet.get());
GetTraceStats(tracing_session).Serialize(packet->set_trace_stats());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
TraceStats TracingServiceImpl::GetTraceStats(TracingSession* tracing_session) {
TraceStats trace_stats;
trace_stats.set_producers_connected(static_cast<uint32_t>(producers_.size()));
trace_stats.set_producers_seen(last_producer_id_);
trace_stats.set_data_sources_registered(
static_cast<uint32_t>(data_sources_.size()));
trace_stats.set_data_sources_seen(last_data_source_instance_id_);
trace_stats.set_tracing_sessions(
static_cast<uint32_t>(tracing_sessions_.size()));
trace_stats.set_total_buffers(static_cast<uint32_t>(buffers_.size()));
trace_stats.set_chunks_discarded(chunks_discarded_);
trace_stats.set_patches_discarded(patches_discarded_);
trace_stats.set_invalid_packets(tracing_session->invalid_packets);
trace_stats.set_flushes_requested(tracing_session->flushes_requested);
trace_stats.set_flushes_succeeded(tracing_session->flushes_succeeded);
trace_stats.set_flushes_failed(tracing_session->flushes_failed);
trace_stats.set_final_flush_outcome(tracing_session->final_flush_outcome);
if (tracing_session->trace_filter) {
auto* filt_stats = trace_stats.mutable_filter_stats();
filt_stats->set_input_packets(tracing_session->filter_input_packets);
filt_stats->set_input_bytes(tracing_session->filter_input_bytes);
filt_stats->set_output_bytes(tracing_session->filter_output_bytes);
filt_stats->set_errors(tracing_session->filter_errors);
filt_stats->set_time_taken_ns(tracing_session->filter_time_taken_ns);
for (uint64_t value : tracing_session->filter_bytes_discarded_per_buffer)
filt_stats->add_bytes_discarded_per_buffer(value);
}
for (BufferID buf_id : tracing_session->buffers_index) {
TraceBuffer* buf = GetBufferByID(buf_id);
if (!buf) {
PERFETTO_DFATAL("Buffer not found.");
continue;
}
*trace_stats.add_buffer_stats() = buf->stats();
} // for (buf in session).
if (!tracing_session->config.builtin_data_sources()
.disable_chunk_usage_histograms()) {
// Emit chunk usage stats broken down by sequence ID (i.e. by trace-writer).
// Writer stats are updated by each TraceBuffer object at ReadBuffers time,
// and there can be >1 buffer per session. A trace writer never writes to
// more than one buffer (it's technically allowed but doesn't happen in the
// current impl of the tracing SDK).
bool has_written_bucket_definition = false;
uint32_t buf_idx = static_cast<uint32_t>(-1);
for (const BufferID buf_id : tracing_session->buffers_index) {
++buf_idx;
const TraceBuffer* buf = GetBufferByID(buf_id);
if (!buf)
continue;
for (auto it = buf->writer_stats().GetIterator(); it; ++it) {
const auto& hist = it.value();
ProducerID p;
WriterID w;
GetProducerAndWriterID(it.key(), &p, &w);
if (!has_written_bucket_definition) {
// Serialize one-off the histogram bucket definition, which is the
// same for all entries in the map.
has_written_bucket_definition = true;
// The -1 in the loop below is to skip the implicit overflow bucket.
for (size_t i = 0; i < hist.num_buckets() - 1; ++i) {
trace_stats.add_chunk_payload_histogram_def(hist.GetBucketThres(i));
}
} // if(!has_written_bucket_definition)
auto* wri_stats = trace_stats.add_writer_stats();
wri_stats->set_sequence_id(
tracing_session->GetPacketSequenceID(kDefaultMachineID, p, w));
wri_stats->set_buffer(buf_idx);
for (size_t i = 0; i < hist.num_buckets(); ++i) {
wri_stats->add_chunk_payload_histogram_counts(hist.GetBucketCount(i));
wri_stats->add_chunk_payload_histogram_sum(hist.GetBucketSum(i));
}
} // for each sequence (writer).
} // for each buffer.
} // if (!disable_chunk_usage_histograms)
return trace_stats;
}
void TracingServiceImpl::EmitUuid(TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
SetServiceTracePacketHeader(packet.get());
auto* uuid = packet->set_trace_uuid();
uuid->set_lsb(tracing_session->trace_uuid.lsb());
uuid->set_msb(tracing_session->trace_uuid.msb());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
void TracingServiceImpl::MaybeEmitTraceConfig(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
if (tracing_session->did_emit_initial_packets)
return;
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
SetServiceTracePacketHeader(packet.get());
tracing_session->config.Serialize(packet->set_trace_config());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
void TracingServiceImpl::EmitSystemInfo(std::vector<TracePacket>* packets) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
auto* info = packet->set_system_info();
base::SystemInfo sys_info = base::GetSystemInfo();
info->set_tracing_service_version(base::GetVersionString());
if (sys_info.timezone_off_mins.has_value())
info->set_timezone_off_mins(*sys_info.timezone_off_mins);
if (sys_info.utsname_info.has_value()) {
auto* utsname_info = info->set_utsname();
utsname_info->set_sysname(sys_info.utsname_info->sysname);
utsname_info->set_version(sys_info.utsname_info->version);
utsname_info->set_machine(sys_info.utsname_info->machine);
utsname_info->set_release(sys_info.utsname_info->release);
}
if (sys_info.page_size.has_value())
info->set_page_size(*sys_info.page_size);
if (sys_info.system_ram_bytes.has_value())
info->set_system_ram_bytes(*sys_info.system_ram_bytes);
if (sys_info.num_cpus.has_value())
info->set_num_cpus(*sys_info.num_cpus);
if (!sys_info.android_build_fingerprint.empty())
info->set_android_build_fingerprint(sys_info.android_build_fingerprint);
if (!sys_info.android_device_manufacturer.empty())
info->set_android_device_manufacturer(sys_info.android_device_manufacturer);
if (sys_info.android_sdk_version.has_value())
info->set_android_sdk_version(*sys_info.android_sdk_version);
if (!sys_info.android_soc_model.empty())
info->set_android_soc_model(sys_info.android_soc_model);
if (!sys_info.android_guest_soc_model.empty())
info->set_android_guest_soc_model(sys_info.android_guest_soc_model);
if (!sys_info.android_hardware_revision.empty())
info->set_android_hardware_revision(sys_info.android_hardware_revision);
if (!sys_info.android_storage_model.empty())
info->set_android_storage_model(sys_info.android_storage_model);
if (!sys_info.android_ram_model.empty())
info->set_android_ram_model(sys_info.android_ram_model);
if (!sys_info.android_serial_console.empty())
info->set_android_serial_console(sys_info.android_serial_console);
SetServiceTracePacketHeader(packet.get());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
void TracingServiceImpl::EmitTraceProvenance(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
packet->set_timestamp(static_cast<uint64_t>(clock_->GetBootTimeNs().count()));
auto* provenance = packet->set_trace_provenance();
for (const BufferID buf_id : tracing_session->buffers_index) {
auto* buffer_proto = provenance->add_buffers();
TraceBuffer* buf = GetBufferByID(buf_id);
if (!buf)
continue;
for (auto it = buf->writer_stats().GetIterator(); it; ++it) {
ProducerID producer_id;
WriterID writer_id;
GetProducerAndWriterID(it.key(), &producer_id, &writer_id);
ProducerEndpointImpl* producer = GetProducer(producer_id);
// TODO(primiano): here we default to kDefaultMachineID for disconnected
// producers. Alternatively we can refactor the writer stats key to
// include the machine ID (e.g. MachineAndProducerAndWriterID instead of
// ProducerAndWriterID)
MachineID machine_id = producer ? producer->client_identity().machine_id()
: kDefaultMachineID;
auto* sequence_proto = buffer_proto->add_sequences();
sequence_proto->set_id(tracing_session->GetPacketSequenceID(
machine_id, producer_id, writer_id));
sequence_proto->set_producer_id(static_cast<int32_t>(producer_id));
}
}
SetServiceTracePacketHeader(packet.get());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
void TracingServiceImpl::MaybeEmitRemoteSystemInfo(
std::vector<TracePacket>* packets) {
std::unordered_set<MachineID> did_emit_machines;
for (const auto& id_and_relay_client : relay_clients_) {
const auto& relay_client = id_and_relay_client.second;
auto machine_id = relay_client->machine_id();
if (did_emit_machines.find(machine_id) != did_emit_machines.end())
continue; // Already emitted for the machine (e.g. multiple clients).
if (relay_client->serialized_system_info().empty()) {
PERFETTO_DLOG("System info not provided for machine ID = %" PRIu32,
machine_id);
continue;
}
// Don't emit twice for the same machine.
did_emit_machines.insert(machine_id);
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
auto& system_info = relay_client->serialized_system_info();
packet->AppendBytes(kTracePacketSystemInfoFieldId, system_info.data(),
system_info.size());
// Relay path: stamps each remote machine's own id, not the adopted local
// one, so it can't use SetServiceTracePacketHeader.
packet->set_machine_id(machine_id);
packet->set_trusted_uid(static_cast<int32_t>(uid_));
packet->set_trusted_packet_sequence_id(kServicePacketSequenceID);
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
}
void TracingServiceImpl::EmitLifecycleEvents(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
using TimestampedPacket =
std::pair<int64_t /* ts */, std::vector<uint8_t> /* serialized packet */>;
std::vector<TimestampedPacket> timestamped_packets;
for (auto& event : tracing_session->lifecycle_events) {
for (int64_t ts : event.timestamps) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
packet->set_timestamp(static_cast<uint64_t>(ts));
SetServiceTracePacketHeader(packet.get());
auto* service_event = packet->set_service_event();
service_event->AppendVarInt(event.field_id, 1);
timestamped_packets.emplace_back(ts, packet.SerializeAsArray());
}
event.timestamps.clear();
}
if (tracing_session->slow_start_event.has_value()) {
const TracingSession::ArbitraryLifecycleEvent& event =
*tracing_session->slow_start_event;
timestamped_packets.emplace_back(event.timestamp, std::move(event.data));
}
tracing_session->slow_start_event.reset();
for (auto& event : tracing_session->last_flush_events) {
timestamped_packets.emplace_back(event.timestamp, std::move(event.data));
}
tracing_session->last_flush_events.clear();
for (size_t i = 0; i < tracing_session->buffer_cloned_timestamps.size();
i++) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
int64_t ts = tracing_session->buffer_cloned_timestamps[i];
packet->set_timestamp(static_cast<uint64_t>(ts));
SetServiceTracePacketHeader(packet.get());
auto* service_event = packet->set_service_event();
service_event->set_buffer_cloned(static_cast<uint32_t>(i));
timestamped_packets.emplace_back(ts, packet.SerializeAsArray());
}
tracing_session->buffer_cloned_timestamps.clear();
// We sort by timestamp here to ensure that the "sequence" of lifecycle
// packets has monotonic timestamps like other sequences in the trace.
// Note that these events could still be out of order with respect to other
// events on the service packet sequence (e.g. trigger received packets).
std::sort(timestamped_packets.begin(), timestamped_packets.end(),
[](const TimestampedPacket& a, const TimestampedPacket& b) {
return a.first < b.first;
});
for (auto& pair : timestamped_packets)
SerializeAndAppendPacket(packets, std::move(pair.second));
}
void TracingServiceImpl::MaybeEmitRemoteClockSync(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
if (tracing_session->did_emit_remote_clock_sync_)
return;
std::unordered_set<MachineID> did_emit_machines;
for (const auto& id_and_relay_client : relay_clients_) {
const auto& relay_client = id_and_relay_client.second;
auto machine_id = relay_client->machine_id();
if (did_emit_machines.find(machine_id) != did_emit_machines.end())
continue; // Already emitted for the machine (e.g. multiple clients).
auto& sync_clock_snapshots = relay_client->synced_clocks();
if (sync_clock_snapshots.empty()) {
PERFETTO_DLOG("Clock not synchronized for machine ID = %" PRIu32,
machine_id);
continue;
}
// Don't emit twice for the same machine.
did_emit_machines.insert(machine_id);
protozero::HeapBuffered<protos::pbzero::TracePacket> sync_packet;
sync_packet->set_machine_id(machine_id);
sync_packet->set_trusted_uid(static_cast<int32_t>(uid_));
auto* remote_clock_sync = sync_packet->set_remote_clock_sync();
for (const auto& sync_exchange : relay_client->synced_clocks()) {
auto* sync_exchange_msg = remote_clock_sync->add_synced_clocks();
auto* client_snapshots = sync_exchange_msg->set_client_clocks();
for (const auto& client_clock : sync_exchange.client_clocks) {
auto* clock = client_snapshots->add_clocks();
clock->set_clock_id(client_clock.clock_id);
clock->set_timestamp(client_clock.timestamp);
}
auto* host_snapshots = sync_exchange_msg->set_host_clocks();
for (const auto& host_clock : sync_exchange.host_clocks) {
auto* clock = host_snapshots->add_clocks();
clock->set_clock_id(host_clock.clock_id);
clock->set_timestamp(host_clock.timestamp);
}
}
SerializeAndAppendPacket(packets, sync_packet.SerializeAsArray());
}
tracing_session->did_emit_remote_clock_sync_ = true;
}
void TracingServiceImpl::MaybeEmitProtoVmInstances(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
if (tracing_session->did_emit_protovm_instances_)
return;
std::optional<protozero::HeapBuffered<protos::pbzero::TracePacket>>
maybe_packet;
auto get_packet_protovms =
[&, protovms = static_cast<protos::pbzero::TracePacket::ProtoVms*>(
nullptr)]() mutable {
if (protovms) {
return protovms;
}
maybe_packet.emplace();
protovms = maybe_packet.value()->set_protovms();
return protovms;
};
for (auto buffer_id : tracing_session->buffers_index) {
auto it = buffers_.find(buffer_id);
PERFETTO_CHECK(it != buffers_.cend());
if (it->second->buf_type() != TraceBuffer::BufType::kV2) {
continue;
}
auto* buffer = static_cast<TraceBufferV2*>(it->second.get());
for (const TraceBufferV2::Vm& vm : buffer->GetProtoVmInstances()) {
auto* vm_instance = get_packet_protovms()->add_instance();
vm_instance->AppendString(
protos::pbzero::TracePacket::ProtoVms::Instance::kProgramFieldNumber,
vm.instance->SerializeProgram());
auto* state = vm_instance->set_state();
vm.instance->SerializeIncrementalState(state);
vm_instance->set_memory_limit_kb(vm.memory_limit_kb);
for (ProducerID producer_id : vm.producers) {
vm_instance->add_producer_id(producer_id);
}
}
}
if (maybe_packet) {
SetServiceTracePacketHeader(maybe_packet->get());
SerializeAndAppendPacket(packets, maybe_packet->SerializeAsArray());
}
tracing_session->did_emit_protovm_instances_ = true;
}
void TracingServiceImpl::EmitExtensionDescriptors(
TracingSession*,
std::vector<TracePacket>* packets) {
for (const auto& desc : init_opts_.extension_descriptors) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
SetServiceTracePacketHeader(packet.get());
auto* ext = packet->set_extension_descriptor();
if (desc.gzipped) {
ext->set_extension_set_gzip(desc.start, desc.size);
} else {
ext->AppendBytes(
protos::pbzero::ExtensionDescriptor::kExtensionSetFieldNumber,
desc.start, desc.size);
}
if (!desc.name.empty()) {
ext->set_file_name(desc.name);
}
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
}
void TracingServiceImpl::MaybeEmitCloneTrigger(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
if (tracing_session->did_emit_initial_packets)
return;
if (tracing_session->clone_trigger.has_value()) {
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
auto* trigger = packet->set_clone_snapshot_trigger();
const auto& info = tracing_session->clone_trigger.value();
trigger->set_trigger_name(info.trigger_name);
trigger->set_producer_name(info.producer_name);
trigger->set_trusted_producer_uid(static_cast<int32_t>(info.producer_uid));
trigger->set_stop_delay_ms(info.trigger_delay_ms);
packet->set_timestamp(info.boot_time_ns);
SetServiceTracePacketHeader(packet.get());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
}
}
void TracingServiceImpl::MaybeEmitReceivedTriggers(
TracingSession* tracing_session,
std::vector<TracePacket>* packets) {
PERFETTO_DCHECK(tracing_session->num_triggers_emitted_into_trace <=
tracing_session->received_triggers.size());
for (size_t i = tracing_session->num_triggers_emitted_into_trace;
i < tracing_session->received_triggers.size(); ++i) {
const auto& info = tracing_session->received_triggers[i];
protozero::HeapBuffered<protos::pbzero::TracePacket> packet;
auto* trigger = packet->set_trigger();
trigger->set_trigger_name(info.trigger_name);
trigger->set_producer_name(info.producer_name);
trigger->set_trusted_producer_uid(static_cast<int32_t>(info.producer_uid));
trigger->set_stop_delay_ms(info.trigger_delay_ms);
packet->set_timestamp(info.boot_time_ns);
SetServiceTracePacketHeader(packet.get());
SerializeAndAppendPacket(packets, packet.SerializeAsArray());
++tracing_session->num_triggers_emitted_into_trace;
}
}
void TracingServiceImpl::MaybeLogUploadEvent(const TraceConfig& cfg,
const base::Uuid& uuid,
PerfettoStatsdAtom atom,
const std::string& trigger_name) {
if (!ShouldLogEvent(cfg))
return;
PERFETTO_CHECK(uuid); // The UUID must be set at this point.
android_stats::MaybeLogUploadEvent(atom, uuid.lsb(), uuid.msb(),
trigger_name);
}
void TracingServiceImpl::MaybeLogTriggerEvent(const TraceConfig& cfg,
const base::Uuid& uuid,
PerfettoTriggerAtom atom,
const std::string& trigger_name) {
if (!ShouldLogEvent(cfg))
return;
PERFETTO_CHECK(uuid); // The UUID must be set at this point.
android_stats::MaybeLogTriggerEvent(atom, uuid.lsb(), trigger_name);
}
size_t TracingServiceImpl::PurgeExpiredAndCountTriggerInWindow(
int64_t now_ns,
uint64_t trigger_name_hash) {
constexpr int64_t kOneDayInNs = 24ll * 60 * 60 * 1000 * 1000 * 1000;
PERFETTO_DCHECK(
std::is_sorted(trigger_history_.begin(), trigger_history_.end()));
size_t remove_count = 0;
size_t trigger_count = 0;
for (const TriggerHistory& h : trigger_history_) {
if (h.timestamp_ns < now_ns - kOneDayInNs) {
remove_count++;
} else if (h.name_hash == trigger_name_hash) {
trigger_count++;
}
}
trigger_history_.erase_front(remove_count);
return trigger_count;
}
base::Status TracingServiceImpl::FlushAndCloneSession(
ConsumerEndpointImpl* consumer,
ConsumerEndpoint::CloneSessionArgs args) {
PERFETTO_DCHECK_THREAD(thread_checker_);
auto clone_target = FlushFlags::CloneTarget::kUnknown;
TracingSession* session = nullptr;
if (args.for_bugreport) {
clone_target = FlushFlags::CloneTarget::kBugreport;
}
if (args.tsid != 0) {
if (args.tsid == kBugreportSessionId) {
// This branch is only here to support the legacy protocol where we could
// clone only a single session using the magic ID kBugreportSessionId.
// The newer perfetto --clone-all-for-bugreport first queries the existing
// sessions and then issues individual clone requests specifying real
// session IDs, setting args.{for_bugreport,skip_trace_filter}=true.
PERFETTO_LOG("Looking for sessions for bugreport");
session = FindTracingSessionWithMaxBugreportScore();
if (!session) {
return base::ErrStatus(
"No tracing sessions eligible for bugreport found");
}
args.tsid = session->id;
clone_target = FlushFlags::CloneTarget::kBugreport;
args.skip_trace_filter = true;
} else {
session = GetTracingSession(args.tsid);
}
} else if (!args.unique_session_name.empty()) {
session = GetTracingSessionByUniqueName(args.unique_session_name);
}
if (!session) {
return base::ErrStatus("Tracing session not found");
}
// Skip the UID check for sessions marked with a bugreport_score > 0.
// Those sessions, by design, can be stolen by any other consumer for the
// sake of creating snapshots for bugreports.
if (!session->IsCloneAllowed(consumer->uid_)) {
return PERFETTO_SVC_ERR("Not allowed to clone a session from another UID");
}
// The new logic we use to clone 'write_into_file' session relies on the
// 'buffer_clone_preserve_read_iter' flag being true; see b/448604718.
//
// The old logic ignored |session->write_into_file| when doing clone.
// Therefore, if the 'buffer_clone_preserve_read_iter' flag is false, we
// ignore the file to make the new logic behave like the old logic.
bool clone_session_write_into_file =
PERFETTO_FLAGS(BUFFER_CLONE_PRESERVE_READ_ITER) &&
session->write_into_file;
if (clone_session_write_into_file) {
if (!args.output_file_fd) {
return PERFETTO_SVC_ERR(
"Failed to clone 'write_into_file' session: a file descriptor is "
"required to copy existing file");
}
base::FlushFile(*session->write_into_file);
base::Status status =
base::CopyFileContents(*session->write_into_file, *args.output_file_fd);
if (!status.ok()) {
return PERFETTO_SVC_ERR(
"Failed to clone 'write_into_file' session: failed to copy existing "
"file: %s",
status.c_message());
}
} else {
// The client always sends a FD because when it asks to CloneSession,
// it doesn't know if the session being cloned is WIF or not. If it's
// not we should just ignore the file, the client will readback via IPC
// as usual in that case.
args.output_file_fd.reset();
}
// If any of the buffers are marked as clear_before_clone, reset them before
// issuing the Flush(kCloneReason).
size_t buf_idx = 0;
for (BufferID src_buf_id : session->buffers_index) {
if (!session->config.buffers()[buf_idx++].clear_before_clone())
continue;
auto buf_iter = buffers_.find(src_buf_id);
PERFETTO_CHECK(buf_iter != buffers_.end());
std::unique_ptr<TraceBuffer>& buf = buf_iter->second;
// No need to reset the buffer if nothing has been written into it yet.
// This is the canonical case if producers behive nicely and don't timeout
// the handling of writes during the flush.
// This check avoids a useless re-mmap upon every Clone() if the buffer is
// already empty (when used in combination with `transfer_on_clone`).
if (!buf->has_data())
continue;
// Some leftover data was left in the buffer. Recreate it to empty it.
const auto buf_policy = buf->overwrite_policy();
const auto buf_size = buf->size();
const auto buf_type = buf->buf_type();
std::unique_ptr<TraceBuffer> old_buf = std::move(buf);
switch (buf_type) {
case TraceBuffer::kV1:
buf = TraceBufferV1::Create(buf_size, buf_policy);
break;
case TraceBuffer::kV2:
buf = TraceBufferV2::Create(buf_size, buf_policy);
break;
}
if (!buf) {
// This is extremely rare but could happen on 32-bit. If the new buffer
// allocation failed, put back the buffer where it was and fail the clone.
// We cannot leave the original tracing session buffer-less as it would
// cause crashes when data sources commit new data.
buf = std::move(old_buf);
return base::ErrStatus(
"Buffer allocation failed while attempting to clone");
}
}
auto weak_consumer = consumer->GetWeakPtr();
const PendingCloneID clone_id = session->last_pending_clone_id_++;
auto& clone_op = session->pending_clones[clone_id];
clone_op.pending_flush_cnt = 0;
// Pre-initialize these vectors just as an optimization to avoid reallocations
// in DoCloneBuffers().
clone_op.buffers.reserve(session->buffers_index.size());
clone_op.buffer_cloned_timestamps.reserve(session->buffers_index.size());
clone_op.weak_consumer = weak_consumer;
clone_op.skip_trace_filter = args.skip_trace_filter;
if (!args.clone_trigger_name.empty()) {
clone_op.clone_trigger = {
args.clone_trigger_boot_time_ns, args.clone_trigger_name,
args.clone_trigger_producer_name,
args.clone_trigger_trusted_producer_uid, args.clone_trigger_delay_ms};
}
if (args.output_file_fd) {
clone_op.output_file_fd = std::move(args.output_file_fd);
}
// Issue separate flush requests for separate buffer groups. The buffer marked
// as transfer_on_clone will be flushed and cloned separately: even if they're
// slower (like in the case of Winscope tracing), they will not delay the
// snapshot of the other buffers.
//
// In the future we might want to split the buffer into more groups and maybe
// allow this to be configurable.
std::array<std::set<BufferID>, 2> bufs_groups;
for (size_t i = 0; i < session->buffers_index.size(); i++) {
if (session->config.buffers()[i].transfer_on_clone()) {
bufs_groups[0].insert(session->buffers_index[i]);
} else {
bufs_groups[1].insert(session->buffers_index[i]);
}
}
SnapshotLifecycleEvent(
session, protos::pbzero::TracingServiceEvent::kFlushStartedFieldNumber,
/*snapshot_clocks=*/true);
clone_op.pending_flush_cnt = bufs_groups.size();
clone_op.clone_started_timestamp_ns = clock_->GetBootTimeNs().count();
for (const std::set<BufferID>& buf_group : bufs_groups) {
FlushDataSourceInstances(
session, 0,
GetFlushableDataSourceInstancesForBuffers(session, buf_group),
[tsid = session->id, clone_id, buf_group, this](bool final_flush) {
OnFlushDoneForClone(tsid, clone_id, buf_group, final_flush);
},
FlushFlags(FlushFlags::Initiator::kTraced,
FlushFlags::Reason::kTraceClone, clone_target));
}
return base::OkStatus();
}
std::map<ProducerID, std::vector<DataSourceInstanceID>>
TracingServiceImpl::GetFlushableDataSourceInstancesForBuffers(
TracingSession* session,
const std::set<BufferID>& bufs) {
std::map<ProducerID, std::vector<DataSourceInstanceID>> data_source_instances;
for (const auto& [producer_id, ds_inst] : session->data_source_instances) {
// TODO(ddiproietto): Consider if we should skip instances if ds_inst.state
// != DataSourceInstance::STARTED
if (ds_inst.no_flush) {
continue;
}
if (!bufs.count(static_cast<BufferID>(ds_inst.config.target_buffer()))) {
continue;
}
data_source_instances[producer_id].push_back(ds_inst.instance_id);
}
return data_source_instances;
}
void TracingServiceImpl::OnFlushDoneForClone(TracingSessionID tsid,
PendingCloneID clone_id,
const std::set<BufferID>& buf_ids,
bool final_flush_outcome) {
TracingSession* src = GetTracingSession(tsid);
// The session might be gone by the time we try to clone it.
if (!src) {
return;
}
auto it = src->pending_clones.find(clone_id);
if (it == src->pending_clones.end()) {
return;
}
auto& clone_op = it->second;
if (final_flush_outcome == false) {
clone_op.flush_failed = true;
}
base::Status result;
base::Uuid uuid;
// First clone the flushed TraceBuffer(s). This can fail because of ENOMEM. If
// it happens bail out early before creating any session.
if (!DoCloneBuffers(*src, buf_ids, &clone_op)) {
result = PERFETTO_SVC_ERR("Buffer allocation failed");
}
bool was_write_into_file = false;
if (result.ok()) {
UpdateMemoryGuardrail();
if (--clone_op.pending_flush_cnt != 0) {
// Wait for more pending flushes.
return;
}
PERFETTO_LOG("FlushAndCloneSession(%" PRIu64 ") started, success=%d", tsid,
final_flush_outcome);
if (clone_op.weak_consumer) {
was_write_into_file = static_cast<bool>(clone_op.output_file_fd);
result = FinishCloneSession(
&*clone_op.weak_consumer, tsid, std::move(clone_op.buffers),
std::move(clone_op.buffer_cloned_timestamps),
clone_op.skip_trace_filter, !clone_op.flush_failed,
clone_op.clone_trigger, &uuid, clone_op.clone_started_timestamp_ns,
std::move(clone_op.output_file_fd));
}
} // if (result.ok())
if (clone_op.weak_consumer) {
clone_op.weak_consumer->consumer_->OnSessionCloned(
{result.ok(), result.message(), uuid, was_write_into_file});
}
src->pending_clones.erase(it);
UpdateMemoryGuardrail();
}
bool TracingServiceImpl::DoCloneBuffers(const TracingSession& src,
const std::set<BufferID>& buf_ids,
PendingClone* clone_op) {
PERFETTO_DCHECK(src.num_buffers() == src.config.buffers().size());
clone_op->buffers.resize(src.buffers_index.size());
clone_op->buffer_cloned_timestamps.resize(src.buffers_index.size());
int64_t now = clock_->GetBootTimeNs().count();
for (size_t buf_idx = 0; buf_idx < src.buffers_index.size(); buf_idx++) {
BufferID src_buf_id = src.buffers_index[buf_idx];
if (buf_ids.count(src_buf_id) == 0)
continue;
auto buf_iter = buffers_.find(src_buf_id);
PERFETTO_CHECK(buf_iter != buffers_.end());
std::unique_ptr<TraceBuffer>& src_buf = buf_iter->second;
std::unique_ptr<TraceBuffer> new_buf;
if (src.config.buffers()[buf_idx].transfer_on_clone()) {
const auto buf_policy = src_buf->overwrite_policy();
const auto buf_size = src_buf->size();
const auto buf_type = src_buf->buf_type();
new_buf = std::move(src_buf);
switch (buf_type) {
case TraceBuffer::kV1:
src_buf = TraceBufferV1::Create(buf_size, buf_policy);
break;
case TraceBuffer::kV2:
src_buf = TraceBufferV2::Create(buf_size, buf_policy);
break;
}
if (!src_buf) {
// If the allocation fails put the buffer back and let the code below
// handle the failure gracefully.
src_buf = std::move(new_buf);
}
} else {
new_buf = src_buf->CloneReadOnly();
}
if (!new_buf.get()) {
return false;
}
clone_op->buffers[buf_idx] = std::move(new_buf);
clone_op->buffer_cloned_timestamps[buf_idx] = now;
}
return true;
}
base::Status TracingServiceImpl::FinishCloneSession(
ConsumerEndpointImpl* consumer,
TracingSessionID src_tsid,
std::vector<std::unique_ptr<TraceBuffer>> buf_snaps,
std::vector<int64_t> buf_cloned_timestamps,
bool skip_trace_filter,
bool final_flush_outcome,
std::optional<TriggerInfo> clone_trigger,
base::Uuid* new_uuid,
int64_t clone_started_timestamp_ns,
base::ScopedFile output_file_fd) {
PERFETTO_DLOG("CloneSession(%" PRIu64
", skip_trace_filter=%d) started, consumer uid: %d",
src_tsid, skip_trace_filter, static_cast<int>(consumer->uid_));
TracingSession* src = GetTracingSession(src_tsid);
// The session might be gone by the time we try to clone it.
if (!src)
return PERFETTO_SVC_ERR("session not found");
if (consumer->tracing_session_id_) {
return PERFETTO_SVC_ERR(
"The consumer is already attached to another tracing session");
}
std::vector<BufferID> buf_ids =
buffer_ids_.AllocateMultiple(buf_snaps.size());
if (buf_ids.size() != buf_snaps.size()) {
return PERFETTO_SVC_ERR("Buffer id allocation failed");
}
PERFETTO_CHECK(std::none_of(
buf_snaps.begin(), buf_snaps.end(),
[](const std::unique_ptr<TraceBuffer>& buf) { return buf == nullptr; }));
const TracingSessionID tsid = ++last_tracing_session_id_;
TracingSession* cloned_session =
&tracing_sessions_
.emplace(std::piecewise_construct, std::forward_as_tuple(tsid),
std::forward_as_tuple(tsid, consumer, src->config,
weak_runner_.task_runner()))
.first->second;
// Generate a new UUID for the cloned session, but preserve the LSB. In some
// contexts the LSB is used to tie the trace back to the statsd subscription
// that triggered it. See the corresponding code in perfetto_cmd.cc which
// reads at triggering_subscription_id().
const int64_t orig_uuid_lsb = src->trace_uuid.lsb();
cloned_session->state = TracingSession::CLONED_READ_ONLY;
cloned_session->trace_uuid = base::Uuidv4();
cloned_session->trace_uuid.set_lsb(orig_uuid_lsb);
*new_uuid = cloned_session->trace_uuid;
for (size_t i = 0; i < buf_snaps.size(); i++) {
BufferID buf_global_id = buf_ids[i];
std::unique_ptr<TraceBuffer>& buf = buf_snaps[i];
// This is only needed for transfer_on_clone. Other buffers are already
// marked as read-only by CloneReadOnly(). We cannot do this early because
// in case of an allocation failure we will put std::move() the original
// buffer back in its place and in that case should not be made read-only.
buf->set_read_only();
buffers_.emplace(buf_global_id, std::move(buf));
cloned_session->buffers_index.emplace_back(buf_global_id);
}
UpdateMemoryGuardrail();
// Copy over relevant state that we want to persist in the cloned session.
// Mostly stats and metadata that is emitted in the trace file by the service.
// Also clear the received trigger list in the main tracing session. A
// CLONE_SNAPSHOT session can go in ring buffer mode for several hours and get
// snapshotted several times. This causes two issues with `received_triggers`:
// 1. Adding noise in the cloned trace emitting triggers that happened too
// far back (see b/290799105).
// 2. Bloating memory (see b/290798988).
cloned_session->should_emit_stats = true;
cloned_session->clone_trigger = clone_trigger;
cloned_session->received_triggers = std::move(src->received_triggers);
src->received_triggers.clear();
src->num_triggers_emitted_into_trace = 0;
cloned_session->lifecycle_events =
std::vector<TracingSession::LifecycleEvent>(src->lifecycle_events);
cloned_session->slow_start_event = src->slow_start_event;
cloned_session->last_flush_events = src->last_flush_events;
cloned_session->initial_clock_snapshot = src->initial_clock_snapshot;
cloned_session->clock_snapshot_ring_buffer = src->clock_snapshot_ring_buffer;
cloned_session->invalid_packets = src->invalid_packets;
cloned_session->flushes_requested = src->flushes_requested;
cloned_session->flushes_succeeded = src->flushes_succeeded;
cloned_session->flushes_failed = src->flushes_failed;
cloned_session->compress_deflate = src->compress_deflate;
if (src->trace_filter && !skip_trace_filter) {
// Copy the trace filter, unless it's a clone-for-bugreport (b/317065412).
cloned_session->trace_filter.reset(
new protozero::MessageFilter(src->trace_filter->config()));
}
cloned_session->buffer_cloned_timestamps = std::move(buf_cloned_timestamps);
SetSingleLifecycleEvent(
cloned_session,
protos::pbzero::TracingServiceEvent::kCloneStartedFieldNumber,
clone_started_timestamp_ns);
SnapshotLifecycleEvent(
cloned_session,
protos::pbzero::TracingServiceEvent::kTracingDisabledFieldNumber,
true /* snapshot_clocks */);
PERFETTO_DLOG("Consumer (uid:%d) cloned tracing session %" PRIu64
" -> %" PRIu64,
static_cast<int>(consumer->uid_), src_tsid, tsid);
consumer->tracing_session_id_ = tsid;
cloned_session->final_flush_outcome = final_flush_outcome
? TraceStats::FINAL_FLUSH_SUCCEEDED
: TraceStats::FINAL_FLUSH_FAILED;
if (output_file_fd) {
cloned_session->write_into_file = std::move(output_file_fd);
cloned_session->write_period_ms = 0;
// Buffers are flushed, no need to flush again before reading into file.
ReadBuffersIntoFile(cloned_session->id,
/* async_flush_buffers_before_read= */ false);
}
return base::OkStatus();
}
} // namespace tracing_service
} // namespace perfetto