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