tracing_muxer_impl.cc | Explore in Territory

/*
 * Copyright (C) 2019 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/internal/tracing_muxer_impl.h"

#include <algorithm>
#include <atomic>
#include <mutex>
#include <optional>
#include <vector>

#include "perfetto/base/build_config.h"
#include "perfetto/base/logging.h"
#include "perfetto/base/task_runner.h"
#include "perfetto/base/time.h"
#include "perfetto/ext/base/hash.h"
#include "perfetto/ext/base/thread_checker.h"
#include "perfetto/ext/base/waitable_event.h"
#include "perfetto/ext/tracing/core/shared_memory_arbiter.h"
#include "perfetto/ext/tracing/core/trace_packet.h"
#include "perfetto/ext/tracing/core/trace_stats.h"
#include "perfetto/ext/tracing/core/trace_writer.h"
#include "perfetto/ext/tracing/core/tracing_service.h"
#include "perfetto/tracing/buffer_exhausted_policy.h"
#include "perfetto/tracing/core/data_source_config.h"
#include "perfetto/tracing/core/tracing_service_state.h"
#include "perfetto/tracing/data_source.h"
#include "perfetto/tracing/internal/data_source_internal.h"
#include "perfetto/tracing/internal/interceptor_trace_writer.h"
#include "perfetto/tracing/internal/tracing_backend_fake.h"
#include "perfetto/tracing/trace_writer_base.h"
#include "perfetto/tracing/tracing.h"
#include "perfetto/tracing/tracing_backend.h"
#include "src/tracing/core/null_trace_writer.h"
#include "src/tracing/internal/tracing_muxer_fake.h"

#include "protos/perfetto/config/interceptor_config.gen.h"

#if PERFETTO_BUILDFLAG(PERFETTO_OS_WIN)
#include <io.h>  // For dup()
#else
#include <unistd.h>  // For dup()
#endif

namespace perfetto {
namespace internal {

namespace {

RegisteredDataSource;

// A task runner which prevents calls to DataSource::Trace() while an operation
// is in progress. Used to guard against unexpected re-entrancy where the
// user-provided task runner implementation tries to enter a trace point under
// the hood.
class NonReentrantTaskRunner : public base::TaskRunner { … };

class StopArgsImpl : public DataSourceBase::StopArgs { … };

class FlushArgsImpl : public DataSourceBase::FlushArgs { … };

// Holds an earlier TracingMuxerImpl instance after ResetForTesting() is called.
static TracingMuxerImpl* g_prev_instance{ … };

template <typename RegisteredBackend>
struct CompareBackendByType { … };

}  // namespace

// ----- Begin of TracingMuxerImpl::ProducerImpl
TracingMuxerImpl::ProducerImpl::ProducerImpl(
    TracingMuxerImpl* muxer,
    TracingBackendId backend_id,
    uint32_t shmem_batch_commits_duration_ms,
    bool shmem_direct_patching_enabled)
    : … { … }

TracingMuxerImpl::ProducerImpl::~ProducerImpl() { … }

void TracingMuxerImpl::ProducerImpl::Initialize(
    std::unique_ptr<ProducerEndpoint> endpoint) { … }

void TracingMuxerImpl::ProducerImpl::OnConnect() { … }

void TracingMuxerImpl::ProducerImpl::OnDisconnect() { … }

void TracingMuxerImpl::ProducerImpl::DisposeConnection() { … }

void TracingMuxerImpl::ProducerImpl::OnTracingSetup() { … }

void TracingMuxerImpl::ProducerImpl::OnStartupTracingSetup() { … }

void TracingMuxerImpl::ProducerImpl::SetupDataSource(
    DataSourceInstanceID id,
    const DataSourceConfig& cfg) { … }

void TracingMuxerImpl::ProducerImpl::StartDataSource(DataSourceInstanceID id,
                                                     const DataSourceConfig&) { … }

void TracingMuxerImpl::ProducerImpl::StopDataSource(DataSourceInstanceID id) { … }

void TracingMuxerImpl::ProducerImpl::Flush(
    FlushRequestID flush_id,
    const DataSourceInstanceID* instances,
    size_t instance_count,
    FlushFlags flush_flags) { … }

void TracingMuxerImpl::ProducerImpl::ClearIncrementalState(
    const DataSourceInstanceID* instances,
    size_t instance_count) { … }

bool TracingMuxerImpl::ProducerImpl::SweepDeadServices() { … }

void TracingMuxerImpl::ProducerImpl::SendOnConnectTriggers() { … }

void TracingMuxerImpl::ProducerImpl::NotifyFlushForDataSourceDone(
    DataSourceInstanceID ds_id,
    FlushRequestID flush_id) { … }

// ----- End of TracingMuxerImpl::ProducerImpl methods.

// ----- Begin of TracingMuxerImpl::ConsumerImpl
TracingMuxerImpl::ConsumerImpl::ConsumerImpl(TracingMuxerImpl* muxer,
                                             BackendType backend_type,
                                             TracingSessionGlobalID session_id)
    : … { … }

TracingMuxerImpl::ConsumerImpl::~ConsumerImpl() { … }

void TracingMuxerImpl::ConsumerImpl::Initialize(
    std::unique_ptr<ConsumerEndpoint> endpoint) { … }

void TracingMuxerImpl::ConsumerImpl::OnConnect() { … }

void TracingMuxerImpl::ConsumerImpl::OnDisconnect() { … }

void TracingMuxerImpl::ConsumerImpl::Disconnect() { … }

void TracingMuxerImpl::ConsumerImpl::OnTracingDisabled(
    const std::string& error) { … }

void TracingMuxerImpl::ConsumerImpl::NotifyStartComplete() { … }

void TracingMuxerImpl::ConsumerImpl::NotifyError(const TracingError& error) { … }

void TracingMuxerImpl::ConsumerImpl::NotifyStopComplete() { … }

void TracingMuxerImpl::ConsumerImpl::OnTraceData(
    std::vector<TracePacket> packets,
    bool has_more) { … }

void TracingMuxerImpl::ConsumerImpl::OnObservableEvents(
    const ObservableEvents& events) { … }

void TracingMuxerImpl::ConsumerImpl::OnSessionCloned(
    const OnSessionClonedArgs&) { … }

void TracingMuxerImpl::ConsumerImpl::OnTraceStats(
    bool success,
    const TraceStats& trace_stats) { … }

// The callbacks below are not used.
void TracingMuxerImpl::ConsumerImpl::OnDetach(bool) { … }
void TracingMuxerImpl::ConsumerImpl::OnAttach(bool, const TraceConfig&) { … }
// ----- End of TracingMuxerImpl::ConsumerImpl

// ----- Begin of TracingMuxerImpl::TracingSessionImpl

// TracingSessionImpl is the RAII object returned to API clients when they
// invoke Tracing::CreateTracingSession. They use it for starting/stopping
// tracing.

TracingMuxerImpl::TracingSessionImpl::TracingSessionImpl(
    TracingMuxerImpl* muxer,
    TracingSessionGlobalID session_id,
    BackendType backend_type)
    : … { … }

// Can be destroyed from any thread.
TracingMuxerImpl::TracingSessionImpl::~TracingSessionImpl() { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::Setup(const TraceConfig& cfg,
                                                 int fd) { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::Start() { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::ChangeTraceConfig(
    const TraceConfig& cfg) { … }

// Can be called from any thread except the service thread.
void TracingMuxerImpl::TracingSessionImpl::StartBlocking() { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::Flush(
    std::function<void(bool)> user_callback,
    uint32_t timeout_ms) { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::Stop() { … }

// Can be called from any thread except the service thread.
void TracingMuxerImpl::TracingSessionImpl::StopBlocking() { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::ReadTrace(ReadTraceCallback cb) { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::SetOnStartCallback(
    std::function<void()> cb) { … }

// Can be called from any thread
void TracingMuxerImpl::TracingSessionImpl::SetOnErrorCallback(
    std::function<void(TracingError)> cb) { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::SetOnStopCallback(
    std::function<void()> cb) { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::GetTraceStats(
    GetTraceStatsCallback cb) { … }

// Can be called from any thread.
void TracingMuxerImpl::TracingSessionImpl::QueryServiceState(
    QueryServiceStateCallback cb) { … }

// ----- End of TracingMuxerImpl::TracingSessionImpl

// ----- Begin of TracingMuxerImpl::StartupTracingSessionImpl

TracingMuxerImpl::StartupTracingSessionImpl::StartupTracingSessionImpl(
    TracingMuxerImpl* muxer,
    TracingSessionGlobalID session_id,
    BackendType backend_type)
    : … { … }

// Can be destroyed from any thread.
TracingMuxerImpl::StartupTracingSessionImpl::~StartupTracingSessionImpl() =
    default;

void TracingMuxerImpl::StartupTracingSessionImpl::Abort() { … }

// Must not be called from the SDK's internal thread.
void TracingMuxerImpl::StartupTracingSessionImpl::AbortBlocking() { … }

// ----- End of TracingMuxerImpl::StartupTracingSessionImpl

// static
TracingMuxer* TracingMuxer::instance_ = …;

// This is called by perfetto::Tracing::Initialize().
// Can be called on any thread. Typically, but not necessarily, that will be
// the embedder's main thread.
TracingMuxerImpl::TracingMuxerImpl(const TracingInitArgs& args)
    : … { … }

void TracingMuxerImpl::Initialize(const TracingInitArgs& args) { … }

void TracingMuxerImpl::AddConsumerBackend(TracingConsumerBackend* backend,
                                          BackendType type) { … }

void TracingMuxerImpl::AddProducerBackend(TracingProducerBackend* backend,
                                          BackendType type,
                                          const TracingInitArgs& args) { … }

TracingMuxerImpl::RegisteredProducerBackend*
TracingMuxerImpl::FindProducerBackendById(TracingBackendId id) { … }

TracingMuxerImpl::RegisteredProducerBackend*
TracingMuxerImpl::FindProducerBackendByType(BackendType type) { … }

TracingMuxerImpl::RegisteredConsumerBackend*
TracingMuxerImpl::FindConsumerBackendByType(BackendType type) { … }

void TracingMuxerImpl::AddBackends(const TracingInitArgs& args) { … }

// Can be called from any thread (but not concurrently).
bool TracingMuxerImpl::RegisterDataSource(
    const DataSourceDescriptor& descriptor,
    DataSourceFactory factory,
    DataSourceParams params,
    bool no_flush,
    DataSourceStaticState* static_state) { … }

// Can be called from any thread (but not concurrently).
void TracingMuxerImpl::UpdateDataSourceDescriptor(
    const DataSourceDescriptor& descriptor,
    const DataSourceStaticState* static_state) { … }

// Can be called from any thread (but not concurrently).
void TracingMuxerImpl::RegisterInterceptor(
    const InterceptorDescriptor& descriptor,
    InterceptorFactory factory,
    InterceptorBase::TLSFactory tls_factory,
    InterceptorBase::TracePacketCallback packet_callback) { … }

void TracingMuxerImpl::ActivateTriggers(
    const std::vector<std::string>& triggers,
    uint32_t ttl_ms) { … }

// Checks if there is any matching startup tracing data source instance for a
// new SetupDataSource call. If so, moves the data source to this tracing
// session (and its target buffer) and returns true, otherwise returns false.
static bool MaybeAdoptStartupTracingInDataSource(
    TracingBackendId backend_id,
    uint32_t backend_connection_id,
    DataSourceInstanceID instance_id,
    const DataSourceConfig& cfg,
    const std::vector<RegisteredDataSource>& data_sources) { … }

// Called by the service of one of the backends.
void TracingMuxerImpl::SetupDataSource(TracingBackendId backend_id,
                                       uint32_t backend_connection_id,
                                       DataSourceInstanceID instance_id,
                                       const DataSourceConfig& cfg) { … }

TracingMuxerImpl::FindDataSourceRes TracingMuxerImpl::SetupDataSourceImpl(
    const RegisteredDataSource& rds,
    TracingBackendId backend_id,
    uint32_t backend_connection_id,
    DataSourceInstanceID instance_id,
    const DataSourceConfig& cfg,
    TracingSessionGlobalID startup_session_id) { … }

// Called by the service of one of the backends.
void TracingMuxerImpl::StartDataSource(TracingBackendId backend_id,
                                       DataSourceInstanceID instance_id) { … }

void TracingMuxerImpl::StartDataSourceImpl(const FindDataSourceRes& ds) { … }

// Called by the service of one of the backends.
void TracingMuxerImpl::StopDataSource_AsyncBegin(
    TracingBackendId backend_id,
    DataSourceInstanceID instance_id) { … }

void TracingMuxerImpl::StopDataSource_AsyncBeginImpl(
    const FindDataSourceRes& ds) { … }

void TracingMuxerImpl::StopDataSource_AsyncEnd(TracingBackendId backend_id,
                                               uint32_t backend_connection_id,
                                               DataSourceInstanceID instance_id,
                                               const FindDataSourceRes& ds) { … }

void TracingMuxerImpl::ClearDataSourceIncrementalState(
    TracingBackendId backend_id,
    DataSourceInstanceID instance_id) { … }

bool TracingMuxerImpl::FlushDataSource_AsyncBegin(
    TracingBackendId backend_id,
    DataSourceInstanceID instance_id,
    FlushRequestID flush_id,
    FlushFlags flush_flags) { … }

void TracingMuxerImpl::FlushDataSource_AsyncEnd(
    TracingBackendId backend_id,
    uint32_t backend_connection_id,
    DataSourceInstanceID instance_id,
    const FindDataSourceRes& ds,
    FlushRequestID flush_id) { … }

void TracingMuxerImpl::SyncProducersForTesting() { … }

void TracingMuxerImpl::DestroyStoppedTraceWritersForCurrentThread() { … }

// Called both when a new data source is registered or when a new backend
// connects. In both cases we want to be sure we reflected the data source
// registrations on the backends.
void TracingMuxerImpl::UpdateDataSourcesOnAllBackends() { … }

void TracingMuxerImpl::UpdateDataSourceOnAllBackends(RegisteredDataSource& rds,
                                                     bool is_changed) { … }

void TracingMuxerImpl::SetupTracingSession(
    TracingSessionGlobalID session_id,
    const std::shared_ptr<TraceConfig>& trace_config,
    base::ScopedFile trace_fd) { … }

void TracingMuxerImpl::StartTracingSession(TracingSessionGlobalID session_id) { … }

void TracingMuxerImpl::ChangeTracingSessionConfig(
    TracingSessionGlobalID session_id,
    const TraceConfig& trace_config) { … }

void TracingMuxerImpl::FlushTracingSession(TracingSessionGlobalID session_id,
                                           uint32_t timeout_ms,
                                           std::function<void(bool)> callback) { … }

void TracingMuxerImpl::StopTracingSession(TracingSessionGlobalID session_id) { … }

void TracingMuxerImpl::DestroyTracingSession(
    TracingSessionGlobalID session_id) { … }

void TracingMuxerImpl::ReadTracingSessionData(
    TracingSessionGlobalID session_id,
    std::function<void(TracingSession::ReadTraceCallbackArgs)> callback) { … }

void TracingMuxerImpl::GetTraceStats(
    TracingSessionGlobalID session_id,
    TracingSession::GetTraceStatsCallback callback) { … }

void TracingMuxerImpl::QueryServiceState(
    TracingSessionGlobalID session_id,
    TracingSession::QueryServiceStateCallback callback) { … }

void TracingMuxerImpl::SetBatchCommitsDurationForTesting(
    uint32_t batch_commits_duration_ms,
    BackendType backend_type) { … }

bool TracingMuxerImpl::EnableDirectSMBPatchingForTesting(
    BackendType backend_type) { … }

TracingMuxerImpl::ConsumerImpl* TracingMuxerImpl::FindConsumer(
    TracingSessionGlobalID session_id) { … }

std::pair<TracingMuxerImpl::ConsumerImpl*,
          TracingMuxerImpl::RegisteredConsumerBackend*>
TracingMuxerImpl::FindConsumerAndBackend(TracingSessionGlobalID session_id) { … }

void TracingMuxerImpl::InitializeConsumer(TracingSessionGlobalID session_id) { … }

void TracingMuxerImpl::OnConsumerDisconnected(ConsumerImpl* consumer) { … }

void TracingMuxerImpl::SetMaxProducerReconnectionsForTesting(uint32_t count) { … }

void TracingMuxerImpl::OnProducerDisconnected(ProducerImpl* producer) { … }

void TracingMuxerImpl::SweepDeadBackends() { … }

TracingMuxerImpl::FindDataSourceRes TracingMuxerImpl::FindDataSource(
    TracingBackendId backend_id,
    DataSourceInstanceID instance_id) { … }

// Can be called from any thread.
std::unique_ptr<TraceWriterBase> TracingMuxerImpl::CreateTraceWriter(
    DataSourceStaticState* static_state,
    uint32_t data_source_instance_index,
    DataSourceState* data_source,
    BufferExhaustedPolicy buffer_exhausted_policy) { … }

// This is called via the public API Tracing::NewTrace().
// Can be called from any thread.
std::unique_ptr<TracingSession> TracingMuxerImpl::CreateTracingSession(
    BackendType requested_backend_type,
    TracingConsumerBackend* (*system_backend_factory)()) { … }

// static
// This is called via the public API Tracing::SetupStartupTracing().
// Can be called from any thread.
std::unique_ptr<StartupTracingSession>
TracingMuxerImpl::CreateStartupTracingSession(
    const TraceConfig& config,
    Tracing::SetupStartupTracingOpts opts) { … }

// Must not be called from the SDK's internal thread.
std::unique_ptr<StartupTracingSession>
TracingMuxerImpl::CreateStartupTracingSessionBlocking(
    const TraceConfig& config,
    Tracing::SetupStartupTracingOpts opts) { … }

void TracingMuxerImpl::AbortStartupTracingSession(
    TracingSessionGlobalID session_id,
    BackendType backend_type) { … }

void TracingMuxerImpl::InitializeInstance(const TracingInitArgs& args) { … }

// static
void TracingMuxerImpl::ResetForTesting() { … }

// static
void TracingMuxerImpl::Shutdown() { … }

void TracingMuxerImpl::AppendResetForTestingCallback(std::function<void()> cb) { … }

TracingMuxer::~TracingMuxer() = default;

static_assert …;

}  // namespace internal
}  // namespace perfetto
chromium/third_party/perfetto/src/tracing/internal/tracing_muxer_impl.cc
