tequilaOS/platform_system_core
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge "Incremental installations in adb, client/host side."

Browse source
This commit is contained in:
Yurii Zubrytskyi 2020-02-19 19:25:10 +00:00 committed by Gerrit Code Review
commit 2bca658a3b
13 changed files with 1108 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
adb/Android.bp
View file

@ -341,6 +341,8 @@ cc_binary_host {
"client/line_printer.cpp",
"client/fastdeploy.cpp",
"client/fastdeploycallbacks.cpp",
"client/incremental.cpp",
"client/incremental_server.cpp",
"shell_service_protocol.cpp",
],
@ -360,6 +362,7 @@ cc_binary_host {
"libfastdeploy_host",
"libdiagnose_usb",
"liblog",
"liblz4",
"libmdnssd",
"libprotobuf-cpp-lite",
"libusb",

32
adb/adb_trace.cpp
View file

@ -118,22 +118,22 @@ static void setup_trace_mask() {
return;
}
std::unordered_map<std::string, int> trace_flags = {
{"1", -1},
{"all", -1},
{"adb", ADB},
{"sockets", SOCKETS},
{"packets", PACKETS},
{"rwx", RWX},
{"usb", USB},
{"sync", SYNC},
{"sysdeps", SYSDEPS},
{"transport", TRANSPORT},
{"jdwp", JDWP},
{"services", SERVICES},
{"auth", AUTH},
{"fdevent", FDEVENT},
{"shell", SHELL}};
std::unordered_map<std::string, int> trace_flags = {{"1", -1},
{"all", -1},
{"adb", ADB},
{"sockets", SOCKETS},
{"packets", PACKETS},
{"rwx", RWX},
{"usb", USB},
{"sync", SYNC},
{"sysdeps", SYSDEPS},
{"transport", TRANSPORT},
{"jdwp", JDWP},
{"services", SERVICES},
{"auth", AUTH},
{"fdevent", FDEVENT},
{"shell", SHELL},
{"incremental", INCREMENTAL}};
std::vector<std::string> elements = android::base::Split(trace_setting, " ");
for (const auto& elem : elements) {

9
adb/adb_trace.h
View file

@ -25,19 +25,20 @@
* the adb_trace_init() function implemented in adb_trace.cpp.
*/
enum AdbTrace {
ADB = 0, /* 0x001 */
ADB = 0, /* 0x001 */
SOCKETS,
PACKETS,
TRANSPORT,
RWX, /* 0x010 */
RWX, /* 0x010 */
USB,
SYNC,
SYSDEPS,
JDWP, /* 0x100 */
JDWP, /* 0x100 */
SERVICES,
AUTH,
FDEVENT,
SHELL
SHELL,
INCREMENTAL,
};
#define VLOG_IS_ON(TAG) \

104
adb/client/adb_install.cpp
View file

@ -36,8 +36,10 @@
#include "client/file_sync_client.h"
#include "commandline.h"
#include "fastdeploy.h"
#include "incremental.h"
static constexpr int kFastDeployMinApi = 24;
static constexpr int kIncrementalMinApi = 29;
namespace {
@ -45,8 +47,8 @@ enum InstallMode {
INSTALL_DEFAULT,
INSTALL_PUSH,
INSTALL_STREAM,
INSTALL_INCREMENTAL,
};
}
static bool can_use_feature(const char* feature) {
@ -70,6 +72,10 @@ static bool is_apex_supported() {
return can_use_feature(kFeatureApex);
}
static bool is_abb_exec_supported() {
return can_use_feature(kFeatureAbbExec);
}
static int pm_command(int argc, const char** argv) {
std::string cmd = "pm";
@ -193,14 +199,14 @@ static int install_app_streamed(int argc, const char** argv, bool use_fastdeploy
posix_fadvise(local_fd.get(), 0, 0, POSIX_FADV_SEQUENTIAL | POSIX_FADV_NOREUSE);
#endif
const bool use_abb = can_use_feature(kFeatureAbbExec);
const bool use_abb_exec = is_abb_exec_supported();
std::string error;
std::vector<std::string> cmd_args = {use_abb ? "package" : "exec:cmd package"};
std::vector<std::string> cmd_args = {use_abb_exec ? "package" : "exec:cmd package"};
cmd_args.reserve(argc + 3);
// don't copy the APK name, but, copy the rest of the arguments as-is
while (argc-- > 1) {
if (use_abb) {
if (use_abb_exec) {
cmd_args.push_back(*argv++);
} else {
cmd_args.push_back(escape_arg(*argv++));
@ -217,7 +223,7 @@ static int install_app_streamed(int argc, const char** argv, bool use_fastdeploy
}
unique_fd remote_fd;
if (use_abb) {
if (use_abb_exec) {
remote_fd = send_abb_exec_command(cmd_args, &error);
} else {
remote_fd.reset(adb_connect(android::base::Join(cmd_args, " "), &error));
@ -287,8 +293,60 @@ static int install_app_legacy(int argc, const char** argv, bool use_fastdeploy)
return result;
}
template <class TimePoint>
static int msBetween(TimePoint start, TimePoint end) {
return std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
}
static int install_app_incremental(int argc, const char** argv) {
printf("Performing Incremental Install\n");
using clock = std::chrono::high_resolution_clock;
const auto start = clock::now();
int first_apk = -1;
int last_apk = -1;
std::string cert_path;
bool wait = false;
std::vector<std::string_view> args = {"package"};
for (int i = 0; i < argc; ++i) {
const auto arg = std::string_view(argv[i]);
if (android::base::EndsWithIgnoreCase(arg, ".apk")) {
last_apk = i;
if (first_apk == -1) {
first_apk = i;
}
} else if (arg == "--wait") {
wait = true;
} else if (arg.starts_with("install-")) {
// incremental installation command on the device is the same for all its variations in
// the adb, e.g. install-multiple or install-multi-package
args.push_back("install");
} else {
args.push_back(arg);
}
}
if (first_apk == -1) error_exit("Need at least one APK file on command line");
const auto afterApk = clock::now();
auto server_process = incremental::install({argv + first_apk, argv + last_apk + 1});
if (!server_process) {
return -1;
}
const auto end = clock::now();
printf("Install command complete (ms: %d total, %d apk prep, %d install)\n",
msBetween(start, end), msBetween(start, afterApk), msBetween(afterApk, end));
if (wait) {
(*server_process).wait();
}
return 0;
}
int install_app(int argc, const char** argv) {
std::vector<int> processedArgIndicies;
std::vector<int> processedArgIndices;
InstallMode installMode = INSTALL_DEFAULT;
bool use_fastdeploy = false;
bool is_reinstall = false;
@ -296,30 +354,42 @@ int install_app(int argc, const char** argv) {
for (int i = 1; i < argc; i++) {
if (!strcmp(argv[i], "--streaming")) {
processedArgIndicies.push_back(i);
processedArgIndices.push_back(i);
installMode = INSTALL_STREAM;
} else if (!strcmp(argv[i], "--no-streaming")) {
processedArgIndicies.push_back(i);
processedArgIndices.push_back(i);
installMode = INSTALL_PUSH;
} else if (!strcmp(argv[i], "-r")) {
// Note that this argument is not added to processedArgIndicies because it
// Note that this argument is not added to processedArgIndices because it
// must be passed through to pm
is_reinstall = true;
} else if (!strcmp(argv[i], "--fastdeploy")) {
processedArgIndicies.push_back(i);
processedArgIndices.push_back(i);
use_fastdeploy = true;
} else if (!strcmp(argv[i], "--no-fastdeploy")) {
processedArgIndicies.push_back(i);
processedArgIndices.push_back(i);
use_fastdeploy = false;
} else if (!strcmp(argv[i], "--force-agent")) {
processedArgIndicies.push_back(i);
processedArgIndices.push_back(i);
agent_update_strategy = FastDeploy_AgentUpdateAlways;
} else if (!strcmp(argv[i], "--date-check-agent")) {
processedArgIndicies.push_back(i);
processedArgIndices.push_back(i);
agent_update_strategy = FastDeploy_AgentUpdateNewerTimeStamp;
} else if (!strcmp(argv[i], "--version-check-agent")) {
processedArgIndicies.push_back(i);
processedArgIndices.push_back(i);
agent_update_strategy = FastDeploy_AgentUpdateDifferentVersion;
} else if (!strcmp(argv[i], "--incremental")) {
processedArgIndices.push_back(i);
installMode = INSTALL_INCREMENTAL;
} else if (!strcmp(argv[i], "--no-incremental")) {
processedArgIndices.push_back(i);
installMode = INSTALL_DEFAULT;
}
}
if (installMode == INSTALL_INCREMENTAL) {
if (get_device_api_level() < kIncrementalMinApi || !is_abb_exec_supported()) {
error_exit("Attempting to use incremental install on unsupported device");
}
}
@ -341,8 +411,8 @@ int install_app(int argc, const char** argv) {
std::vector<const char*> passthrough_argv;
for (int i = 0; i < argc; i++) {
if (std::find(processedArgIndicies.begin(), processedArgIndicies.end(), i) ==
processedArgIndicies.end()) {
if (std::find(processedArgIndices.begin(), processedArgIndices.end(), i) ==
processedArgIndices.end()) {
passthrough_argv.push_back(argv[i]);
}
}
@ -357,6 +427,8 @@ int install_app(int argc, const char** argv) {
case INSTALL_STREAM:
return install_app_streamed(passthrough_argv.size(), passthrough_argv.data(),
use_fastdeploy);
case INSTALL_INCREMENTAL:
return install_app_incremental(passthrough_argv.size(), passthrough_argv.data());
case INSTALL_DEFAULT:
default:
return 1;

13
adb/client/commandline.cpp
View file

@ -60,6 +60,7 @@
#include "client/file_sync_client.h"
#include "commandline.h"
#include "fastdeploy.h"
#include "incremental_server.h"
#include "services.h"
#include "shell_protocol.h"
#include "sysdeps/chrono.h"
@ -1959,6 +1960,18 @@ int adb_commandline(int argc, const char** argv) {
error_exit("usage: adb reconnect [device|offline]");
}
}
} else if (!strcmp(argv[0], "inc-server")) {
if (argc < 3) {
error_exit("usage: adb inc-server FD FILE1 FILE2 ...");
}
int fd = atoi(argv[1]);
if (fd < 3) {
// Disallow invalid FDs and stdin/out/err as well.
error_exit("Invalid fd number given: %d", fd);
}
fd = adb_register_socket(fd);
close_on_exec(fd);
return incremental::serve(fd, argc - 2, argv + 2);
}
error_exit("unknown command %s", argv[0]);

22
adb/client/fastdeploy.cpp
View file

@ -81,17 +81,21 @@ struct FileDeleter {
} // namespace
int get_device_api_level() {
REPORT_FUNC_TIME();
std::vector<char> sdk_version_output_buffer;
std::vector<char> sdk_version_error_buffer;
int api_level = -1;
static const int api_level = [] {
REPORT_FUNC_TIME();
std::vector<char> sdk_version_output_buffer;
std::vector<char> sdk_version_error_buffer;
int api_level = -1;
int statusCode = capture_shell_command("getprop ro.build.version.sdk",
&sdk_version_output_buffer, &sdk_version_error_buffer);
if (statusCode == 0 && sdk_version_output_buffer.size() > 0) {
api_level = strtol((char*)sdk_version_output_buffer.data(), NULL, 10);
}
int status_code =
capture_shell_command("getprop ro.build.version.sdk", &sdk_version_output_buffer,
&sdk_version_error_buffer);
if (status_code == 0 && sdk_version_output_buffer.size() > 0) {
api_level = strtol((char*)sdk_version_output_buffer.data(), nullptr, 10);
}
return api_level;
}();
return api_level;
}

213
adb/client/incremental.cpp Normal file
View file

@ -0,0 +1,213 @@
/*
* Copyright (C) 2020 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 "incremental.h"
#include <android-base/file.h>
#include <android-base/stringprintf.h>
#include <openssl/base64.h>
#include "adb_client.h"
#include "adb_io.h"
#include "adb_utils.h"
#include "commandline.h"
#include "sysdeps.h"
#ifndef _WIN32
#include <endian.h>
#else
#define be32toh(x) _byteswap_ulong(x)
#endif
using namespace std::literals;
namespace incremental {
namespace {
static constexpr auto IDSIG = ".idsig"sv;
using android::base::StringPrintf;
using Size = int64_t;
static inline int32_t read_int32(borrowed_fd fd) {
int32_t result;
ReadFully(fd, &result, sizeof(result));
return result;
}
static inline int32_t read_be_int32(borrowed_fd fd) {
return int32_t(be32toh(read_int32(fd)));
}
static inline void append_bytes_with_size(borrowed_fd fd, std::vector<char>* bytes) {
int32_t be_size = read_int32(fd);
int32_t size = int32_t(be32toh(be_size));
auto old_size = bytes->size();
bytes->resize(old_size + sizeof(be_size) + size);
memcpy(bytes->data() + old_size, &be_size, sizeof(be_size));
ReadFully(fd, bytes->data() + old_size + sizeof(be_size), size);
}
static inline std::pair<std::vector<char>, int32_t> read_id_sig_headers(borrowed_fd fd) {
std::vector<char> result;
append_bytes_with_size(fd, &result); // verityRootHash
append_bytes_with_size(fd, &result); // v3Digest
append_bytes_with_size(fd, &result); // pkcs7SignatureBlock
auto tree_size = read_be_int32(fd); // size of the verity tree
return {std::move(result), tree_size};
}
static inline Size verity_tree_size_for_file(Size fileSize) {
constexpr int INCFS_DATA_FILE_BLOCK_SIZE = 4096;
constexpr int SHA256_DIGEST_SIZE = 32;
constexpr int digest_size = SHA256_DIGEST_SIZE;
constexpr int hash_per_block = INCFS_DATA_FILE_BLOCK_SIZE / digest_size;
Size total_tree_block_count = 0;
auto block_count = 1 + (fileSize - 1) / INCFS_DATA_FILE_BLOCK_SIZE;
auto hash_block_count = block_count;
for (auto i = 0; hash_block_count > 1; i++) {
hash_block_count = (hash_block_count + hash_per_block - 1) / hash_per_block;
total_tree_block_count += hash_block_count;
}
return total_tree_block_count * INCFS_DATA_FILE_BLOCK_SIZE;
}
// Base64-encode signature bytes. Keeping fd at the position of start of verity tree.
static std::pair<unique_fd, std::string> read_and_encode_signature(Size file_size,
std::string signature_file) {
signature_file += IDSIG;
struct stat st;
if (stat(signature_file.c_str(), &st)) {
fprintf(stderr, "Failed to stat signature file %s. Abort.\n", signature_file.c_str());
return {};
}
unique_fd fd(adb_open(signature_file.c_str(), O_RDONLY | O_CLOEXEC));
if (fd < 0) {
fprintf(stderr, "Failed to open signature file: %s. Abort.\n", signature_file.c_str());
return {};
}
auto [signature, tree_size] = read_id_sig_headers(fd);
if (auto expected = verity_tree_size_for_file(file_size); tree_size != expected) {
fprintf(stderr,
"Verity tree size mismatch in signature file: %s [was %lld, expected %lld].\n",
signature_file.c_str(), (long long)tree_size, (long long)expected);
return {};
}
size_t base64_len = 0;
if (!EVP_EncodedLength(&base64_len, signature.size())) {
fprintf(stderr, "Fail to estimate base64 encoded length. Abort.\n");
return {};
}
std::string encoded_signature;
encoded_signature.resize(base64_len);
encoded_signature.resize(EVP_EncodeBlock((uint8_t*)encoded_signature.data(),
(const uint8_t*)signature.data(), signature.size()));
return {std::move(fd), std::move(encoded_signature)};
}
// Send install-incremental to the device along with properly configured file descriptors in
// streaming format. Once connection established, send all fs-verity tree bytes.
static unique_fd start_install(const std::vector<std::string>& files) {
std::vector<std::string> command_args{"package", "install-incremental"};
// fd's with positions at the beginning of fs-verity
std::vector<unique_fd> signature_fds;
signature_fds.reserve(files.size());
for (int i = 0, size = files.size(); i < size; ++i) {
const auto& file = files[i];
struct stat st;
if (stat(file.c_str(), &st)) {
fprintf(stderr, "Failed to stat input file %s. Abort.\n", file.c_str());
return {};
}
auto [signature_fd, signature] = read_and_encode_signature(st.st_size, file);
if (!signature_fd.ok()) {
return {};
}
auto file_desc =
StringPrintf("%s:%lld:%s:%s", android::base::Basename(file).c_str(),
(long long)st.st_size, std::to_string(i).c_str(), signature.c_str());
command_args.push_back(std::move(file_desc));
signature_fds.push_back(std::move(signature_fd));
}
std::string error;
auto connection_fd = unique_fd(send_abb_exec_command(command_args, &error));
if (connection_fd < 0) {
fprintf(stderr, "Failed to run: %s, error: %s\n",
android::base::Join(command_args, " ").c_str(), error.c_str());
return {};
}
// Pushing verity trees for all installation files.
for (auto&& local_fd : signature_fds) {
if (!copy_to_file(local_fd.get(), connection_fd.get())) {
fprintf(stderr, "Failed to stream tree bytes: %s. Abort.\n", strerror(errno));
return {};
}
}
return connection_fd;
}
} // namespace
std::optional<Process> install(std::vector<std::string> files) {
auto connection_fd = start_install(files);
if (connection_fd < 0) {
fprintf(stderr, "adb: failed to initiate installation on device.\n");
return {};
}
std::string adb_path = android::base::GetExecutablePath();
auto osh = adb_get_os_handle(connection_fd.get());
#ifdef _WIN32
auto fd_param = std::to_string(reinterpret_cast<intptr_t>(osh));
#else /* !_WIN32 a.k.a. Unix */
auto fd_param = std::to_string(osh);
#endif
std::vector<std::string> args(std::move(files));
args.insert(args.begin(), {"inc-server", fd_param});
auto child = adb_launch_process(adb_path, std::move(args), {connection_fd.get()});
if (!child) {
fprintf(stderr, "adb: failed to fork: %s\n", strerror(errno));
return {};
}
auto killOnExit = [](Process* p) { p->kill(); };
std::unique_ptr<Process, decltype(killOnExit)> serverKiller(&child, killOnExit);
// TODO: Terminate server process if installation fails.
serverKiller.release();
return child;
}
} // namespace incremental

30
adb/client/incremental.h Normal file
View file

@ -0,0 +1,30 @@
/*
* Copyright (C) 2020 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.
*/
#pragma once
#include "adb_unique_fd.h"
#include <optional>
#include <string>
#include "sysdeps.h"
namespace incremental {
std::optional<Process> install(std::vector<std::string> files);
} // namespace incremental

545
adb/client/incremental_server.cpp Normal file
View file

@ -0,0 +1,545 @@
/*
* Copyright (C) 2020 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.
*/
#define TRACE_TAG INCREMENTAL
#include "incremental_server.h"
#include "adb.h"
#include "adb_io.h"
#include "adb_trace.h"
#include "adb_unique_fd.h"
#include "adb_utils.h"
#include "sysdeps.h"
#ifndef _WIN32
#include <endian.h>
#else
#define be64toh(x) _byteswap_uint64(x)
#define be32toh(x) _byteswap_ulong(x)
#define be16toh(x) _byteswap_ushort(x)
#define htobe64(x) _byteswap_uint64(x)
#define htobe32(x) _byteswap_ulong(x)
#define htobe16(x) _byteswap_ushort(x)
#endif
#include <android-base/strings.h>
#include <lz4.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <array>
#include <deque>
#include <fstream>
#include <thread>
#include <type_traits>
#include <unordered_set>
namespace incremental {
static constexpr int kBlockSize = 4096;
static constexpr int kCompressedSizeMax = kBlockSize * 0.95;
static constexpr short kCompressionNone = 0;
static constexpr short kCompressionLZ4 = 1;
static constexpr int kCompressBound = std::max(kBlockSize, LZ4_COMPRESSBOUND(kBlockSize));
static constexpr auto kReadBufferSize = 128 * 1024;
using BlockSize = int16_t;
using FileId = int16_t;
using BlockIdx = int32_t;
using NumBlocks = int32_t;
using CompressionType = int16_t;
using RequestType = int16_t;
using ChunkHeader = int32_t;
using MagicType = uint32_t;
static constexpr MagicType INCR = 0x494e4352; // LE INCR
static constexpr RequestType EXIT = 0;
static constexpr RequestType BLOCK_MISSING = 1;
static constexpr RequestType PREFETCH = 2;
static constexpr inline int64_t roundDownToBlockOffset(int64_t val) {
return val & ~(kBlockSize - 1);
}
static constexpr inline int64_t roundUpToBlockOffset(int64_t val) {
return roundDownToBlockOffset(val + kBlockSize - 1);
}
static constexpr inline NumBlocks numBytesToNumBlocks(int64_t bytes) {
return roundUpToBlockOffset(bytes) / kBlockSize;
}
static constexpr inline off64_t blockIndexToOffset(BlockIdx blockIdx) {
return static_cast<off64_t>(blockIdx) * kBlockSize;
}
template <typename T>
static inline constexpr T toBigEndian(T t) {
using unsigned_type = std::make_unsigned_t<T>;
if constexpr (std::is_same_v<T, int16_t>) {
return htobe16(static_cast<unsigned_type>(t));
} else if constexpr (std::is_same_v<T, int32_t>) {
return htobe32(static_cast<unsigned_type>(t));
} else if constexpr (std::is_same_v<T, int64_t>) {
return htobe64(static_cast<unsigned_type>(t));
} else {
return t;
}
}
template <typename T>
static inline constexpr T readBigEndian(void* data) {
using unsigned_type = std::make_unsigned_t<T>;
if constexpr (std::is_same_v<T, int16_t>) {
return static_cast<T>(be16toh(*reinterpret_cast<unsigned_type*>(data)));
} else if constexpr (std::is_same_v<T, int32_t>) {
return static_cast<T>(be32toh(*reinterpret_cast<unsigned_type*>(data)));
} else if constexpr (std::is_same_v<T, int64_t>) {
return static_cast<T>(be64toh(*reinterpret_cast<unsigned_type*>(data)));
} else {
return T();
}
}
// Received from device
// !Does not include magic!
struct RequestCommand {
RequestType request_type; // 2 bytes
FileId file_id; // 2 bytes
union {
BlockIdx block_idx;
NumBlocks num_blocks;
}; // 4 bytes
} __attribute__((packed));
// Placed before actual data bytes of each block
struct ResponseHeader {
FileId file_id; // 2 bytes
CompressionType compression_type; // 2 bytes
BlockIdx block_idx; // 4 bytes
BlockSize block_size; // 2 bytes
} __attribute__((packed));
// Holds streaming state for a file
class File {
public:
// Plain file
File(const char* filepath, FileId id, int64_t size, unique_fd fd) : File(filepath, id, size) {
this->fd_ = std::move(fd);
}
int64_t ReadBlock(BlockIdx block_idx, void* buf, bool* is_zip_compressed,
std::string* error) const {
char* buf_ptr = static_cast<char*>(buf);
int64_t bytes_read = -1;
const off64_t offsetStart = blockIndexToOffset(block_idx);
bytes_read = adb_pread(fd_, &buf_ptr[sizeof(ResponseHeader)], kBlockSize, offsetStart);
return bytes_read;
}
const unique_fd& RawFd() const { return fd_; }
std::vector<bool> sentBlocks;
NumBlocks sentBlocksCount;
const char* const filepath;
const FileId id;
const int64_t size;
private:
File(const char* filepath, FileId id, int64_t size) : filepath(filepath), id(id), size(size) {
sentBlocks.resize(numBytesToNumBlocks(size));
}
unique_fd fd_;
};
class IncrementalServer {
public:
IncrementalServer(unique_fd fd, std::vector<File> files)
: adb_fd_(std::move(fd)), files_(std::move(files)) {
buffer_.reserve(kReadBufferSize);
}
bool Serve();
private:
struct PrefetchState {
const File* file;
BlockIdx overallIndex = 0;
BlockIdx overallEnd = 0;
PrefetchState(const File& f) : file(&f), overallEnd((BlockIdx)f.sentBlocks.size()) {}
PrefetchState(const File& f, BlockIdx start, int count)
: file(&f),
overallIndex(start),
overallEnd(std::min<BlockIdx>(start + count, f.sentBlocks.size())) {}
bool done() const { return overallIndex >= overallEnd; }
};
bool SkipToRequest(void* buffer, size_t* size, bool blocking);
std::optional<RequestCommand> ReadRequest(bool blocking);
void erase_buffer_head(int count) { buffer_.erase(buffer_.begin(), buffer_.begin() + count); }
enum class SendResult { Sent, Skipped, Error };
SendResult SendBlock(FileId fileId, BlockIdx blockIdx, bool flush = false);
bool SendDone();
void RunPrefetching();
void Send(const void* data, size_t size, bool flush);
void Flush();
using TimePoint = decltype(std::chrono::high_resolution_clock::now());
bool Exit(std::optional<TimePoint> startTime, int missesCount, int missesSent);
unique_fd const adb_fd_;
std::vector<File> files_;
// Incoming data buffer.
std::vector<char> buffer_;
std::deque<PrefetchState> prefetches_;
int compressed_ = 0, uncompressed_ = 0;
long long sentSize_ = 0;
std::vector<char> pendingBlocks_;
};
bool IncrementalServer::SkipToRequest(void* buffer, size_t* size, bool blocking) {
while (true) {
// Looking for INCR magic.
bool magic_found = false;
int bcur = 0;
for (int bsize = buffer_.size(); bcur + 4 < bsize; ++bcur) {
uint32_t magic = be32toh(*(uint32_t*)(buffer_.data() + bcur));
if (magic == INCR) {
magic_found = true;
break;
}
}
if (bcur > 0) {
// Stream the rest to stderr.
fprintf(stderr, "%.*s", bcur, buffer_.data());
erase_buffer_head(bcur);
}
if (magic_found && buffer_.size() >= *size + sizeof(INCR)) {
// fine, return
memcpy(buffer, buffer_.data() + sizeof(INCR), *size);
erase_buffer_head(*size + sizeof(INCR));
return true;
}
adb_pollfd pfd = {adb_fd_.get(), POLLIN, 0};
auto res = adb_poll(&pfd, 1, blocking ? -1 : 0);
if (res != 1) {
if (res < 0) {
fprintf(stderr, "Failed to poll: %s\n", strerror(errno));
return false;
}
*size = 0;
return true;
}
auto bsize = buffer_.size();
buffer_.resize(kReadBufferSize);
int r = adb_read(adb_fd_, buffer_.data() + bsize, kReadBufferSize - bsize);
if (r > 0) {
buffer_.resize(bsize + r);
continue;
}
if (r == -1) {
fprintf(stderr, "Failed to read from fd %d: %d. Exit\n", adb_fd_.get(), errno);
return true;
}
// socket is closed
return false;
}
}
std::optional<RequestCommand> IncrementalServer::ReadRequest(bool blocking) {
uint8_t commandBuf[sizeof(RequestCommand)];
auto size = sizeof(commandBuf);
if (!SkipToRequest(&commandBuf, &size, blocking)) {
return {{EXIT}};
}
if (size < sizeof(RequestCommand)) {
return {};
}
RequestCommand request;
request.request_type = readBigEndian<RequestType>(&commandBuf[0]);
request.file_id = readBigEndian<FileId>(&commandBuf[2]);
request.block_idx = readBigEndian<BlockIdx>(&commandBuf[4]);
return request;
}
auto IncrementalServer::SendBlock(FileId fileId, BlockIdx blockIdx, bool flush) -> SendResult {
auto& file = files_[fileId];
if (blockIdx >= static_cast<long>(file.sentBlocks.size())) {
fprintf(stderr, "Failed to read file %s at block %" PRId32 " (past end).\n", file.filepath,
blockIdx);
return SendResult::Error;
}
if (file.sentBlocks[blockIdx]) {
return SendResult::Skipped;
}
std::string error;
char raw[sizeof(ResponseHeader) + kBlockSize];
bool isZipCompressed = false;
const int64_t bytesRead = file.ReadBlock(blockIdx, &raw, &isZipCompressed, &error);
if (bytesRead < 0) {
fprintf(stderr, "Failed to get data for %s at blockIdx=%d (%s).\n", file.filepath, blockIdx,
error.c_str());
return SendResult::Error;
}
ResponseHeader* header = nullptr;
char data[sizeof(ResponseHeader) + kCompressBound];
char* compressed = data + sizeof(*header);
int16_t compressedSize = 0;
if (!isZipCompressed) {
compressedSize =
LZ4_compress_default(raw + sizeof(*header), compressed, bytesRead, kCompressBound);
}
int16_t blockSize;
if (compressedSize > 0 && compressedSize < kCompressedSizeMax) {
++compressed_;
blockSize = compressedSize;
header = reinterpret_cast<ResponseHeader*>(data);
header->compression_type = toBigEndian(kCompressionLZ4);
} else {
++uncompressed_;
blockSize = bytesRead;
header = reinterpret_cast<ResponseHeader*>(raw);
header->compression_type = toBigEndian(kCompressionNone);
}
header->file_id = toBigEndian(fileId);
header->block_size = toBigEndian(blockSize);
header->block_idx = toBigEndian(blockIdx);
file.sentBlocks[blockIdx] = true;
file.sentBlocksCount += 1;
Send(header, sizeof(*header) + blockSize, flush);
return SendResult::Sent;
}
bool IncrementalServer::SendDone() {
ResponseHeader header;
header.file_id = -1;
header.compression_type = 0;
header.block_idx = 0;
header.block_size = 0;
Send(&header, sizeof(header), true);
return true;
}
void IncrementalServer::RunPrefetching() {
constexpr auto kPrefetchBlocksPerIteration = 128;
int blocksToSend = kPrefetchBlocksPerIteration;
while (!prefetches_.empty() && blocksToSend > 0) {
auto& prefetch = prefetches_.front();
const auto& file = *prefetch.file;
for (auto& i = prefetch.overallIndex; blocksToSend > 0 && i < prefetch.overallEnd; ++i) {
if (auto res = SendBlock(file.id, i); res == SendResult::Sent) {
--blocksToSend;
} else if (res == SendResult::Error) {
fprintf(stderr, "Failed to send block %" PRId32 "\n", i);
}
}
if (prefetch.done()) {
prefetches_.pop_front();
}
}
}
void IncrementalServer::Send(const void* data, size_t size, bool flush) {
constexpr auto kChunkFlushSize = 31 * kBlockSize;
if (pendingBlocks_.empty()) {
pendingBlocks_.resize(sizeof(ChunkHeader));
}
pendingBlocks_.insert(pendingBlocks_.end(), static_cast<const char*>(data),
static_cast<const char*>(data) + size);
if (flush || pendingBlocks_.size() > kChunkFlushSize) {
Flush();
}
}
void IncrementalServer::Flush() {
if (pendingBlocks_.empty()) {
return;
}
*(ChunkHeader*)pendingBlocks_.data() =
toBigEndian<int32_t>(pendingBlocks_.size() - sizeof(ChunkHeader));
if (!WriteFdExactly(adb_fd_, pendingBlocks_.data(), pendingBlocks_.size())) {
fprintf(stderr, "Failed to write %d bytes\n", int(pendingBlocks_.size()));
}
sentSize_ += pendingBlocks_.size();
pendingBlocks_.clear();
}
bool IncrementalServer::Exit(std::optional<TimePoint> startTime, int missesCount, int missesSent) {
using namespace std::chrono;
auto endTime = high_resolution_clock::now();
fprintf(stderr,
"Connection failed or received exit command. Exit.\n"
"Misses: %d, of those unique: %d; sent compressed: %d, uncompressed: "
"%d, mb: %.3f\n"
"Total time taken: %.3fms\n",
missesCount, missesSent, compressed_, uncompressed_, sentSize_ / 1024.0 / 1024.0,
duration_cast<microseconds>(endTime - (startTime ? *startTime : endTime)).count() /
1000.0);
return true;
}
bool IncrementalServer::Serve() {
// Initial handshake to verify connection is still alive
if (!SendOkay(adb_fd_)) {
fprintf(stderr, "Connection is dead. Abort.\n");
return false;
}
std::unordered_set<FileId> prefetchedFiles;
bool doneSent = false;
int missesCount = 0;
int missesSent = 0;
using namespace std::chrono;
std::optional<TimePoint> startTime;
while (true) {
if (!doneSent && prefetches_.empty() &&
std::all_of(files_.begin(), files_.end(), [](const File& f) {
return f.sentBlocksCount == NumBlocks(f.sentBlocks.size());
})) {
fprintf(stdout, "All files should be loaded. Notifying the device.\n");
SendDone();
doneSent = true;
}
const bool blocking = prefetches_.empty();
if (blocking) {
// We've no idea how long the blocking call is, so let's flush whatever is still unsent.
Flush();
}
auto request = ReadRequest(blocking);
if (!startTime) {
startTime = high_resolution_clock::now();
}
if (request) {
FileId fileId = request->file_id;
BlockIdx blockIdx = request->block_idx;
switch (request->request_type) {
case EXIT: {
// Stop everything.
return Exit(startTime, missesCount, missesSent);
}
case BLOCK_MISSING: {
++missesCount;
// Sends one single block ASAP.
if (fileId < 0 || fileId >= (FileId)files_.size() || blockIdx < 0 ||
blockIdx >= (BlockIdx)files_[fileId].sentBlocks.size()) {
fprintf(stderr,
"Received invalid data request for file_id %" PRId16
" block_idx %" PRId32 ".\n",
fileId, blockIdx);
break;
}
// fprintf(stderr, "\treading file %d block %04d\n", (int)fileId,
// (int)blockIdx);
if (auto res = SendBlock(fileId, blockIdx, true); res == SendResult::Error) {
fprintf(stderr, "Failed to send block %" PRId32 ".\n", blockIdx);
} else if (res == SendResult::Sent) {
++missesSent;
// Make sure we send more pages from this place onward, in case if the OS is
// reading a bigger block.
prefetches_.emplace_front(files_[fileId], blockIdx + 1, 7);
}
break;
}
case PREFETCH: {
// Start prefetching for a file
if (fileId < 0) {
fprintf(stderr,
"Received invalid prefetch request for file_id %" PRId16 "\n",
fileId);
break;
}
if (!prefetchedFiles.insert(fileId).second) {
fprintf(stderr,
"Received duplicate prefetch request for file_id %" PRId16 "\n",
fileId);
break;
}
D("Received prefetch request for file_id %" PRId16 ".\n", fileId);
prefetches_.emplace_back(files_[fileId]);
break;
}
default:
fprintf(stderr, "Invalid request %" PRId16 ",%" PRId16 ",%" PRId32 ".\n",
request->request_type, fileId, blockIdx);
break;
}
}
RunPrefetching();
}
}
bool serve(int adb_fd, int argc, const char** argv) {
auto connection_fd = unique_fd(adb_fd);
if (argc <= 0) {
error_exit("inc-server: must specify at least one file.");
}
std::vector<File> files;
files.reserve(argc);
for (int i = 0; i < argc; ++i) {
auto filepath = argv[i];
struct stat st;
if (stat(filepath, &st)) {
fprintf(stderr, "Failed to stat input file %s. Abort.\n", filepath);
return {};
}
unique_fd fd(adb_open(filepath, O_RDONLY));
if (fd < 0) {
error_exit("inc-server: failed to open file '%s'.", filepath);
}
files.emplace_back(filepath, i, st.st_size, std::move(fd));
}
IncrementalServer server(std::move(connection_fd), std::move(files));
printf("Serving...\n");
fclose(stdin);
fclose(stdout);
return server.Serve();
}
} // namespace incremental

26
adb/client/incremental_server.h Normal file
View file

@ -0,0 +1,26 @@
/*
* Copyright (C) 2020 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.
*/
#pragma once
namespace incremental {
// Expecting arguments like:
// {FILE1 FILE2 ...}
// Where FILE* are files to serve.
bool serve(int adbFd, int argc, const char** argv);
} // namespace incremental

65
adb/sysdeps.h
View file

@ -267,6 +267,39 @@ inline void seekdir(DIR*, long) {
#define getcwd adb_getcwd
// A very simple wrapper over a launched child process
class Process {
public:
constexpr explicit Process(HANDLE h = nullptr) : h_(h) {}
~Process() { close(); }
constexpr explicit operator bool() const { return h_ != nullptr; }
void wait() {
if (*this) {
::WaitForSingleObject(h_, INFINITE);
close();
}
}
void kill() {
if (*this) {
::TerminateProcess(h_, -1);
}
}
private:
void close() {
if (*this) {
::CloseHandle(h_);
h_ = nullptr;
}
}
HANDLE h_;
};
Process adb_launch_process(std::string_view executable, std::vector<std::string> args,
std::initializer_list<int> fds_to_inherit = {});
// Helper class to convert UTF-16 argv from wmain() to UTF-8 args that can be
// passed to main().
class NarrowArgs {
@ -432,11 +465,11 @@ static __inline__ int adb_read(borrowed_fd fd, void* buf, size_t len) {
return TEMP_FAILURE_RETRY(read(fd.get(), buf, len));
}
static __inline__ int adb_pread(int fd, void* buf, size_t len, off64_t offset) {
static __inline__ int adb_pread(borrowed_fd fd, void* buf, size_t len, off64_t offset) {
#if defined(__APPLE__)
return TEMP_FAILURE_RETRY(pread(fd, buf, len, offset));
return TEMP_FAILURE_RETRY(pread(fd.get(), buf, len, offset));
#else
return TEMP_FAILURE_RETRY(pread64(fd, buf, len, offset));
return TEMP_FAILURE_RETRY(pread64(fd.get(), buf, len, offset));
#endif
}
@ -612,6 +645,32 @@ static __inline__ int adb_get_os_handle(borrowed_fd fd) {
return fd.get();
}
// A very simple wrapper over a launched child process
class Process {
public:
constexpr explicit Process(pid_t pid) : pid_(pid) {}
constexpr explicit operator bool() const { return pid_ >= 0; }
void wait() {
if (*this) {
int status;
::waitpid(pid_, &status, 0);
pid_ = -1;
}
}
void kill() {
if (*this) {
::kill(pid_, SIGTERM);
}
}
private:
pid_t pid_;
};
Process adb_launch_process(std::string_view executable, std::vector<std::string> args,
std::initializer_list<int> fds_to_inherit = {});
#endif /* !_WIN32 */
static inline void disable_tcp_nagle(borrowed_fd fd) {

34
adb/sysdeps_unix.cpp
View file

@ -56,3 +56,37 @@ bool set_tcp_keepalive(borrowed_fd fd, int interval_sec) {
return true;
}
static __inline__ void disable_close_on_exec(borrowed_fd fd) {
const auto oldFlags = fcntl(fd.get(), F_GETFD);
const auto newFlags = (oldFlags & ~FD_CLOEXEC);
if (newFlags != oldFlags) {
fcntl(fd.get(), F_SETFD, newFlags);
}
}
Process adb_launch_process(std::string_view executable, std::vector<std::string> args,
std::initializer_list<int> fds_to_inherit) {
const auto pid = fork();
if (pid != 0) {
// parent, includes the case when failed to fork()
return Process(pid);
}
// child
std::vector<std::string> copies;
copies.reserve(args.size() + 1);
copies.emplace_back(executable);
copies.insert(copies.end(), std::make_move_iterator(args.begin()),
std::make_move_iterator(args.end()));
std::vector<char*> rawArgs;
rawArgs.reserve(copies.size() + 1);
for (auto&& str : copies) {
rawArgs.push_back(str.data());
}
rawArgs.push_back(nullptr);
for (auto fd : fds_to_inherit) {
disable_close_on_exec(fd);
}
exit(execv(copies.front().data(), rawArgs.data()));
}

60
adb/sysdeps_win32.cpp
View file

@ -2771,6 +2771,66 @@ char* adb_getcwd(char* buf, int size) {
return buf;
}
void enable_inherit(borrowed_fd fd) {
auto osh = adb_get_os_handle(fd);
const auto h = reinterpret_cast<HANDLE>(osh);
::SetHandleInformation(h, HANDLE_FLAG_INHERIT, HANDLE_FLAG_INHERIT);
}
void disable_inherit(borrowed_fd fd) {
auto osh = adb_get_os_handle(fd);
const auto h = reinterpret_cast<HANDLE>(osh);
::SetHandleInformation(h, HANDLE_FLAG_INHERIT, 0);
}
Process adb_launch_process(std::string_view executable, std::vector<std::string> args,
std::initializer_list<int> fds_to_inherit) {
std::wstring wexe;
if (!android::base::UTF8ToWide(executable.data(), executable.size(), &wexe)) {
return Process();
}
std::wstring wargs = L"\"" + wexe + L"\"";
std::wstring warg;
for (auto arg : args) {
warg.clear();
if (!android::base::UTF8ToWide(arg.data(), arg.size(), &warg)) {
return Process();
}
wargs += L" \"";
wargs += warg;
wargs += L'\"';
}
STARTUPINFOW sinfo = {sizeof(sinfo)};
PROCESS_INFORMATION pinfo = {};
// TODO: use the Vista+ API to pass the list of inherited handles explicitly;
// see http://blogs.msdn.com/b/oldnewthing/archive/2011/12/16/10248328.aspx
for (auto fd : fds_to_inherit) {
enable_inherit(fd);
}
const auto created = CreateProcessW(wexe.c_str(), wargs.data(),
nullptr, // process attributes
nullptr, // thread attributes
fds_to_inherit.size() > 0, // inherit any handles?
0, // flags
nullptr, // environment
nullptr, // current directory
&sinfo, // startup info
&pinfo);
for (auto fd : fds_to_inherit) {
disable_inherit(fd);
}
if (!created) {
return Process();
}
::CloseHandle(pinfo.hThread);
return Process(pinfo.hProcess);
}
// The SetThreadDescription API was brought in version 1607 of Windows 10.
typedef HRESULT(WINAPI* SetThreadDescription)(HANDLE hThread, PCWSTR lpThreadDescription);