1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
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 "cmd_executor.h"
16 #include "service_controller.h"
17
18 #include <seq_packet_socket_server.h>
19
20 #include <algorithm>
21 #include <cstring>
22 #include <iostream>
23 #include <memory>
24 #include <mutex>
25 #include <thread>
26
27 #include <poll.h>
28 #include <sys/prctl.h>
29 #include <sys/stat.h>
30 #include <unistd.h>
31
32 namespace OHOS {
33 namespace HiviewDFX {
34 static const int MAX_CLIENT_CONNECTIONS = 100;
35
~CmdExecutor()36 CmdExecutor::~CmdExecutor()
37 {
38 std::lock_guard<std::mutex> lg(m_clientAccess);
39 for (auto& client : m_clients) {
40 client->m_stopThread.store(true);
41 }
42 for (auto& client : m_clients) {
43 if (client->m_clientThread.joinable()) {
44 client->m_clientThread.join();
45 }
46 }
47 }
48
MainLoop()49 void CmdExecutor::MainLoop()
50 {
51 SeqPacketSocketServer cmdServer(CONTROL_SOCKET_NAME, MAX_CLIENT_CONNECTIONS);
52 if (cmdServer.Init() < 0) {
53 std::cerr << "Failed to init control socket ! \n";
54 return;
55 }
56 std::cout << "Begin to cmd accept !\n";
57 int listeningStatus = cmdServer.Listen(MAX_CLIENT_CONNECTIONS);
58 if (listeningStatus < 0) {
59 std::cerr << "Socket listen failed: ";
60 HilogPrintError(listeningStatus);
61 return;
62 }
63 std::cout << "Server started to listen !\n";
64
65 using namespace std::chrono_literals;
66 for (;;) {
67 const auto maxtime = 3000ms;
68 short outEvent = 0;
69 auto pollResult = cmdServer.Poll(POLLIN, outEvent, maxtime);
70 if (pollResult == 0) { // poll == 0 means timeout
71 CleanFinishedClients();
72 continue;
73 } else if (pollResult < 0) {
74 std::cerr << "Socket polling error: ";
75 HilogPrintError(errno);
76 break;
77 } else if (pollResult != 1 || outEvent != POLLIN) {
78 std::cerr << "Wrong poll result data."
79 " Result: " << pollResult <<
80 " OutEvent: " << outEvent << "\n";
81 break;
82 }
83
84 int acceptResult = cmdServer.Accept();
85 if (acceptResult > 0) {
86 int acceptedSockedFd = acceptResult;
87 std::unique_ptr<Socket> handler = std::make_unique<Socket>(SOCK_SEQPACKET);
88 handler->setHandler(acceptedSockedFd);
89 OnAcceptedConnection(std::move(handler));
90 } else {
91 std::cerr << "Socket accept failed: ";
92 HilogPrintError(errno);
93 break;
94 }
95 }
96 }
97
OnAcceptedConnection(std::unique_ptr<Socket> handler)98 void CmdExecutor::OnAcceptedConnection(std::unique_ptr<Socket> handler)
99 {
100 std::lock_guard<std::mutex> lg(m_clientAccess);
101 auto newVal = std::make_unique<ClientThread>();
102 newVal->m_stopThread.store(false);
103 newVal->m_clientThread = std::thread(&CmdExecutor::ClientEventLoop, this, std::move(handler));
104 m_clients.push_back(std::move(newVal));
105 }
106
ClientEventLoop(std::unique_ptr<Socket> handler)107 void CmdExecutor::ClientEventLoop(std::unique_ptr<Socket> handler)
108 {
109 decltype(m_clients)::iterator clientInfoIt;
110 {
111 std::lock_guard<std::mutex> lg(m_clientAccess);
112 clientInfoIt = std::find_if(m_clients.begin(), m_clients.end(),
113 [](const std::unique_ptr<ClientThread>& ct) {
114 return ct->m_clientThread.get_id() == std::this_thread::get_id();
115 });
116 }
117 if (clientInfoIt == m_clients.end()) {
118 std::cerr << "Failed to find client\n";
119 return;
120 }
121
122 prctl(PR_SET_NAME, "hilogd.query");
123 ServiceController serviceCtrl(std::move(handler), m_hilogBuffer);
124 serviceCtrl.CommunicationLoop((*clientInfoIt)->m_stopThread);
125
126 std::lock_guard<std::mutex> ul(m_finishedClientAccess);
127 m_finishedClients.push_back(std::this_thread::get_id());
128 }
129
CleanFinishedClients()130 void CmdExecutor::CleanFinishedClients()
131 {
132 std::list<std::thread> threadsToJoin;
133 {
134 // select clients to clean up - pick threads that we have to be sure are ended
135 std::scoped_lock sl(m_finishedClientAccess, m_clientAccess);
136 for (auto threadId : m_finishedClients) {
137 auto clientInfoIt = std::find_if(m_clients.begin(), m_clients.end(),
138 [&threadId](const std::unique_ptr<ClientThread>& ct) {
139 return ct->m_clientThread.get_id() == threadId;
140 });
141 if (clientInfoIt != m_clients.end()) {
142 threadsToJoin.push_back(std::move((*clientInfoIt)->m_clientThread));
143 m_clients.erase(clientInfoIt);
144 }
145 }
146 m_finishedClients.clear();
147 }
148 for (auto& thread : threadsToJoin) {
149 if (thread.joinable()) {
150 thread.join();
151 }
152 }
153 }
154 } // namespace HiviewDFX
155 } // namespace OHOS
156