ACPR/android_system_core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge "adb: defer acknowledgement of pushed files until the end."

Browse source
This commit is contained in:
Josh Gao 2020-03-10 01:55:51 +00:00 • committed by Gerrit Code Review
commit 2263bcb4d0
1 changed files with 90 additions and 98 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

188
adb/client/file_sync_client.cpp
View file

@ -29,6 +29,7 @@
#include <utime.h> #include <utime.h>
#include <chrono> #include <chrono>
#include <deque>
#include <functional> #include <functional>
#include <memory> #include <memory>
#include <sstream> #include <sstream>
@ -203,7 +204,7 @@ struct TransferLedger {
class SyncConnection { class SyncConnection {
public: public:
SyncConnection() : expect_done_(false) { SyncConnection() {
max = SYNC_DATA_MAX; // TODO: decide at runtime. max = SYNC_DATA_MAX; // TODO: decide at runtime.
std::string error; std::string error;
@ -239,16 +240,6 @@ class SyncConnection {
bool IsValid() { return fd >= 0; } bool IsValid() { return fd >= 0; }
bool ReceivedError(const char* from, const char* to) {
adb_pollfd pfd = {.fd = fd.get(), .events = POLLIN};
int rc = adb_poll(&pfd, 1, 0);
if (rc < 0) {
Error("failed to poll: %s", strerror(errno));
return true;
}
return rc != 0;
}
void NewTransfer() { void NewTransfer() {
current_ledger_.Reset(); current_ledger_.Reset();
} }
@ -258,6 +249,11 @@ class SyncConnection {
global_ledger_.bytes_transferred += bytes; global_ledger_.bytes_transferred += bytes;
} }
void RecordFileSent(std::string from, std::string to) {
RecordFilesTransferred(1);
deferred_acknowledgements_.emplace_back(std::move(from), std::move(to));
}
void RecordFilesTransferred(size_t files) { void RecordFilesTransferred(size_t files) {
current_ledger_.files_transferred += files; current_ledger_.files_transferred += files;
global_ledger_.files_transferred += files; global_ledger_.files_transferred += files;
@ -283,39 +279,38 @@ class SyncConnection {
} }
} }
bool SendRequest(int id, const char* path_and_mode) { bool SendRequest(int id, const std::string& path) {
size_t path_length = strlen(path_and_mode); if (path.length() > 1024) {
if (path_length > 1024) { Error("SendRequest failed: path too long: %zu", path.length());
Error("SendRequest failed: path too long: %zu", path_length);
errno = ENAMETOOLONG; errno = ENAMETOOLONG;
return false; return false;
} }
// Sending header and payload in a single write makes a noticeable // Sending header and payload in a single write makes a noticeable
// difference to "adb sync" performance. // difference to "adb sync" performance.
std::vector<char> buf(sizeof(SyncRequest) + path_length); std::vector<char> buf(sizeof(SyncRequest) + path.length());
SyncRequest* req = reinterpret_cast<SyncRequest*>(&buf[0]); SyncRequest* req = reinterpret_cast<SyncRequest*>(&buf[0]);
req->id = id; req->id = id;
req->path_length = path_length; req->path_length = path.length();
char* data = reinterpret_cast<char*>(req + 1); char* data = reinterpret_cast<char*>(req + 1);
memcpy(data, path_and_mode, path_length); memcpy(data, path.data(), path.length());
return WriteFdExactly(fd, &buf[0], buf.size()); return WriteFdExactly(fd, buf.data(), buf.size());
} }
bool SendStat(const char* path_and_mode) { bool SendStat(const std::string& path) {
if (!have_stat_v2_) { if (!have_stat_v2_) {
errno = ENOTSUP; errno = ENOTSUP;
return false; return false;
} }
return SendRequest(ID_STAT_V2, path_and_mode); return SendRequest(ID_STAT_V2, path);
} }
bool SendLstat(const char* path_and_mode) { bool SendLstat(const std::string& path) {
if (have_stat_v2_) { if (have_stat_v2_) {
return SendRequest(ID_LSTAT_V2, path_and_mode); return SendRequest(ID_LSTAT_V2, path);
} else { } else {
return SendRequest(ID_LSTAT_V1, path_and_mode); return SendRequest(ID_LSTAT_V1, path);
} }
} }
@ -374,7 +369,7 @@ class SyncConnection {
return true; return true;
} }
bool SendLs(const char* path) { bool SendLs(const std::string& path) {
return SendRequest(have_ls_v2_ ? ID_LIST_V2 : ID_LIST_V1, path); return SendRequest(have_ls_v2_ ? ID_LIST_V2 : ID_LIST_V1, path);
} }
@ -415,28 +410,26 @@ class SyncConnection {
// Sending header, payload, and footer in a single write makes a huge // Sending header, payload, and footer in a single write makes a huge
// difference to "adb sync" performance. // difference to "adb sync" performance.
bool SendSmallFile(const char* path_and_mode, bool SendSmallFile(const std::string& path, mode_t mode, const std::string& lpath,
const char* lpath, const char* rpath, const std::string& rpath, unsigned mtime, const char* data,
unsigned mtime, size_t data_length) {
const char* data, size_t data_length) { std::string path_and_mode = android::base::StringPrintf("%s,%d", path.c_str(), mode);
size_t path_length = strlen(path_and_mode); if (path_and_mode.length() > 1024) {
if (path_length > 1024) { Error("SendSmallFile failed: path too long: %zu", path_and_mode.length());
Error("SendSmallFile failed: path too long: %zu", path_length);
errno = ENAMETOOLONG; errno = ENAMETOOLONG;
return false; return false;
} }
std::vector<char> buf(sizeof(SyncRequest) + path_length + std::vector<char> buf(sizeof(SyncRequest) + path_and_mode.length() + sizeof(SyncRequest) +
sizeof(SyncRequest) + data_length + data_length + sizeof(SyncRequest));
sizeof(SyncRequest));
char* p = &buf[0]; char* p = &buf[0];
SyncRequest* req_send = reinterpret_cast<SyncRequest*>(p); SyncRequest* req_send = reinterpret_cast<SyncRequest*>(p);
req_send->id = ID_SEND; req_send->id = ID_SEND;
req_send->path_length = path_length; req_send->path_length = path_and_mode.length();
p += sizeof(SyncRequest); p += sizeof(SyncRequest);
memcpy(p, path_and_mode, path_length); memcpy(p, path_and_mode.data(), path_and_mode.size());
p += path_length; p += path_and_mode.length();
SyncRequest* req_data = reinterpret_cast<SyncRequest*>(p); SyncRequest* req_data = reinterpret_cast<SyncRequest*>(p);
req_data->id = ID_DATA; req_data->id = ID_DATA;
@ -451,34 +444,34 @@ class SyncConnection {
p += sizeof(SyncRequest); p += sizeof(SyncRequest);
WriteOrDie(lpath, rpath, &buf[0], (p - &buf[0])); WriteOrDie(lpath, rpath, &buf[0], (p - &buf[0]));
expect_done_ = true;
// RecordFilesTransferred gets called in CopyDone. RecordFileSent(lpath, rpath);
RecordBytesTransferred(data_length); RecordBytesTransferred(data_length);
ReportProgress(rpath, data_length, data_length); ReportProgress(rpath, data_length, data_length);
return true; return true;
} }
bool SendLargeFile(const char* path_and_mode, bool SendLargeFile(const std::string& path, mode_t mode, const std::string& lpath,
const char* lpath, const char* rpath, const std::string& rpath, unsigned mtime) {
unsigned mtime) { std::string path_and_mode = android::base::StringPrintf("%s,%d", path.c_str(), mode);
if (!SendRequest(ID_SEND, path_and_mode)) { if (!SendRequest(ID_SEND, path_and_mode)) {
Error("failed to send ID_SEND message '%s': %s", path_and_mode, strerror(errno)); Error("failed to send ID_SEND message '%s': %s", path_and_mode.c_str(),
strerror(errno));
return false; return false;
} }
struct stat st; struct stat st;
if (stat(lpath, &st) == -1) { if (stat(lpath.c_str(), &st) == -1) {
Error("cannot stat '%s': %s", lpath, strerror(errno)); Error("cannot stat '%s': %s", lpath.c_str(), strerror(errno));
return false; return false;
} }
uint64_t total_size = st.st_size; uint64_t total_size = st.st_size;
uint64_t bytes_copied = 0; uint64_t bytes_copied = 0;
unique_fd lfd(adb_open(lpath, O_RDONLY)); unique_fd lfd(adb_open(lpath.c_str(), O_RDONLY));
if (lfd < 0) { if (lfd < 0) {
Error("opening '%s' locally failed: %s", lpath, strerror(errno)); Error("opening '%s' locally failed: %s", lpath.c_str(), strerror(errno));
return false; return false;
} }
@ -487,7 +480,7 @@ class SyncConnection {
while (true) { while (true) {
int bytes_read = adb_read(lfd, sbuf.data, max - sizeof(SyncRequest)); int bytes_read = adb_read(lfd, sbuf.data, max - sizeof(SyncRequest));
if (bytes_read == -1) { if (bytes_read == -1) {
Error("reading '%s' locally failed: %s", lpath, strerror(errno)); Error("reading '%s' locally failed: %s", lpath.c_str(), strerror(errno));
return false; return false;
} else if (bytes_read == 0) { } else if (bytes_read == 0) {
break; break;
@ -499,55 +492,53 @@ class SyncConnection {
RecordBytesTransferred(bytes_read); RecordBytesTransferred(bytes_read);
bytes_copied += bytes_read; bytes_copied += bytes_read;
// Check to see if we've received an error from the other side.
if (ReceivedError(lpath, rpath)) {
break;
}
ReportProgress(rpath, bytes_copied, total_size); ReportProgress(rpath, bytes_copied, total_size);
} }
syncmsg msg; syncmsg msg;
msg.data.id = ID_DONE; msg.data.id = ID_DONE;
msg.data.size = mtime; msg.data.size = mtime;
expect_done_ = true; RecordFileSent(lpath, rpath);
// RecordFilesTransferred gets called in CopyDone.
return WriteOrDie(lpath, rpath, &msg.data, sizeof(msg.data)); return WriteOrDie(lpath, rpath, &msg.data, sizeof(msg.data));
} }
bool CopyDone(const char* from, const char* to) { bool ReadAcknowledgments() {
bool result = true;
while (!deferred_acknowledgements_.empty()) {
auto [from, to] = std::move(deferred_acknowledgements_.front());
deferred_acknowledgements_.pop_front();
result &= CopyDone(from, to);
}
return result;
}
bool CopyDone(const std::string& from, const std::string& to) {
syncmsg msg; syncmsg msg;
if (!ReadFdExactly(fd, &msg.status, sizeof(msg.status))) { if (!ReadFdExactly(fd, &msg.status, sizeof(msg.status))) {
Error("failed to copy '%s' to '%s': couldn't read from device", from, to); Error("failed to copy '%s' to '%s': couldn't read from device", from.c_str(),
to.c_str());
return false; return false;
} }
if (msg.status.id == ID_OKAY) { if (msg.status.id == ID_OKAY) {
if (expect_done_) { return true;
expect_done_ = false;
RecordFilesTransferred(1);
return true;
} else {
Error("failed to copy '%s' to '%s': received premature success", from, to);
return true;
}
} }
if (msg.status.id != ID_FAIL) { if (msg.status.id != ID_FAIL) {
Error("failed to copy '%s' to '%s': unknown reason %d", from, to, msg.status.id); Error("failed to copy '%s' to '%s': unknown reason %d", from.c_str(), to.c_str(),
msg.status.id);
return false; return false;
} }
return ReportCopyFailure(from, to, msg); return ReportCopyFailure(from, to, msg);
} }
bool ReportCopyFailure(const char* from, const char* to, const syncmsg& msg) { bool ReportCopyFailure(const std::string& from, const std::string& to, const syncmsg& msg) {
std::vector<char> buf(msg.status.msglen + 1); std::vector<char> buf(msg.status.msglen + 1);
if (!ReadFdExactly(fd, &buf[0], msg.status.msglen)) { if (!ReadFdExactly(fd, &buf[0], msg.status.msglen)) {
Error("failed to copy '%s' to '%s'; failed to read reason (!): %s", Error("failed to copy '%s' to '%s'; failed to read reason (!): %s", from.c_str(),
from, to, strerror(errno)); to.c_str(), strerror(errno));
return false; return false;
} }
buf[msg.status.msglen] = 0; buf[msg.status.msglen] = 0;
Error("failed to copy '%s' to '%s': remote %s", from, to, &buf[0]); Error("failed to copy '%s' to '%s': remote %s", from.c_str(), to.c_str(), &buf[0]);
return false; return false;
} }
@ -616,7 +607,7 @@ class SyncConnection {
size_t max; size_t max;
private: private:
bool expect_done_; std::deque<std::pair<std::string, std::string>> deferred_acknowledgements_;
FeatureSet features_; FeatureSet features_;
bool have_stat_v2_; bool have_stat_v2_;
bool have_ls_v2_; bool have_ls_v2_;
@ -629,16 +620,19 @@ class SyncConnection {
return SendRequest(ID_QUIT, ""); // TODO: add a SendResponse? return SendRequest(ID_QUIT, ""); // TODO: add a SendResponse?
} }
bool WriteOrDie(const char* from, const char* to, const void* data, size_t data_length) { bool WriteOrDie(const std::string& from, const std::string& to, const void* data,
size_t data_length) {
if (!WriteFdExactly(fd, data, data_length)) { if (!WriteFdExactly(fd, data, data_length)) {
if (errno == ECONNRESET) { if (errno == ECONNRESET) {
// Assume adbd told us why it was closing the connection, and // Assume adbd told us why it was closing the connection, and
// try to read failure reason from adbd. // try to read failure reason from adbd.
syncmsg msg; syncmsg msg;
if (!ReadFdExactly(fd, &msg.status, sizeof(msg.status))) { if (!ReadFdExactly(fd, &msg.status, sizeof(msg.status))) {
Error("failed to copy '%s' to '%s': no response: %s", from, to, strerror(errno)); Error("failed to copy '%s' to '%s': no response: %s", from.c_str(), to.c_str(),
strerror(errno));
} else if (msg.status.id != ID_FAIL) { } else if (msg.status.id != ID_FAIL) {
Error("failed to copy '%s' to '%s': not ID_FAIL: %d", from, to, msg.status.id); Error("failed to copy '%s' to '%s': not ID_FAIL: %d", from.c_str(), to.c_str(),
msg.status.id);
} else { } else {
ReportCopyFailure(from, to, msg); ReportCopyFailure(from, to, msg);
} }
@ -651,20 +645,20 @@ class SyncConnection {
} }
}; };
static bool sync_ls(SyncConnection& sc, const char* path, static bool sync_ls(SyncConnection& sc, const std::string& path,
const std::function<sync_ls_cb>& func) { const std::function<sync_ls_cb>& func) {
return sc.SendLs(path) && sc.FinishLs(func); return sc.SendLs(path) && sc.FinishLs(func);
} }
static bool sync_stat(SyncConnection& sc, const char* path, struct stat* st) { static bool sync_stat(SyncConnection& sc, const std::string& path, struct stat* st) {
return sc.SendStat(path) && sc.FinishStat(st); return sc.SendStat(path) && sc.FinishStat(st);
} }
static bool sync_lstat(SyncConnection& sc, const char* path, struct stat* st) { static bool sync_lstat(SyncConnection& sc, const std::string& path, struct stat* st) {
return sc.SendLstat(path) && sc.FinishStat(st); return sc.SendLstat(path) && sc.FinishStat(st);
} }
static bool sync_stat_fallback(SyncConnection& sc, const char* path, struct stat* st) { static bool sync_stat_fallback(SyncConnection& sc, const std::string& path, struct stat* st) {
if (sync_stat(sc, path, st)) { if (sync_stat(sc, path, st)) {
return true; return true;
} }
@ -688,7 +682,7 @@ static bool sync_stat_fallback(SyncConnection& sc, const char* path, struct stat
struct stat tmp_st; struct stat tmp_st;
st->st_mode &= ~S_IFMT; st->st_mode &= ~S_IFMT;
if (sync_lstat(sc, dir_path.c_str(), &tmp_st)) { if (sync_lstat(sc, dir_path, &tmp_st)) {
st->st_mode |= S_IFDIR; st->st_mode |= S_IFDIR;
} else { } else {
st->st_mode |= S_IFREG; st->st_mode |= S_IFREG;
@ -697,10 +691,8 @@ static bool sync_stat_fallback(SyncConnection& sc, const char* path, struct stat
return true; return true;
} }
static bool sync_send(SyncConnection& sc, const char* lpath, const char* rpath, unsigned mtime, static bool sync_send(SyncConnection& sc, const std::string& lpath, const std::string& rpath,
mode_t mode, bool sync) { unsigned mtime, mode_t mode, bool sync) {
std::string path_and_mode = android::base::StringPrintf("%s,%d", rpath, mode);
if (sync) { if (sync) {
struct stat st; struct stat st;
if (sync_lstat(sc, rpath, &st)) { if (sync_lstat(sc, rpath, &st)) {
@ -714,41 +706,40 @@ static bool sync_send(SyncConnection& sc, const char* lpath, const char* rpath,
if (S_ISLNK(mode)) { if (S_ISLNK(mode)) {
#if !defined(_WIN32) #if !defined(_WIN32)
char buf[PATH_MAX]; char buf[PATH_MAX];
ssize_t data_length = readlink(lpath, buf, PATH_MAX - 1); ssize_t data_length = readlink(lpath.c_str(), buf, PATH_MAX - 1);
if (data_length == -1) { if (data_length == -1) {
sc.Error("readlink '%s' failed: %s", lpath, strerror(errno)); sc.Error("readlink '%s' failed: %s", lpath.c_str(), strerror(errno));
return false; return false;
} }
buf[data_length++] = '\0'; buf[data_length++] = '\0';
if (!sc.SendSmallFile(path_and_mode.c_str(), lpath, rpath, mtime, buf, data_length)) { if (!sc.SendSmallFile(rpath, mode, lpath, rpath, mtime, buf, data_length)) {
return false; return false;
} }
return sc.CopyDone(lpath, rpath); return true;
#endif #endif
} }
struct stat st; struct stat st;
if (stat(lpath, &st) == -1) { if (stat(lpath.c_str(), &st) == -1) {
sc.Error("failed to stat local file '%s': %s", lpath, strerror(errno)); sc.Error("failed to stat local file '%s': %s", lpath.c_str(), strerror(errno));
return false; return false;
} }
if (st.st_size < SYNC_DATA_MAX) { if (st.st_size < SYNC_DATA_MAX) {
std::string data; std::string data;
if (!android::base::ReadFileToString(lpath, &data, true)) { if (!android::base::ReadFileToString(lpath, &data, true)) {
sc.Error("failed to read all of '%s': %s", lpath, strerror(errno)); sc.Error("failed to read all of '%s': %s", lpath.c_str(), strerror(errno));
return false; return false;
} }
if (!sc.SendSmallFile(path_and_mode.c_str(), lpath, rpath, mtime, if (!sc.SendSmallFile(rpath, mode, lpath, rpath, mtime, data.data(), data.size())) {
data.data(), data.size())) {
return false; return false;
} }
} else { } else {
if (!sc.SendLargeFile(path_and_mode.c_str(), lpath, rpath, mtime)) { if (!sc.SendLargeFile(rpath, mode, lpath, rpath, mtime)) {
return false; return false;
} }
} }
return sc.CopyDone(lpath, rpath); return true;
} }
static bool sync_recv(SyncConnection& sc, const char* rpath, const char* lpath, static bool sync_recv(SyncConnection& sc, const char* rpath, const char* lpath,
@ -943,7 +934,7 @@ static bool copy_local_dir_remote(SyncConnection& sc, std::string lpath,
if (check_timestamps) { if (check_timestamps) {
for (const copyinfo& ci : file_list) { for (const copyinfo& ci : file_list) {
if (!sc.SendLstat(ci.rpath.c_str())) { if (!sc.SendLstat(ci.rpath)) {
sc.Error("failed to send lstat"); sc.Error("failed to send lstat");
return false; return false;
} }
@ -965,7 +956,7 @@ static bool copy_local_dir_remote(SyncConnection& sc, std::string lpath,
if (list_only) { if (list_only) {
sc.Println("would push: %s -> %s", ci.lpath.c_str(), ci.rpath.c_str()); sc.Println("would push: %s -> %s", ci.lpath.c_str(), ci.rpath.c_str());
} else { } else {
if (!sync_send(sc, ci.lpath.c_str(), ci.rpath.c_str(), ci.time, ci.mode, false)) { if (!sync_send(sc, ci.lpath, ci.rpath, ci.time, ci.mode, false)) {
return false; return false;
} }
} }
@ -1069,6 +1060,7 @@ bool do_sync_push(const std::vector<const char*>& srcs, const char* dst, bool sy
sc.ReportTransferRate(src_path, TransferDirection::push); sc.ReportTransferRate(src_path, TransferDirection::push);
} }
success &= sc.ReadAcknowledgments();
sc.ReportOverallTransferRate(TransferDirection::push); sc.ReportOverallTransferRate(TransferDirection::push);
return success; return success;
} }