btif/src/btif_sock_thread.c

   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, &param);
 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, &param);
 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 }