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 }