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