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

adb: switch asocket::enqueue to std::string.

Browse source 
Switch asocket over to taking a std::string instead of apacket* for
data. This allows us to remove asocket specific fields from apacket*.

Test: python test_device.py with x86_64 emulator, walleye
Test: adb_test on host
Change-Id: I9d157ff331a75ba49a54fdd4194e3f6cdff722f4
This commit is contained in:
Josh Gao 2018-02-01 13:17:50 -08:00
parent 5caaebdc3d
commit 27cb7dca77
8 changed files with 166 additions and 126 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
adb/adb.cpp
View file

@ -474,13 +474,14 @@ void handle_packet(apacket *p, atransport *t)
asocket* s = find_local_socket(p->msg.arg1, p->msg.arg0);
if (s) {
unsigned rid = p->msg.arg0;
p->len = p->msg.data_length;
if (s->enqueue(s, p) == 0) {
// TODO: Convert apacket::data to a type that we can move out of.
std::string copy(p->data, p->data + p->msg.data_length);
if (s->enqueue(s, std::move(copy)) == 0) {
D("Enqueue the socket");
send_ready(s->id, rid, t);
}
return;
}
}
break;

5
adb/adb.h
View file

@ -73,11 +73,6 @@ struct amessage {
};
struct apacket {
apacket* next;
size_t len;
char* ptr;
amessage msg;
char data[MAX_PAYLOAD];
};

32
adb/jdwp_service.cpp
View file

@ -470,10 +470,9 @@ static void jdwp_socket_close(asocket* s) {
free(s);
}
static int jdwp_socket_enqueue(asocket* s, apacket* p) {
static int jdwp_socket_enqueue(asocket* s, std::string) {
/* you can't write to this asocket */
D("LS(%d): JDWP socket received data?", s->id);
put_apacket(p);
s->peer->close(s->peer);
return -1;
}
@ -486,9 +485,11 @@ static void jdwp_socket_ready(asocket* s) {
* on the second one, close the connection
*/
if (!jdwp->pass) {
apacket* p = get_apacket();
p->len = jdwp_process_list((char*)p->data, s->get_max_payload());
peer->enqueue(peer, p);
std::string data;
data.resize(s->get_max_payload());
size_t len = jdwp_process_list(&data[0], data.size());
data.resize(len);
peer->enqueue(peer, std::move(data));
jdwp->pass = true;
} else {
peer->close(peer);
@ -524,17 +525,14 @@ struct JdwpTracker : public asocket {
static std::vector<std::unique_ptr<JdwpTracker>> _jdwp_trackers;
static void jdwp_process_list_updated(void) {
char buffer[1024];
int len = jdwp_process_list_msg(buffer, sizeof(buffer));
std::string data;
data.resize(1024);
data.resize(jdwp_process_list_msg(&data[0], data.size()));
for (auto& t : _jdwp_trackers) {
apacket* p = get_apacket();
memcpy(p->data, buffer, len);
p->len = len;
if (t->peer) {
// The tracker might not have been connected yet.
t->peer->enqueue(t->peer, p);
t->peer->enqueue(t->peer, data);
}
}
}
@ -560,17 +558,17 @@ static void jdwp_tracker_ready(asocket* s) {
JdwpTracker* t = (JdwpTracker*)s;
if (t->need_initial) {
apacket* p = get_apacket();
std::string data;
data.resize(s->get_max_payload());
data.resize(jdwp_process_list_msg(&data[0], data.size()));
t->need_initial = false;
p->len = jdwp_process_list_msg((char*)p->data, s->get_max_payload());
s->peer->enqueue(s->peer, p);
s->peer->enqueue(s->peer, std::move(data));
}
}
static int jdwp_tracker_enqueue(asocket* s, apacket* p) {
static int jdwp_tracker_enqueue(asocket* s, std::string) {
/* you can't write to this socket */
D("LS(%d): JDWP tracker received data?", s->id);
put_apacket(p);
s->peer->close(s->peer);
return -1;
}

65
adb/range.h Normal file
View file

@ -0,0 +1,65 @@
#pragma once
/*
* Copyright (C) 2018 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 <string>
#include <android-base/logging.h>
struct Range {
explicit Range(std::string data) : data_(std::move(data)) {}
Range(const Range& copy) = delete;
Range& operator=(const Range& copy) = delete;
Range(Range&& move) = default;
Range& operator=(Range&& move) = default;
bool empty() const {
return size() == 0;
}
size_t size() const {
return data_.size() - begin_offset_ - end_offset_;
};
void drop_front(size_t n) {
CHECK_GE(size(), n);
begin_offset_ += n;
}
void drop_end(size_t n) {
CHECK_GE(size(), n);
end_offset_ += n;
}
char* data() {
return &data_[0] + begin_offset_;
}
std::string::iterator begin() {
return data_.begin() + begin_offset_;
}
std::string::iterator end() {
return data_.end() - end_offset_;
}
std::string data_;
size_t begin_offset_ = 0;
size_t end_offset_ = 0;
};

12
adb/socket.h
View file

@ -19,9 +19,12 @@
#include <stddef.h>
#include <deque>
#include <memory>
#include <string>
#include "fdevent.h"
#include "range.h"
struct apacket;
class atransport;
@ -59,9 +62,10 @@ struct asocket {
fdevent fde;
int fd;
// queue of apackets waiting to be written
apacket* pkt_first;
apacket* pkt_last;
// queue of data waiting to be written
std::deque<Range> packet_queue;
std::string smart_socket_data;
/* enqueue is called by our peer when it has data
* for us. It should return 0 if we can accept more
@ -69,7 +73,7 @@ struct asocket {
* peer->ready() when we once again are ready to
* receive data.
*/
int (*enqueue)(asocket* s, apacket* pkt);
int (*enqueue)(asocket* s, std::string data);
/* ready is called by the peer when it is ready for
* us to send data via enqueue again

8
adb/socket_test.cpp
View file

@ -114,10 +114,10 @@ static void CloseWithPacketThreadFunc(CloseWithPacketArg* arg) {
ASSERT_TRUE(s != nullptr);
arg->bytes_written = 0;
while (true) {
apacket* p = get_apacket();
p->len = sizeof(p->data);
arg->bytes_written += p->len;
int ret = s->enqueue(s, p);
std::string data;
data.resize(MAX_PAYLOAD);
arg->bytes_written += data.size();
int ret = s->enqueue(s, std::move(data));
if (ret == 1) {
// The writer has one packet waiting to send.
break;

148
adb/sockets.cpp
View file

@ -37,6 +37,7 @@
#include "adb.h"
#include "adb_io.h"
#include "range.h"
#include "transport.h"
static std::recursive_mutex& local_socket_list_lock = *new std::recursive_mutex();
@ -105,55 +106,47 @@ restart:
}
}
static int local_socket_enqueue(asocket* s, apacket* p) {
D("LS(%d): enqueue %zu", s->id, p->len);
static int local_socket_enqueue(asocket* s, std::string data) {
D("LS(%d): enqueue %zu", s->id, data.size());
p->ptr = p->data;
Range r(std::move(data));
/* if there is already data queue'd, we will receive
** events when it's time to write. just add this to
** the tail
*/
if (s->pkt_first) {
if (!s->packet_queue.empty()) {
goto enqueue;
}
/* write as much as we can, until we
** would block or there is an error/eof
*/
while (p->len > 0) {
int r = adb_write(s->fd, p->ptr, p->len);
if (r > 0) {
p->len -= r;
p->ptr += r;
while (!r.empty()) {
int rc = adb_write(s->fd, r.data(), r.size());
if (rc > 0) {
r.drop_front(rc);
continue;
}
if ((r == 0) || (errno != EAGAIN)) {
if (rc == 0 || errno != EAGAIN) {
D("LS(%d): not ready, errno=%d: %s", s->id, errno, strerror(errno));
put_apacket(p);
s->has_write_error = true;
s->close(s);
return 1; /* not ready (error) */
} else {
// errno == EAGAIN
break;
}
}
if (p->len == 0) {
put_apacket(p);
if (r.empty()) {
return 0; /* ready for more data */
}
enqueue:
p->next = 0;
if (s->pkt_first) {
s->pkt_last->next = p;
} else {
s->pkt_first = p;
}
s->pkt_last = p;
/* make sure we are notified when we can drain the queue */
s->packet_queue.push_back(std::move(r));
fdevent_add(&s->fde, FDE_WRITE);
return 1; /* not ready (backlog) */
@ -167,7 +160,6 @@ static void local_socket_ready(asocket* s) {
// be sure to hold the socket list lock when calling this
static void local_socket_destroy(asocket* s) {
apacket *p, *n;
int exit_on_close = s->exit_on_close;
D("LS(%d): destroying fde.fd=%d", s->id, s->fde.fd);
@ -177,12 +169,6 @@ static void local_socket_destroy(asocket* s) {
*/
fdevent_remove(&s->fde);
/* dispose of any unwritten data */
for (p = s->pkt_first; p; p = n) {
D("LS(%d): discarding %zu bytes", s->id, p->len);
n = p->next;
put_apacket(p);
}
remove_socket(s);
free(s);
@ -212,7 +198,7 @@ static void local_socket_close(asocket* s) {
/* If we are already closing, or if there are no
** pending packets, destroy immediately
*/
if (s->closing || s->has_write_error || s->pkt_first == NULL) {
if (s->closing || s->has_write_error || s->packet_queue.empty()) {
int id = s->id;
local_socket_destroy(s);
D("LS(%d): closed", id);
@ -238,35 +224,30 @@ static void local_socket_event_func(int fd, unsigned ev, void* _s) {
** in order to simplify the code.
*/
if (ev & FDE_WRITE) {
apacket* p;
while ((p = s->pkt_first) != nullptr) {
while (p->len > 0) {
int r = adb_write(fd, p->ptr, p->len);
if (r == -1) {
while (!s->packet_queue.empty()) {
Range& r = s->packet_queue.front();
while (!r.empty()) {
int rc = adb_write(fd, r.data(), r.size());
if (rc == -1) {
/* returning here is ok because FDE_READ will
** be processed in the next iteration loop
*/
if (errno == EAGAIN) {
return;
}
} else if (r > 0) {
p->ptr += r;
p->len -= r;
} else if (rc > 0) {
r.drop_front(rc);
continue;
}
D(" closing after write because r=%d and errno is %d", r, errno);
D(" closing after write because rc=%d and errno is %d", rc, errno);
s->has_write_error = true;
s->close(s);
return;
}
if (p->len == 0) {
s->pkt_first = p->next;
if (s->pkt_first == 0) {
s->pkt_last = 0;
}
put_apacket(p);
if (r.empty()) {
s->packet_queue.pop_front();
}
}
@ -288,9 +269,10 @@ static void local_socket_event_func(int fd, unsigned ev, void* _s) {
}
if (ev & FDE_READ) {
apacket* p = get_apacket();
char* x = p->data;
const size_t max_payload = s->get_max_payload();
std::string data;
data.resize(max_payload);
char* x = &data[0];
size_t avail = max_payload;
int r = 0;
int is_eof = 0;
@ -315,16 +297,15 @@ static void local_socket_event_func(int fd, unsigned ev, void* _s) {
}
D("LS(%d): fd=%d post avail loop. r=%d is_eof=%d forced_eof=%d", s->id, s->fd, r, is_eof,
s->fde.force_eof);
if ((avail == max_payload) || (s->peer == 0)) {
put_apacket(p);
} else {
p->len = max_payload - avail;
if (avail != max_payload && s->peer) {
data.resize(max_payload - avail);
// s->peer->enqueue() may call s->close() and free s,
// so save variables for debug printing below.
unsigned saved_id = s->id;
int saved_fd = s->fd;
r = s->peer->enqueue(s->peer, p);
r = s->peer->enqueue(s->peer, std::move(data));
D("LS(%u): fd=%d post peer->enqueue(). r=%d", saved_id, saved_fd, r);
if (r < 0) {
@ -428,12 +409,22 @@ static asocket* create_host_service_socket(const char* name, const char* serial,
}
#endif /* ADB_HOST */
static int remote_socket_enqueue(asocket* s, apacket* p) {
static int remote_socket_enqueue(asocket* s, std::string data) {
D("entered remote_socket_enqueue RS(%d) WRITE fd=%d peer.fd=%d", s->id, s->fd, s->peer->fd);
apacket* p = get_apacket();
p->msg.command = A_WRTE;
p->msg.arg0 = s->peer->id;
p->msg.arg1 = s->id;
p->msg.data_length = p->len;
p->msg.data_length = data.size();
if (data.size() > sizeof(p->data)) {
put_apacket(p);
return -1;
}
// TODO: Convert apacket::data to a type that we can move into.
memcpy(p->data, data.data(), data.size());
send_packet(p, s->transport);
return 1;
}
@ -533,7 +524,7 @@ static void local_socket_close_notify(asocket* s) {
s->close(s);
}
static unsigned unhex(char* s, int len) {
static unsigned unhex(const char* s, int len) {
unsigned n = 0, c;
while (len-- > 0) {
@ -637,8 +628,7 @@ char* skip_host_serial(char* service) {
#endif // ADB_HOST
static int smart_socket_enqueue(asocket* s, apacket* p) {
unsigned len;
static int smart_socket_enqueue(asocket* s, std::string data) {
#if ADB_HOST
char* service = nullptr;
char* serial = nullptr;
@ -646,49 +636,38 @@ static int smart_socket_enqueue(asocket* s, apacket* p) {
TransportType type = kTransportAny;
#endif
D("SS(%d): enqueue %zu", s->id, p->len);
D("SS(%d): enqueue %zu", s->id, data.size());
if (s->pkt_first == 0) {
s->pkt_first = p;
s->pkt_last = p;
if (s->smart_socket_data.empty()) {
s->smart_socket_data = std::move(data);
} else {
if ((s->pkt_first->len + p->len) > s->get_max_payload()) {
D("SS(%d): overflow", s->id);
put_apacket(p);
goto fail;
}
memcpy(s->pkt_first->data + s->pkt_first->len, p->data, p->len);
s->pkt_first->len += p->len;
put_apacket(p);
p = s->pkt_first;
std::copy(data.begin(), data.end(), std::back_inserter(s->smart_socket_data));
}
/* don't bother if we can't decode the length */
if (p->len < 4) {
if (s->smart_socket_data.size() < 4) {
return 0;
}
len = unhex(p->data, 4);
if ((len < 1) || (len > MAX_PAYLOAD)) {
D("SS(%d): bad size (%d)", s->id, len);
uint32_t len = unhex(s->smart_socket_data.data(), 4);
if (len == 0 || len > MAX_PAYLOAD) {
D("SS(%d): bad size (%u)", s->id, len);
goto fail;
}
D("SS(%d): len is %d", s->id, len);
D("SS(%d): len is %u", s->id, len);
/* can't do anything until we have the full header */
if ((len + 4) > p->len) {
D("SS(%d): waiting for %zu more bytes", s->id, len + 4 - p->len);
if ((len + 4) > s->smart_socket_data.size()) {
D("SS(%d): waiting for %zu more bytes", s->id, len + 4 - s->smart_socket_data.size());
return 0;
}
p->data[len + 4] = 0;
s->smart_socket_data[len + 4] = 0;
D("SS(%d): '%s'", s->id, (char*)(p->data + 4));
D("SS(%d): '%s'", s->id, (char*)(s->smart_socket_data.data() + 4));
#if ADB_HOST
service = (char*)p->data + 4;
service = &s->smart_socket_data[4];
if (!strncmp(service, "host-serial:", strlen("host-serial:"))) {
char* serial_end;
service += strlen("host-serial:");
@ -736,7 +715,7 @@ static int smart_socket_enqueue(asocket* s, apacket* p) {
}
if (!strncmp(service, "transport", strlen("transport"))) {
D("SS(%d): okay transport", s->id);
p->len = 0;
s->smart_socket_data.clear();
return 0;
}
@ -807,7 +786,7 @@ static int smart_socket_enqueue(asocket* s, apacket* p) {
/* give him our transport and upref it */
s->peer->transport = s->transport;
connect_to_remote(s->peer, (char*)(p->data + 4));
connect_to_remote(s->peer, s->smart_socket_data.data() + 4);
s->peer = 0;
s->close(s);
return 1;
@ -827,9 +806,6 @@ static void smart_socket_ready(asocket* s) {
static void smart_socket_close(asocket* s) {
D("SS(%d): closed", s->id);
if (s->pkt_first) {
put_apacket(s->pkt_first);
}
if (s->peer) {
s->peer->peer = 0;
s->peer->close(s->peer);

15
adb/transport.cpp
View file

@ -414,21 +414,22 @@ static void device_tracker_close(asocket* socket) {
free(tracker);
}
static int device_tracker_enqueue(asocket* socket, apacket* p) {
static int device_tracker_enqueue(asocket* socket, std::string) {
/* you can't read from a device tracker, close immediately */
put_apacket(p);
device_tracker_close(socket);
return -1;
}
static int device_tracker_send(device_tracker* tracker, const std::string& string) {
apacket* p = get_apacket();
asocket* peer = tracker->socket.peer;
snprintf(reinterpret_cast<char*>(p->data), 5, "%04x", static_cast<int>(string.size()));
memcpy(&p->data[4], string.data(), string.size());
p->len = 4 + string.size();
return peer->enqueue(peer, p);
std::string data;
data.resize(4 + string.size());
char buf[5];
snprintf(buf, sizeof(buf), "%04x", static_cast<int>(string.size()));
memcpy(&data[0], buf, 4);
memcpy(&data[4], string.data(), string.size());
return peer->enqueue(peer, std::move(data));
}
static void device_tracker_ready(asocket* socket) {