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