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
16 #include "event_demultiplexer.h"
17 #include "event_reactor.h"
18 #include "event_handler.h"
19 #include "common_timer_errors.h"
20 #include "utils_log.h"
21
22 #include <vector>
23 #include <sys/epoll.h>
24
25 namespace OHOS {
26 namespace Utils {
27
28 static const int EPOLL_MAX_EVENS_INIT = 8;
29 static const int HALF_OF_MAX_EVENT = 2;
30 static const int EPOLL_INVALID_FD = -1;
31 static const int INTERRUPTED_SYS_CALL = 4;
32 static const int EPOLL_ERROR_BADF = 9;
33 static const int EPOLL_ERROR_EINVAL = 22;
34 static const int QUEUE_MAX_SIZE = 10;
35
EventDemultiplexer()36 EventDemultiplexer::EventDemultiplexer()
37 : epollFd_(epoll_create1(EPOLL_CLOEXEC)), maxEvents_(EPOLL_MAX_EVENS_INIT), mutex_(), eventHandlers_(),
38 epollCtlErrQueue_()
39 {
40 }
41
~EventDemultiplexer()42 EventDemultiplexer::~EventDemultiplexer()
43 {
44 CleanUp();
45 }
46
StartUp()47 uint32_t EventDemultiplexer::StartUp()
48 {
49 if (epollFd_ < 0) {
50 epollFd_ = epoll_create1(EPOLL_CLOEXEC);
51 if (epollFd_ < 0) {
52 return TIMER_ERR_BADF;
53 }
54 }
55 return TIMER_ERR_OK;
56 }
57
CleanUp()58 void EventDemultiplexer::CleanUp()
59 {
60 if (epollFd_ != EPOLL_INVALID_FD) {
61 close(epollFd_);
62 epollFd_ = EPOLL_INVALID_FD;
63 }
64 }
65
UpdateEventHandler(EventHandler * handler)66 uint32_t EventDemultiplexer::UpdateEventHandler(EventHandler* handler)
67 {
68 if (handler == nullptr) {
69 return TIMER_ERR_INVALID_VALUE;
70 }
71
72 std::lock_guard<std::recursive_mutex> lock(mutex_);
73 auto itor = eventHandlers_.find(handler->GetHandle());
74 if (itor == eventHandlers_.end()) {
75 eventHandlers_.insert(std::make_pair(handler->GetHandle(), handler->shared_from_this()));
76 return Update(EPOLL_CTL_ADD, handler);
77 }
78
79 if (handler->Events() == EventReactor::NONE_EVENT) {
80 eventHandlers_.erase(itor);
81 return Update(EPOLL_CTL_DEL, handler);
82 }
83
84 if (handler != itor->second.get()) {
85 return TIMER_ERR_DEAL_FAILED;
86 }
87 return Update(EPOLL_CTL_MOD, handler);
88 }
89
Update(int operation,EventHandler * handler)90 uint32_t EventDemultiplexer::Update(int operation, EventHandler* handler)
91 {
92 struct epoll_event event;
93 bzero(&event, sizeof(event));
94 event.events = Reactor2Epoll(handler->Events());
95 event.data.fd = handler->GetHandle();
96
97 if (epoll_ctl(epollFd_, operation, handler->GetHandle(), &event) != 0) {
98 UTILS_LOGE("epoll_ctl %{public}d operation %{public}d on handle %{public}d failed, errno %{public}d",
99 epollFd_, operation, handler->GetHandle(), errno);
100
101 if (epollCtlErrQueue_.size() < QUEUE_MAX_SIZE) {
102 epollCtlErrQueue_.push(std::make_tuple(operation, errno, handler->GetHandle()));
103 }
104 return TIMER_ERR_DEAL_FAILED;
105 }
106 return TIMER_ERR_OK;
107 }
108
Polling(int timeout)109 int EventDemultiplexer::Polling(int timeout /* ms */)
110 {
111 std::vector<struct epoll_event> epollEvents(maxEvents_);
112 std::vector<std::shared_ptr<EventHandler>> taskQue;
113 std::vector<uint32_t> eventQue;
114
115 int nfds = epoll_wait(epollFd_, &epollEvents[0], static_cast<int>(epollEvents.size()), timeout);
116 if (nfds == 0) {
117 return nfds;
118 }
119 if (nfds == -1) {
120 if (errno != INTERRUPTED_SYS_CALL) {
121 UTILS_LOGE("epoll_wait failed, errno %{public}d, epollFd_: %{public}d, pollEvents.size: %{public}zu",
122 errno, epollFd_, epollEvents.size());
123 }
124 if (errno == EPOLL_ERROR_BADF || errno == EPOLL_ERROR_EINVAL) {
125 UTILS_LOGE("epoll_wait critical error, thread exit");
126 return EPOLL_CRITICAL_ERROR;
127 }
128 return nfds;
129 }
130
131 {
132 std::lock_guard<std::recursive_mutex> lock(mutex_);
133 for (int idx = 0; idx < nfds; ++idx) {
134 int targetFd = epollEvents[idx].data.fd;
135 uint32_t events = epollEvents[idx].events;
136
137 auto itor = eventHandlers_.find(targetFd);
138 if (itor != eventHandlers_.end()) {
139 taskQue.emplace_back(itor->second);
140 eventQue.emplace_back(events);
141 } else {
142 UTILS_LOGE("fd not found in eventHandlers_, fd=%{public}d, events=%{public}d", targetFd, events);
143
144 while (!epollCtlErrQueue_.empty()) {
145 auto [operation, errnoNum, fd] = epollCtlErrQueue_.front();
146 UTILS_LOGE("epoll_ctl: %{public}d, errno: %{public}d, handle: %{public}d", operation, errnoNum, fd);
147 epollCtlErrQueue_.pop();
148 }
149 }
150 }
151 }
152
153 for (size_t idx = 0u; idx < taskQue.size() && idx < eventQue.size(); idx++) {
154 taskQue[idx]->HandleEvents(eventQue[idx]);
155 }
156
157 if (nfds == maxEvents_) {
158 maxEvents_ *= HALF_OF_MAX_EVENT;
159 }
160 return nfds;
161 }
162
Epoll2Reactor(uint32_t epollEvents)163 uint32_t EventDemultiplexer::Epoll2Reactor(uint32_t epollEvents)
164 {
165 if (epollEvents & (EPOLLIN | EPOLLPRI | EPOLLRDHUP)) {
166 return EventReactor::READ_EVENT;
167 }
168
169 return EventReactor::NONE_EVENT;
170 }
171
Reactor2Epoll(uint32_t reactorEvent)172 uint32_t EventDemultiplexer::Reactor2Epoll(uint32_t reactorEvent)
173 {
174 switch (reactorEvent) {
175 case EventReactor::NONE_EVENT:
176 return TIMER_ERR_OK;
177 case EventReactor::READ_EVENT:
178 return EPOLLIN | EPOLLPRI;
179 default:
180 UTILS_LOGD("invalid event %{public}u.", reactorEvent);
181 return TIMER_ERR_DEAL_FAILED;
182 }
183 }
184
185 }
186 }
187