Change FuseAppLoop so that it can process messages asynchronously.

Previously FuseAppLoopCallback needs to return values in a synchrnous
manner. The CL changes it to asynchronous mannger so that apps can
process FUSE message asynchrnously.

Bug: 35229514
Test: FuseAppLoopTest
Change-Id: I8edcfdb003a25cfd5e9c490ec871140220b21e35
diff --git a/libappfuse/FuseAppLoop.cc b/libappfuse/FuseAppLoop.cc
index a31880e..b6bc191 100644
--- a/libappfuse/FuseAppLoop.cc
+++ b/libappfuse/FuseAppLoop.cc
@@ -16,205 +16,232 @@
 
 #include "libappfuse/FuseAppLoop.h"
 
+#include <sys/eventfd.h>
 #include <sys/stat.h>
 
 #include <android-base/logging.h>
 #include <android-base/unique_fd.h>
 
+#include "libappfuse/EpollController.h"
+
 namespace android {
 namespace fuse {
 
 namespace {
 
-void HandleLookUp(FuseBuffer* buffer, FuseAppLoopCallback* callback) {
-  // AppFuse does not support directory structure now.
-  // It can lookup only files under the mount point.
-  if (buffer->request.header.nodeid != FUSE_ROOT_ID) {
-    LOG(ERROR) << "Nodeid is not FUSE_ROOT_ID.";
-    buffer->response.Reset(0, -ENOENT, buffer->request.header.unique);
-    return;
-  }
-
-  // Ensure that the filename ends with 0.
-  const size_t filename_length =
-      buffer->request.header.len - sizeof(fuse_in_header);
-  if (buffer->request.lookup_name[filename_length - 1] != 0) {
-    LOG(ERROR) << "File name does not end with 0.";
-    buffer->response.Reset(0, -ENOENT, buffer->request.header.unique);
-    return;
-  }
-
-  const uint64_t inode =
-      static_cast<uint64_t>(atol(buffer->request.lookup_name));
-  if (inode == 0 || inode == LONG_MAX) {
-    LOG(ERROR) << "Invalid filename";
-    buffer->response.Reset(0, -ENOENT, buffer->request.header.unique);
-    return;
-  }
-
-  const int64_t size = callback->OnGetSize(inode);
-  if (size < 0) {
-    buffer->response.Reset(0, size, buffer->request.header.unique);
-    return;
-  }
-
-  buffer->response.Reset(sizeof(fuse_entry_out), 0,
-                         buffer->request.header.unique);
-  buffer->response.entry_out.nodeid = inode;
-  buffer->response.entry_out.attr_valid = 10;
-  buffer->response.entry_out.entry_valid = 10;
-  buffer->response.entry_out.attr.ino = inode;
-  buffer->response.entry_out.attr.mode = S_IFREG | 0777;
-  buffer->response.entry_out.attr.size = size;
-}
-
-void HandleGetAttr(FuseBuffer* buffer, FuseAppLoopCallback* callback) {
-  const uint64_t nodeid = buffer->request.header.nodeid;
-  int64_t size;
-  uint32_t mode;
-  if (nodeid == FUSE_ROOT_ID) {
-    size = 0;
-    mode = S_IFDIR | 0777;
-  } else {
-    size = callback->OnGetSize(buffer->request.header.nodeid);
-    if (size < 0) {
-      buffer->response.Reset(0, size, buffer->request.header.unique);
-      return;
+bool HandleLookUp(FuseAppLoop* loop, FuseBuffer* buffer, FuseAppLoopCallback* callback) {
+    // AppFuse does not support directory structure now.
+    // It can lookup only files under the mount point.
+    if (buffer->request.header.nodeid != FUSE_ROOT_ID) {
+        LOG(ERROR) << "Nodeid is not FUSE_ROOT_ID.";
+        return loop->ReplySimple(buffer->request.header.unique, -ENOENT);
     }
-    mode = S_IFREG | 0777;
-  }
 
-  buffer->response.Reset(sizeof(fuse_attr_out), 0,
-                         buffer->request.header.unique);
-  buffer->response.attr_out.attr_valid = 10;
-  buffer->response.attr_out.attr.ino = nodeid;
-  buffer->response.attr_out.attr.mode = mode;
-  buffer->response.attr_out.attr.size = size;
+    // Ensure that the filename ends with 0.
+    const size_t filename_length = buffer->request.header.len - sizeof(fuse_in_header);
+    if (buffer->request.lookup_name[filename_length - 1] != 0) {
+        LOG(ERROR) << "File name does not end with 0.";
+        return loop->ReplySimple(buffer->request.header.unique, -ENOENT);
+    }
+
+    const uint64_t inode = static_cast<uint64_t>(atol(buffer->request.lookup_name));
+    if (inode == 0 || inode == LONG_MAX) {
+        LOG(ERROR) << "Invalid filename";
+        return loop->ReplySimple(buffer->request.header.unique, -ENOENT);
+    }
+
+    callback->OnLookup(buffer->request.header.unique, inode);
+    return true;
 }
 
-void HandleOpen(FuseBuffer* buffer, FuseAppLoopCallback* callback) {
-  const int32_t file_handle = callback->OnOpen(buffer->request.header.nodeid);
-  if (file_handle < 0) {
-    buffer->response.Reset(0, file_handle, buffer->request.header.unique);
-    return;
-  }
-  buffer->response.Reset(sizeof(fuse_open_out), kFuseSuccess,
-                         buffer->request.header.unique);
-  buffer->response.open_out.fh = file_handle;
+bool HandleGetAttr(FuseAppLoop* loop, FuseBuffer* buffer, FuseAppLoopCallback* callback) {
+    if (buffer->request.header.nodeid == FUSE_ROOT_ID) {
+        return loop->ReplyGetAttr(buffer->request.header.unique, buffer->request.header.nodeid, 0,
+                                  S_IFDIR | 0777);
+    } else {
+        callback->OnGetAttr(buffer->request.header.unique, buffer->request.header.nodeid);
+        return true;
+    }
 }
 
-void HandleFsync(FuseBuffer* buffer, FuseAppLoopCallback* callback) {
-  buffer->response.Reset(0, callback->OnFsync(buffer->request.header.nodeid),
-                         buffer->request.header.unique);
+bool HandleRead(FuseAppLoop* loop, FuseBuffer* buffer, FuseAppLoopCallback* callback) {
+    if (buffer->request.read_in.size > kFuseMaxRead) {
+        return loop->ReplySimple(buffer->request.header.unique, -EINVAL);
+    }
+
+    callback->OnRead(buffer->request.header.unique, buffer->request.header.nodeid,
+                     buffer->request.read_in.offset, buffer->request.read_in.size);
+    return true;
 }
 
-void HandleRelease(FuseBuffer* buffer, FuseAppLoopCallback* callback) {
-  buffer->response.Reset(0, callback->OnRelease(buffer->request.header.nodeid),
-                         buffer->request.header.unique);
+bool HandleWrite(FuseAppLoop* loop, FuseBuffer* buffer, FuseAppLoopCallback* callback) {
+    if (buffer->request.write_in.size > kFuseMaxWrite) {
+        return loop->ReplySimple(buffer->request.header.unique, -EINVAL);
+    }
+
+    callback->OnWrite(buffer->request.header.unique, buffer->request.header.nodeid,
+                      buffer->request.write_in.offset, buffer->request.write_in.size,
+                      buffer->request.write_data);
+    return true;
 }
 
-void HandleRead(FuseBuffer* buffer, FuseAppLoopCallback* callback) {
-  const uint64_t unique = buffer->request.header.unique;
-  const uint64_t nodeid = buffer->request.header.nodeid;
-  const uint64_t offset = buffer->request.read_in.offset;
-  const uint32_t size = buffer->request.read_in.size;
+bool HandleMessage(FuseAppLoop* loop, FuseBuffer* buffer, int fd, FuseAppLoopCallback* callback) {
+    if (!buffer->request.Read(fd)) {
+        return false;
+    }
 
-  if (size > kFuseMaxRead) {
-    buffer->response.Reset(0, -EINVAL, buffer->request.header.unique);
-    return;
-  }
+    const uint32_t opcode = buffer->request.header.opcode;
+    LOG(VERBOSE) << "Read a fuse packet, opcode=" << opcode;
+    switch (opcode) {
+        case FUSE_FORGET:
+            // Do not reply to FUSE_FORGET.
+            return true;
 
-  const int32_t read_size = callback->OnRead(nodeid, offset, size,
-                                             buffer->response.read_data);
-  if (read_size < 0) {
-    buffer->response.Reset(0, read_size, buffer->request.header.unique);
-    return;
-  }
+        case FUSE_LOOKUP:
+            return HandleLookUp(loop, buffer, callback);
 
-  buffer->response.ResetHeader(read_size, kFuseSuccess, unique);
-}
+        case FUSE_GETATTR:
+            return HandleGetAttr(loop, buffer, callback);
 
-void HandleWrite(FuseBuffer* buffer, FuseAppLoopCallback* callback) {
-  const uint64_t unique = buffer->request.header.unique;
-  const uint64_t nodeid = buffer->request.header.nodeid;
-  const uint64_t offset = buffer->request.write_in.offset;
-  const uint32_t size = buffer->request.write_in.size;
+        case FUSE_OPEN:
+            callback->OnOpen(buffer->request.header.unique, buffer->request.header.nodeid);
+            return true;
 
-  if (size > kFuseMaxWrite) {
-    buffer->response.Reset(0, -EINVAL, buffer->request.header.unique);
-    return;
-  }
+        case FUSE_READ:
+            return HandleRead(loop, buffer, callback);
 
-  const int32_t write_size = callback->OnWrite(nodeid, offset, size,
-                                               buffer->request.write_data);
-  if (write_size < 0) {
-    buffer->response.Reset(0, write_size, buffer->request.header.unique);
-    return;
-  }
+        case FUSE_WRITE:
+            return HandleWrite(loop, buffer, callback);
 
-  buffer->response.Reset(sizeof(fuse_write_out), kFuseSuccess, unique);
-  buffer->response.write_out.size = write_size;
+        case FUSE_RELEASE:
+            callback->OnRelease(buffer->request.header.unique, buffer->request.header.nodeid);
+            return true;
+
+        case FUSE_FSYNC:
+            callback->OnFsync(buffer->request.header.unique, buffer->request.header.nodeid);
+            return true;
+
+        default:
+            buffer->HandleNotImpl();
+            return buffer->response.Write(fd);
+    }
 }
 
 } // namespace
 
-bool StartFuseAppLoop(int raw_fd, FuseAppLoopCallback* callback) {
-  base::unique_fd fd(raw_fd);
-  FuseBuffer buffer;
+FuseAppLoopCallback::~FuseAppLoopCallback() = default;
 
-  LOG(DEBUG) << "Start fuse loop.";
-  while (callback->IsActive()) {
-    if (!buffer.request.Read(fd)) {
-      return false;
+FuseAppLoop::FuseAppLoop(base::unique_fd&& fd) : fd_(std::move(fd)) {}
+
+void FuseAppLoop::Break() {
+    const int64_t value = 1;
+    if (write(break_fd_, &value, sizeof(value)) == -1) {
+        PLOG(ERROR) << "Failed to send a break event";
+    }
+}
+
+bool FuseAppLoop::ReplySimple(uint64_t unique, int32_t result) {
+    if (result == -ENOSYS) {
+        // We should not return -ENOSYS because the kernel stops delivering FUSE
+        // command after receiving -ENOSYS as a result for the command.
+        result = -EBADF;
+    }
+    FuseSimpleResponse response;
+    response.Reset(0, result, unique);
+    return response.Write(fd_);
+}
+
+bool FuseAppLoop::ReplyLookup(uint64_t unique, uint64_t inode, int64_t size) {
+    FuseSimpleResponse response;
+    response.Reset(sizeof(fuse_entry_out), 0, unique);
+    response.entry_out.nodeid = inode;
+    response.entry_out.attr_valid = 10;
+    response.entry_out.entry_valid = 10;
+    response.entry_out.attr.ino = inode;
+    response.entry_out.attr.mode = S_IFREG | 0777;
+    response.entry_out.attr.size = size;
+    return response.Write(fd_);
+}
+
+bool FuseAppLoop::ReplyGetAttr(uint64_t unique, uint64_t inode, int64_t size, int mode) {
+    CHECK(mode == (S_IFREG | 0777) || mode == (S_IFDIR | 0777));
+    FuseSimpleResponse response;
+    response.Reset(sizeof(fuse_attr_out), 0, unique);
+    response.attr_out.attr_valid = 10;
+    response.attr_out.attr.ino = inode;
+    response.attr_out.attr.mode = mode;
+    response.attr_out.attr.size = size;
+    return response.Write(fd_);
+}
+
+bool FuseAppLoop::ReplyOpen(uint64_t unique, uint64_t fh) {
+    FuseSimpleResponse response;
+    response.Reset(sizeof(fuse_open_out), kFuseSuccess, unique);
+    response.open_out.fh = fh;
+    return response.Write(fd_);
+}
+
+bool FuseAppLoop::ReplyWrite(uint64_t unique, uint32_t size) {
+    CHECK(size <= kFuseMaxWrite);
+    FuseSimpleResponse response;
+    response.Reset(sizeof(fuse_write_out), kFuseSuccess, unique);
+    response.write_out.size = size;
+    return response.Write(fd_);
+}
+
+bool FuseAppLoop::ReplyRead(uint64_t unique, uint32_t size, const void* data) {
+    CHECK(size <= kFuseMaxRead);
+    FuseSimpleResponse response;
+    response.ResetHeader(size, kFuseSuccess, unique);
+    return response.WriteWithBody(fd_, sizeof(FuseResponse), data);
+}
+
+void FuseAppLoop::Start(FuseAppLoopCallback* callback) {
+    break_fd_.reset(eventfd(/* initval */ 0, EFD_CLOEXEC));
+    if (break_fd_.get() == -1) {
+        PLOG(ERROR) << "Failed to open FD for break event";
+        return;
     }
 
-    const uint32_t opcode = buffer.request.header.opcode;
-    LOG(VERBOSE) << "Read a fuse packet, opcode=" << opcode;
-    switch (opcode) {
-      case FUSE_FORGET:
-        // Do not reply to FUSE_FORGET.
-        continue;
-
-      case FUSE_LOOKUP:
-        HandleLookUp(&buffer, callback);
-        break;
-
-      case FUSE_GETATTR:
-        HandleGetAttr(&buffer, callback);
-        break;
-
-      case FUSE_OPEN:
-        HandleOpen(&buffer, callback);
-        break;
-
-      case FUSE_READ:
-        HandleRead(&buffer, callback);
-        break;
-
-      case FUSE_WRITE:
-        HandleWrite(&buffer, callback);
-        break;
-
-      case FUSE_RELEASE:
-        HandleRelease(&buffer, callback);
-        break;
-
-      case FUSE_FSYNC:
-        HandleFsync(&buffer, callback);
-        break;
-
-      default:
-        buffer.HandleNotImpl();
-        break;
+    base::unique_fd epoll_fd(epoll_create1(EPOLL_CLOEXEC));
+    if (epoll_fd.get() == -1) {
+        PLOG(ERROR) << "Failed to open FD for epoll";
+        return;
     }
 
-    if (!buffer.response.Write(fd)) {
-      LOG(ERROR) << "Failed to write a response to the device.";
-      return false;
-    }
-  }
+    int last_event;
+    int break_event;
 
-  return true;
+    std::unique_ptr<EpollController> epoll_controller(new EpollController(std::move(epoll_fd)));
+    if (!epoll_controller->AddFd(fd_, EPOLLIN, &last_event)) {
+        return;
+    }
+    if (!epoll_controller->AddFd(break_fd_, EPOLLIN, &break_event)) {
+        return;
+    }
+
+    last_event = 0;
+    break_event = 0;
+
+    FuseBuffer buffer;
+    while (true) {
+        if (!epoll_controller->Wait(1)) {
+            break;
+        }
+        last_event = 0;
+        *reinterpret_cast<int*>(epoll_controller->events()[0].data.ptr) =
+            epoll_controller->events()[0].events;
+
+        if (break_event != 0 || (last_event & ~EPOLLIN) != 0) {
+            break;
+        }
+
+        if (!HandleMessage(this, &buffer, fd_, callback)) {
+            break;
+        }
+    }
+
+    LOG(VERBOSE) << "FuseAppLoop exit";
 }
 
 }  // namespace fuse