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 #include "stream_plugin.h"
16
17 #include <sys/syscall.h>
18 #include <sys/types.h>
19 #include <unistd.h>
20
21 #include "securec.h"
22 #include "stream_plugin_result.pbencoder.h"
23
24 using namespace OHOS::Developtools::Profiler;
25
StreamPlugin()26 StreamPlugin::StreamPlugin() {}
27
~StreamPlugin()28 StreamPlugin::~StreamPlugin() {}
29
Start(const uint8_t* configData, uint32_t configSize)30 int StreamPlugin::Start(const uint8_t* configData, uint32_t configSize)
31 {
32 // 反序列化
33 CHECK_TRUE(protoConfig_.ParseFromArray(configData, configSize) > 0, -1, "%s:parseFromArray failed!", __func__);
34 // 启动线程写数据
35 std::unique_lock<std::mutex> locker(mutex_);
36 running_ = true;
37 writeThread_ = std::thread([this] { this->Loop(); });
38
39 return 0;
40 }
41
Stop()42 int StreamPlugin::Stop()
43 {
44 std::unique_lock<std::mutex> locker(mutex_);
45 running_ = false;
46 locker.unlock();
47 if (writeThread_.joinable()) {
48 writeThread_.join();
49 }
50 PROFILER_LOG_INFO(LOG_CORE, "%s:stop success!", __func__);
51 return 0;
52 }
53
SetWriter(WriterStruct* writer)54 int StreamPlugin::SetWriter(WriterStruct* writer)
55 {
56 resultWriter_ = writer;
57 return 0;
58 }
59
Loop(void)60 void StreamPlugin::Loop(void)
61 {
62 PROFILER_LOG_INFO(LOG_CORE, "%s:transporter thread %d start !!!!!", __func__, gettid());
63 CHECK_NOTNULL(resultWriter_, NO_RETVAL, "%s: resultWriter_ nullptr", __func__);
64
65 uint32_t index = 0;
66 while (running_) {
67 if (resultWriter_->isProtobufSerialize) {
68 StreamData dataProto;
69 dataProto.set_intdata(index);
70 dataProto.set_stringdata(std::to_string(index));
71 buffer_.resize(dataProto.ByteSizeLong());
72 dataProto.SerializeToArray(buffer_.data(), buffer_.size());
73 if (index < 50) { // 50: count of loop
74 resultWriter_->write(resultWriter_, buffer_.data(), buffer_.size());
75 resultWriter_->flush(resultWriter_);
76 }
77 } else {
78 ProtoEncoder::StreamData dataProto(resultWriter_->startReport(resultWriter_));
79 dataProto.set_intdata(index);
80 dataProto.set_stringdata(std::to_string(index));
81 int messageLen = dataProto.Finish();
82 resultWriter_->finishReport(resultWriter_, messageLen);
83 if (index < 50) { // 50: count of loop
84 resultWriter_->flush(resultWriter_);
85 }
86 }
87
88 index++;
89 }
90 PROFILER_LOG_INFO(LOG_CORE, "%s:transporter thread %d exit !!!!!", __func__, gettid());
91 }