1 /*
2 * Copyright (c) Huawei Technologies Co., Ltd. 2021. All rights reserved.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "plugin_service.h"
17
18 #include <cinttypes>
19 #include <fcntl.h>
20 #include <sys/wait.h>
21 #include <unistd.h>
22
23 #include "plugin_command_builder.h"
24 #include "plugin_service_impl.h"
25 #include "plugin_session_manager.h"
26 #include "profiler_capability_manager.h"
27 #include "profiler_data_repeater.h"
28 #include "securec.h"
29 #include "share_memory_allocator.h"
30 #include "socket_context.h"
31
32 namespace {
33 const int PAGE_BYTES = 4096;
34 const int DEFAULT_EVENT_POLLING_INTERVAL = 5000;
35 constexpr uint32_t FLUSH_BASELINE = (1U << 21); // need to flush data size with offline mode
36 constexpr uint32_t STOP_BASELINE = (1U << 22); // need to stop take data size with offline mode
37 } // namespace
38
PluginService()39 PluginService::PluginService()
40 {
41 pluginIdCounter_ = 0;
42 if (getuid() == 0) {
43 StartService(DEFAULT_UNIX_SOCKET_FULL_PATH);
44 } else {
45 StartService(DEFAULT_UNIX_SOCKET_PATH);
46 }
47
48 pluginCommandBuilder_ = std::make_shared<PluginCommandBuilder>();
49
50 eventPoller_ = std::make_unique<EpollEventPoller>(DEFAULT_EVENT_POLLING_INTERVAL);
51 CHECK_NOTNULL(eventPoller_, NO_RETVAL, "create event poller FAILED!");
52
53 eventPoller_->Init();
54 }
55
~PluginService()56 PluginService::~PluginService()
57 {
58 if (eventPoller_) {
59 eventPoller_->Stop();
60 eventPoller_->Finalize();
61 }
62 }
63
StartEpollThread()64 bool PluginService::StartEpollThread()
65 {
66 if (eventPoller_) {
67 return eventPoller_->Start();
68 }
69 return false;
70 }
71
StopEpollThread()72 bool PluginService::StopEpollThread()
73 {
74 if (eventPoller_) {
75 return eventPoller_->Stop();
76 }
77 return false;
78 }
79
SetPluginSessionManager(const PluginSessionManagerPtr& pluginSessionManager)80 void PluginService::SetPluginSessionManager(const PluginSessionManagerPtr& pluginSessionManager)
81 {
82 pluginSessionManager_ = pluginSessionManager;
83 }
84
SetProfilerSessionConfig(const std::shared_ptr<ProfilerSessionConfig> profilerSessionConfig, const std::vector<std::string>& pluginNames)85 void PluginService::SetProfilerSessionConfig(const std::shared_ptr<ProfilerSessionConfig> profilerSessionConfig,
86 const std::vector<std::string>& pluginNames)
87 {
88 for (const std::string& name : pluginNames) {
89 profilerSessionConfigs_[name] = profilerSessionConfig;
90 }
91 }
92
GetSemaphore(uint32_t id) const93 SemaphorePtr PluginService::GetSemaphore(uint32_t id) const
94 {
95 std::unique_lock<std::mutex> lock(mutex_);
96 auto it = waitSemphores_.find(id);
97 if (it != waitSemphores_.end()) {
98 return it->second;
99 }
100 return nullptr;
101 }
102
StartService(const std::string& unixSocketName)103 bool PluginService::StartService(const std::string& unixSocketName)
104 {
105 pluginServiceImpl_ = std::make_shared<PluginServiceImpl>(*this);
106 serviceEntry_ = std::make_shared<ServiceEntry>();
107 if (!serviceEntry_->StartServer(unixSocketName)) {
108 pluginServiceImpl_ = nullptr;
109 serviceEntry_ = nullptr;
110 PROFILER_LOG_DEBUG(LOG_CORE, "Start IPC Service FAIL");
111 return false;
112 }
113 serviceEntry_->RegisterService(*pluginServiceImpl_.get());
114 return true;
115 }
116
GetReusePolicy(const ProfilerSessionConfig::BufferConfig& bufferConfig)117 static ShareMemoryBlock::ReusePolicy GetReusePolicy(const ProfilerSessionConfig::BufferConfig& bufferConfig)
118 {
119 if (bufferConfig.policy() == ProfilerSessionConfig::BufferConfig::RECYCLE) {
120 return ShareMemoryBlock::DROP_OLD;
121 }
122 return ShareMemoryBlock::DROP_NONE;
123 }
124
125 // create plugin session with buffer config
CreatePluginSession(const ProfilerPluginConfig& pluginConfig, const ProfilerSessionConfig::BufferConfig& bufferConfig, const ProfilerDataRepeaterPtr& dataRepeater)126 bool PluginService::CreatePluginSession(const ProfilerPluginConfig& pluginConfig,
127 const ProfilerSessionConfig::BufferConfig& bufferConfig,
128 const ProfilerDataRepeaterPtr& dataRepeater)
129 {
130 uint32_t pluginId = 0;
131 PluginContextPtr pluginCtx = nullptr;
132 std::string pluginName = pluginConfig.name();
133 isProtobufSerialize_ = pluginConfig.is_protobuf_serialize();
134 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
135 CHECK_NOTNULL(pluginCtx, false, "get PluginContext failed!");
136
137 pluginCtx->profilerDataRepeater = dataRepeater;
138
139 uint32_t bufferSize = bufferConfig.pages() * PAGE_BYTES;
140 auto cmd = pluginCommandBuilder_->BuildCreateSessionCmd(pluginConfig, bufferSize);
141 CHECK_TRUE(cmd != nullptr, false, "CreatePluginSession BuildCreateSessionCmd FAIL %s", pluginName.c_str());
142
143 auto smb = ShareMemoryAllocator::GetInstance().CreateMemoryBlockLocal(pluginName, bufferSize);
144 CHECK_TRUE(smb != nullptr, false, "CreateMemoryBlockLocal FAIL %s", pluginName.c_str());
145
146 auto policy = GetReusePolicy(bufferConfig);
147 PROFILER_LOG_DEBUG(LOG_CORE, "CreatePluginSession policy = %d", (int)policy);
148 smb->SetReusePolicy(policy);
149
150 auto notifier = EventNotifier::Create(0, EventNotifier::NONBLOCK);
151 CHECK_NOTNULL(notifier, false, "create EventNotifier for %s failed!", pluginName.c_str());
152 CHECK_TRUE(pluginServiceImpl_->SendHeartBeat(*pluginCtx->context), false,
153 "%s hiprofiler_plugin process is off line!", __func__);
154 pluginCtx->shareMemoryBlock = smb;
155 pluginCtx->eventNotifier = notifier;
156 pluginCtx->profilerPluginState.set_state(ProfilerPluginState::LOADED);
157 pluginServiceImpl_->PushCommand(*pluginCtx->context, cmd);
158 pluginCtx->context->SendFileDescriptor(smb->GetfileDescriptor());
159 pluginCtx->context->SendFileDescriptor(notifier->GetFd());
160
161 auto configIter = profilerSessionConfigs_.find(pluginName);
162 if (configIter == profilerSessionConfigs_.end()) {
163 PROFILER_LOG_ERROR(LOG_CORE, "profiler session config not set fot plugin name: %s", pluginName.c_str());
164 return false;
165 }
166 if (configIter->second->session_mode() == ProfilerSessionConfig::OFFLINE) {
167 eventPoller_->AddFileDescriptor(notifier->GetFd(),
168 [this, pluginCtx] { this->ReadShareMemoryOffline(*pluginCtx); });
169 } else if (configIter->second->session_mode() == ProfilerSessionConfig::ONLINE) {
170 eventPoller_->AddFileDescriptor(notifier->GetFd(),
171 [this, pluginCtx] { this->ReadShareMemoryOnline(*pluginCtx); });
172 }
173 PROFILER_LOG_DEBUG(LOG_CORE, "CreatePluginSession %s done, shmem fd = %d",
174 pluginName.c_str(), smb->GetfileDescriptor());
175 return true;
176 }
177
178 // create plugin session without buffer config
CreatePluginSession(const ProfilerPluginConfig& pluginConfig, const ProfilerDataRepeaterPtr& dataRepeater)179 bool PluginService::CreatePluginSession(const ProfilerPluginConfig& pluginConfig,
180 const ProfilerDataRepeaterPtr& dataRepeater)
181 {
182 uint32_t pluginId = 0;
183 PluginContextPtr pluginCtx = nullptr;
184 std::string pluginName = pluginConfig.name();
185 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
186 CHECK_NOTNULL(pluginCtx, false, "get PluginContext failed!");
187
188 pluginCtx->profilerDataRepeater = dataRepeater;
189 pluginCtx->shareMemoryBlock = nullptr;
190
191 auto cmd = pluginCommandBuilder_->BuildCreateSessionCmd(pluginConfig, 0);
192 CHECK_TRUE(cmd != nullptr, false, "CreatePluginSession BuildCreateSessionCmd FAIL %s", pluginName.c_str());
193 CHECK_TRUE(pluginServiceImpl_->SendHeartBeat(*pluginCtx->context), false,
194 "%s hiprofiler_plugin process is off line!", __func__);
195 pluginCtx->profilerPluginState.set_state(ProfilerPluginState::LOADED);
196 pluginServiceImpl_->PushCommand(*pluginCtx->context, cmd);
197 PROFILER_LOG_DEBUG(LOG_CORE, "CreatePluginSession %s done!", pluginName.c_str());
198 return true;
199 }
200
StartPluginSession(const ProfilerPluginConfig& config)201 bool PluginService::StartPluginSession(const ProfilerPluginConfig& config)
202 {
203 uint32_t pluginId = 0;
204 PluginContextPtr pluginCtx = nullptr;
205 std::string pluginName = config.name();
206 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
207 CHECK_NOTNULL(pluginCtx, false, "get PluginContext failed!");
208
209 auto cmd = pluginCommandBuilder_->BuildStartSessionCmd(config, pluginId);
210 CHECK_TRUE(cmd != nullptr, false, "StartPluginSession BuildStartSessionCmd FAIL %s", pluginName.c_str());
211 CHECK_TRUE(pluginServiceImpl_->SendHeartBeat(*pluginCtx->context), false,
212 "%s hiprofiler_plugin process is off line!", __func__);
213 pluginCtx->profilerPluginState.set_state(ProfilerPluginState::IN_SESSION);
214 pluginServiceImpl_->PushCommand(*pluginCtx->context, cmd);
215 PROFILER_LOG_INFO(LOG_CORE, "StartPluginSession %s done!", pluginName.c_str());
216 return true;
217 }
218
StopPluginSession(const std::string& pluginName)219 bool PluginService::StopPluginSession(const std::string& pluginName)
220 {
221 uint32_t pluginId = 0;
222 PluginContextPtr pluginCtx = nullptr;
223 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
224 CHECK_NOTNULL(pluginCtx, false, "get PluginContext failed!");
225
226 auto cmd = pluginCommandBuilder_->BuildStopSessionCmd(pluginId);
227 CHECK_TRUE(cmd != nullptr, false, "StopPluginSession BuildStopSessionCmd FAIL %s", pluginName.c_str());
228 CHECK_TRUE(pluginServiceImpl_->SendHeartBeat(*pluginCtx->context), false,
229 "%s hiprofiler_plugin process is off line!", __func__);
230 pluginCtx->profilerPluginState.set_state(ProfilerPluginState::LOADED);
231 pluginServiceImpl_->PushCommand(*pluginCtx->context, cmd);
232 auto sem = GetSemaphoreFactory().Create(0);
233 CHECK_NOTNULL(sem, false, "create Semaphore for stop %s FAILED!", pluginName.c_str());
234
235 waitSemphores_[cmd->command_id()] = sem;
236 PROFILER_LOG_DEBUG(LOG_CORE, "=== StopPluginSession %s Waiting ... ===", pluginName.c_str());
237 // wait on semaphore at most 30 seconds.
238 if (!sem->TimedWait(30)) {
239 // semaphore timeout
240 PROFILER_LOG_DEBUG(LOG_CORE, "=== StopPluginSession Waiting FAIL ===");
241 return false;
242 }
243 PROFILER_LOG_DEBUG(LOG_CORE, "StopPluginSession %s done!", pluginName.c_str());
244 return true;
245 }
246
DestroyPluginSession(const std::string& pluginName)247 bool PluginService::DestroyPluginSession(const std::string& pluginName)
248 {
249 uint32_t pluginId = 0;
250 PluginContextPtr pluginCtx = nullptr;
251 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
252 CHECK_NOTNULL(pluginCtx, false, "get PluginContext failed!");
253
254 auto cmd = pluginCommandBuilder_->BuildDestroySessionCmd(pluginId);
255 CHECK_TRUE(cmd != nullptr, false, "DestroyPluginSession BuildDestroySessionCmd FAIL %s", pluginName.c_str());
256
257 if (profilerSessionConfigs_.find(pluginName) != profilerSessionConfigs_.end()) {
258 profilerSessionConfigs_.erase(profilerSessionConfigs_.find(pluginName));
259 }
260 if (pluginCtx->shareMemoryBlock) {
261 ShareMemoryAllocator::GetInstance().ReleaseMemoryBlockLocal(pluginName);
262 }
263
264 if (pluginCtx->eventNotifier) {
265 eventPoller_->RemoveFileDescriptor(pluginCtx->eventNotifier->GetFd());
266 pluginCtx->eventNotifier = nullptr;
267 }
268 CHECK_TRUE(pluginServiceImpl_->SendHeartBeat(*pluginCtx->context), false,
269 "%s hiprofiler_plugin process is off line!", __func__);
270 pluginCtx->profilerPluginState.set_state(ProfilerPluginState::REGISTERED);
271 pluginServiceImpl_->PushCommand(*pluginCtx->context, cmd);
272 PROFILER_LOG_INFO(LOG_CORE, "DestroyPluginSession %s done!", pluginName.c_str());
273 return true;
274 }
275
RefreshPluginSession(const std::string& pluginName)276 bool PluginService::RefreshPluginSession(const std::string& pluginName)
277 {
278 uint32_t pluginId = 0;
279 PluginContextPtr pluginCtx = nullptr;
280 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
281 CHECK_NOTNULL(pluginCtx, false, "get PluginContext failed!");
282
283 auto cmd = pluginCommandBuilder_->BuildRefreshSessionCmd(pluginId);
284 CHECK_TRUE(cmd != nullptr, false, "RefreshPluginSession BuildRefreshSessionCmd FAIL %s", pluginName.c_str());
285 CHECK_TRUE(pluginServiceImpl_->SendHeartBeat(*pluginCtx->context), false,
286 "%s hiprofiler_plugin process is off line!", __func__);
287 pluginServiceImpl_->PushCommand(*pluginCtx->context, cmd);
288 PROFILER_LOG_INFO(LOG_CORE, "RefreshPluginSession %s done!", pluginName.c_str());
289 return true;
290 }
291
RemovePluginSessionCtx(const std::string& pluginName)292 bool PluginService::RemovePluginSessionCtx(const std::string& pluginName)
293 {
294 PluginContextPtr pluginCtx = GetPluginContext(pluginName).second;
295 CHECK_NOTNULL(pluginCtx, false, "get PluginContext failed!");
296
297 if (pluginCtx->shareMemoryBlock) {
298 ShareMemoryAllocator::GetInstance().ReleaseMemoryBlockLocal(pluginName);
299 pluginCtx->shareMemoryBlock = nullptr;
300 }
301
302 if (pluginCtx->eventNotifier) {
303 eventPoller_->RemoveFileDescriptor(pluginCtx->eventNotifier->GetFd());
304 pluginCtx->eventNotifier = nullptr;
305 }
306
307 pluginCtx->profilerPluginState.set_state(ProfilerPluginState::INITED);
308 PROFILER_LOG_INFO(LOG_CORE, "RemovePluginSessionCtx %s done!", pluginName.c_str());
309 return true;
310 }
311
GetPluginContext(const std::string& pluginName)312 std::pair<uint32_t, PluginContextPtr> PluginService::GetPluginContext(const std::string& pluginName)
313 {
314 std::unique_lock<std::mutex> lock(mutex_);
315 CHECK_TRUE(nameIndex_.count(pluginName) > 0, std::make_pair(0, nullptr),
316 "GetPluginContext failed, plugin name `%s` not found!", pluginName.c_str());
317 uint32_t id = nameIndex_[pluginName];
318
319 CHECK_TRUE(pluginContext_.count(id) > 0, std::make_pair(id, nullptr), "plugin id %u not found!", id);
320 return std::make_pair(id, pluginContext_[id]);
321 }
322
GetPluginContextById(uint32_t id)323 PluginContextPtr PluginService::GetPluginContextById(uint32_t id)
324 {
325 std::unique_lock<std::mutex> lock(mutex_);
326 CHECK_TRUE(pluginContext_.count(id) > 0, nullptr, "plugin id %u not found!", id);
327 return pluginContext_[id];
328 }
329
AddPluginInfo(const PluginInfo& pluginInfo)330 bool PluginService::AddPluginInfo(const PluginInfo& pluginInfo)
331 {
332 if (nameIndex_.find(pluginInfo.name) == nameIndex_.end()) { // add new plugin
333 auto pluginCtx = std::make_shared<PluginContext>();
334 CHECK_NOTNULL(pluginCtx, false, "create PluginContext failed!");
335
336 ProfilerPluginCapability capability;
337 capability.set_path(pluginInfo.path);
338 capability.set_name(pluginInfo.name);
339 CHECK_TRUE(ProfilerCapabilityManager::GetInstance().AddCapability(capability), false,
340 "AddPluginInfo AddCapability FAIL");
341
342 pluginCtx->name = pluginInfo.name;
343 pluginCtx->path = pluginInfo.path;
344 pluginCtx->context = pluginInfo.context;
345 pluginCtx->config.set_name(pluginInfo.name);
346 pluginCtx->config.set_plugin_sha256(pluginInfo.sha256);
347 pluginCtx->profilerPluginState.set_name(pluginInfo.name);
348 pluginCtx->profilerPluginState.set_state(ProfilerPluginState::REGISTERED);
349 pluginCtx->sha256 = pluginInfo.sha256;
350 pluginCtx->bufferSizeHint = pluginInfo.bufferSizeHint;
351 pluginCtx->isStandaloneFileData = pluginInfo.isStandaloneFileData;
352 pluginCtx->outFileName = pluginInfo.outFileName;
353 pluginCtx->pluginVersion = pluginInfo.pluginVersion;
354
355 uint32_t pluginId = ++pluginIdCounter_;
356 std::unique_lock<std::mutex> lock(mutex_);
357 pluginContext_[pluginId] = pluginCtx;
358 nameIndex_[pluginInfo.name] = pluginId;
359 } else { // update sha256 or bufferSizeHint
360 std::unique_lock<std::mutex> lock(mutex_);
361 CHECK_TRUE(nameIndex_.count(pluginInfo.name) > 0, false, "plugin name %s not found!", pluginInfo.name.c_str());
362
363 uint32_t pluginId = nameIndex_[pluginInfo.name];
364 CHECK_TRUE(pluginContext_.count(pluginId) > 0, false, "plugin id %u not found!", pluginId);
365 auto pluginCtx = pluginContext_[pluginId];
366
367 if (pluginInfo.sha256 != "") {
368 pluginCtx->sha256 = pluginInfo.sha256;
369 }
370 if (pluginInfo.bufferSizeHint != 0) {
371 pluginCtx->bufferSizeHint = pluginInfo.bufferSizeHint;
372 }
373 if (pluginInfo.isStandaloneFileData != false) {
374 pluginCtx->isStandaloneFileData = pluginInfo.isStandaloneFileData;
375 }
376 if (pluginInfo.outFileName != "") {
377 pluginCtx->outFileName = pluginInfo.outFileName;
378 }
379 if (pluginInfo.pluginVersion != "") {
380 pluginCtx->pluginVersion = pluginInfo.pluginVersion;
381 }
382 }
383 PROFILER_LOG_DEBUG(LOG_CORE, "AddPluginInfo for %s done!", pluginInfo.name.c_str());
384
385 return true;
386 }
387
GetPluginInfo(const std::string& pluginName, PluginInfo& pluginInfo)388 bool PluginService::GetPluginInfo(const std::string& pluginName, PluginInfo& pluginInfo)
389 {
390 uint32_t pluginId = 0;
391 PluginContextPtr pluginCtx = nullptr;
392 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
393 CHECK_TRUE(pluginId, false, "plugin name %s not found!", pluginName.c_str());
394 CHECK_TRUE(pluginCtx, false, "plugin id %u not found!", pluginId);
395
396 pluginInfo.id = pluginId;
397 pluginInfo.name = pluginCtx->name;
398 pluginInfo.path = pluginCtx->path;
399 pluginInfo.sha256 = pluginCtx->sha256;
400 pluginInfo.bufferSizeHint = pluginCtx->bufferSizeHint;
401 return true;
402 }
403
RemovePluginInfo(const PluginInfo& pluginInfo)404 bool PluginService::RemovePluginInfo(const PluginInfo& pluginInfo)
405 {
406 uint32_t pluginId = pluginInfo.id;
407 PluginContextPtr pluginCtx = GetPluginContextById(pluginId);
408 CHECK_NOTNULL(pluginCtx, false, "RemovePluginInfo failed, id %d not found!", pluginId);
409
410 std::string pluginName = pluginCtx->config.name();
411 CHECK_TRUE(ProfilerCapabilityManager::GetInstance().RemoveCapability(pluginName), false,
412 "RemovePluginInfo RemoveCapability FAIL %d", pluginId);
413
414 auto pluginState = pluginCtx->profilerPluginState.state();
415 if (pluginState == ProfilerPluginState::LOADED || pluginState == ProfilerPluginState::IN_SESSION) {
416 std::vector<std::string> pluginNames = {pluginName};
417 pluginSessionManager_->InvalidatePluginSessions(pluginNames);
418 pluginSessionManager_->RemovePluginSessions(pluginNames);
419 this->RemovePluginSessionCtx(pluginName);
420 }
421
422 std::unique_lock<std::mutex> lock(mutex_);
423 nameIndex_.erase(pluginName);
424 pluginContext_.erase(pluginId);
425 PROFILER_LOG_DEBUG(LOG_CORE, "RemovePluginInfo for %s done!", pluginName.c_str());
426 return true;
427 }
428
ReadShareMemoryOffline(PluginContext& context)429 void PluginService::ReadShareMemoryOffline(PluginContext& context)
430 {
431 CHECK_NOTNULL(context.shareMemoryBlock, NO_RETVAL, "smb of %s is null!", context.path.c_str());
432 CHECK_NOTNULL(traceWriter_, NO_RETVAL, "traceWriter_ is null!");
433 if (context.eventNotifier) {
434 context.eventNotifier->Take();
435 }
436
437 uint32_t stopTakeDataSize = 0;
438 while (true) {
439 int retval = 0;
440 bool ret = context.shareMemoryBlock->TakeData([&](const int8_t data[], uint32_t size) -> bool {
441 CHECK_NOTNULL(data, false, "memory block data is null!");
442 retval = traceWriter_->Write(data, size);
443 CHECK_TRUE(retval != -1, false, "need to splite file");
444 CHECK_TRUE(retval > 0, false, "write %d bytes failed!", size);
445 return true;
446 }, isProtobufSerialize_);
447
448 if (retval == -1) {
449 PROFILER_LOG_DEBUG(LOG_CORE, "need to clear share memory block and report the basic data");
450 pluginSessionManager_->RefreshPluginSession();
451 break;
452 }
453
454 dataFlushSize_ += static_cast<uint32_t>(retval);
455 stopTakeDataSize += static_cast<uint32_t>(retval);
456 if (stopTakeDataSize > STOP_BASELINE) {
457 break;
458 } else if (dataFlushSize_ > FLUSH_BASELINE) {
459 traceWriter_->Flush();
460 traceWriter_->Finish();
461 dataFlushSize_ = 0;
462 }
463
464 if (!ret) { // no data to read
465 break;
466 }
467 }
468 traceWriter_->Flush();
469 traceWriter_->Finish();
470 }
471
ReadShareMemoryOnline(PluginContext& context)472 void PluginService::ReadShareMemoryOnline(PluginContext& context)
473 {
474 CHECK_NOTNULL(context.shareMemoryBlock, NO_RETVAL, "smb of %s is null!", context.path.c_str());
475 if (context.eventNotifier) {
476 context.eventNotifier->Take();
477 }
478
479 while (true) {
480 auto pluginData = std::make_shared<ProfilerPluginData>();
481 bool ret = context.shareMemoryBlock->TakeData([&](const int8_t data[], uint32_t size) -> bool {
482 int retval = pluginData->ParseFromArray(reinterpret_cast<const char*>(data), size);
483 CHECK_TRUE(retval, false, "parse %d bytes failed!", size);
484 return true;
485 }, isProtobufSerialize_);
486 if (!ret) {
487 break;
488 }
489 if (!context.profilerDataRepeater->PutPluginData(pluginData)) {
490 break;
491 }
492 }
493 }
494
FlushShareMemory(PluginContext& context)495 void PluginService::FlushShareMemory(PluginContext& context)
496 {
497 CHECK_NOTNULL(context.shareMemoryBlock, NO_RETVAL, "smb of %s is null!", context.path.c_str());
498 CHECK_NOTNULL(traceWriter_, NO_RETVAL, "traceWriter_ is null!");
499
500 while (true) {
501 bool ret = context.shareMemoryBlock->TakeData([&](const int8_t data[], uint32_t size) -> bool {
502 int retval = traceWriter_->Write(data, size);
503 CHECK_TRUE(retval > 0, false, "write %d bytes failed!", size);
504 return true;
505 }, isProtobufSerialize_);
506 if (!ret) { // no data to read
507 break;
508 }
509 }
510 traceWriter_->Finish();
511 }
512
FlushAllData(const std::string& pluginName)513 void PluginService::FlushAllData(const std::string& pluginName)
514 {
515 PROFILER_LOG_INFO(LOG_CORE, "FlushAllData for %s start!", pluginName.c_str());
516 auto configIter = profilerSessionConfigs_.find(pluginName);
517 if (configIter == profilerSessionConfigs_.end()) {
518 PROFILER_LOG_ERROR(LOG_CORE, "profiler session config not set fot plugin name: %s", pluginName.c_str());
519 return;
520 }
521 if (!configIter->second->discard_cache_data()) {
522 uint32_t pluginId = 0;
523 PluginContextPtr pluginCtx = nullptr;
524 std::tie(pluginId, pluginCtx) = GetPluginContext(pluginName);
525 CHECK_NOTNULL(pluginCtx, NO_RETVAL, "%s: get PluginContext(%s) failed!", __func__, pluginName.c_str());
526 if (configIter->second->session_mode() == ProfilerSessionConfig::OFFLINE) {
527 FlushShareMemory(*pluginCtx);
528 } else if (configIter->second->session_mode() == ProfilerSessionConfig::ONLINE) {
529 ReadShareMemoryOnline(*pluginCtx);
530 }
531 }
532 PROFILER_LOG_INFO(LOG_CORE, "FlushAllData for %s done!", pluginName.c_str());
533 }
534
AppendResult(NotifyResultRequest& request)535 bool PluginService::AppendResult(NotifyResultRequest& request)
536 {
537 pluginCommandBuilder_->GetedCommandResponse(request.command_id());
538 auto sem = GetSemaphore(request.command_id());
539 if (sem) {
540 sem->Post();
541 }
542
543 int size = request.result_size();
544 PROFILER_LOG_DEBUG(LOG_CORE, "AppendResult size:%d, cmd id:%d", size, request.command_id());
545 for (int i = 0; i < size; i++) {
546 PluginResult pr = request.result(i);
547 if (pr.data().size() > 0) {
548 PROFILER_LOG_DEBUG(LOG_CORE, "AppendResult Size : %zu", pr.data().size());
549 uint32_t pluginId = pr.plugin_id();
550 PluginContextPtr pluginCtx = GetPluginContextById(pluginId);
551 CHECK_NOTNULL(pluginCtx, false, "plugin id %u not found!", pluginId);
552 if (pluginCtx->profilerDataRepeater == nullptr) {
553 PROFILER_LOG_DEBUG(LOG_CORE, "AppendResult profilerDataRepeater==nullptr %s %d",
554 pr.status().name().c_str(), pluginId);
555 return false;
556 }
557 auto pluginData = std::make_shared<ProfilerPluginData>();
558 pluginData->set_name(pr.status().name());
559 pluginData->set_status(0);
560 pluginData->set_data(pr.data());
561 if (!pluginCtx->profilerDataRepeater->PutPluginData(pluginData)) {
562 return false;
563 }
564 } else if (pr.out_file_name() != "") { // updata plugin outFileName
565 std::unique_lock<std::mutex> lock(mutex_);
566 auto pluginId = pr.plugin_id();
567 CHECK_TRUE(pluginContext_.count(pluginId) > 0, false, "plugin id %u not found!", pluginId);
568 pluginContext_[pluginId]->outFileName = pr.out_file_name();
569 } else {
570 PROFILER_LOG_DEBUG(LOG_CORE, "Flush?Data From ShareMemory?");
571 }
572 }
573 return true;
574 }
575
GetPluginStatus()576 std::vector<ProfilerPluginState> PluginService::GetPluginStatus()
577 {
578 std::vector<ProfilerPluginState> status;
579 std::unique_lock<std::mutex> lock(mutex_);
580 for (auto& entry : pluginContext_) {
581 status.push_back(entry.second->profilerPluginState);
582 }
583 return status;
584 }
585
GetPluginIdByName(std::string name)586 uint32_t PluginService::GetPluginIdByName(std::string name)
587 {
588 std::unique_lock<std::mutex> lock(mutex_);
589 if (nameIndex_.find(name) == nameIndex_.end()) {
590 return 0;
591 }
592 return nameIndex_[name];
593 }
594
SetTraceWriter(const TraceFileWriterPtr& traceWriter)595 void PluginService::SetTraceWriter(const TraceFileWriterPtr& traceWriter)
596 {
597 traceWriter_ = traceWriter;
598 }