diff --git a/adb/Android.mk b/adb/Android.mk
index e841205..a05bb55 100644
--- a/adb/Android.mk
+++ b/adb/Android.mk
@@ -128,7 +128,7 @@
 
 # Even though we're building a static library (and thus there's no link step for
 # this to take effect), this adds the includes to our path.
-LOCAL_STATIC_LIBRARIES := libcrypto_utils libcrypto libbase
+LOCAL_STATIC_LIBRARIES := libcrypto_utils libcrypto libqemu_pipe libbase
 
 LOCAL_WHOLE_STATIC_LIBRARIES := libadbd_usb
 
@@ -361,6 +361,7 @@
 LOCAL_STATIC_LIBRARIES := \
     libadbd \
     libbase \
+    libqemu_pipe \
     libbootloader_message \
     libfs_mgr \
     libfec \
diff --git a/adb/transport_local.cpp b/adb/transport_local.cpp
index 4198a52..12b98ba 100644
--- a/adb/transport_local.cpp
+++ b/adb/transport_local.cpp
@@ -289,7 +289,7 @@
 #define open    adb_open
 #define read    adb_read
 #define write   adb_write
-#include <system/qemu_pipe.h>
+#include <qemu_pipe.h>
 #undef open
 #undef read
 #undef write
diff --git a/include/system/qemu_pipe.h b/include/system/qemu_pipe.h
deleted file mode 100644
index af25079..0000000
--- a/include/system/qemu_pipe.h
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * Copyright (C) 2011 The Android Open Source Project
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-#ifndef ANDROID_INCLUDE_SYSTEM_QEMU_PIPE_H
-#define ANDROID_INCLUDE_SYSTEM_QEMU_PIPE_H
-
-#include <unistd.h>
-#include <fcntl.h>
-#include <string.h>
-#include <errno.h>
-
-// Define QEMU_PIPE_DEBUG if you want to print error messages when an error
-// occurs during pipe operations. The macro should simply take a printf-style
-// formatting string followed by optional arguments.
-#ifndef QEMU_PIPE_DEBUG
-#  define  QEMU_PIPE_DEBUG(...)   (void)0
-#endif
-
-// Try to open a new Qemu fast-pipe. This function returns a file descriptor
-// that can be used to communicate with a named service managed by the
-// emulator.
-//
-// This file descriptor can be used as a standard pipe/socket descriptor.
-//
-// 'pipeName' is the name of the emulator service you want to connect to,
-// and must begin with 'pipe:' (e.g. 'pipe:camera' or 'pipe:opengles').
-//
-// On success, return a valid file descriptor, or -1/errno on failure. E.g.:
-//
-// EINVAL  -> unknown/unsupported pipeName
-// ENOSYS  -> fast pipes not available in this system.
-//
-// ENOSYS should never happen, except if you're trying to run within a
-// misconfigured emulator.
-//
-// You should be able to open several pipes to the same pipe service,
-// except for a few special cases (e.g. GSM modem), where EBUSY will be
-// returned if more than one client tries to connect to it.
-static __inline__ int qemu_pipe_open(const char* pipeName) {
-    // Sanity check.
-    if (!pipeName || memcmp(pipeName, "pipe:", 5) != 0) {
-        errno = EINVAL;
-        return -1;
-    }
-
-    int fd = TEMP_FAILURE_RETRY(open("/dev/qemu_pipe", O_RDWR));
-    if (fd < 0) {
-        QEMU_PIPE_DEBUG("%s: Could not open /dev/qemu_pipe: %s", __FUNCTION__,
-                        strerror(errno));
-        return -1;
-    }
-
-    // Write the pipe name, *including* the trailing zero which is necessary.
-    size_t pipeNameLen = strlen(pipeName);
-    ssize_t ret = TEMP_FAILURE_RETRY(write(fd, pipeName, pipeNameLen + 1U));
-    if (ret != (ssize_t)pipeNameLen + 1) {
-        QEMU_PIPE_DEBUG("%s: Could not connect to %s pipe service: %s",
-                        __FUNCTION__, pipeName, strerror(errno));
-        if (ret == 0) {
-            errno = ECONNRESET;
-        } else if (ret > 0) {
-            errno = EINVAL;
-        }
-        return -1;
-    }
-    return fd;
-}
-
-// Send a framed message |buff| of |len| bytes through the |fd| descriptor.
-// This really adds a 4-hexchar prefix describing the payload size.
-// Returns 0 on success, and -1 on error.
-static int __inline__ qemu_pipe_frame_send(int fd,
-                                           const void* buff,
-                                           size_t len) {
-    char header[5];
-    snprintf(header, sizeof(header), "%04zx", len);
-    ssize_t ret = TEMP_FAILURE_RETRY(write(fd, header, 4));
-    if (ret != 4) {
-        QEMU_PIPE_DEBUG("Can't write qemud frame header: %s", strerror(errno));
-        return -1;
-    }
-    ret = TEMP_FAILURE_RETRY(write(fd, buff, len));
-    if (ret != (ssize_t)len) {
-        QEMU_PIPE_DEBUG("Can't write qemud frame payload: %s", strerror(errno));
-        return -1;
-    }
-    return 0;
-}
-
-// Read a frame message from |fd|, and store it into |buff| of |len| bytes.
-// If the framed message is larger than |len|, then this returns -1 and the
-// content is lost. Otherwise, this returns the size of the message. NOTE:
-// empty messages are possible in a framed wire protocol and do not mean
-// end-of-stream.
-static int __inline__ qemu_pipe_frame_recv(int fd, void* buff, size_t len) {
-    char header[5];
-    ssize_t ret = TEMP_FAILURE_RETRY(read(fd, header, 4));
-    if (ret != 4) {
-        QEMU_PIPE_DEBUG("Can't read qemud frame header: %s", strerror(errno));
-        return -1;
-    }
-    header[4] = '\0';
-    size_t size;
-    if (sscanf(header, "%04zx", &size) != 1) {
-        QEMU_PIPE_DEBUG("Malformed qemud frame header: [%.*s]", 4, header);
-        return -1;
-    }
-    if (size > len) {
-        QEMU_PIPE_DEBUG("Oversized qemud frame (% bytes, expected <= %)", size,
-                        len);
-        return -1;
-    }
-    ret = TEMP_FAILURE_RETRY(read(fd, buff, size));
-    if (ret != (ssize_t)size) {
-        QEMU_PIPE_DEBUG("Could not read qemud frame payload: %s",
-                        strerror(errno));
-        return -1;
-    }
-    return size;
-}
-
-#endif /* ANDROID_INCLUDE_HARDWARE_QEMUD_PIPE_H */
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
diff --git a/libappfuse/include/libappfuse/FuseAppLoop.h b/libappfuse/include/libappfuse/FuseAppLoop.h
index c3edfcc..f2ef2b5 100644
--- a/libappfuse/include/libappfuse/FuseAppLoop.h
+++ b/libappfuse/include/libappfuse/FuseAppLoop.h
@@ -17,23 +17,51 @@
 #ifndef ANDROID_LIBAPPFUSE_FUSEAPPLOOP_H_
 #define ANDROID_LIBAPPFUSE_FUSEAPPLOOP_H_
 
+#include <memory>
+#include <mutex>
+
+#include <android-base/unique_fd.h>
+
 #include "libappfuse/FuseBuffer.h"
 
 namespace android {
 namespace fuse {
 
+class EpollController;
+
 class FuseAppLoopCallback {
  public:
-  virtual bool IsActive() = 0;
-  virtual int64_t OnGetSize(uint64_t inode) = 0;
-  virtual int32_t OnFsync(uint64_t inode) = 0;
-  virtual int32_t OnWrite(
-      uint64_t inode, uint64_t offset, uint32_t size, const void* data) = 0;
-  virtual int32_t OnRead(
-      uint64_t inode, uint64_t offset, uint32_t size, void* data) = 0;
-  virtual int32_t OnOpen(uint64_t inode) = 0;
-  virtual int32_t OnRelease(uint64_t inode) = 0;
-  virtual ~FuseAppLoopCallback() = default;
+   virtual void OnLookup(uint64_t unique, uint64_t inode) = 0;
+   virtual void OnGetAttr(uint64_t unique, uint64_t inode) = 0;
+   virtual void OnFsync(uint64_t unique, uint64_t inode) = 0;
+   virtual void OnWrite(uint64_t unique, uint64_t inode, uint64_t offset, uint32_t size,
+                        const void* data) = 0;
+   virtual void OnRead(uint64_t unique, uint64_t inode, uint64_t offset, uint32_t size) = 0;
+   virtual void OnOpen(uint64_t unique, uint64_t inode) = 0;
+   virtual void OnRelease(uint64_t unique, uint64_t inode) = 0;
+   virtual ~FuseAppLoopCallback();
+};
+
+class FuseAppLoop final {
+  public:
+    FuseAppLoop(base::unique_fd&& fd);
+
+    void Start(FuseAppLoopCallback* callback);
+    void Break();
+
+    bool ReplySimple(uint64_t unique, int32_t result);
+    bool ReplyLookup(uint64_t unique, uint64_t inode, int64_t size);
+    bool ReplyGetAttr(uint64_t unique, uint64_t inode, int64_t size, int mode);
+    bool ReplyOpen(uint64_t unique, uint64_t fh);
+    bool ReplyWrite(uint64_t unique, uint32_t size);
+    bool ReplyRead(uint64_t unique, uint32_t size, const void* data);
+
+  private:
+    base::unique_fd fd_;
+    base::unique_fd break_fd_;
+
+    // Lock for multi-threading.
+    std::mutex mutex_;
 };
 
 bool StartFuseAppLoop(int fd, FuseAppLoopCallback* callback);
diff --git a/libappfuse/tests/FuseAppLoopTest.cc b/libappfuse/tests/FuseAppLoopTest.cc
index 64dd813..98e3665 100644
--- a/libappfuse/tests/FuseAppLoopTest.cc
+++ b/libappfuse/tests/FuseAppLoopTest.cc
@@ -23,6 +23,9 @@
 #include <gtest/gtest.h>
 #include <thread>
 
+#include "libappfuse/EpollController.h"
+#include "libappfuse/FuseBridgeLoop.h"
+
 namespace android {
 namespace fuse {
 namespace {
@@ -37,82 +40,61 @@
 class Callback : public FuseAppLoopCallback {
  public:
   std::vector<CallbackRequest> requests;
+  FuseAppLoop* loop;
 
-  bool IsActive() override {
-    return true;
+  void OnGetAttr(uint64_t seq, uint64_t inode) override {
+      EXPECT_NE(FUSE_ROOT_ID, static_cast<int>(inode));
+      EXPECT_TRUE(loop->ReplyGetAttr(seq, inode, kTestFileSize, S_IFREG | 0777));
   }
 
-  int64_t OnGetSize(uint64_t inode) override {
-    if (inode == FUSE_ROOT_ID) {
-      return 0;
-    } else {
-      return kTestFileSize;
-    }
+  void OnLookup(uint64_t unique, uint64_t inode) override {
+      EXPECT_NE(FUSE_ROOT_ID, static_cast<int>(inode));
+      EXPECT_TRUE(loop->ReplyLookup(unique, inode, kTestFileSize));
   }
 
-  int32_t OnFsync(uint64_t inode) override {
-    requests.push_back({
-      .code = FUSE_FSYNC,
-      .inode = inode
-    });
-    return 0;
+  void OnFsync(uint64_t seq, uint64_t inode) override {
+      requests.push_back({.code = FUSE_FSYNC, .inode = inode});
+      loop->ReplySimple(seq, 0);
   }
 
-  int32_t OnWrite(uint64_t inode,
-                  uint64_t offset ATTRIBUTE_UNUSED,
-                  uint32_t size ATTRIBUTE_UNUSED,
-                  const void* data ATTRIBUTE_UNUSED) override {
-    requests.push_back({
-      .code = FUSE_WRITE,
-      .inode = inode
-    });
-    return 0;
+  void OnWrite(uint64_t seq, uint64_t inode, uint64_t offset ATTRIBUTE_UNUSED,
+               uint32_t size ATTRIBUTE_UNUSED, const void* data ATTRIBUTE_UNUSED) override {
+      requests.push_back({.code = FUSE_WRITE, .inode = inode});
+      loop->ReplyWrite(seq, 0);
   }
 
-  int32_t OnRead(uint64_t inode,
-                 uint64_t offset ATTRIBUTE_UNUSED,
-                 uint32_t size ATTRIBUTE_UNUSED,
-                 void* data ATTRIBUTE_UNUSED) override {
-    requests.push_back({
-      .code = FUSE_READ,
-      .inode = inode
-    });
-    return 0;
+  void OnRead(uint64_t seq, uint64_t inode, uint64_t offset ATTRIBUTE_UNUSED,
+              uint32_t size ATTRIBUTE_UNUSED) override {
+      requests.push_back({.code = FUSE_READ, .inode = inode});
+      loop->ReplySimple(seq, 0);
   }
 
-  int32_t OnOpen(uint64_t inode) override {
-    requests.push_back({
-      .code = FUSE_OPEN,
-      .inode = inode
-    });
-    return 0;
+  void OnOpen(uint64_t seq, uint64_t inode) override {
+      requests.push_back({.code = FUSE_OPEN, .inode = inode});
+      loop->ReplyOpen(seq, inode);
   }
 
-  int32_t OnRelease(uint64_t inode) override {
-    requests.push_back({
-      .code = FUSE_RELEASE,
-      .inode = inode
-    });
-    return 0;
+  void OnRelease(uint64_t seq, uint64_t inode) override {
+      requests.push_back({.code = FUSE_RELEASE, .inode = inode});
+      loop->ReplySimple(seq, 0);
   }
 };
 
 class FuseAppLoopTest : public ::testing::Test {
- private:
-  std::thread thread_;
-
  protected:
-  base::unique_fd sockets_[2];
-  Callback callback_;
-  FuseRequest request_;
-  FuseResponse response_;
+   std::thread thread_;
+   base::unique_fd sockets_[2];
+   Callback callback_;
+   FuseRequest request_;
+   FuseResponse response_;
+   std::unique_ptr<FuseAppLoop> loop_;
 
-  void SetUp() override {
-    base::SetMinimumLogSeverity(base::VERBOSE);
-    ASSERT_TRUE(SetupMessageSockets(&sockets_));
-    thread_ = std::thread([this] {
-      StartFuseAppLoop(sockets_[1].release(), &callback_);
-    });
+   void SetUp() override {
+       base::SetMinimumLogSeverity(base::VERBOSE);
+       ASSERT_TRUE(SetupMessageSockets(&sockets_));
+       loop_.reset(new FuseAppLoop(std::move(sockets_[1])));
+       callback_.loop = loop_.get();
+       thread_ = std::thread([this] { loop_->Start(&callback_); });
   }
 
   void CheckCallback(
@@ -300,5 +282,18 @@
   CheckCallback(sizeof(fuse_write_in), FUSE_WRITE, sizeof(fuse_write_out));
 }
 
+TEST_F(FuseAppLoopTest, Break) {
+    // Ensure that the loop started.
+    request_.Reset(sizeof(fuse_open_in), FUSE_OPEN, 1);
+    request_.header.nodeid = 10;
+    ASSERT_TRUE(request_.Write(sockets_[0]));
+    ASSERT_TRUE(response_.Read(sockets_[0]));
+
+    loop_->Break();
+    if (thread_.joinable()) {
+        thread_.join();
+    }
+}
+
 }  // namespace fuse
 }  // namespace android
diff --git a/qemu_pipe/Android.mk b/qemu_pipe/Android.mk
new file mode 100644
index 0000000..6e0144c
--- /dev/null
+++ b/qemu_pipe/Android.mk
@@ -0,0 +1,19 @@
+# Copyright 2011 The Android Open Source Project
+
+LOCAL_PATH:= $(call my-dir)
+
+common_static_libraries := \
+    libbase
+include $(CLEAR_VARS)
+LOCAL_CLANG := true
+LOCAL_SANITIZE := integer
+LOCAL_SRC_FILES:= \
+    qemu_pipe.cpp
+LOCAL_C_INCLUDES := \
+    $(LOCAL_PATH)/include \
+    system/base/include
+LOCAL_MODULE:= libqemu_pipe
+LOCAL_STATIC_LIBRARIES := $(common_static_libraries)
+LOCAL_EXPORT_C_INCLUDE_DIRS := $(LOCAL_PATH)/include
+LOCAL_CFLAGS := -Werror
+include $(BUILD_STATIC_LIBRARY)
diff --git a/qemu_pipe/include/qemu_pipe.h b/qemu_pipe/include/qemu_pipe.h
new file mode 100644
index 0000000..0987498
--- /dev/null
+++ b/qemu_pipe/include/qemu_pipe.h
@@ -0,0 +1,64 @@
+/*
+ * Copyright (C) 2011 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef ANDROID_CORE_INCLUDE_QEMU_PIPE_H
+#define ANDROID_CORE_INCLUDE_QEMU_PIPE_H
+
+#include <stddef.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+// Try to open a new Qemu fast-pipe. This function returns a file descriptor
+// that can be used to communicate with a named service managed by the
+// emulator.
+//
+// This file descriptor can be used as a standard pipe/socket descriptor.
+//
+// 'pipeName' is the name of the emulator service you want to connect to,
+// and should begin with 'pipe:' (e.g. 'pipe:camera' or 'pipe:opengles').
+// For backward compatibility, the 'pipe:' prefix can be omitted, and in
+// that case, qemu_pipe_open will add it for you.
+
+// On success, return a valid file descriptor, or -1/errno on failure. E.g.:
+//
+// EINVAL  -> unknown/unsupported pipeName
+// ENOSYS  -> fast pipes not available in this system.
+//
+// ENOSYS should never happen, except if you're trying to run within a
+// misconfigured emulator.
+//
+// You should be able to open several pipes to the same pipe service,
+// except for a few special cases (e.g. GSM modem), where EBUSY will be
+// returned if more than one client tries to connect to it.
+int qemu_pipe_open(const char* pipeName);
+
+// Send a framed message |buff| of |len| bytes through the |fd| descriptor.
+// This really adds a 4-hexchar prefix describing the payload size.
+// Returns 0 on success, and -1 on error.
+int qemu_pipe_frame_send(int fd, const void* buff, size_t len);
+
+// Read a frame message from |fd|, and store it into |buff| of |len| bytes.
+// If the framed message is larger than |len|, then this returns -1 and the
+// content is lost. Otherwise, this returns the size of the message. NOTE:
+// empty messages are possible in a framed wire protocol and do not mean
+// end-of-stream.
+int qemu_pipe_frame_recv(int fd, void* buff, size_t len);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* ANDROID_CORE_INCLUDE_QEMU_PIPE_H */
diff --git a/qemu_pipe/qemu_pipe.cpp b/qemu_pipe/qemu_pipe.cpp
new file mode 100644
index 0000000..beeccb0
--- /dev/null
+++ b/qemu_pipe/qemu_pipe.cpp
@@ -0,0 +1,108 @@
+/*
+ * Copyright (C) 2011 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "qemu_pipe.h"
+
+#include <unistd.h>
+#include <fcntl.h>
+#include <string.h>
+#include <errno.h>
+#include <stdio.h>
+
+#include <android-base/file.h>
+
+using android::base::ReadFully;
+using android::base::WriteFully;
+
+// Define QEMU_PIPE_DEBUG if you want to print error messages when an error
+// occurs during pipe operations. The macro should simply take a printf-style
+// formatting string followed by optional arguments.
+#ifndef QEMU_PIPE_DEBUG
+#  define  QEMU_PIPE_DEBUG(...)   (void)0
+#endif
+
+int qemu_pipe_open(const char* pipeName) {
+    // Sanity check.
+    if (!pipeName) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    int fd = TEMP_FAILURE_RETRY(open("/dev/qemu_pipe", O_RDWR));
+    if (fd < 0) {
+        QEMU_PIPE_DEBUG("%s: Could not open /dev/qemu_pipe: %s", __FUNCTION__,
+                        strerror(errno));
+        return -1;
+    }
+
+    // Write the pipe name, *including* the trailing zero which is necessary.
+    size_t pipeNameLen = strlen(pipeName);
+    if (WriteFully(fd, pipeName, pipeNameLen + 1U)) {
+        return fd;
+    }
+
+    // now, add 'pipe:' prefix and try again
+    // Note: host side will wait for the trailing '\0' to start
+    // service lookup.
+    const char pipe_prefix[] = "pipe:";
+    if (WriteFully(fd, pipe_prefix, strlen(pipe_prefix)) &&
+            WriteFully(fd, pipeName, pipeNameLen + 1U)) {
+        return fd;
+    }
+    QEMU_PIPE_DEBUG("%s: Could not write to %s pipe service: %s",
+            __FUNCTION__, pipeName, strerror(errno));
+    close(fd);
+    return -1;
+}
+
+int qemu_pipe_frame_send(int fd, const void* buff, size_t len) {
+    char header[5];
+    snprintf(header, sizeof(header), "%04zx", len);
+    if (!WriteFully(fd, header, 4)) {
+        QEMU_PIPE_DEBUG("Can't write qemud frame header: %s", strerror(errno));
+        return -1;
+    }
+    if (!WriteFully(fd, buff, len)) {
+        QEMU_PIPE_DEBUG("Can't write qemud frame payload: %s", strerror(errno));
+        return -1;
+    }
+    return 0;
+}
+
+int qemu_pipe_frame_recv(int fd, void* buff, size_t len) {
+    char header[5];
+    if (!ReadFully(fd, header, 4)) {
+        QEMU_PIPE_DEBUG("Can't read qemud frame header: %s", strerror(errno));
+        return -1;
+    }
+    header[4] = '\0';
+    size_t size;
+    if (sscanf(header, "%04zx", &size) != 1) {
+        QEMU_PIPE_DEBUG("Malformed qemud frame header: [%.*s]", 4, header);
+        return -1;
+    }
+    if (size > len) {
+        QEMU_PIPE_DEBUG("Oversized qemud frame (% bytes, expected <= %)", size,
+                        len);
+        return -1;
+    }
+    if (!ReadFully(fd, buff, size)) {
+        QEMU_PIPE_DEBUG("Could not read qemud frame payload: %s",
+                        strerror(errno));
+        return -1;
+    }
+    return size;
+}