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