1 //
2 // Copyright 2016 The Android Open Source Project
3 //
4 // Licensed under the Apache License, Version 2.0 (the "License");
5 // you may not use this file except in compliance with the License.
6 // You may obtain a copy of the License at
7 //
8 // http://www.apache.org/licenses/LICENSE-2.0
9 //
10 // Unless required by applicable law or agreed to in writing, software
11 // distributed under the License is distributed on an "AS IS" BASIS,
12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 // See the License for the specific language governing permissions and
14 // limitations under the License.
15 //
16 
17 #include "async_fd_watcher.h"
18 
19 #include <algorithm>
20 #include <atomic>
21 #include <condition_variable>
22 #include <log/log.h>
23 #include <map>
24 #include <mutex>
25 #include <thread>
26 #include <vector>
27 #include "fcntl.h"
28 #include "sys/select.h"
29 #include "unistd.h"
30 
31 static const int INVALID_FD = -1;
32 static const int BT_RT_PRIORITY = 1;
33 
34 namespace android {
35 namespace hardware {
36 namespace bluetooth {
37 namespace async {
38 
WatchFdForNonBlockingReads(int file_descriptor,const ReadCallback & on_read_fd_ready_callback)39 int AsyncFdWatcher::WatchFdForNonBlockingReads(
40     int file_descriptor, const ReadCallback& on_read_fd_ready_callback) {
41   // Add file descriptor and callback
42   {
43     std::unique_lock<std::mutex> guard(internal_mutex_);
44     watched_fds_[file_descriptor] = on_read_fd_ready_callback;
45   }
46 
47   // Start the thread if not started yet
48   return tryStartThread();
49 }
50 
ConfigureTimeout(const std::chrono::milliseconds timeout,const TimeoutCallback & on_timeout_callback)51 int AsyncFdWatcher::ConfigureTimeout(
52     const std::chrono::milliseconds timeout,
53     const TimeoutCallback& on_timeout_callback) {
54   // Add timeout and callback
55   {
56     std::unique_lock<std::mutex> guard(timeout_mutex_);
57     timeout_cb_ = on_timeout_callback;
58     timeout_ms_ = timeout;
59   }
60 
61   notifyThread();
62   return 0;
63 }
64 
StopWatchingFileDescriptors()65 void AsyncFdWatcher::StopWatchingFileDescriptors() { stopThread(); }
66 
~AsyncFdWatcher()67 AsyncFdWatcher::~AsyncFdWatcher() {}
68 
69 // Make sure to call this with at least one file descriptor ready to be
70 // watched upon or the thread routine will return immediately
tryStartThread()71 int AsyncFdWatcher::tryStartThread() {
72   if (std::atomic_exchange(&running_, true)) return 0;
73 
74   // Set up the communication channel
75   int pipe_fds[2];
76   if (pipe2(pipe_fds, O_NONBLOCK)) return -1;
77 
78   notification_listen_fd_ = pipe_fds[0];
79   notification_write_fd_ = pipe_fds[1];
80 
81   thread_ = std::thread([this]() { ThreadRoutine(); });
82   if (!thread_.joinable()) return -1;
83 
84   return 0;
85 }
86 
stopThread()87 int AsyncFdWatcher::stopThread() {
88   if (!std::atomic_exchange(&running_, false)) return 0;
89 
90   notifyThread();
91   if (std::this_thread::get_id() != thread_.get_id()) {
92     thread_.join();
93   }
94 
95   {
96     std::unique_lock<std::mutex> guard(internal_mutex_);
97     watched_fds_.clear();
98   }
99 
100   {
101     std::unique_lock<std::mutex> guard(timeout_mutex_);
102     timeout_cb_ = nullptr;
103   }
104 
105   return 0;
106 }
107 
notifyThread()108 int AsyncFdWatcher::notifyThread() {
109   uint8_t buffer[] = {0};
110   if (TEMP_FAILURE_RETRY(write(notification_write_fd_, &buffer, 1)) < 0) {
111     return -1;
112   }
113   return 0;
114 }
115 
ThreadRoutine()116 void AsyncFdWatcher::ThreadRoutine() {
117 
118   // Make watching thread RT.
119   struct sched_param rt_params;
120   rt_params.sched_priority = BT_RT_PRIORITY;
121   if (sched_setscheduler(gettid(), SCHED_FIFO, &rt_params)) {
122     ALOGE("%s unable to set SCHED_FIFO for pid %d, tid %d, error %s", __func__,
123           getpid(), gettid(), strerror(errno));
124   }
125 
126   while (running_) {
127     fd_set read_fds;
128     FD_ZERO(&read_fds);
129     FD_SET(notification_listen_fd_, &read_fds);
130     int max_read_fd = INVALID_FD;
131     for (auto& it : watched_fds_) {
132       FD_SET(it.first, &read_fds);
133       max_read_fd = std::max(max_read_fd, it.first);
134     }
135 
136     struct timeval timeout;
137     struct timeval* timeout_ptr = NULL;
138     if (timeout_ms_ > std::chrono::milliseconds(0)) {
139       timeout.tv_sec = timeout_ms_.count() / 1000;
140       timeout.tv_usec = (timeout_ms_.count() % 1000) * 1000;
141       timeout_ptr = &timeout;
142     }
143 
144     // Wait until there is data available to read on some FD.
145     int nfds = std::max(notification_listen_fd_, max_read_fd);
146     int retval = select(nfds + 1, &read_fds, NULL, NULL, timeout_ptr);
147 
148     // There was some error.
149     if (retval < 0) continue;
150 
151     // Timeout.
152     if (retval == 0) {
153       // Allow the timeout callback to modify the timeout.
154       TimeoutCallback saved_cb;
155       {
156         std::unique_lock<std::mutex> guard(timeout_mutex_);
157         if (timeout_ms_ > std::chrono::milliseconds(0))
158           saved_cb = timeout_cb_;
159       }
160       if (saved_cb != nullptr)
161         saved_cb();
162       continue;
163     }
164 
165     // Read data from the notification FD.
166     if (FD_ISSET(notification_listen_fd_, &read_fds)) {
167       char buffer[] = {0};
168       TEMP_FAILURE_RETRY(read(notification_listen_fd_, buffer, 1));
169       continue;
170     }
171 
172     // Invoke the data ready callbacks if appropriate.
173     {
174       // Hold the mutex to make sure that the callbacks are still valid.
175       std::unique_lock<std::mutex> guard(internal_mutex_);
176       for (auto& it : watched_fds_) {
177         if (FD_ISSET(it.first, &read_fds)) {
178         it.second(it.first);
179         }
180       }
181     }
182   }
183 }
184 
185 } // namespace async
186 } // namespace bluetooth
187 } // namespace hardware
188 } // namespace android
189