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