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.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
79typedef struct {
80  struct pollfd pfd;
81  uint32_t user_id;
82  int type;
83  int flags;
84} poll_slot_t;
85typedef 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;
95static thread_slot_t ts[MAX_THREAD];
96
97static void* sock_poll_thread(void* arg);
98static inline void close_cmd_fd(int h);
99
100static inline void add_poll(int h, int fd, int type, int flags,
101                            uint32_t user_id);
102
103static std::recursive_mutex thread_slot_lock;
104
105static 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}
130static void init_poll(int cmd_fd);
131static 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}
145static 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}
152int 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}
169int 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 */
193static 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}
204static 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}
214typedef struct {
215  int id;
216  int fd;
217  int type;
218  int flags;
219  uint32_t user_id;
220} sock_cmd_t;
221int 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
251bool 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
269int 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}
302int 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}
318int 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}
342static 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}
354static 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
362static 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}
375static 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}
399static 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}
412static 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
454static 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
466static 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
495static 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}
518static 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