1 /******************************************************************************
2  *
3  *  Copyright 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 <bluetooth/log.h>
33 #include <fcntl.h>
34 #include <features.h>
35 #include <poll.h>
36 #include <pthread.h>
37 #include <stdlib.h>
38 #include <string.h>
39 #include <sys/select.h>
40 #include <sys/socket.h>
41 #include <sys/types.h>
42 #include <sys/un.h>
43 #include <time.h>
44 #include <unistd.h>
45 
46 #include <array>
47 #include <mutex>
48 #include <optional>
49 
50 #include "os/log.h"
51 #include "osi/include/osi.h"  // OSI_NO_INTR
52 
53 #define asrt(s)                                         \
54   do {                                                  \
55     if (!(s)) log::error("## assert {} failed ##", #s); \
56   } while (0)
57 
58 #define MAX_THREAD 8
59 #define MAX_POLL 64
60 #define POLL_EXCEPTION_EVENTS (POLLHUP | POLLRDHUP | POLLERR | POLLNVAL)
61 #define IS_EXCEPTION(e) ((e)&POLL_EXCEPTION_EVENTS)
62 #define IS_READ(e) ((e)&POLLIN)
63 #define IS_WRITE(e) ((e)&POLLOUT)
64 /*cmd executes in socket poll thread */
65 #define CMD_WAKEUP 1
66 #define CMD_EXIT 2
67 #define CMD_ADD_FD 3
68 #define CMD_REMOVE_FD 4
69 #define CMD_USER_PRIVATE 5
70 
71 using namespace bluetooth;
72 
73 struct poll_slot_t {
74   struct pollfd pfd;
75   uint32_t user_id;
76   int type;
77   int flags;
78 };
79 struct thread_slot_t {
80   int cmd_fdr, cmd_fdw;
81   int poll_count;
82   poll_slot_t ps[MAX_POLL];
83   int psi[MAX_POLL];  // index of poll slot
84   std::optional<pthread_t> thread_id;
85   btsock_signaled_cb callback;
86   btsock_cmd_cb cmd_callback;
87   int used;
88 };
89 static thread_slot_t ts[MAX_THREAD];
90 
91 static void* sock_poll_thread(void* arg);
92 static inline void close_cmd_fd(int h);
93 
94 static inline void add_poll(int h, int fd, int type, int flags,
95                             uint32_t user_id);
96 
97 static std::recursive_mutex thread_slot_lock;
98 
create_thread(void * (* start_routine)(void *),void * arg,pthread_t * thread_id)99 static inline int create_thread(void* (*start_routine)(void*), void* arg,
100                                 pthread_t* thread_id) {
101   pthread_attr_t thread_attr;
102   pthread_attr_init(&thread_attr);
103   pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_JOINABLE);
104   int policy;
105   int min_pri = 0;
106   int ret = -1;
107   struct sched_param param;
108 
109   ret = pthread_create(thread_id, &thread_attr, start_routine, arg);
110   if (ret != 0) {
111     log::error("pthread_create : {}", strerror(errno));
112     return ret;
113   }
114   /* We need to lower the priority of this thread to ensure the stack gets
115    * priority over transfer to a socket */
116   pthread_getschedparam(*thread_id, &policy, &param);
117   min_pri = sched_get_priority_min(policy);
118   if (param.sched_priority > min_pri) {
119     param.sched_priority -= 1;
120   }
121   pthread_setschedparam(*thread_id, policy, &param);
122   return ret;
123 }
124 static void init_poll(int cmd_fd);
alloc_thread_slot()125 static int alloc_thread_slot() {
126   std::unique_lock<std::recursive_mutex> lock(thread_slot_lock);
127   int i;
128   // reversed order to save guard uninitialized access to 0 index
129   for (i = MAX_THREAD - 1; i >= 0; i--) {
130     if (!ts[i].used) {
131       ts[i].used = 1;
132       return i;
133     }
134   }
135   log::error("execeeded max thread count");
136   return -1;
137 }
free_thread_slot(int h)138 static void free_thread_slot(int h) {
139   if (0 <= h && h < MAX_THREAD) {
140     close_cmd_fd(h);
141     ts[h].used = 0;
142   } else
143     log::error("invalid thread handle:{}", h);
144 }
btsock_thread_init()145 void btsock_thread_init() {
146   static int initialized;
147   if (!initialized) {
148     initialized = 1;
149     int h;
150     for (h = 0; h < MAX_THREAD; h++) {
151       ts[h].cmd_fdr = ts[h].cmd_fdw = -1;
152       ts[h].used = 0;
153       ts[h].thread_id = std::nullopt;
154       ts[h].poll_count = 0;
155       ts[h].callback = NULL;
156       ts[h].cmd_callback = NULL;
157     }
158   }
159 }
btsock_thread_create(btsock_signaled_cb callback,btsock_cmd_cb cmd_callback)160 int btsock_thread_create(btsock_signaled_cb callback,
161                          btsock_cmd_cb cmd_callback) {
162   asrt(callback || cmd_callback);
163   int h = alloc_thread_slot();
164   if (h >= 0) {
165     init_poll(h);
166     pthread_t thread;
167     int status = create_thread(sock_poll_thread, (void*)(uintptr_t)h, &thread);
168     if (status) {
169       log::error("create_thread failed: {}", strerror(status));
170       free_thread_slot(h);
171       return -1;
172     }
173 
174     ts[h].thread_id = thread;
175     ts[h].callback = callback;
176     ts[h].cmd_callback = cmd_callback;
177   }
178   return h;
179 }
180 
181 /* create dummy socket pair used to wake up select loop */
init_cmd_fd(int h)182 static inline void init_cmd_fd(int h) {
183   asrt(ts[h].cmd_fdr == -1 && ts[h].cmd_fdw == -1);
184   if (socketpair(AF_UNIX, SOCK_STREAM, 0, &ts[h].cmd_fdr) < 0) {
185     log::error("socketpair failed: {}", strerror(errno));
186     return;
187   }
188   // add the cmd fd for read & write
189   add_poll(h, ts[h].cmd_fdr, 0, SOCK_THREAD_FD_RD, 0);
190 }
close_cmd_fd(int h)191 static inline void close_cmd_fd(int h) {
192   if (ts[h].cmd_fdr != -1) {
193     close(ts[h].cmd_fdr);
194     ts[h].cmd_fdr = -1;
195   }
196   if (ts[h].cmd_fdw != -1) {
197     close(ts[h].cmd_fdw);
198     ts[h].cmd_fdw = -1;
199   }
200 }
201 typedef struct {
202   int id;
203   int fd;
204   int type;
205   int flags;
206   uint32_t user_id;
207 } sock_cmd_t;
btsock_thread_add_fd(int h,int fd,int type,int flags,uint32_t user_id)208 int btsock_thread_add_fd(int h, int fd, int type, int flags, uint32_t user_id) {
209   if (h < 0 || h >= MAX_THREAD) {
210     log::error("invalid bt thread handle:{}", h);
211     return false;
212   }
213   if (ts[h].cmd_fdw == -1) {
214     log::error("cmd socket is not created. socket thread may not initialized");
215     return false;
216   }
217   if (flags & SOCK_THREAD_ADD_FD_SYNC) {
218     // must executed in socket poll thread
219     if (ts[h].thread_id.value() == pthread_self()) {
220       // cleanup one-time flags
221       flags &= ~SOCK_THREAD_ADD_FD_SYNC;
222       add_poll(h, fd, type, flags, user_id);
223       return true;
224     }
225     log::warn(
226         "THREAD_ADD_FD_SYNC is not called in poll thread, fallback to async");
227   }
228   sock_cmd_t cmd = {CMD_ADD_FD, fd, type, flags, user_id};
229 
230   ssize_t ret;
231   OSI_NO_INTR(ret = send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0));
232 
233   return ret == sizeof(cmd);
234 }
btsock_thread_wakeup(int h)235 int btsock_thread_wakeup(int h) {
236   if (h < 0 || h >= MAX_THREAD) {
237     log::error("invalid bt thread handle:{}", h);
238     return false;
239   }
240   if (ts[h].cmd_fdw == -1) {
241     log::error("thread handle:{}, cmd socket is not created", h);
242     return false;
243   }
244   sock_cmd_t cmd = {CMD_WAKEUP, 0, 0, 0, 0};
245 
246   ssize_t ret;
247   OSI_NO_INTR(ret = send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0));
248 
249   return ret == sizeof(cmd);
250 }
btsock_thread_exit(int h)251 int btsock_thread_exit(int h) {
252   if (h < 0 || h >= MAX_THREAD) {
253     log::error("invalid bt thread slot:{}", h);
254     return false;
255   }
256   if (ts[h].cmd_fdw == -1) {
257     log::error("cmd socket is not created");
258     return false;
259   }
260   sock_cmd_t cmd = {CMD_EXIT, 0, 0, 0, 0};
261 
262   ssize_t ret;
263   OSI_NO_INTR(ret = send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0));
264 
265   if (ret == sizeof(cmd)) {
266     if (ts[h].thread_id != std::nullopt) {
267       pthread_join(ts[h].thread_id.value(), 0);
268       ts[h].thread_id = std::nullopt;
269     }
270     free_thread_slot(h);
271     return true;
272   }
273   return false;
274 }
init_poll(int h)275 static void init_poll(int h) {
276   int i;
277   ts[h].poll_count = 0;
278   ts[h].thread_id = std::nullopt;
279   ts[h].callback = NULL;
280   ts[h].cmd_callback = NULL;
281   for (i = 0; i < MAX_POLL; i++) {
282     ts[h].ps[i].pfd.fd = -1;
283     ts[h].psi[i] = -1;
284   }
285   init_cmd_fd(h);
286 }
flags2pevents(int flags)287 static inline unsigned int flags2pevents(int flags) {
288   unsigned int pevents = 0;
289   if (flags & SOCK_THREAD_FD_WR) pevents |= POLLOUT;
290   if (flags & SOCK_THREAD_FD_RD) pevents |= POLLIN;
291   pevents |= POLL_EXCEPTION_EVENTS;
292   return pevents;
293 }
294 
set_poll(poll_slot_t * ps,int fd,int type,int flags,uint32_t user_id)295 static inline void set_poll(poll_slot_t* ps, int fd, int type, int flags,
296                             uint32_t user_id) {
297   ps->pfd.fd = fd;
298   ps->user_id = user_id;
299   if (ps->type != 0 && ps->type != type)
300     log::error("poll socket type should not changed! type was:{}, type now:{}",
301                ps->type, type);
302   ps->type = type;
303   ps->flags = flags;
304   ps->pfd.events = flags2pevents(flags);
305   ps->pfd.revents = 0;
306 }
add_poll(int h,int fd,int type,int flags,uint32_t user_id)307 static inline void add_poll(int h, int fd, int type, int flags,
308                             uint32_t user_id) {
309   asrt(fd != -1);
310   int i;
311   int empty = -1;
312   poll_slot_t* ps = ts[h].ps;
313 
314   for (i = 0; i < MAX_POLL; i++) {
315     if (ps[i].pfd.fd == fd) {
316       asrt(ts[h].poll_count < MAX_POLL);
317 
318       set_poll(&ps[i], fd, type, flags | ps[i].flags, user_id);
319       return;
320     } else if (empty < 0 && ps[i].pfd.fd == -1)
321       empty = i;
322   }
323   if (empty >= 0) {
324     asrt(ts[h].poll_count < MAX_POLL);
325     set_poll(&ps[empty], fd, type, flags, user_id);
326     ++ts[h].poll_count;
327     return;
328   }
329   log::error("exceeded max poll slot:{}!", MAX_POLL);
330 }
remove_poll(int h,poll_slot_t * ps,int flags)331 static inline void remove_poll(int h, poll_slot_t* ps, int flags) {
332   if (flags == ps->flags) {
333     // all monitored events signaled. To remove it, just clear the slot
334     --ts[h].poll_count;
335     memset(ps, 0, sizeof(*ps));
336     ps->pfd.fd = -1;
337   } else {
338     // one read or one write monitor event signaled, removed the accordding bit
339     ps->flags &= ~flags;
340     // update the poll events mask
341     ps->pfd.events = flags2pevents(ps->flags);
342   }
343 }
process_cmd_sock(int h)344 static int process_cmd_sock(int h) {
345   sock_cmd_t cmd = {-1, 0, 0, 0, 0};
346   int fd = ts[h].cmd_fdr;
347 
348   ssize_t ret;
349   OSI_NO_INTR(ret = recv(fd, &cmd, sizeof(cmd), MSG_WAITALL));
350 
351   if (ret != sizeof(cmd)) {
352     log::error("recv cmd errno:{}", errno);
353     return false;
354   }
355   switch (cmd.id) {
356     case CMD_ADD_FD:
357       add_poll(h, cmd.fd, cmd.type, cmd.flags, cmd.user_id);
358       break;
359     case CMD_REMOVE_FD:
360       for (int i = 1; i < MAX_POLL; ++i) {
361         poll_slot_t* poll_slot = &ts[h].ps[i];
362         if (poll_slot->pfd.fd == cmd.fd) {
363           remove_poll(h, poll_slot, poll_slot->flags);
364           break;
365         }
366       }
367       close(cmd.fd);
368       break;
369     case CMD_WAKEUP:
370       break;
371     case CMD_USER_PRIVATE:
372       asrt(ts[h].cmd_callback);
373       if (ts[h].cmd_callback)
374         ts[h].cmd_callback(fd, cmd.type, cmd.flags, cmd.user_id);
375       break;
376     case CMD_EXIT:
377       return false;
378     default:
379       log::warn("unknown cmd: {}", cmd.id);
380       break;
381   }
382   return true;
383 }
384 
process_data_sock(int h,struct pollfd * pfds,int pfds_count,int event_count)385 static void process_data_sock(int h, struct pollfd* pfds, int pfds_count,
386                               int event_count) {
387   asrt(event_count <= pfds_count);
388   int i;
389   for (i = 1; i < pfds_count; i++) {
390     if (pfds[i].revents) {
391       int ps_i = ts[h].psi[i];
392       if (ts[h].ps[ps_i].pfd.fd == -1) {
393         log::info("Socket has been removed from poll set");
394         continue;
395       }
396       asrt(pfds[i].fd == ts[h].ps[ps_i].pfd.fd);
397       uint32_t user_id = ts[h].ps[ps_i].user_id;
398       int type = ts[h].ps[ps_i].type;
399       int flags = 0;
400       if (IS_READ(pfds[i].revents)) {
401         flags |= SOCK_THREAD_FD_RD;
402       }
403       if (IS_WRITE(pfds[i].revents)) {
404         flags |= SOCK_THREAD_FD_WR;
405       }
406       if (IS_EXCEPTION(pfds[i].revents)) {
407         flags |= SOCK_THREAD_FD_EXCEPTION;
408         // remove the whole slot not flags
409         remove_poll(h, &ts[h].ps[ps_i], ts[h].ps[ps_i].flags);
410       } else if (flags)
411         remove_poll(h, &ts[h].ps[ps_i],
412                     flags);  // remove the monitor flags that already processed
413       if (flags) ts[h].callback(pfds[i].fd, type, flags, user_id);
414     }
415   }
416 }
417 
prepare_poll_fds(int h,struct pollfd * pfds)418 static void prepare_poll_fds(int h, struct pollfd* pfds) {
419   int count = 0;
420   int ps_i = 0;
421   int pfd_i = 0;
422   asrt(ts[h].poll_count <= MAX_POLL);
423   while (count < ts[h].poll_count) {
424     if (ps_i >= MAX_POLL) {
425       log::error(
426           "exceed max poll range, ps_i:{}, MAX_POLL:{}, count:{}, "
427           "ts[h].poll_count:{}",
428           ps_i, MAX_POLL, count, ts[h].poll_count);
429       return;
430     }
431     if (ts[h].ps[ps_i].pfd.fd >= 0) {
432       pfds[pfd_i] = ts[h].ps[ps_i].pfd;
433       ts[h].psi[pfd_i] = ps_i;
434       count++;
435       pfd_i++;
436     }
437     ps_i++;
438   }
439 }
sock_poll_thread(void * arg)440 static void* sock_poll_thread(void* arg) {
441   std::array<struct pollfd, MAX_POLL> pfds;
442 
443   int h = (intptr_t)arg;
444   for (;;) {
445     pfds = {};
446     prepare_poll_fds(h, pfds.data());
447     int ret;
448     OSI_NO_INTR(ret = poll(pfds.data(), ts[h].poll_count, -1));
449     if (ret == -1) {
450       log::error("poll ret -1, exit the thread, errno:{}, err:{}", errno,
451                  strerror(errno));
452       break;
453     }
454     if (ret != 0) {
455       int need_process_data_fd = true;
456       int pfds_count = ts[h].poll_count;
457       if (pfds[0].revents)  // cmd fd always is the first one
458       {
459         asrt(pfds[0].fd == ts[h].cmd_fdr);
460         if (!process_cmd_sock(h)) {
461           log::info("h:{}, process_cmd_sock return false, exit...", h);
462           break;
463         }
464         if (ret == 1)
465           need_process_data_fd = false;
466         else
467           ret--;  // exclude the cmd fd
468       }
469       if (need_process_data_fd)
470         process_data_sock(h, pfds.data(), pfds_count, ret);
471     } else {
472       log::info("no data, select ret: {}", ret);
473     };
474   }
475   log::info("socket poll thread exiting, h:{}", h);
476   return 0;
477 }
478