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