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 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 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 63 void AsyncFdWatcher::StopWatchingFileDescriptors() { stopThread(); } 64 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 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 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 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 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
