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
82typedef struct {
83    struct pollfd pfd;
84    uint32_t user_id;
85    int type;
86    int flags;
87} poll_slot_t;
88typedef 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;
98static thread_slot_t ts[MAX_THREAD];
99
100static void *sock_poll_thread(void *arg);
101static inline void close_cmd_fd(int h);
102
103static inline void add_poll(int h, int fd, int type, int flags, uint32_t user_id);
104
105static pthread_mutex_t thread_slot_lock = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
106
107static 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}
133static void init_poll(int cmd_fd);
134static 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}
150static 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}
159int 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}
179int 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 */
207static 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}
219static 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}
232typedef struct
233{
234    int id;
235    int fd;
236    int type;
237    int flags;
238    uint32_t user_id;
239} sock_cmd_t;
240int 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
273bool 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
294int 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}
331int 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}
350int 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}
376static 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}
390static 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
401static 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}
412static 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}
440static 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}
457static 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}
503static 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
539static 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}
564static 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