android adb源码分析(4)

本篇分析transport.c。transport的原理是使用了fevent机制进行数据传输，atransport的定义如下：
struct atransport
{
atransport *next;
atransport *prev;

int (*read_from_remote)(apacket *p, atransport *t);
int (*write_to_remote)(apacket *p, atransport *t);
void (*close)(atransport *t);
void (*kick)(atransport *t);

int fd;
int transport_socket;
fdevent transport_fde;
int ref_count;
unsigned sync_token;
int connection_state;
int online;
transport_type type;

/* usb handle or socket fd as needed */
usb_handle *usb;
int sfd;

/* used to identify transports for clients */
char *serial;
char *product;
char *model;
char *device;
char *devpath;
int adb_port; // Use for emulators (local transport)

/* a list of adisconnect callbacks called when the transport is kicked */
int kicked;
adisconnect disconnects;

void *key;
unsigned char token[TOKEN_SIZE];
fdevent auth_fde;
unsigned failed_auth_attempts;
};
它是一个双向链表，所有的实体都保存在全局数据transport_list中：
static atransport transport_list = {
.next = &transport_list,
.prev = &transport_list,
};
atransport的数据传输的两个函数是：
static int
read_packet(int fd, const char* name, apacket** ppacket)
{
char *p = (char*)ppacket; /* really read a packet address */
int r;
int len = sizeof(*ppacket);
char buff[8];
if (!name) {
snprintf(buff, sizeof buff, "fd=%d", fd);
name = buff;
}
while(len > 0) {
r = adb_read(fd, p, len);
if(r > 0) {
len -= r;
p += r;
} else {
D("%s: read_packet (fd=%d), error ret=%d errno=%d: %s\n", name, fd, r, errno, strerror(errno));
if((r < 0) && (errno == EINTR)) continue;
return -1;
}
}

#if ADB_TRACE
if (ADB_TRACING) {
dump_packet(name, "from remote", *ppacket);
}
#endif
return 0;
}

static int
write_packet(int fd, const char* name, apacket** ppacket)
{
char *p = (char*) ppacket; /* we really write the packet address */
int r, len = sizeof(ppacket);
char buff[8];
if (!name) {
snprintf(buff, sizeof buff, "fd=%d", fd);
name = buff;
}

#if ADB_TRACE
if (ADB_TRACING) {
dump_packet(name, "to remote", *ppacket);
}
#endif
len = sizeof(ppacket);
while(len > 0) {
r = adb_write(fd, p, len);
if(r > 0) {
len -= r;
p += r;
} else {
D("%s: write_packet (fd=%d) error ret=%d errno=%d: %s\n", name, fd, r, errno, strerror(errno));
if((r < 0) && (errno == EINTR)) continue;
return -1;
}
}
return 0;
}


这是它的内部传输接口，注意传输的是apacket的数据内部。
外部接口的数据传输接口是：
static void transport_socket_events(int fd, unsigned events, void *_t)
{
atransport *t = _t;
D("transport_socket_events(fd=%d, events=%04x,...)\n", fd, events);
if(events & FDE_READ){
apacket *p = 0;
if(read_packet(fd, t->serial, &p)){
D("%s: failed to read packet from transport socket on fd %d\n", t->serial, fd);
} else {
handle_packet(p, (atransport *) _t);
}
}
}

void send_packet(apacket *p, atransport *t)
{
unsigned char *x;
unsigned sum;
unsigned count;

p->msg.magic = p->msg.command ^ 0xffffffff;

count = p->msg.data_length;
x = (unsigned char *) p->data;
sum = 0;
while(count-- > 0){
sum += *x++;
}
p->msg.data_check = sum;

print_packet("send", p);

if (t == NULL) {
D("Transport is null \n");
// Zap errno because print_packet() and other stuff have errno effect.
errno = 0;
fatal_errno("Transport is null");
}

if(write_packet(t->transport_socket, t->serial, &p)){
fatal_errno("cannot enqueue packet on transport socket");
}
}
物理接口到atransport的接口是：
/* The transport is opened by transport_register_func before
** the input and output threads are started.
**
** The output thread issues a SYNC(1, token) message to let
** the input thread know to start things up. In the event
** of transport IO failure, the output thread will post a
** SYNC(0,0) message to ensure shutdown.
**
** The transport will not actually be closed until both
** threads exit, but the input thread will kick the transport
** on its way out to disconnect the underlying device.
*/

static void *output_thread(void *_t)
{
atransport *t = _t;
apacket *p;

D("%s: starting transport output thread on fd %d, SYNC online (%d)\n",
t->serial, t->fd, t->sync_token + 1);
p = get_apacket();
p->msg.command = A_SYNC;
p->msg.arg0 = 1;
p->msg.arg1 = ++(t->sync_token);
p->msg.magic = A_SYNC ^ 0xffffffff;
if(write_packet(t->fd, t->serial, &p)) {
put_apacket(p);
D("%s: failed to write SYNC packet\n", t->serial);
goto oops;
}

D("%s: data pump started\n", t->serial);
for(;;) {
p = get_apacket();

if(t->read_from_remote(p, t) == 0){
D("%s: received remote packet, sending to transport\n",
t->serial);
if(write_packet(t->fd, t->serial, &p)){
put_apacket(p);
D("%s: failed to write apacket to transport\n", t->serial);
goto oops;
}
} else {
D("%s: remote read failed for transport\n", t->serial);
put_apacket(p);
break;
}
}

D("%s: SYNC offline for transport\n", t->serial);
p = get_apacket();
p->msg.command = A_SYNC;
p->msg.arg0 = 0;
p->msg.arg1 = 0;
p->msg.magic = A_SYNC ^ 0xffffffff;
if(write_packet(t->fd, t->serial, &p)) {
put_apacket(p);
D("%s: failed to write SYNC apacket to transport", t->serial);
}

oops:
D("%s: transport output thread is exiting\n", t->serial);
kick_transport(t);
transport_unref(t);
return 0;
}

static void *input_thread(void *_t)
{
atransport *t = _t;
apacket *p;
int active = 0;

D("%s: starting transport input thread, reading from fd %d\n",
t->serial, t->fd);

for(;;){
if(read_packet(t->fd, t->serial, &p)) {
D("%s: failed to read apacket from transport on fd %d\n",
t->serial, t->fd );
break;
}
if(p->msg.command == A_SYNC){
if(p->msg.arg0 == 0) {
D("%s: transport SYNC offline\n", t->serial);
put_apacket(p);
break;
} else {
if(p->msg.arg1 == t->sync_token) {
D("%s: transport SYNC online\n", t->serial);
active = 1;
} else {
D("%s: transport ignoring SYNC %d != %d\n",
t->serial, p->msg.arg1, t->sync_token);
}
}
} else {
if(active) {
D("%s: transport got packet, sending to remote\n", t->serial);
t->write_to_remote(p, t);
} else {
D("%s: transport ignoring packet while offline\n", t->serial);
}
}

put_apacket(p);
}

// this is necessary to avoid a race condition that occured when a transport closes
// while a client socket is still active.
close_all_sockets(t);

D("%s: transport input thread is exiting, fd %d\n", t->serial, t->fd);
kick_transport(t);
transport_unref(t);
return 0;
}
这里，源码做了一个物理到抽象的转换。最上层还有一层是tmsg的传输（作者为什么要这么绕，实在是看得不明白）：void init_transport_registration(void)
{
int s[2];

if(adb_socketpair(s)){
fatal_errno("cannot open transport registration socketpair");
}

transport_registration_send = s[0];
transport_registration_recv = s[1];

fdevent_install(&transport_registration_fde,
transport_registration_recv,
transport_registration_func,
0);

fdevent_set(&transport_registration_fde, FDE_READ);
}

/* the fdevent select pump is single threaded */
static void register_transport(atransport *transport)
{
tmsg m;
m.transport = transport;
m.action = 1;
D("transport: %s registered\n", transport->serial);
if(transport_write_action(transport_registration_send, &m)) {
fatal_errno("cannot write transport registration socket\n");
}
}

static void remove_transport(atransport *transport)
{
tmsg m;
m.transport = transport;
m.action = 0;
D("transport: %s removed\n", transport->serial);
if(transport_write_action(transport_registration_send, &m)) {
fatal_errno("cannot write transport registration socket\n");
}
}

tmsg的作用是用于atransport的管理。