xref: /aosp_15_r20/external/perfetto/test/fake_producer.cc (revision 6dbdd20afdafa5e3ca9b8809fa73465d530080dc)
1 /*
2  * Copyright (C) 2018 The Android Open Source Project
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  *      http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 
17 #include "test/fake_producer.h"
18 
19 #include <mutex>
20 #include <thread>
21 
22 #include "perfetto/base/logging.h"
23 #include "perfetto/base/time.h"
24 #include "perfetto/ext/base/utils.h"
25 #include "perfetto/ext/tracing/core/commit_data_request.h"
26 #include "perfetto/ext/tracing/core/shared_memory_arbiter.h"
27 #include "perfetto/ext/tracing/core/trace_packet.h"
28 #include "perfetto/ext/tracing/core/trace_writer.h"
29 #include "perfetto/tracing/core/data_source_config.h"
30 #include "src/ipc/client_impl.h"
31 #include "src/tracing/ipc/producer/producer_ipc_client_impl.h"
32 
33 #include "protos/perfetto/config/test_config.gen.h"
34 #include "protos/perfetto/trace/test_event.pbzero.h"
35 #include "protos/perfetto/trace/trace_packet.pbzero.h"
36 
37 namespace perfetto {
38 
39 namespace {
40 const MaybeUnboundBufferID kStartupTargetBufferReservationId = 1;
41 }  // namespace
42 
FakeProducer(const std::string & name,base::TaskRunner * task_runner)43 FakeProducer::FakeProducer(const std::string& name,
44                            base::TaskRunner* task_runner)
45     : name_(name), task_runner_(task_runner) {}
46 FakeProducer::~FakeProducer() = default;
47 
Connect(const char * socket_name,std::function<void ()> on_connect,std::function<void ()> on_setup_data_source_instance,std::function<void ()> on_create_data_source_instance,std::unique_ptr<SharedMemory> shm,std::unique_ptr<SharedMemoryArbiter> shm_arbiter)48 void FakeProducer::Connect(const char* socket_name,
49                            std::function<void()> on_connect,
50                            std::function<void()> on_setup_data_source_instance,
51                            std::function<void()> on_create_data_source_instance,
52                            std::unique_ptr<SharedMemory> shm,
53                            std::unique_ptr<SharedMemoryArbiter> shm_arbiter) {
54   PERFETTO_DCHECK_THREAD(thread_checker_);
55   endpoint_ = ProducerIPCClient::Connect(
56       socket_name, this, "android.perfetto.FakeProducer", task_runner_,
57       TracingService::ProducerSMBScrapingMode::kDefault,
58       /*shared_memory_size_hint_bytes=*/0,
59       /*shared_memory_page_size_hint_bytes=*/4096, std::move(shm),
60       std::move(shm_arbiter));
61   on_connect_ = std::move(on_connect);
62   on_setup_data_source_instance_ = std::move(on_setup_data_source_instance);
63   on_create_data_source_instance_ = std::move(on_create_data_source_instance);
64 }
65 
OnConnect()66 void FakeProducer::OnConnect() {
67   PERFETTO_DCHECK_THREAD(thread_checker_);
68   DataSourceDescriptor descriptor;
69   descriptor.set_name(name_);
70   endpoint_->RegisterDataSource(descriptor);
71   auto on_connect_callback = std::move(on_connect_);
72   auto task_runner = task_runner_;
73   endpoint_->Sync([task_runner, on_connect_callback] {
74     task_runner->PostTask(on_connect_callback);
75   });
76 }
77 
OnDisconnect()78 void FakeProducer::OnDisconnect() {
79   PERFETTO_DCHECK_THREAD(thread_checker_);
80   PERFETTO_FATAL("Producer unexpectedly disconnected from the service");
81 }
82 
SetupDataSource(DataSourceInstanceID,const DataSourceConfig &)83 void FakeProducer::SetupDataSource(DataSourceInstanceID,
84                                    const DataSourceConfig&) {
85   task_runner_->PostTask(on_setup_data_source_instance_);
86 }
87 
StartDataSource(DataSourceInstanceID,const DataSourceConfig & source_config)88 void FakeProducer::StartDataSource(DataSourceInstanceID,
89                                    const DataSourceConfig& source_config) {
90   PERFETTO_DCHECK_THREAD(thread_checker_);
91   if (trace_writer_) {
92     // Startup tracing was already active, just bind the target buffer.
93     endpoint_->MaybeSharedMemoryArbiter()->BindStartupTargetBuffer(
94         kStartupTargetBufferReservationId,
95         static_cast<BufferID>(source_config.target_buffer()));
96   } else {
97     // Common case: Start tracing now.
98     trace_writer_ = endpoint_->CreateTraceWriter(
99         static_cast<BufferID>(source_config.target_buffer()));
100     SetupFromConfig(source_config.for_testing());
101   }
102   if (source_config.for_testing().send_batch_on_register()) {
103     ProduceEventBatch(on_create_data_source_instance_);
104   } else {
105     task_runner_->PostTask(on_create_data_source_instance_);
106   }
107 }
108 
StopDataSource(DataSourceInstanceID)109 void FakeProducer::StopDataSource(DataSourceInstanceID) {
110   PERFETTO_DCHECK_THREAD(thread_checker_);
111   trace_writer_.reset();
112 }
113 
114 // Note: this can be called on a different thread.
ProduceStartupEventBatch(const protos::gen::TestConfig & config,SharedMemoryArbiter * arbiter,std::function<void ()> callback)115 void FakeProducer::ProduceStartupEventBatch(
116     const protos::gen::TestConfig& config,
117     SharedMemoryArbiter* arbiter,
118     std::function<void()> callback) {
119   task_runner_->PostTask([this, config, arbiter, callback] {
120     SetupFromConfig(config);
121 
122     PERFETTO_CHECK(!trace_writer_);
123     trace_writer_ =
124         arbiter->CreateStartupTraceWriter(kStartupTargetBufferReservationId);
125 
126     EmitEventBatchOnTaskRunner({});
127 
128     // Issue callback right after writing - cannot wait for flush yet because
129     // we're not connected yet.
130     callback();
131   });
132 }
133 
134 // Note: this can be called on a different thread.
ProduceEventBatch(std::function<void ()> callback)135 void FakeProducer::ProduceEventBatch(std::function<void()> callback) {
136   task_runner_->PostTask(
137       [this, callback] { EmitEventBatchOnTaskRunner(callback); });
138 }
139 
RegisterDataSource(const DataSourceDescriptor & desc)140 void FakeProducer::RegisterDataSource(const DataSourceDescriptor& desc) {
141   task_runner_->PostTask([this, desc] { endpoint_->RegisterDataSource(desc); });
142 }
143 
CommitData(const CommitDataRequest & req,std::function<void ()> callback)144 void FakeProducer::CommitData(const CommitDataRequest& req,
145                               std::function<void()> callback) {
146   task_runner_->PostTask(
147       [this, req, callback] { endpoint_->CommitData(req, callback); });
148 }
149 
Sync(std::function<void ()> callback)150 void FakeProducer::Sync(std::function<void()> callback) {
151   task_runner_->PostTask([this, callback] { endpoint_->Sync(callback); });
152 }
153 
ActivateTrigger(const std::string & trigger_name)154 void FakeProducer::ActivateTrigger(const std::string& trigger_name) {
155   task_runner_->PostTask(
156       [this, trigger_name] { endpoint_->ActivateTriggers({trigger_name}); });
157 }
158 
OnTracingSetup()159 void FakeProducer::OnTracingSetup() {}
160 
Flush(FlushRequestID flush_request_id,const DataSourceInstanceID *,size_t num_data_sources,FlushFlags)161 void FakeProducer::Flush(FlushRequestID flush_request_id,
162                          const DataSourceInstanceID*,
163                          size_t num_data_sources,
164                          FlushFlags) {
165   PERFETTO_DCHECK(num_data_sources > 0);
166   if (trace_writer_)
167     trace_writer_->Flush();
168   endpoint_->NotifyFlushComplete(flush_request_id);
169 }
170 
unix_socket_fd()171 base::SocketHandle FakeProducer::unix_socket_fd() {
172   // Since FakeProducer is only used in tests we can include and assume the
173   // implementation.
174   auto* producer = static_cast<ProducerIPCClientImpl*>(endpoint_.get());
175   auto* ipc_client =
176       static_cast<ipc::ClientImpl*>(producer->GetClientForTesting());
177   return ipc_client->GetUnixSocketForTesting()->fd();
178 }
179 
SetupFromConfig(const protos::gen::TestConfig & config)180 void FakeProducer::SetupFromConfig(const protos::gen::TestConfig& config) {
181   rnd_engine_ = std::minstd_rand0(config.seed());
182   message_count_ = config.message_count();
183   message_size_ = config.message_size();
184   max_messages_per_second_ = config.max_messages_per_second();
185 }
186 
EmitEventBatchOnTaskRunner(std::function<void ()> callback)187 void FakeProducer::EmitEventBatchOnTaskRunner(std::function<void()> callback) {
188   PERFETTO_CHECK(trace_writer_);
189   PERFETTO_CHECK(message_size_ > 1);
190   std::unique_ptr<char, base::FreeDeleter> payload(
191       static_cast<char*>(malloc(message_size_)));
192   memset(payload.get(), '.', message_size_);
193   payload.get()[message_size_ - 1] = 0;
194 
195   base::TimeMillis start = base::GetWallTimeMs();
196   int64_t iterations = 0;
197   uint32_t messages_to_emit = message_count_;
198   while (messages_to_emit > 0) {
199     uint32_t messages_in_minibatch =
200         max_messages_per_second_ == 0
201             ? messages_to_emit
202             : std::min(max_messages_per_second_, messages_to_emit);
203     PERFETTO_DCHECK(messages_to_emit >= messages_in_minibatch);
204 
205     for (uint32_t i = 0; i < messages_in_minibatch; i++) {
206       auto handle = trace_writer_->NewTracePacket();
207       handle->set_for_testing()->set_seq_value(
208           static_cast<uint32_t>(rnd_engine_()));
209       handle->set_for_testing()->set_str(payload.get(), message_size_);
210     }
211     messages_to_emit -= messages_in_minibatch;
212     iterations++;
213 
214     // Pause until the second boundary to make sure that we are adhering to
215     // the speed limitation.
216     if (max_messages_per_second_ > 0) {
217       int64_t expected_time_taken = iterations * 1000;
218       base::TimeMillis time_taken = base::GetWallTimeMs() - start;
219       while (time_taken.count() < expected_time_taken) {
220         std::this_thread::sleep_for(
221             base::TimeMillis(expected_time_taken - time_taken.count()));
222         time_taken = base::GetWallTimeMs() - start;
223       }
224     }
225     trace_writer_->Flush(messages_to_emit > 0 ? [] {} : callback);
226   }
227 }
228 
229 }  // namespace perfetto
230