async_fd_watcher.cc revision 761dc49c591d9fe6cc1a3a2e90026678af3ad41a 
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
30static const int INVALID_FD = -1;
31
32namespace android {
33namespace hardware {
34namespace bluetooth {
35namespace async {
36
37int 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
49int 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
63void AsyncFdWatcher::StopWatchingFileDescriptors() { stopThread(); }
64
65AsyncFdWatcher::~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
69int 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
85int 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
106int 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
114void 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