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