diff --git a/src/frontends/android/app/src/main/jni/libandroidbridge/routing/direct_tcp.c b/src/frontends/android/app/src/main/jni/libandroidbridge/routing/direct_tcp.c index e62230e44..605a576b8 100644 --- a/src/frontends/android/app/src/main/jni/libandroidbridge/routing/direct_tcp.c +++ b/src/frontends/android/app/src/main/jni/libandroidbridge/routing/direct_tcp.c @@ -59,6 +59,13 @@ */ #define MAX_BUFFERED (256 * 1024) +/** + * Stop reading from the remote socket while this much is waiting for the + * app: the data stays in the kernel socket and TCP throttles the sender, + * instead of the flow hitting MAX_BUFFERED and being aborted + */ +#define RELAY_HIGH_WATER (64 * 1024) + /** * Give up on a direct connect after this many milliseconds (a destination * that silently drops SYNs would otherwise hang the app's connection @@ -126,6 +133,7 @@ static lwip_err_t on_accept(void *arg, struct tcp_pcb *newpcb, lwip_err_t err); static lwip_err_t on_recv(void *arg, struct tcp_pcb *pcb, struct pbuf *p, lwip_err_t err); static void on_err(void *arg, lwip_err_t err); static lwip_err_t on_poll(void *arg, struct tcp_pcb *pcb); +static lwip_err_t on_sent(void *arg, struct tcp_pcb *pcb, u16_t len); /** * One relayed connection @@ -330,13 +338,20 @@ static void relay_flush_to_sock(tcp_relay_t *relay) { while (relay->to_sock_len) { - ssize_t n = write(relay->fd, relay->to_sock, relay->to_sock_len); + ssize_t n = write(relay->fd, relay->to_sock, + min(relay->to_sock_len, 0xffff)); if (n < 0) { return; } buf_consume(relay->to_sock, &relay->to_sock_len, n); + if (relay->pcb) + { /* reopen the app's send window only for what the socket took: + * acknowledging on receipt let a fast app outrun a slow uplink + * until the flow hit MAX_BUFFERED and was aborted */ + tcp_recved(relay->pcb, n); + } } } @@ -729,6 +744,7 @@ static lwip_err_t on_accept(void *arg, struct tcp_pcb *newpcb, lwip_err_t err) tcp_arg(newpcb, relay); tcp_recv(newpcb, on_recv); tcp_err(newpcb, on_err); + tcp_sent(newpcb, on_sent); tcp_poll(newpcb, on_poll, 4); this->relays->insert_last(this->relays, relay); DIRECT_STATS_ADD(this->stats->tcp_flows, 1); @@ -767,7 +783,6 @@ static lwip_err_t on_recv(void *arg, struct tcp_pcb *pcb, struct pbuf *p, lwip_e relay->dead = TRUE; return ERR_ABRT; } - tcp_recved(pcb, p->tot_len); pbuf_free(p); relay_flush_to_sock(relay); return ERR_OK; @@ -802,6 +817,22 @@ static lwip_err_t on_poll(void *arg, struct tcp_pcb *pcb) return ERR_OK; } +/** + * lwIP sent callback: the app acknowledged data, so the send buffer has + * room again. Without it the rest of a response only moved on the next + * socket read or the 2 s poll callback. + */ +static lwip_err_t on_sent(void *arg, struct tcp_pcb *pcb, u16_t len) +{ + tcp_relay_t *relay = arg; + + if (relay && relay->pcb && !relay->dead) + { + relay_flush_to_pcb(relay); + } + return ERR_OK; +} + /** * netif output: a finished IP packet for the kernel. lwIP keeps ownership * of the pbuf (unacked segments reference it for retransmission and free @@ -1020,7 +1051,16 @@ METHOD(direct_tcp_t, fill_poll, size_t, while (n < cap && enumerator->enumerate(enumerator, &relay)) { fds[n].fd = relay->fd; - fds[n].events = POLLIN | (relay->connected ? 0 : POLLOUT); + fds[n].events = 0; + if (!relay->connected || relay->to_sock_len) + { /* connect result, or room for data the socket refused before */ + fds[n].events |= POLLOUT; + } + if (!relay->remote_eof && relay->to_pcb_len < RELAY_HIGH_WATER) + { /* a socket at EOF stays readable forever: polling it for input + * made the engine thread spin until the app took the last byte */ + fds[n].events |= POLLIN; + } fds[n].revents = 0; n++; } @@ -1057,6 +1097,13 @@ METHOD(direct_tcp_t, process_poll, void, relay_destroy(this, relay, TRUE); continue; } + if (relay->remote_eof && (re & (POLLERR | POLLHUP))) + { /* fully closed: HUP is reported whatever we ask for, so drop the + * socket (poll ignores fd -1); what is buffered still goes to the app */ + close(relay->fd); + relay->fd = -1; + re = 0; + } if (!relay->connected) { uint64_t now = now_ms(); @@ -1142,13 +1189,16 @@ METHOD(direct_tcp_t, process_poll, void, } } } - if (re & (POLLIN | POLLHUP)) + if (re & (POLLIN | POLLHUP | POLLERR)) { uint8_t buf[16384]; - ssize_t n; + ssize_t n = 1; bool failed = FALSE; - while ((n = read(relay->fd, buf, sizeof(buf))) > 0) + /* a broken socket is read out whatever is buffered: its error + * must surface, or poll would report it round after round */ + while ((relay->to_pcb_len < RELAY_HIGH_WATER || (re & (POLLHUP | POLLERR))) && + (n = read(relay->fd, buf, sizeof(buf))) > 0) { if (!buf_append(relay, FALSE, buf, n)) {