1 /*
2 * Copyright (c) 2022-2023 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
16 #include "wrapper_listener.h"
17
18 #include <cerrno>
19 #include <cstdlib>
20 #include <fcntl.h>
21 #include <memory>
22 #include <poll.h>
23 #include <sys/socket.h>
24 #include <thread>
25 #include <unistd.h>
26 #include <vector>
27 #include <pthread.h>
28
29 #include "ffrt.h"
30 #include "ffrt_inner.h"
31 #include "netlink_define.h"
32 #include "netnative_log_wrapper.h"
33
34 namespace OHOS::nmd {
35 namespace {
36 constexpr int32_t PRET_SIZE = 2;
37 } // namespace
38
WrapperListener(int32_t socket,std::function<void (int32_t)> recvFunc)39 WrapperListener::WrapperListener(int32_t socket, std::function<void(int32_t)> recvFunc)
40 {
41 socket_ = socket;
42 startReceiveFunc_ = recvFunc;
43 }
44
~WrapperListener()45 WrapperListener::~WrapperListener()
46 {
47 Stop();
48 }
49
Start()50 int32_t WrapperListener::Start()
51 {
52 if (socket_ < 0) {
53 NETNATIVE_LOGE("listener socket_ < 0 %{public}d", socket_);
54 return NetlinkResult::ERROR;
55 }
56
57 int pipeRet = pipe2(pipe_, O_CLOEXEC);
58 if (pipeRet != 0) {
59 NETNATIVE_LOGE("pipeRes = %{public}d, pipe create failed errno = %{public}d, %{public}s", pipeRet, errno,
60 strerror(errno));
61 return NetlinkResult::ERROR;
62 }
63 ffrt::submit([this]() { WrapperListener::ListenThread(this); }, {}, {}, ffrt::task_attr().name("WrapListen"));
64 return NetlinkResult::OK;
65 }
66
Stop()67 int32_t WrapperListener::Stop()
68 {
69 NETNATIVE_LOGI("WrapperListener: Stop");
70 char pipe = PIPE_SHUTDOWN;
71 if (TEMP_FAILURE_RETRY(write(pipe_[1], &pipe, sizeof(pipe))) != 1) {
72 NETNATIVE_LOGE("write pipe failed errno = %{public}d, %{public}s", errno, strerror(errno));
73 return NetlinkResult::ERROR;
74 }
75
76 for (auto &pi : pipe_) {
77 if (pi > 0) {
78 close(pi);
79 }
80 }
81
82 if (socket_ > -1) {
83 close(socket_);
84 }
85 const int32_t exitDelay = 1500;
86 std::this_thread::sleep_for(std::chrono::milliseconds(exitDelay));
87
88 return NetlinkResult::OK;
89 }
90
ListenThread(WrapperListener * listener)91 void WrapperListener::ListenThread(WrapperListener *listener)
92 {
93 listener->Listen();
94 }
95
Listen()96 void WrapperListener::Listen()
97 {
98 if (startReceiveFunc_ == nullptr) {
99 NETNATIVE_LOGE("startReceiveFunc_ is nullptr start listen failed");
100 return;
101 }
102 while (true) {
103 std::vector<pollfd> pollFds;
104 std::lock_guard<ffrt::mutex> lock(clientsLock_);
105 pollFds.reserve(PRET_SIZE + 1);
106 pollfd polfd;
107 polfd.fd = pipe_[0];
108 polfd.events = POLLIN;
109 pollFds.emplace_back(polfd);
110 polfd.fd = socket_;
111 polfd.events = POLLIN;
112 pollFds.emplace_back(polfd);
113 ffrt::sync_io(socket_);
114 int32_t ret = TEMP_FAILURE_RETRY(poll(pollFds.data(), pollFds.size(), -1));
115 if (ret < 0) {
116 ffrt::this_task::sleep_for(std::chrono::seconds(1));
117 }
118
119 if (pollFds[0].revents & (POLLIN | POLLERR)) {
120 char ctlp = PIPE_SHUTDOWN;
121 TEMP_FAILURE_RETRY(read(pipe_[0], &ctlp, sizeof(ctlp)));
122 if (ctlp == PIPE_SHUTDOWN) {
123 break;
124 }
125 continue;
126 }
127 startReceiveFunc_(socket_);
128 }
129 }
130 } // namespace OHOS::nmd
131