1 /******************************************************************************
2  *
3  *  Copyright 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 
create_thread(void * (* start_routine)(void *),void * arg,pthread_t * thread_id)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, &param);
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, &param);
128   return ret;
129 }
130 static void init_poll(int cmd_fd);
alloc_thread_slot()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 }
free_thread_slot(int h)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 }
btsock_thread_init()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 }
btsock_thread_create(btsock_signaled_cb callback,btsock_cmd_cb cmd_callback)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 */
init_cmd_fd(int h)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 }
close_cmd_fd(int h)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;
btsock_thread_add_fd(int h,int fd,int type,int flags,uint32_t user_id)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 
btsock_thread_remove_fd_and_close(int thread_handle,int fd)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 
btsock_thread_post_cmd(int h,int type,const unsigned char * data,int size,uint32_t user_id)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 }
btsock_thread_wakeup(int h)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 }
btsock_thread_exit(int h)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 }
init_poll(int h)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 }
flags2pevents(int flags)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 
set_poll(poll_slot_t * ps,int fd,int type,int flags,uint32_t user_id)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 }
add_poll(int h,int fd,int type,int flags,uint32_t user_id)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 }
remove_poll(int h,poll_slot_t * ps,int flags)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 }
process_cmd_sock(int h)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 
print_events(short events)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 
process_data_sock(int h,struct pollfd * pfds,int count)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 
prepare_poll_fds(int h,struct pollfd * pfds)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 }
sock_poll_thread(void * arg)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