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 #ifndef NO_THREAD_PRIORITY
33 static const int BT_RT_PRIORITY = 1;
34 #endif
35
36 namespace android {
37 namespace hardware {
38 namespace bluetooth {
39 namespace async {
40
WatchFdForNonBlockingReads(int file_descriptor,const ReadCallback & on_read_fd_ready_callback)41 int AsyncFdWatcher::WatchFdForNonBlockingReads(
42 int file_descriptor, const ReadCallback& on_read_fd_ready_callback) {
43 // Add file descriptor and callback
44 {
45 std::unique_lock<std::mutex> guard(internal_mutex_);
46 watched_fds_[file_descriptor] = on_read_fd_ready_callback;
47 }
48
49 // Start the thread if not started yet
50 return tryStartThread();
51 }
52
ConfigureTimeout(const std::chrono::milliseconds timeout,const TimeoutCallback & on_timeout_callback)53 int AsyncFdWatcher::ConfigureTimeout(
54 const std::chrono::milliseconds timeout,
55 const TimeoutCallback& on_timeout_callback) {
56 // Add timeout and callback
57 {
58 std::unique_lock<std::mutex> guard(timeout_mutex_);
59 timeout_cb_ = on_timeout_callback;
60 timeout_ms_ = timeout;
61 }
62
63 notifyThread();
64 return 0;
65 }
66
StopWatchingFileDescriptors()67 void AsyncFdWatcher::StopWatchingFileDescriptors() { stopThread(); }
68
~AsyncFdWatcher()69 AsyncFdWatcher::~AsyncFdWatcher() {}
70
71 // Make sure to call this with at least one file descriptor ready to be
72 // watched upon or the thread routine will return immediately
tryStartThread()73 int AsyncFdWatcher::tryStartThread() {
74 if (std::atomic_exchange(&running_, true)) return 0;
75
76 // Set up the communication channel
77 int pipe_fds[2];
78 if (pipe2(pipe_fds, O_NONBLOCK)) return -1;
79
80 notification_listen_fd_ = pipe_fds[0];
81 notification_write_fd_ = pipe_fds[1];
82
83 thread_ = std::thread([this]() { ThreadRoutine(); });
84 if (!thread_.joinable()) return -1;
85
86 return 0;
87 }
88
stopThread()89 int AsyncFdWatcher::stopThread() {
90 if (!std::atomic_exchange(&running_, false)) return 0;
91
92 notifyThread();
93 if (std::this_thread::get_id() != thread_.get_id()) {
94 thread_.join();
95 }
96
97 {
98 std::unique_lock<std::mutex> guard(internal_mutex_);
99 watched_fds_.clear();
100 }
101
102 {
103 std::unique_lock<std::mutex> guard(timeout_mutex_);
104 timeout_cb_ = nullptr;
105 }
106
107 close(notification_listen_fd_);
108 close(notification_write_fd_);
109
110 return 0;
111 }
112
notifyThread()113 int AsyncFdWatcher::notifyThread() {
114 uint8_t buffer[] = {0};
115 if (TEMP_FAILURE_RETRY(write(notification_write_fd_, &buffer, 1)) < 0) {
116 return -1;
117 }
118 return 0;
119 }
120
ThreadRoutine()121 void AsyncFdWatcher::ThreadRoutine() {
122 #ifndef NO_THREAD_PRIORITY
123 // Make watching thread RT.
124 struct sched_param rt_params;
125 rt_params.sched_priority = BT_RT_PRIORITY;
126 if (sched_setscheduler(gettid(), SCHED_FIFO, &rt_params)) {
127 ALOGE("%s unable to set SCHED_FIFO for pid %d, tid %d, error %s", __func__,
128 getpid(), gettid(), strerror(errno));
129 }
130 #endif
131
132 while (running_) {
133 fd_set read_fds;
134 FD_ZERO(&read_fds);
135 FD_SET(notification_listen_fd_, &read_fds);
136 int max_read_fd = INVALID_FD;
137 for (auto& it : watched_fds_) {
138 FD_SET(it.first, &read_fds);
139 max_read_fd = std::max(max_read_fd, it.first);
140 }
141
142 struct timeval timeout;
143 struct timeval* timeout_ptr = NULL;
144 if (timeout_ms_ > std::chrono::milliseconds(0)) {
145 timeout.tv_sec = timeout_ms_.count() / 1000;
146 timeout.tv_usec = (timeout_ms_.count() % 1000) * 1000;
147 timeout_ptr = &timeout;
148 }
149
150 // Wait until there is data available to read on some FD.
151 int nfds = std::max(notification_listen_fd_, max_read_fd);
152 int retval = select(nfds + 1, &read_fds, NULL, NULL, timeout_ptr);
153
154 // There was some error.
155 if (retval < 0) continue;
156
157 // Timeout.
158 if (retval == 0) {
159 // Allow the timeout callback to modify the timeout.
160 TimeoutCallback saved_cb;
161 {
162 std::unique_lock<std::mutex> guard(timeout_mutex_);
163 if (timeout_ms_ > std::chrono::milliseconds(0))
164 saved_cb = timeout_cb_;
165 }
166 if (saved_cb != nullptr)
167 saved_cb();
168 continue;
169 }
170
171 // Read data from the notification FD.
172 if (FD_ISSET(notification_listen_fd_, &read_fds)) {
173 char buffer[] = {0};
174 TEMP_FAILURE_RETRY(read(notification_listen_fd_, buffer, 1));
175 continue;
176 }
177
178 // Invoke the data ready callbacks if appropriate.
179 {
180 // Hold the mutex to make sure that the callbacks are still valid.
181 std::unique_lock<std::mutex> guard(internal_mutex_);
182 for (auto& it : watched_fds_) {
183 if (FD_ISSET(it.first, &read_fds)) {
184 it.second(it.first);
185 }
186 }
187 }
188 }
189 }
190
191 } // namespace async
192 } // namespace bluetooth
193 } // namespace hardware
194 } // namespace android
195