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