1 /******************************************************************************
2 *
3 * Copyright 2014 Google, Inc.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at:
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 ******************************************************************************/
18
19 #define LOG_TAG "bt_osi_reactor"
20
21 #include "osi/include/reactor.h"
22
23 #include <base/logging.h>
24 #include <errno.h>
25 #include <pthread.h>
26 #include <stdlib.h>
27 #include <string.h>
28 #include <sys/epoll.h>
29 #include <sys/eventfd.h>
30 #include <unistd.h>
31
32 #include <mutex>
33
34 #include "osi/include/allocator.h"
35 #include "osi/include/list.h"
36 #include "osi/include/log.h"
37
38 #if !defined(EFD_SEMAPHORE)
39 #define EFD_SEMAPHORE (1 << 0)
40 #endif
41
42 struct reactor_t {
43 int epoll_fd;
44 int event_fd;
45 std::mutex* list_mutex;
46 list_t* invalidation_list; // reactor objects that have been unregistered.
47 pthread_t run_thread; // the pthread on which reactor_run is executing.
48 bool is_running; // indicates whether |run_thread| is valid.
49 bool object_removed;
50 };
51
52 struct reactor_object_t {
53 int fd; // the file descriptor to monitor for events.
54 void* context; // a context that's passed back to the *_ready functions.
55 reactor_t* reactor; // the reactor instance this object is registered with.
56 std::mutex* mutex; // protects the lifetime of this object and all variables.
57
58 void (*read_ready)(void* context); // function to call when the file
59 // descriptor becomes readable.
60 void (*write_ready)(void* context); // function to call when the file
61 // descriptor becomes writeable.
62 };
63
64 static reactor_status_t run_reactor(reactor_t* reactor, int iterations);
65
66 static const size_t MAX_EVENTS = 64;
67 static const eventfd_t EVENT_REACTOR_STOP = 1;
68
reactor_new(void)69 reactor_t* reactor_new(void) {
70 reactor_t* ret = (reactor_t*)osi_calloc(sizeof(reactor_t));
71
72 ret->epoll_fd = INVALID_FD;
73 ret->event_fd = INVALID_FD;
74
75 ret->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
76 if (ret->epoll_fd == INVALID_FD) {
77 LOG_ERROR("%s unable to create epoll instance: %s", __func__,
78 strerror(errno));
79 goto error;
80 }
81
82 ret->event_fd = eventfd(0, 0);
83 if (ret->event_fd == INVALID_FD) {
84 LOG_ERROR("%s unable to create eventfd: %s", __func__, strerror(errno));
85 goto error;
86 }
87
88 ret->list_mutex = new std::mutex;
89 ret->invalidation_list = list_new(NULL);
90 if (!ret->invalidation_list) {
91 LOG_ERROR("%s unable to allocate object invalidation list.", __func__);
92 goto error;
93 }
94
95 struct epoll_event event;
96 memset(&event, 0, sizeof(event));
97 event.events = EPOLLIN;
98 event.data.ptr = NULL;
99 if (epoll_ctl(ret->epoll_fd, EPOLL_CTL_ADD, ret->event_fd, &event) == -1) {
100 LOG_ERROR("%s unable to register eventfd with epoll set: %s", __func__,
101 strerror(errno));
102 goto error;
103 }
104
105 return ret;
106
107 error:;
108 reactor_free(ret);
109 return NULL;
110 }
111
reactor_free(reactor_t * reactor)112 void reactor_free(reactor_t* reactor) {
113 if (!reactor) return;
114
115 list_free(reactor->invalidation_list);
116 close(reactor->event_fd);
117 close(reactor->epoll_fd);
118 osi_free(reactor);
119 }
120
reactor_start(reactor_t * reactor)121 reactor_status_t reactor_start(reactor_t* reactor) {
122 CHECK(reactor != NULL);
123 return run_reactor(reactor, 0);
124 }
125
reactor_run_once(reactor_t * reactor)126 reactor_status_t reactor_run_once(reactor_t* reactor) {
127 CHECK(reactor != NULL);
128 return run_reactor(reactor, 1);
129 }
130
reactor_stop(reactor_t * reactor)131 void reactor_stop(reactor_t* reactor) {
132 CHECK(reactor != NULL);
133
134 eventfd_write(reactor->event_fd, EVENT_REACTOR_STOP);
135 }
136
reactor_register(reactor_t * reactor,int fd,void * context,void (* read_ready)(void * context),void (* write_ready)(void * context))137 reactor_object_t* reactor_register(reactor_t* reactor, int fd, void* context,
138 void (*read_ready)(void* context),
139 void (*write_ready)(void* context)) {
140 CHECK(reactor != NULL);
141 CHECK(fd != INVALID_FD);
142
143 reactor_object_t* object =
144 (reactor_object_t*)osi_calloc(sizeof(reactor_object_t));
145
146 object->reactor = reactor;
147 object->fd = fd;
148 object->context = context;
149 object->read_ready = read_ready;
150 object->write_ready = write_ready;
151 object->mutex = new std::mutex;
152
153 struct epoll_event event;
154 memset(&event, 0, sizeof(event));
155 if (read_ready) event.events |= (EPOLLIN | EPOLLRDHUP);
156 if (write_ready) event.events |= EPOLLOUT;
157 event.data.ptr = object;
158
159 if (epoll_ctl(reactor->epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1) {
160 LOG_ERROR("%s unable to register fd %d to epoll set: %s", __func__, fd,
161 strerror(errno));
162 delete object->mutex;
163 osi_free(object);
164 return NULL;
165 }
166
167 return object;
168 }
169
reactor_change_registration(reactor_object_t * object,void (* read_ready)(void * context),void (* write_ready)(void * context))170 bool reactor_change_registration(reactor_object_t* object,
171 void (*read_ready)(void* context),
172 void (*write_ready)(void* context)) {
173 CHECK(object != NULL);
174
175 struct epoll_event event;
176 memset(&event, 0, sizeof(event));
177 if (read_ready) event.events |= (EPOLLIN | EPOLLRDHUP);
178 if (write_ready) event.events |= EPOLLOUT;
179 event.data.ptr = object;
180
181 if (epoll_ctl(object->reactor->epoll_fd, EPOLL_CTL_MOD, object->fd, &event) ==
182 -1) {
183 LOG_ERROR("%s unable to modify interest set for fd %d: %s", __func__,
184 object->fd, strerror(errno));
185 return false;
186 }
187
188 std::lock_guard<std::mutex> lock(*object->mutex);
189 object->read_ready = read_ready;
190 object->write_ready = write_ready;
191
192 return true;
193 }
194
reactor_unregister(reactor_object_t * obj)195 void reactor_unregister(reactor_object_t* obj) {
196 CHECK(obj != NULL);
197
198 reactor_t* reactor = obj->reactor;
199
200 if (epoll_ctl(reactor->epoll_fd, EPOLL_CTL_DEL, obj->fd, NULL) == -1)
201 LOG_ERROR("%s unable to unregister fd %d from epoll set: %s", __func__,
202 obj->fd, strerror(errno));
203
204 if (reactor->is_running &&
205 pthread_equal(pthread_self(), reactor->run_thread)) {
206 reactor->object_removed = true;
207 return;
208 }
209
210 {
211 std::unique_lock<std::mutex> lock(*reactor->list_mutex);
212 list_append(reactor->invalidation_list, obj);
213 }
214
215 // Taking the object lock here makes sure a callback for |obj| isn't
216 // currently executing. The reactor thread must then either be before
217 // the callbacks or after. If after, we know that the object won't be
218 // referenced because it has been taken out of the epoll set. If before,
219 // it won't be referenced because the reactor thread will check the
220 // invalidation_list and find it in there. So by taking this lock, we
221 // are waiting until the reactor thread drops all references to |obj|.
222 // One the wait completes, we can unlock and destroy |obj| safely.
223 obj->mutex->lock();
224 obj->mutex->unlock();
225 delete obj->mutex;
226 osi_free(obj);
227 }
228
229 // Runs the reactor loop for a maximum of |iterations|.
230 // 0 |iterations| means loop forever.
231 // |reactor| may not be NULL.
run_reactor(reactor_t * reactor,int iterations)232 static reactor_status_t run_reactor(reactor_t* reactor, int iterations) {
233 CHECK(reactor != NULL);
234
235 reactor->run_thread = pthread_self();
236 reactor->is_running = true;
237
238 struct epoll_event events[MAX_EVENTS];
239 for (int i = 0; iterations == 0 || i < iterations; ++i) {
240 {
241 std::lock_guard<std::mutex> lock(*reactor->list_mutex);
242 list_clear(reactor->invalidation_list);
243 }
244
245 int ret;
246 OSI_NO_INTR(ret = epoll_wait(reactor->epoll_fd, events, MAX_EVENTS, -1));
247 if (ret == -1) {
248 LOG_ERROR("%s error in epoll_wait: %s", __func__, strerror(errno));
249 reactor->is_running = false;
250 return REACTOR_STATUS_ERROR;
251 }
252
253 for (int j = 0; j < ret; ++j) {
254 // The event file descriptor is the only one that registers with
255 // a NULL data pointer. We use the NULL to identify it and break
256 // out of the reactor loop.
257 if (events[j].data.ptr == NULL) {
258 eventfd_t value;
259 eventfd_read(reactor->event_fd, &value);
260 reactor->is_running = false;
261 return REACTOR_STATUS_STOP;
262 }
263
264 reactor_object_t* object = (reactor_object_t*)events[j].data.ptr;
265
266 std::unique_lock<std::mutex> lock(*reactor->list_mutex);
267 if (list_contains(reactor->invalidation_list, object)) {
268 continue;
269 }
270
271 // Downgrade the list lock to an object lock.
272 {
273 std::lock_guard<std::mutex> obj_lock(*object->mutex);
274 lock.unlock();
275
276 reactor->object_removed = false;
277 if (events[j].events & (EPOLLIN | EPOLLHUP | EPOLLRDHUP | EPOLLERR) &&
278 object->read_ready)
279 object->read_ready(object->context);
280 if (!reactor->object_removed && events[j].events & EPOLLOUT &&
281 object->write_ready)
282 object->write_ready(object->context);
283 }
284
285 if (reactor->object_removed) {
286 delete object->mutex;
287 osi_free(object);
288 }
289 }
290 }
291
292 reactor->is_running = false;
293 return REACTOR_STATUS_DONE;
294 }
295