1 /******************************************************************************
2 *
3 * Copyright 2009-2012 Broadcom Corporation
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 /*******************************************************************************
20 *
21 * Filename: btif_sock_thread.cc
22 *
23 * Description: socket select thread
24 *
25 ******************************************************************************/
26
27 #define LOG_TAG "bt_btif_sock"
28
29 #include "btif_sock_thread.h"
30
31 #include <alloca.h>
32 #include <ctype.h>
33 #include <errno.h>
34 #include <fcntl.h>
35 #include <features.h>
36 #include <pthread.h>
37 #include <signal.h>
38 #include <stdio.h>
39 #include <stdlib.h>
40 #include <string.h>
41 #include <sys/poll.h>
42 #include <sys/select.h>
43 #include <sys/socket.h>
44 #include <sys/types.h>
45 #include <sys/un.h>
46 #include <time.h>
47 #include <unistd.h>
48
49 #include <mutex>
50 #include <optional>
51 #include <string>
52
53 #include "bta_api.h"
54 #include "btif_common.h"
55 #include "btif_sock.h"
56 #include "btif_sock_util.h"
57 #include "btif_util.h"
58 #include "osi/include/socket_utils/sockets.h"
59
60 #define asrt(s) \
61 do { \
62 if (!(s)) \
63 APPL_TRACE_ERROR("## %s assert %s failed at line:%d ##", __func__, #s, \
64 __LINE__) \
65 } while (0)
66
67 #define MAX_THREAD 8
68 #define MAX_POLL 64
69 #define POLL_EXCEPTION_EVENTS (POLLHUP | POLLRDHUP | POLLERR | POLLNVAL)
70 #define IS_EXCEPTION(e) ((e)&POLL_EXCEPTION_EVENTS)
71 #define IS_READ(e) ((e)&POLLIN)
72 #define IS_WRITE(e) ((e)&POLLOUT)
73 /*cmd executes in socket poll thread */
74 #define CMD_WAKEUP 1
75 #define CMD_EXIT 2
76 #define CMD_ADD_FD 3
77 #define CMD_REMOVE_FD 4
78 #define CMD_USER_PRIVATE 5
79
80 struct poll_slot_t {
81 struct pollfd pfd;
82 uint32_t user_id;
83 int type;
84 int flags;
85 };
86 struct thread_slot_t {
87 int cmd_fdr, cmd_fdw;
88 int poll_count;
89 poll_slot_t ps[MAX_POLL];
90 int psi[MAX_POLL]; // index of poll slot
91 std::optional<pthread_t> thread_id;
92 btsock_signaled_cb callback;
93 btsock_cmd_cb cmd_callback;
94 int used;
95 };
96 static thread_slot_t ts[MAX_THREAD];
97
98 static void* sock_poll_thread(void* arg);
99 static inline void close_cmd_fd(int h);
100
101 static inline void add_poll(int h, int fd, int type, int flags,
102 uint32_t user_id);
103
104 static std::recursive_mutex thread_slot_lock;
105
create_thread(void * (* start_routine)(void *),void * arg,pthread_t * thread_id)106 static inline int create_thread(void* (*start_routine)(void*), void* arg,
107 pthread_t* thread_id) {
108 pthread_attr_t thread_attr;
109 pthread_attr_init(&thread_attr);
110 pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_JOINABLE);
111 int policy;
112 int min_pri = 0;
113 int ret = -1;
114 struct sched_param param;
115
116 ret = pthread_create(thread_id, &thread_attr, start_routine, arg);
117 if (ret != 0) {
118 APPL_TRACE_ERROR("pthread_create : %s", strerror(errno));
119 return ret;
120 }
121 /* We need to lower the priority of this thread to ensure the stack gets
122 * priority over transfer to a socket */
123 pthread_getschedparam(*thread_id, &policy, ¶m);
124 min_pri = sched_get_priority_min(policy);
125 if (param.sched_priority > min_pri) {
126 param.sched_priority -= 1;
127 }
128 pthread_setschedparam(*thread_id, policy, ¶m);
129 return ret;
130 }
131 static void init_poll(int cmd_fd);
alloc_thread_slot()132 static int alloc_thread_slot() {
133 std::unique_lock<std::recursive_mutex> lock(thread_slot_lock);
134 int i;
135 // reversed order to save guard uninitialized access to 0 index
136 for (i = MAX_THREAD - 1; i >= 0; i--) {
137 if (!ts[i].used) {
138 ts[i].used = 1;
139 return i;
140 }
141 }
142 APPL_TRACE_ERROR("execeeded max thread count");
143 return -1;
144 }
free_thread_slot(int h)145 static void free_thread_slot(int h) {
146 if (0 <= h && h < MAX_THREAD) {
147 close_cmd_fd(h);
148 ts[h].used = 0;
149 } else
150 APPL_TRACE_ERROR("invalid thread handle:%d", h);
151 }
btsock_thread_init()152 void btsock_thread_init() {
153 static int initialized;
154 if (!initialized) {
155 initialized = 1;
156 int h;
157 for (h = 0; h < MAX_THREAD; h++) {
158 ts[h].cmd_fdr = ts[h].cmd_fdw = -1;
159 ts[h].used = 0;
160 ts[h].thread_id = std::nullopt;
161 ts[h].poll_count = 0;
162 ts[h].callback = NULL;
163 ts[h].cmd_callback = NULL;
164 }
165 }
166 }
btsock_thread_create(btsock_signaled_cb callback,btsock_cmd_cb cmd_callback)167 int btsock_thread_create(btsock_signaled_cb callback,
168 btsock_cmd_cb cmd_callback) {
169 asrt(callback || cmd_callback);
170 int h = alloc_thread_slot();
171 if (h >= 0) {
172 init_poll(h);
173 pthread_t thread;
174 int status = create_thread(sock_poll_thread, (void*)(uintptr_t)h, &thread);
175 if (status) {
176 APPL_TRACE_ERROR("create_thread failed: %s", strerror(status));
177 free_thread_slot(h);
178 return -1;
179 }
180
181 ts[h].thread_id = thread;
182 ts[h].callback = callback;
183 ts[h].cmd_callback = cmd_callback;
184 }
185 return h;
186 }
187
188 /* create dummy socket pair used to wake up select loop */
init_cmd_fd(int h)189 static inline void init_cmd_fd(int h) {
190 asrt(ts[h].cmd_fdr == -1 && ts[h].cmd_fdw == -1);
191 if (socketpair(AF_UNIX, SOCK_STREAM, 0, &ts[h].cmd_fdr) < 0) {
192 APPL_TRACE_ERROR("socketpair failed: %s", strerror(errno));
193 return;
194 }
195 // add the cmd fd for read & write
196 add_poll(h, ts[h].cmd_fdr, 0, SOCK_THREAD_FD_RD, 0);
197 }
close_cmd_fd(int h)198 static inline void close_cmd_fd(int h) {
199 if (ts[h].cmd_fdr != -1) {
200 close(ts[h].cmd_fdr);
201 ts[h].cmd_fdr = -1;
202 }
203 if (ts[h].cmd_fdw != -1) {
204 close(ts[h].cmd_fdw);
205 ts[h].cmd_fdw = -1;
206 }
207 }
208 typedef struct {
209 int id;
210 int fd;
211 int type;
212 int flags;
213 uint32_t user_id;
214 } sock_cmd_t;
btsock_thread_add_fd(int h,int fd,int type,int flags,uint32_t user_id)215 int btsock_thread_add_fd(int h, int fd, int type, int flags, uint32_t user_id) {
216 if (h < 0 || h >= MAX_THREAD) {
217 APPL_TRACE_ERROR("invalid bt thread handle:%d", h);
218 return false;
219 }
220 if (ts[h].cmd_fdw == -1) {
221 APPL_TRACE_ERROR(
222 "cmd socket is not created. socket thread may not initialized");
223 return false;
224 }
225 if (flags & SOCK_THREAD_ADD_FD_SYNC) {
226 // must executed in socket poll thread
227 if (ts[h].thread_id.value() == pthread_self()) {
228 // cleanup one-time flags
229 flags &= ~SOCK_THREAD_ADD_FD_SYNC;
230 add_poll(h, fd, type, flags, user_id);
231 return true;
232 }
233 LOG_WARN(
234 "THREAD_ADD_FD_SYNC is not called in poll thread, fallback to async");
235 }
236 sock_cmd_t cmd = {CMD_ADD_FD, fd, type, flags, user_id};
237
238 ssize_t ret;
239 OSI_NO_INTR(ret = send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0));
240
241 return ret == sizeof(cmd);
242 }
243
btsock_thread_remove_fd_and_close(int thread_handle,int fd)244 bool btsock_thread_remove_fd_and_close(int thread_handle, int fd) {
245 if (thread_handle < 0 || thread_handle >= MAX_THREAD) {
246 APPL_TRACE_ERROR("%s invalid thread handle: %d", __func__, thread_handle);
247 return false;
248 }
249 if (fd == -1) {
250 APPL_TRACE_ERROR("%s invalid file descriptor.", __func__);
251 return false;
252 }
253
254 sock_cmd_t cmd = {CMD_REMOVE_FD, fd, 0, 0, 0};
255
256 ssize_t ret;
257 OSI_NO_INTR(ret = send(ts[thread_handle].cmd_fdw, &cmd, sizeof(cmd), 0));
258
259 return ret == sizeof(cmd);
260 }
261
btsock_thread_post_cmd(int h,int type,const unsigned char * data,int size,uint32_t user_id)262 int btsock_thread_post_cmd(int h, int type, const unsigned char* data, int size,
263 uint32_t user_id) {
264 if (h < 0 || h >= MAX_THREAD) {
265 APPL_TRACE_ERROR("invalid bt thread handle:%d", h);
266 return false;
267 }
268 if (ts[h].cmd_fdw == -1) {
269 APPL_TRACE_ERROR(
270 "cmd socket is not created. socket thread may not initialized");
271 return false;
272 }
273 sock_cmd_t cmd = {CMD_USER_PRIVATE, 0, type, size, user_id};
274 sock_cmd_t* cmd_send = &cmd;
275 int size_send = sizeof(cmd);
276 if (data && size) {
277 size_send = sizeof(cmd) + size;
278 cmd_send = (sock_cmd_t*)alloca(size_send);
279 if (cmd_send) {
280 *cmd_send = cmd;
281 memcpy(cmd_send + 1, data, size);
282 } else {
283 APPL_TRACE_ERROR("alloca failed at h:%d, cmd type:%d, size:%d", h, type,
284 size_send);
285 return false;
286 }
287 }
288
289 ssize_t ret;
290 OSI_NO_INTR(ret = send(ts[h].cmd_fdw, cmd_send, size_send, 0));
291
292 return ret == size_send;
293 }
btsock_thread_wakeup(int h)294 int btsock_thread_wakeup(int h) {
295 if (h < 0 || h >= MAX_THREAD) {
296 APPL_TRACE_ERROR("invalid bt thread handle:%d", h);
297 return false;
298 }
299 if (ts[h].cmd_fdw == -1) {
300 APPL_TRACE_ERROR("thread handle:%d, cmd socket is not created", h);
301 return false;
302 }
303 sock_cmd_t cmd = {CMD_WAKEUP, 0, 0, 0, 0};
304
305 ssize_t ret;
306 OSI_NO_INTR(ret = send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0));
307
308 return ret == sizeof(cmd);
309 }
btsock_thread_exit(int h)310 int btsock_thread_exit(int h) {
311 if (h < 0 || h >= MAX_THREAD) {
312 APPL_TRACE_ERROR("invalid bt thread slot:%d", h);
313 return false;
314 }
315 if (ts[h].cmd_fdw == -1) {
316 APPL_TRACE_ERROR("cmd socket is not created");
317 return false;
318 }
319 sock_cmd_t cmd = {CMD_EXIT, 0, 0, 0, 0};
320
321 ssize_t ret;
322 OSI_NO_INTR(ret = send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0));
323
324 if (ret == sizeof(cmd)) {
325 if (ts[h].thread_id != std::nullopt) {
326 pthread_join(ts[h].thread_id.value(), 0);
327 ts[h].thread_id = std::nullopt;
328 }
329 free_thread_slot(h);
330 return true;
331 }
332 return false;
333 }
init_poll(int h)334 static void init_poll(int h) {
335 int i;
336 ts[h].poll_count = 0;
337 ts[h].thread_id = std::nullopt;
338 ts[h].callback = NULL;
339 ts[h].cmd_callback = NULL;
340 for (i = 0; i < MAX_POLL; i++) {
341 ts[h].ps[i].pfd.fd = -1;
342 ts[h].psi[i] = -1;
343 }
344 init_cmd_fd(h);
345 }
flags2pevents(int flags)346 static inline unsigned int flags2pevents(int flags) {
347 unsigned int pevents = 0;
348 if (flags & SOCK_THREAD_FD_WR) pevents |= POLLOUT;
349 if (flags & SOCK_THREAD_FD_RD) pevents |= POLLIN;
350 pevents |= POLL_EXCEPTION_EVENTS;
351 return pevents;
352 }
353
set_poll(poll_slot_t * ps,int fd,int type,int flags,uint32_t user_id)354 static inline void set_poll(poll_slot_t* ps, int fd, int type, int flags,
355 uint32_t user_id) {
356 ps->pfd.fd = fd;
357 ps->user_id = user_id;
358 if (ps->type != 0 && ps->type != type)
359 APPL_TRACE_ERROR(
360 "poll socket type should not changed! type was:%d, type now:%d",
361 ps->type, type);
362 ps->type = type;
363 ps->flags = flags;
364 ps->pfd.events = flags2pevents(flags);
365 ps->pfd.revents = 0;
366 }
add_poll(int h,int fd,int type,int flags,uint32_t user_id)367 static inline void add_poll(int h, int fd, int type, int flags,
368 uint32_t user_id) {
369 asrt(fd != -1);
370 int i;
371 int empty = -1;
372 poll_slot_t* ps = ts[h].ps;
373
374 for (i = 0; i < MAX_POLL; i++) {
375 if (ps[i].pfd.fd == fd) {
376 asrt(ts[h].poll_count < MAX_POLL);
377
378 set_poll(&ps[i], fd, type, flags | ps[i].flags, user_id);
379 return;
380 } else if (empty < 0 && ps[i].pfd.fd == -1)
381 empty = i;
382 }
383 if (empty >= 0) {
384 asrt(ts[h].poll_count < MAX_POLL);
385 set_poll(&ps[empty], fd, type, flags, user_id);
386 ++ts[h].poll_count;
387 return;
388 }
389 APPL_TRACE_ERROR("exceeded max poll slot:%d!", MAX_POLL);
390 }
remove_poll(int h,poll_slot_t * ps,int flags)391 static inline void remove_poll(int h, poll_slot_t* ps, int flags) {
392 if (flags == ps->flags) {
393 // all monitored events signaled. To remove it, just clear the slot
394 --ts[h].poll_count;
395 memset(ps, 0, sizeof(*ps));
396 ps->pfd.fd = -1;
397 } else {
398 // one read or one write monitor event signaled, removed the accordding bit
399 ps->flags &= ~flags;
400 // update the poll events mask
401 ps->pfd.events = flags2pevents(ps->flags);
402 }
403 }
process_cmd_sock(int h)404 static int process_cmd_sock(int h) {
405 sock_cmd_t cmd = {-1, 0, 0, 0, 0};
406 int fd = ts[h].cmd_fdr;
407
408 ssize_t ret;
409 OSI_NO_INTR(ret = recv(fd, &cmd, sizeof(cmd), MSG_WAITALL));
410
411 if (ret != sizeof(cmd)) {
412 LOG_ERROR("recv cmd errno:%d", errno);
413 return false;
414 }
415 switch (cmd.id) {
416 case CMD_ADD_FD:
417 add_poll(h, cmd.fd, cmd.type, cmd.flags, cmd.user_id);
418 break;
419 case CMD_REMOVE_FD:
420 for (int i = 1; i < MAX_POLL; ++i) {
421 poll_slot_t* poll_slot = &ts[h].ps[i];
422 if (poll_slot->pfd.fd == cmd.fd) {
423 remove_poll(h, poll_slot, poll_slot->flags);
424 break;
425 }
426 }
427 close(cmd.fd);
428 break;
429 case CMD_WAKEUP:
430 break;
431 case CMD_USER_PRIVATE:
432 asrt(ts[h].cmd_callback);
433 if (ts[h].cmd_callback)
434 ts[h].cmd_callback(fd, cmd.type, cmd.flags, cmd.user_id);
435 break;
436 case CMD_EXIT:
437 return false;
438 default:
439 LOG_WARN("unknown cmd: %d", cmd.id);
440 break;
441 }
442 return true;
443 }
444
process_data_sock(int h,struct pollfd * pfds,int count)445 static void process_data_sock(int h, struct pollfd* pfds, int count) {
446 asrt(count <= ts[h].poll_count);
447 int i;
448 for (i = 1; i < ts[h].poll_count; i++) {
449 if (pfds[i].revents) {
450 int ps_i = ts[h].psi[i];
451 asrt(pfds[i].fd == ts[h].ps[ps_i].pfd.fd);
452 uint32_t user_id = ts[h].ps[ps_i].user_id;
453 int type = ts[h].ps[ps_i].type;
454 int flags = 0;
455 if (IS_READ(pfds[i].revents)) {
456 flags |= SOCK_THREAD_FD_RD;
457 }
458 if (IS_WRITE(pfds[i].revents)) {
459 flags |= SOCK_THREAD_FD_WR;
460 }
461 if (IS_EXCEPTION(pfds[i].revents)) {
462 flags |= SOCK_THREAD_FD_EXCEPTION;
463 // remove the whole slot not flags
464 remove_poll(h, &ts[h].ps[ps_i], ts[h].ps[ps_i].flags);
465 } else if (flags)
466 remove_poll(h, &ts[h].ps[ps_i],
467 flags); // remove the monitor flags that already processed
468 if (flags) ts[h].callback(pfds[i].fd, type, flags, user_id);
469 }
470 }
471 }
472
prepare_poll_fds(int h,struct pollfd * pfds)473 static void prepare_poll_fds(int h, struct pollfd* pfds) {
474 int count = 0;
475 int ps_i = 0;
476 int pfd_i = 0;
477 asrt(ts[h].poll_count <= MAX_POLL);
478 memset(pfds, 0, sizeof(pfds[0]) * ts[h].poll_count);
479 while (count < ts[h].poll_count) {
480 if (ps_i >= MAX_POLL) {
481 APPL_TRACE_ERROR(
482 "exceed max poll range, ps_i:%d, MAX_POLL:%d, count:%d, "
483 "ts[h].poll_count:%d",
484 ps_i, MAX_POLL, count, ts[h].poll_count);
485 return;
486 }
487 if (ts[h].ps[ps_i].pfd.fd >= 0) {
488 pfds[pfd_i] = ts[h].ps[ps_i].pfd;
489 ts[h].psi[pfd_i] = ps_i;
490 count++;
491 pfd_i++;
492 }
493 ps_i++;
494 }
495 }
sock_poll_thread(void * arg)496 static void* sock_poll_thread(void* arg) {
497 struct pollfd pfds[MAX_POLL];
498 memset(pfds, 0, sizeof(pfds));
499 int h = (intptr_t)arg;
500 for (;;) {
501 prepare_poll_fds(h, pfds);
502 int ret;
503 OSI_NO_INTR(ret = poll(pfds, ts[h].poll_count, -1));
504 if (ret == -1) {
505 APPL_TRACE_ERROR("poll ret -1, exit the thread, errno:%d, err:%s", errno,
506 strerror(errno));
507 break;
508 }
509 if (ret != 0) {
510 int need_process_data_fd = true;
511 if (pfds[0].revents) // cmd fd always is the first one
512 {
513 asrt(pfds[0].fd == ts[h].cmd_fdr);
514 if (!process_cmd_sock(h)) {
515 LOG_INFO("h:%d, process_cmd_sock return false, exit...", h);
516 break;
517 }
518 if (ret == 1)
519 need_process_data_fd = false;
520 else
521 ret--; // exclude the cmd fd
522 }
523 if (need_process_data_fd) process_data_sock(h, pfds, ret);
524 } else {
525 LOG_INFO("no data, select ret: %d", ret);
526 };
527 }
528 LOG_INFO("socket poll thread exiting, h:%d", h);
529 return 0;
530 }
531