adb: Use kernel aio for functionfs.

This method works around the downsides of
ENDPOINT_ALLOC, namely that it is not affected
by memory fragmentation and it uses an upstream
interface.

Also add libasyncio to provide the necessary syscalls
to both adb and mtp.

Add some small optimizations to file_sync.

Bug: 37916658
Test: run adb push/pull
Change-Id: If3b3be02b5e2d4f9cffec1b8ddc02a5768a51a1f

adb/Android.mk

@@ -116,7 +116,7 @@ LOCAL_SANITIZE := $(adb_target_sanitize)
 
 # 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 libbase libasyncio
 
 include $(BUILD_STATIC_LIBRARY)
 
@@ -362,6 +362,7 @@ LOCAL_SANITIZE := $(adb_target_sanitize)
 LOCAL_STRIP_MODULE := keep_symbols
 LOCAL_STATIC_LIBRARIES := \
     libadbd \
+    libasyncio \
     libavb_user \
     libbase \
     libbootloader_message \

adb/adb.h

@@ -31,8 +31,7 @@
 #include "usb.h"
 
 constexpr size_t MAX_PAYLOAD_V1 = 4 * 1024;
-constexpr size_t MAX_PAYLOAD_V2 = 256 * 1024;
-constexpr size_t MAX_PAYLOAD = MAX_PAYLOAD_V2;
+constexpr size_t MAX_PAYLOAD = 1024 * 1024;
 
 constexpr size_t LINUX_MAX_SOCKET_SIZE = 4194304;
 

adb/daemon/usb.cpp

@@ -26,6 +26,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/ioctl.h>
+#include <sys/mman.h>
 #include <sys/types.h>
 #include <unistd.h>
 
@@ -49,16 +50,11 @@ using namespace std::chrono_literals;
 #define MAX_PACKET_SIZE_HS 512
 #define MAX_PACKET_SIZE_SS 1024
 
-// Kernels before 3.3 have a 16KiB transfer limit  That limit was replaced
-// with a 16MiB global limit in 3.3, but each URB submitted required a
-// contiguous kernel allocation, so you would get ENOMEM if you tried to
-// send something larger than the biggest available contiguous kernel
-// memory region. Large contiguous allocations could be unreliable
-// on a device kernel that has been running for a while fragmenting its
-// memory so we start with a larger allocation, and shrink the amount if
-// necessary.
 #define USB_FFS_BULK_SIZE 16384
 
+// Number of buffers needed to fit MAX_PAYLOAD, with an extra for ZLPs.
+#define USB_FFS_NUM_BUFS ((MAX_PAYLOAD / USB_FFS_BULK_SIZE) + 1)
+
 #define cpu_to_le16(x) htole16(x)
 #define cpu_to_le32(x) htole32(x)
 
@@ -234,6 +230,26 @@ static const struct {
     },
 };
 
+static void aio_block_init(aio_block* aiob) {
+    aiob->iocb.resize(USB_FFS_NUM_BUFS);
+    aiob->iocbs.resize(USB_FFS_NUM_BUFS);
+    aiob->events.resize(USB_FFS_NUM_BUFS);
+    aiob->num_submitted = 0;
+    for (unsigned i = 0; i < USB_FFS_NUM_BUFS; i++) {
+        aiob->iocbs[i] = &aiob->iocb[i];
+    }
+}
+
+static int getMaxPacketSize(int ffs_fd) {
+    usb_endpoint_descriptor desc;
+    if (ioctl(ffs_fd, FUNCTIONFS_ENDPOINT_DESC, reinterpret_cast<unsigned long>(&desc))) {
+        D("[ could not get endpoint descriptor! (%d) ]", errno);
+        return MAX_PACKET_SIZE_HS;
+    } else {
+        return desc.wMaxPacketSize;
+    }
+}
+
 bool init_functionfs(struct usb_handle* h) {
     LOG(INFO) << "initializing functionfs";
 
@@ -301,6 +317,14 @@ bool init_functionfs(struct usb_handle* h) {
         goto err;
     }
 
+    if (io_setup(USB_FFS_NUM_BUFS, &h->read_aiob.ctx) ||
+        io_setup(USB_FFS_NUM_BUFS, &h->write_aiob.ctx)) {
+        D("[ aio: got error on io_setup (%d) ]", errno);
+    }
+
+    h->read_aiob.fd = h->bulk_out;
+    h->write_aiob.fd = h->bulk_in;
+
     h->max_rw = MAX_PAYLOAD;
     while (h->max_rw >= USB_FFS_BULK_SIZE && retries < ENDPOINT_ALLOC_RETRIES) {
         int ret_in = ioctl(h->bulk_in, FUNCTIONFS_ENDPOINT_ALLOC, static_cast<__u32>(h->max_rw));
@@ -410,6 +434,65 @@ static int usb_ffs_read(usb_handle* h, void* data, int len) {
     return 0;
 }
 
+static int usb_ffs_do_aio(usb_handle* h, const void* data, int len, bool read) {
+    aio_block* aiob = read ? &h->read_aiob : &h->write_aiob;
+    bool zero_packet = false;
+
+    int num_bufs = len / USB_FFS_BULK_SIZE + (len % USB_FFS_BULK_SIZE == 0 ? 0 : 1);
+    const char* cur_data = reinterpret_cast<const char*>(data);
+    int packet_size = getMaxPacketSize(aiob->fd);
+
+    if (posix_madvise(const_cast<void*>(data), len, POSIX_MADV_SEQUENTIAL | POSIX_MADV_WILLNEED) <
+        0) {
+        D("[ Failed to madvise: %d ]", errno);
+    }
+
+    for (int i = 0; i < num_bufs; i++) {
+        int buf_len = std::min(len, USB_FFS_BULK_SIZE);
+        io_prep(&aiob->iocb[i], aiob->fd, cur_data, buf_len, 0, read);
+
+        len -= buf_len;
+        cur_data += buf_len;
+
+        if (len == 0 && buf_len % packet_size == 0 && read) {
+            // adb does not expect the device to send a zero packet after data transfer,
+            // but the host *does* send a zero packet for the device to read.
+            zero_packet = true;
+        }
+    }
+    if (zero_packet) {
+        io_prep(&aiob->iocb[num_bufs], aiob->fd, reinterpret_cast<const void*>(cur_data),
+                packet_size, 0, read);
+        num_bufs += 1;
+    }
+
+    if (io_submit(aiob->ctx, num_bufs, aiob->iocbs.data()) < num_bufs) {
+        D("[ aio: got error submitting %s (%d) ]", read ? "read" : "write", errno);
+        return -1;
+    }
+    if (TEMP_FAILURE_RETRY(
+            io_getevents(aiob->ctx, num_bufs, num_bufs, aiob->events.data(), nullptr)) < num_bufs) {
+        D("[ aio: got error waiting %s (%d) ]", read ? "read" : "write", errno);
+        return -1;
+    }
+    for (int i = 0; i < num_bufs; i++) {
+        if (aiob->events[i].res < 0) {
+            errno = aiob->events[i].res;
+            D("[ aio: got error event on %s (%d) ]", read ? "read" : "write", errno);
+            return -1;
+        }
+    }
+    return 0;
+}
+
+static int usb_ffs_aio_read(usb_handle* h, void* data, int len) {
+    return usb_ffs_do_aio(h, data, len, true);
+}
+
+static int usb_ffs_aio_write(usb_handle* h, const void* data, int len) {
+    return usb_ffs_do_aio(h, data, len, false);
+}
+
 static void usb_ffs_kick(usb_handle* h) {
     int err;
 
@@ -438,6 +521,9 @@ static void usb_ffs_close(usb_handle* h) {
     h->kicked = false;
     adb_close(h->bulk_out);
     adb_close(h->bulk_in);
+    io_destroy(h->read_aiob.ctx);
+    io_destroy(h->write_aiob.ctx);
+
     // Notify usb_adb_open_thread to open a new connection.
     h->lock.lock();
     h->open_new_connection = true;
@@ -450,8 +536,17 @@ static void usb_ffs_init() {
 
     usb_handle* h = new usb_handle();
 
-    h->write = usb_ffs_write;
-    h->read = usb_ffs_read;
+    if (android::base::GetBoolProperty("sys.usb.ffs.aio_compat", false)) {
+        // Devices on older kernels (< 3.18) will not have aio support for ffs
+        // unless backported. Fall back on the non-aio functions instead.
+        h->write = usb_ffs_write;
+        h->read = usb_ffs_read;
+    } else {
+        h->write = usb_ffs_aio_write;
+        h->read = usb_ffs_aio_read;
+        aio_block_init(&h->read_aiob);
+        aio_block_init(&h->write_aiob);
+    }
     h->kick = usb_ffs_kick;
     h->close = usb_ffs_close;
 

adb/daemon/usb.h

@@ -20,6 +20,17 @@
 #include <condition_variable>
 #include <mutex>
 
+#include <asyncio/AsyncIO.h>
+
+struct aio_block {
+    std::vector<struct iocb> iocb;
+    std::vector<struct iocb*> iocbs;
+    std::vector<struct io_event> events;
+    aio_context_t ctx;
+    int num_submitted;
+    int fd;
+};
+
 struct usb_handle {
     usb_handle() : kicked(false) {
     }
@@ -39,7 +50,11 @@ struct usb_handle {
     int bulk_out = -1; /* "out" from the host's perspective => source for adbd */
     int bulk_in = -1;  /* "in" from the host's perspective => sink for adbd */
 
+    // Access to these blocks is very not thread safe. Have one block for both the
+    // read and write threads.
+    struct aio_block read_aiob;
+    struct aio_block write_aiob;
+
     int max_rw;
 };
 
-bool init_functionfs(struct usb_handle* h);

adb/file_sync_client.cpp

@@ -441,7 +441,7 @@ class SyncConnection {
         syncsendbuf sbuf;
         sbuf.id = ID_DATA;
         while (true) {
-            int bytes_read = adb_read(lfd, sbuf.data, max);
+            int bytes_read = adb_read(lfd, sbuf.data, max - sizeof(SyncRequest));
             if (bytes_read == -1) {
                 Error("reading '%s' locally failed: %s", lpath, strerror(errno));
                 adb_close(lfd);

adb/file_sync_service.cpp

@@ -206,6 +206,12 @@ static bool handle_send_file(int s, const char* path, uid_t uid, gid_t gid, uint
     __android_log_security_bswrite(SEC_TAG_ADB_SEND_FILE, path);
 
     int fd = adb_open_mode(path, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC, mode);
+
+    if (posix_fadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL | POSIX_FADV_NOREUSE | POSIX_FADV_WILLNEED) <
+        0) {
+        D("[ Failed to fadvise: %d ]", errno);
+    }
+
     if (fd < 0 && errno == ENOENT) {
         if (!secure_mkdirs(android::base::Dirname(path))) {
             SendSyncFailErrno(s, "secure_mkdirs failed");
@@ -413,10 +419,14 @@ static bool do_recv(int s, const char* path, std::vector<char>& buffer) {
         return false;
     }
 
+    if (posix_fadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL | POSIX_FADV_NOREUSE) < 0) {
+        D("[ Failed to fadvise: %d ]", errno);
+    }
+
     syncmsg msg;
     msg.data.id = ID_DATA;
     while (true) {
-        int r = adb_read(fd, &buffer[0], buffer.size());
+        int r = adb_read(fd, &buffer[0], buffer.size() - sizeof(msg.data));
         if (r <= 0) {
             if (r == 0) break;
             SendSyncFailErrno(s, "read failed");

libasyncio/Android.bp

@@ -0,0 +1,44 @@
+//
+// Copyright (C) 2017 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.
+//
+
+libasyncio_cppflags = [
+    "-Wall",
+    "-Wextra",
+    "-Werror",
+]
+
+cc_library {
+    name: "libasyncio",
+    vendor_available: true,
+    host_supported: true,
+    srcs: [
+        "AsyncIO.cpp",
+    ],
+    cppflags: libasyncio_cppflags,
+
+    export_include_dirs: ["include"],
+    target: {
+        darwin: {
+            enabled: false,
+        },
+        linux_bionic: {
+            enabled: true,
+        },
+        windows: {
+            enabled: false,
+        },
+    },
+}

libasyncio/AsyncIO.cpp

@@ -0,0 +1,50 @@
+/*
+ * Copyright (C) 2016 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 <asyncio/AsyncIO.h>
+#include <sys/syscall.h>
+#include <unistd.h>
+
+int io_setup(unsigned nr, aio_context_t* ctxp) {
+    memset(ctxp, 0, sizeof(*ctxp));
+    return syscall(__NR_io_setup, nr, ctxp);
+}
+
+int io_destroy(aio_context_t ctx) {
+    return syscall(__NR_io_destroy, ctx);
+}
+
+int io_submit(aio_context_t ctx, long nr, iocb** iocbpp) {
+    return syscall(__NR_io_submit, ctx, nr, iocbpp);
+}
+
+int io_getevents(aio_context_t ctx, long min_nr, long max_nr, io_event* events, timespec* timeout) {
+    return syscall(__NR_io_getevents, ctx, min_nr, max_nr, events, timeout);
+}
+
+int io_cancel(aio_context_t ctx, iocb* iocbp, io_event* result) {
+    return syscall(__NR_io_cancel, ctx, iocbp, result);
+}
+
+void io_prep(iocb* iocb, int fd, const void* buf, uint64_t count, int64_t offset, bool read) {
+    memset(iocb, 0, sizeof(*iocb));
+    iocb->aio_fildes = fd;
+    iocb->aio_lio_opcode = read ? IOCB_CMD_PREAD : IOCB_CMD_PWRITE;
+    iocb->aio_reqprio = 0;
+    iocb->aio_buf = reinterpret_cast<uint64_t>(buf);
+    iocb->aio_nbytes = count;
+    iocb->aio_offset = offset;
+}

libasyncio/include/asyncio/AsyncIO.h

@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2016 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 _ASYNCIO_H
+#define _ASYNCIO_H
+
+#include <cstring>
+#include <cstdint>
+#include <linux/aio_abi.h>
+#include <sys/cdefs.h>
+#include <sys/types.h>
+#include <time.h>
+#include <unistd.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * Provides kernel aio operations.
+ */
+
+int io_setup(unsigned nr, aio_context_t* ctxp);
+int io_destroy(aio_context_t ctx);
+int io_submit(aio_context_t ctx, long nr, iocb** iocbpp);
+int io_getevents(aio_context_t ctx, long min_nr, long max_nr, io_event* events, timespec* timeout);
+int io_cancel(aio_context_t ctx, iocb*, io_event* result);
+void io_prep(iocb* iocb, int fd, const void* buf, uint64_t count, int64_t offset, bool read);
+
+#ifdef __cplusplus
+};
+#endif
+
+#endif  // ASYNCIO_H