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