1 /* 2 * Copyright (C) 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 specic language governing permissions and 14 * limitations under the License. 15 */ 16 17 #include "libappfuse/FuseBridgeLoop.h" 18 19 #include <sys/epoll.h> 20 #include <sys/socket.h> 21 22 #include <unordered_map> 23 24 #include <android-base/logging.h> 25 #include <android-base/unique_fd.h> 26 27 #include "libappfuse/EpollController.h" 28 29 namespace android { 30 namespace fuse { 31 namespace { 32 33 enum class FuseBridgeState { kWaitToReadEither, kWaitToReadProxy, kWaitToWriteProxy, kClosing }; 34 35 struct FuseBridgeEntryEvent { 36 FuseBridgeEntry* entry; 37 int events; 38 }; 39 40 void GetObservedEvents(FuseBridgeState state, int* device_events, int* proxy_events) { 41 switch (state) { 42 case FuseBridgeState::kWaitToReadEither: 43 *device_events = EPOLLIN; 44 *proxy_events = EPOLLIN; 45 return; 46 case FuseBridgeState::kWaitToReadProxy: 47 *device_events = 0; 48 *proxy_events = EPOLLIN; 49 return; 50 case FuseBridgeState::kWaitToWriteProxy: 51 *device_events = 0; 52 *proxy_events = EPOLLOUT; 53 return; 54 case FuseBridgeState::kClosing: 55 *device_events = 0; 56 *proxy_events = 0; 57 return; 58 } 59 } 60 61 void LogResponseError(const std::string& message, const FuseResponse& response) { 62 LOG(ERROR) << message << ": header.len=" << response.header.len 63 << " header.error=" << response.header.error 64 << " header.unique=" << response.header.unique; 65 } 66 } 67 68 class FuseBridgeEntry { 69 public: 70 FuseBridgeEntry(int mount_id, base::unique_fd&& dev_fd, base::unique_fd&& proxy_fd) 71 : mount_id_(mount_id), 72 device_fd_(std::move(dev_fd)), 73 proxy_fd_(std::move(proxy_fd)), 74 state_(FuseBridgeState::kWaitToReadEither), 75 last_state_(FuseBridgeState::kWaitToReadEither), 76 last_device_events_({this, 0}), 77 last_proxy_events_({this, 0}), 78 open_count_(0) {} 79 80 // Transfer bytes depends on availability of FDs and the internal |state_|. 81 void Transfer(FuseBridgeLoopCallback* callback) { 82 constexpr int kUnexpectedEventMask = ~(EPOLLIN | EPOLLOUT); 83 const bool unexpected_event = (last_device_events_.events & kUnexpectedEventMask) || 84 (last_proxy_events_.events & kUnexpectedEventMask); 85 const bool device_read_ready = last_device_events_.events & EPOLLIN; 86 const bool proxy_read_ready = last_proxy_events_.events & EPOLLIN; 87 const bool proxy_write_ready = last_proxy_events_.events & EPOLLOUT; 88 89 last_device_events_.events = 0; 90 last_proxy_events_.events = 0; 91 92 LOG(VERBOSE) << "Transfer device_read_ready=" << device_read_ready 93 << " proxy_read_ready=" << proxy_read_ready 94 << " proxy_write_ready=" << proxy_write_ready; 95 96 if (unexpected_event) { 97 LOG(ERROR) << "Invalid epoll event is observed"; 98 state_ = FuseBridgeState::kClosing; 99 return; 100 } 101 102 switch (state_) { 103 case FuseBridgeState::kWaitToReadEither: 104 if (proxy_read_ready) { 105 state_ = ReadFromProxy(); 106 } else if (device_read_ready) { 107 state_ = ReadFromDevice(callback); 108 } 109 return; 110 111 case FuseBridgeState::kWaitToReadProxy: 112 CHECK(proxy_read_ready); 113 state_ = ReadFromProxy(); 114 return; 115 116 case FuseBridgeState::kWaitToWriteProxy: 117 CHECK(proxy_write_ready); 118 state_ = WriteToProxy(); 119 return; 120 121 case FuseBridgeState::kClosing: 122 return; 123 } 124 } 125 126 bool IsClosing() const { return state_ == FuseBridgeState::kClosing; } 127 128 int mount_id() const { return mount_id_; } 129 130 private: 131 friend class BridgeEpollController; 132 133 FuseBridgeState ReadFromProxy() { 134 switch (buffer_.response.ReadOrAgain(proxy_fd_)) { 135 case ResultOrAgain::kSuccess: 136 break; 137 case ResultOrAgain::kFailure: 138 return FuseBridgeState::kClosing; 139 case ResultOrAgain::kAgain: 140 return FuseBridgeState::kWaitToReadProxy; 141 } 142 143 if (!buffer_.response.Write(device_fd_)) { 144 LogResponseError("Failed to write a reply from proxy to device", buffer_.response); 145 return FuseBridgeState::kClosing; 146 } 147 148 auto it = opcode_map_.find(buffer_.response.header.unique); 149 if (it != opcode_map_.end()) { 150 switch (it->second) { 151 case FUSE_OPEN: 152 if (buffer_.response.header.error == fuse::kFuseSuccess) { 153 open_count_++; 154 } 155 break; 156 157 case FUSE_RELEASE: 158 if (open_count_ > 0) { 159 open_count_--; 160 } else { 161 LOG(WARNING) << "Unexpected FUSE_RELEASE before opening a file."; 162 break; 163 } 164 if (open_count_ == 0) { 165 return FuseBridgeState::kClosing; 166 } 167 break; 168 } 169 opcode_map_.erase(it); 170 } 171 172 return FuseBridgeState::kWaitToReadEither; 173 } 174 175 FuseBridgeState ReadFromDevice(FuseBridgeLoopCallback* callback) { 176 LOG(VERBOSE) << "ReadFromDevice"; 177 if (!buffer_.request.Read(device_fd_)) { 178 return FuseBridgeState::kClosing; 179 } 180 181 const uint32_t opcode = buffer_.request.header.opcode; 182 const uint64_t unique = buffer_.request.header.unique; 183 LOG(VERBOSE) << "Read a fuse packet, opcode=" << opcode << " unique=" << unique; 184 if (unique == 0) { 185 return FuseBridgeState::kWaitToReadEither; 186 } 187 switch (opcode) { 188 case FUSE_FORGET: 189 // Do not reply to FUSE_FORGET. 190 return FuseBridgeState::kWaitToReadEither; 191 192 case FUSE_LOOKUP: 193 case FUSE_GETATTR: 194 case FUSE_OPEN: 195 case FUSE_READ: 196 case FUSE_WRITE: 197 case FUSE_RELEASE: 198 case FUSE_FSYNC: 199 if (opcode == FUSE_OPEN || opcode == FUSE_RELEASE) { 200 opcode_map_.emplace(buffer_.request.header.unique, opcode); 201 } 202 return WriteToProxy(); 203 204 case FUSE_INIT: 205 buffer_.HandleInit(); 206 break; 207 208 default: 209 buffer_.HandleNotImpl(); 210 break; 211 } 212 213 if (!buffer_.response.Write(device_fd_)) { 214 LogResponseError("Failed to write a response to device", buffer_.response); 215 return FuseBridgeState::kClosing; 216 } 217 218 if (opcode == FUSE_INIT) { 219 callback->OnMount(mount_id_); 220 } 221 222 return FuseBridgeState::kWaitToReadEither; 223 } 224 225 FuseBridgeState WriteToProxy() { 226 switch (buffer_.request.WriteOrAgain(proxy_fd_)) { 227 case ResultOrAgain::kSuccess: 228 return FuseBridgeState::kWaitToReadEither; 229 case ResultOrAgain::kFailure: 230 LOG(ERROR) << "Failed to write a request to proxy:" 231 << " header.len=" << buffer_.request.header.len 232 << " header.opcode=" << buffer_.request.header.opcode 233 << " header.unique=" << buffer_.request.header.unique 234 << " header.nodeid=" << buffer_.request.header.nodeid; 235 return FuseBridgeState::kClosing; 236 case ResultOrAgain::kAgain: 237 return FuseBridgeState::kWaitToWriteProxy; 238 } 239 } 240 241 const int mount_id_; 242 base::unique_fd device_fd_; 243 base::unique_fd proxy_fd_; 244 FuseBuffer buffer_; 245 FuseBridgeState state_; 246 FuseBridgeState last_state_; 247 FuseBridgeEntryEvent last_device_events_; 248 FuseBridgeEntryEvent last_proxy_events_; 249 250 // Remember map between unique and opcode in fuse_in_header so that we can 251 // refer the opcode later. 252 std::unordered_map<uint64_t, uint32_t> opcode_map_; 253 254 int open_count_; 255 256 DISALLOW_COPY_AND_ASSIGN(FuseBridgeEntry); 257 }; 258 259 class BridgeEpollController : private EpollController { 260 public: 261 BridgeEpollController(base::unique_fd&& poll_fd) : EpollController(std::move(poll_fd)) {} 262 263 bool AddBridgePoll(FuseBridgeEntry* bridge) const { 264 return InvokeControl(EPOLL_CTL_ADD, bridge); 265 } 266 267 bool UpdateOrDeleteBridgePoll(FuseBridgeEntry* bridge) const { 268 return InvokeControl( 269 bridge->state_ != FuseBridgeState::kClosing ? EPOLL_CTL_MOD : EPOLL_CTL_DEL, bridge); 270 } 271 272 bool Wait(size_t bridge_count, std::unordered_set<FuseBridgeEntry*>* entries_out) { 273 CHECK(entries_out); 274 const size_t event_count = std::max<size_t>(bridge_count * 2, 1); 275 if (!EpollController::Wait(event_count)) { 276 return false; 277 } 278 entries_out->clear(); 279 for (const auto& event : events()) { 280 FuseBridgeEntryEvent* const entry_event = 281 reinterpret_cast<FuseBridgeEntryEvent*>(event.data.ptr); 282 entry_event->events = event.events; 283 entries_out->insert(entry_event->entry); 284 } 285 return true; 286 } 287 288 private: 289 bool InvokeControl(int op, FuseBridgeEntry* bridge) const { 290 LOG(VERBOSE) << "InvokeControl op=" << op << " bridge=" << bridge->mount_id_ 291 << " state=" << static_cast<int>(bridge->state_) 292 << " last_state=" << static_cast<int>(bridge->last_state_); 293 294 int last_device_events; 295 int last_proxy_events; 296 int device_events; 297 int proxy_events; 298 GetObservedEvents(bridge->last_state_, &last_device_events, &last_proxy_events); 299 GetObservedEvents(bridge->state_, &device_events, &proxy_events); 300 bool result = true; 301 if (op != EPOLL_CTL_MOD || last_device_events != device_events) { 302 result &= EpollController::InvokeControl(op, bridge->device_fd_, device_events, 303 &bridge->last_device_events_); 304 } 305 if (op != EPOLL_CTL_MOD || last_proxy_events != proxy_events) { 306 result &= EpollController::InvokeControl(op, bridge->proxy_fd_, proxy_events, 307 &bridge->last_proxy_events_); 308 } 309 return result; 310 } 311 }; 312 313 FuseBridgeLoop::FuseBridgeLoop() : opened_(true) { 314 base::unique_fd epoll_fd(epoll_create1(/* no flag */ 0)); 315 if (epoll_fd.get() == -1) { 316 PLOG(ERROR) << "Failed to open FD for epoll"; 317 opened_ = false; 318 return; 319 } 320 epoll_controller_.reset(new BridgeEpollController(std::move(epoll_fd))); 321 } 322 323 FuseBridgeLoop::~FuseBridgeLoop() { CHECK(bridges_.empty()); } 324 325 bool FuseBridgeLoop::AddBridge(int mount_id, base::unique_fd dev_fd, base::unique_fd proxy_fd) { 326 LOG(VERBOSE) << "Adding bridge " << mount_id; 327 328 std::unique_ptr<FuseBridgeEntry> bridge( 329 new FuseBridgeEntry(mount_id, std::move(dev_fd), std::move(proxy_fd))); 330 std::lock_guard<std::mutex> lock(mutex_); 331 if (!opened_) { 332 LOG(ERROR) << "Tried to add a mount to a closed bridge"; 333 return false; 334 } 335 if (bridges_.count(mount_id)) { 336 LOG(ERROR) << "Tried to add a mount point that has already been added"; 337 return false; 338 } 339 if (!epoll_controller_->AddBridgePoll(bridge.get())) { 340 return false; 341 } 342 343 bridges_.emplace(mount_id, std::move(bridge)); 344 return true; 345 } 346 347 bool FuseBridgeLoop::ProcessEventLocked(const std::unordered_set<FuseBridgeEntry*>& entries, 348 FuseBridgeLoopCallback* callback) { 349 for (auto entry : entries) { 350 entry->Transfer(callback); 351 if (!epoll_controller_->UpdateOrDeleteBridgePoll(entry)) { 352 return false; 353 } 354 if (entry->IsClosing()) { 355 const int mount_id = entry->mount_id(); 356 callback->OnClosed(mount_id); 357 bridges_.erase(mount_id); 358 if (bridges_.size() == 0) { 359 // All bridges are now closed. 360 return false; 361 } 362 } 363 } 364 return true; 365 } 366 367 void FuseBridgeLoop::Start(FuseBridgeLoopCallback* callback) { 368 LOG(DEBUG) << "Start fuse bridge loop"; 369 std::unordered_set<FuseBridgeEntry*> entries; 370 while (true) { 371 const bool wait_result = epoll_controller_->Wait(bridges_.size(), &entries); 372 LOG(VERBOSE) << "Receive epoll events"; 373 { 374 std::lock_guard<std::mutex> lock(mutex_); 375 if (!(wait_result && ProcessEventLocked(entries, callback))) { 376 for (auto it = bridges_.begin(); it != bridges_.end();) { 377 callback->OnClosed(it->second->mount_id()); 378 it = bridges_.erase(it); 379 } 380 opened_ = false; 381 return; 382 } 383 } 384 } 385 } 386 387 } // namespace fuse 388 } // namespace android 389