jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

redesign TCP close: closing_tun/closing_rem flags, poll-driven cleanup, proper half-close

congestion
Evgeny 5 months ago
parent
238fb0eb23
commit
cec736a4b0
3 changed files with 26 additions and 33 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 53
      src/tcp_proxy.c
  2. 4
      src/tcp_proxy.h
  3. 2
      tests/test_tcp_proxy.c

53
src/tcp_proxy.c
Unescape View File

@ -211,7 +211,7 @@ static err_t proxy_accept_cb(void *arg, struct tcp_pcb *newpcb, err_t err) {
struct proxy_conn *pc = u_calloc(1, sizeof(struct proxy_conn)); struct proxy_conn *pc = u_calloc(1, sizeof(struct proxy_conn));
if (!pc) return LERR_MEM; if (!pc) return LERR_MEM;
pc->proxy = p; pc->pcb = newpcb; pc->closing = 0; pc->half_closed = 0; pc->proxy = p; pc->pcb = newpcb; pc->closing_tun = 0; pc->closing_rem = 0;
struct tcp_proxy_mapping *m = find_mapping_by_port(p, newpcb->local_port); struct tcp_proxy_mapping *m = find_mapping_by_port(p, newpcb->local_port);
if (m) { memcpy(pc->dest_ip, m->remote_ip, 4); pc->dest_port = m->remote_port; } if (m) { memcpy(pc->dest_ip, m->remote_ip, 4); pc->dest_port = m->remote_port; }
@ -250,22 +250,9 @@ static err_t proxy_recv_cb(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err_t
if (!pc) { if (p) pbuf_free(p); return LERR_OK; } if (!pc) { if (p) pbuf_free(p); return LERR_OK; }
if (p == NULL || err != LERR_OK) { if (p == NULL || err != LERR_OK) {
struct tcp_proxy *proxy = pc->proxy; DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "TCP proxy: TUN FIN received, stream=%s active=%d",
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "TCP proxy: remote closed (FIN/EOF), stream=%s active=%d",
pc->active ? "active" : "passive", pc->active); pc->active ? "active" : "passive", pc->active);
pc->closing = 1; pc->closing_tun = 1;
if (pc->transport) { pc->transport->ops->destroy(pc->transport); pc->transport = NULL; }
struct tcp_pcb *pcb = pc->pcb;
if (pcb) { tcp_arg(pcb, NULL); pc->pcb = NULL; tcp_close(pcb); }
struct proxy_conn **prev = &proxy->conns;
while (*prev) {
if (*prev == pc) { *prev = pc->next; proxy->conn_count--; break; }
prev = &(*prev)->next;
}
if (pc->eim_mapping) { pc->eim_mapping->delete_at_tb = get_time_tb() + proxy->eim_timeout_tb; pc->eim_mapping = NULL; }
if (pc->uip_to_transport) { struct ll_entry *e2; while ((e2 = queue_data_get(pc->uip_to_transport))) { queue_dgram_free(e2); queue_entry_free(e2); } queue_free(pc->uip_to_transport); }
if (pc->transport_to_uip) { struct ll_entry *e2; while ((e2 = queue_data_get(pc->transport_to_uip))) { queue_dgram_free(e2); queue_entry_free(e2); } queue_free(pc->transport_to_uip); }
u_free(pc);
return LERR_OK; return LERR_OK;
} }
@ -275,7 +262,7 @@ static err_t proxy_recv_cb(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err_t
if (e) queue_data_put(pc->uip_to_transport, e); if (e) queue_data_put(pc->uip_to_transport, e);
tcp_recved(pcb, p->tot_len); tcp_recved(pcb, p->tot_len);
pbuf_free(p); pbuf_free(p);
} else if (pc->transport && !pc->half_closed) { } else if (pc->transport && !pc->closing_rem) {
uint16_t len = p->tot_len; uint16_t len = p->tot_len;
uint8_t *data = u_malloc(len); uint8_t *data = u_malloc(len);
if (data) { if (data) {
@ -305,7 +292,7 @@ static void proxy_err_cb(void *arg, err_t err) {
struct proxy_conn *pc = (struct proxy_conn *)arg; struct proxy_conn *pc = (struct proxy_conn *)arg;
if (!pc) return; if (!pc) return;
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: tcp error %d", err); DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: tcp error %d", err);
pc->closing = 1; pc->closing_tun = 1;
if (pc->transport) pc->transport->ops->close(pc->transport); if (pc->transport) pc->transport->ops->close(pc->transport);
} }
@ -313,8 +300,16 @@ static err_t proxy_poll_cb(void *arg, struct tcp_pcb *pcb) {
struct proxy_conn *pc = (struct proxy_conn *)arg; struct proxy_conn *pc = (struct proxy_conn *)arg;
if (!pc) return LERR_OK; if (!pc) return LERR_OK;
proxy_feed_from_transport(pc); proxy_feed_from_transport(pc);
if (pc->closing && pc->transport == NULL && pc->pcb) { if (pc->closing_tun && pc->closing_rem && pc->pcb) {
tcp_arg(pc->pcb, NULL); pc->pcb = NULL; tcp_close(pcb); struct tcp_pcb *save = pc->pcb;
tcp_arg(save, NULL); pc->pcb = NULL; tcp_close(save);
struct tcp_proxy *proxy = pc->proxy;
struct proxy_conn **prev = &proxy->conns;
while (*prev) { if (*prev == pc) { *prev = pc->next; proxy->conn_count--; break; } prev = &(*prev)->next; }
if (pc->eim_mapping) { pc->eim_mapping->delete_at_tb = get_time_tb() + proxy->eim_timeout_tb; pc->eim_mapping = NULL; }
if (pc->uip_to_transport) { struct ll_entry *e2; while ((e2 = queue_data_get(pc->uip_to_transport))) { queue_dgram_free(e2); queue_entry_free(e2); } queue_free(pc->uip_to_transport); }
if (pc->transport_to_uip) { struct ll_entry *e2; while ((e2 = queue_data_get(pc->transport_to_uip))) { queue_dgram_free(e2); queue_entry_free(e2); } queue_free(pc->transport_to_uip); }
u_free(pc);
} }
return LERR_OK; return LERR_OK;
} }
@ -324,7 +319,7 @@ static err_t proxy_connected_cb(void *arg, struct tcp_pcb *pcb, err_t err) {
if (!pc) return LERR_ABRT; if (!pc) return LERR_ABRT;
if (err != LERR_OK) { if (err != LERR_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: active connect failed: %d", err); DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: active connect failed: %d", err);
pc->closing = 1; pc->closing_tun = 1;
return LERR_ABRT; return LERR_ABRT;
} }
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP proxy: active connect established to %d.%d.%d.%d:%d cwnd=%u snd_wnd=%u snd_buf=%u", DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP proxy: active connect established to %d.%d.%d.%d:%d cwnd=%u snd_wnd=%u snd_buf=%u",
@ -486,7 +481,7 @@ static void sock_transport_close(struct tcp_proxy_transport* t) {
if(st->read_id) { uasync_remove_socket_t(st->ua, st->sock); st->read_id = NULL; } if(st->read_id) { uasync_remove_socket_t(st->ua, st->sock); st->read_id = NULL; }
if(st->write_id) { uasync_remove_socket_t(st->ua, st->sock); st->write_id = NULL; } if(st->write_id) { uasync_remove_socket_t(st->ua, st->sock); st->write_id = NULL; }
socket_close_wrapper(st->sock); st->sock = SOCKET_INVALID; st->connected = 0; socket_close_wrapper(st->sock); st->sock = SOCKET_INVALID; st->connected = 0;
if(pc && !pc->closing) pc->closing = 1; if(pc && !pc->closing_rem) pc->closing_rem = 1;
if(pc) pc->transport = NULL; if(pc) pc->transport = NULL;
} }
@ -591,7 +586,7 @@ static void etcp_transport_close(struct tcp_proxy_transport* t) {
etcp_route_send(et->inst, et->remote_node_id, e); } else queue_entry_free(e); etcp_route_send(et->inst, et->remote_node_id, e); } else queue_entry_free(e);
} }
et->connected = 0; et->connected = 0;
if (et->conn) et->conn->transport = NULL; if (et->conn) { et->conn->transport = NULL; et->conn->closing_rem = 1; }
} }
static void etcp_transport_destroy(struct tcp_proxy_transport* t) { static void etcp_transport_destroy(struct tcp_proxy_transport* t) {
@ -654,7 +649,7 @@ static void handle_connected(struct tcp_proxy* p, uint64_t stream_id, struct ll_
} }
} }
else { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: remote refused stream=%016llx", (unsigned long long)stream_id); else { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: remote refused stream=%016llx", (unsigned long long)stream_id);
pc->closing = 1; } pc->closing_rem = 1; }
} }
queue_entry_free(entry); queue_dgram_free(entry); queue_entry_free(entry); queue_dgram_free(entry);
} }
@ -699,10 +694,8 @@ void tcp_proxy_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (subcmd == TCP_PROXY_SUBCMD_CLOSE) { if (subcmd == TCP_PROXY_SUBCMD_CLOSE) {
struct proxy_conn* pc = find_pc_by_stream(proxy, stream_id); struct proxy_conn* pc = find_pc_by_stream(proxy, stream_id);
if (pc) { if (pc) {
pc->closing = 1; pc->closing_rem = 1;
if (pc->transport) { pc->transport->ops->destroy(pc->transport); pc->transport = NULL; } if (pc->transport) { pc->transport->ops->destroy(pc->transport); pc->transport = NULL; }
struct tcp_pcb *pcb = pc->pcb;
if (pcb) { tcp_arg(pcb, NULL); pc->pcb = NULL; tcp_close(pcb); }
} }
} }
} }
@ -817,7 +810,7 @@ struct proxy_conn* tcp_proxy_active_open(struct tcp_proxy* p, const char* dest_i
struct proxy_conn* pc = u_calloc(1, sizeof(struct proxy_conn)); struct proxy_conn* pc = u_calloc(1, sizeof(struct proxy_conn));
if (!pc) return NULL; if (!pc) return NULL;
pc->proxy = p; pc->active = 1; pc->closing = 0; pc->half_closed = 0; pc->proxy = p; pc->active = 1; pc->closing_tun = 0; pc->closing_rem = 0;
pc->dest_port = htons(dest_port); pc->dest_port = htons(dest_port);
pc->dest_ip[0] = (uint8_t)a0; pc->dest_ip[1] = (uint8_t)a1; pc->dest_ip[2] = (uint8_t)a2; pc->dest_ip[3] = (uint8_t)a3; pc->dest_ip[0] = (uint8_t)a0; pc->dest_ip[1] = (uint8_t)a1; pc->dest_ip[2] = (uint8_t)a2; pc->dest_ip[3] = (uint8_t)a3;
pc->tun_ip = htonl((a0 << 24) | (a1 << 16) | (a2 << 8) | a3); pc->tun_ip = htonl((a0 << 24) | (a1 << 16) | (a2 << 8) | a3);
@ -869,7 +862,7 @@ ssize_t tcp_proxy_active_recv(struct tcp_proxy* p, struct proxy_conn* pc, uint8_
int tcp_proxy_active_close(struct tcp_proxy* p, struct proxy_conn* pc) { int tcp_proxy_active_close(struct tcp_proxy* p, struct proxy_conn* pc) {
(void)p; (void)p;
if (!pc || !pc->active) return -1; if (!pc || !pc->active) return -1;
pc->closing = 1; pc->closing_tun = 1;
tcp_close(pc->pcb); tcp_close(pc->pcb);
return 0; return 0;
} }
@ -877,7 +870,7 @@ int tcp_proxy_active_close(struct tcp_proxy* p, struct proxy_conn* pc) {
int tcp_proxy_active_send_done(struct tcp_proxy* p, struct proxy_conn* pc) { int tcp_proxy_active_send_done(struct tcp_proxy* p, struct proxy_conn* pc) {
(void)p; (void)p;
if (!pc || !pc->active || !pc->pcb) return 1; if (!pc || !pc->active || !pc->pcb) return 1;
if (pc->closing) return 1; if (pc->closing_tun) return 1;
int q = queue_entry_count(pc->transport_to_uip); int q = queue_entry_count(pc->transport_to_uip);
if (q > 0) return 0; if (q > 0) return 0;
if (pc->pcb->unsent || pc->pcb->unacked) return 0; if (pc->pcb->unsent || pc->pcb->unacked) return 0;

4
src/tcp_proxy.h
Unescape View File

@ -57,8 +57,8 @@ struct proxy_conn {
uint8_t dest_ip[4]; // proxy target IP (where to forward OS socket) uint8_t dest_ip[4]; // proxy target IP (where to forward OS socket)
uint32_t tun_ip; // TUN IP (what client connected to, for uip_hostaddr) uint32_t tun_ip; // TUN IP (what client connected to, for uip_hostaddr)
uint16_t dest_port; uint16_t dest_port;
int closing; int closing_tun; // 1 = TUN/lwIP side sent FIN (waiting for remote)
int half_closed; int closing_rem; // 1 = remote/transport closed (waiting for TUN)
int active; // 1 = outgoing connection int active; // 1 = outgoing connection
struct tcp_proxy_mapping* eim_mapping; struct tcp_proxy_mapping* eim_mapping;
uint64_t remote_stream_id; // stream ID for remote proxy (0=local) uint64_t remote_stream_id; // stream ID for remote proxy (0=local)

2
tests/test_tcp_proxy.c
Unescape View File

@ -115,7 +115,7 @@ int main(void) {
uasync_poll(ua, 100); timeout_ms -= 10; uasync_poll(ua, 100); timeout_ms -= 10;
ssize_t n = tcp_proxy_active_recv(a, pc, recv_buf + total_rcvd, TEST_SIZE - total_rcvd); ssize_t n = tcp_proxy_active_recv(a, pc, recv_buf + total_rcvd, TEST_SIZE - total_rcvd);
if(n > 0) total_rcvd += n; if(n > 0) total_rcvd += n;
if(pc->closing) break; if(pc->closing_rem && total_rcvd < TEST_SIZE) break;
} }
// 11. Half-close after all data received // 11. Half-close after all data received

Write Preview
Loading…
Cancel
Save