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

add non-blocking send with EAGAIN+retry timer, normalizer flow control on recv, remove client-side send_buf buffering

congestion
Evgeny 4 months ago
parent
c4e80e6fd7
commit
0f0857c22a
6 changed files with 165 additions and 52 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. 7
      src/etcp_router.c
  2. 3
      src/etcp_router.h
  3. 139
      src/remote_proxy.c
  4. 14
      src/remote_proxy.h
  5. 52
      src/tcp_proxy.c
  6. 2
      src/tcp_proxy.h

7
src/etcp_router.c
Unescape View File

@ -153,3 +153,10 @@ int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_
svc_id, (unsigned long long)dst_node_id, conn->log_name, total_len); svc_id, (unsigned long long)dst_node_id, conn->log_name, total_len);
return etcp_send(conn, new_entry); return etcp_send(conn, new_entry);
} }
int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t node_id) {
if (!inst || !inst->bgp) return -1;
struct ETCP_CONN* conn = route_bgp_find_conn_for_node(inst->bgp, node_id);
if (!conn || !conn->normalizer || !conn->normalizer->input) return -1;
return queue_entry_count(conn->normalizer->input);
}

3
src/etcp_router.h
Unescape View File

@ -35,4 +35,7 @@ int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id);
// Принимает ownership entry. // Принимает ownership entry.
int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_entry* entry); int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_entry* entry);
// Возвращает количество пакетов в очереди normalizer->input для узла node_id, -1 если нет соединения
int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t node_id);
#endif // ETCP_ROUTER_H #endif // ETCP_ROUTER_H

139
src/remote_proxy.c
Unescape View File

@ -31,8 +31,13 @@ static struct remote_proxy_ctx* g_rp_ctx = NULL;
static void rp_sock_read_cb(socket_t sock, void* arg); static void rp_sock_read_cb(socket_t sock, void* arg);
static void rp_sock_write_cb(socket_t sock, void* arg); static void rp_sock_write_cb(socket_t sock, void* arg);
static void rp_sock_error_cb(socket_t sock, void* arg); static void rp_sock_error_cb(socket_t sock, void* arg);
static void rp_sock_retry_cb(void* arg);
static void rp_sock_pause_cb(void* arg);
static void rp_sock_send(struct remote_proxy_conn* rc, const uint8_t* data, size_t len);
void rp_conn_free(struct remote_proxy_conn* rc); void rp_conn_free(struct remote_proxy_conn* rc);
#define RP_NORMALIZER_Q_THRESHOLD 64
static int rp_send_msg(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t subcmd, static int rp_send_msg(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t subcmd,
uint32_t sid, uint16_t seq, const uint8_t* data, size_t len) { uint32_t sid, uint16_t seq, const uint8_t* data, size_t len) {
struct ll_entry* e = queue_entry_new(0); struct ll_entry* e = queue_entry_new(0);
@ -62,6 +67,89 @@ static int rp_conn_total(struct remote_proxy_conn* rc) {
return n; return n;
} }
// ====================================================================
// Неблокирующий send с EAGAIN и таймером повтора
// ====================================================================
static int rp_sock_try_send(struct remote_proxy_conn* rc) {
if (!rc || rc->sock == SOCKET_INVALID || !rc->out_buf) return 0;
while (rc->out_off < rc->out_len) {
ssize_t n = send(rc->sock, rc->out_buf + rc->out_off, rc->out_len - rc->out_off, MSG_NOSIGNAL);
if (n > 0) {
rc->out_off += n;
if (rc->out_off >= rc->out_len) {
u_free(rc->out_buf); rc->out_buf = NULL; rc->out_len = rc->out_off = 0;
return 0;
}
continue;
}
if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) return 1;
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:SEND_ERR fd=%d sid=%08x err=%s", (int)rc->sock, rc->stream_id, strerror(errno));
return -1;
}
return 0;
}
static void rp_sock_retry_cb(void* arg) {
struct remote_proxy_conn* rc = (struct remote_proxy_conn*)arg;
if (!rc || !rc->out_buf) { rc->out_timer = NULL; return; }
int ret = rp_sock_try_send(rc);
if (ret == 0) {
rc->out_timer = NULL; rc->out_backoff = 0;
} else if (ret == 1) {
rc->out_backoff = rc->out_backoff < 20 ? 20 : rc->out_backoff * 2;
if (rc->out_backoff > 5000) rc->out_backoff = 5000;
rc->out_timer = uasync_set_timeout(rc->ua, rc->out_backoff, rc, rp_sock_retry_cb, "rp_send");
} else {
u_free(rc->out_buf); rc->out_buf = NULL; rc->out_len = rc->out_off = 0;
rc->out_timer = NULL; rc->out_backoff = 0;
}
}
static void rp_sock_send(struct remote_proxy_conn* rc, const uint8_t* data, size_t len) {
if (!rc || !data || len == 0) return;
if (rc->out_buf) {
size_t new_len = rc->out_len + len;
uint8_t* new_buf = u_realloc(rc->out_buf, new_len);
if (!new_buf) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:SEND realloc(%zu) failed sid=%08x", new_len, rc->stream_id); return; }
memcpy(new_buf + rc->out_len, data, len);
rc->out_buf = new_buf; rc->out_len = new_len;
return;
}
rc->out_buf = u_malloc(len);
if (!rc->out_buf) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:SEND malloc(%zu) failed sid=%08x", len, rc->stream_id); return; }
memcpy(rc->out_buf, data, len); rc->out_len = len; rc->out_off = 0;
int ret = rp_sock_try_send(rc);
if (ret == 0) return;
if (ret == 1) {
rc->out_backoff = 10;
rc->out_timer = uasync_set_timeout(rc->ua, rc->out_backoff, rc, rp_sock_retry_cb, "rp_send");
} else {
u_free(rc->out_buf); rc->out_buf = NULL; rc->out_len = rc->out_off = 0;
}
}
static void rp_sock_pause_cb(void* arg) {
struct remote_proxy_conn* rc = (struct remote_proxy_conn*)arg;
if (!rc || rc->sock == SOCKET_INVALID) { rc->pause_timer = NULL; return; }
struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
int qcnt = -1;
if (inst) qcnt = etcp_router_input_q_count(inst, rc->peer_node_id);
if (qcnt > RP_NORMALIZER_Q_THRESHOLD) {
rc->pause_backoff = rc->pause_backoff * 2;
if (rc->pause_backoff > 5000) rc->pause_backoff = 5000;
rc->pause_timer = uasync_set_timeout(rc->ua, rc->pause_backoff, rc, rp_sock_pause_cb, "rp_pause");
return;
}
if (rc->pause_buf && rc->pause_len > 0) {
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:PAUSE_FLUSH fd=%d sid=%08x len=%zu qcnt=%d", (int)rc->sock, rc->stream_id, rc->pause_len, qcnt);
rp_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->send_seq, rc->pause_buf, rc->pause_len);
rc->send_seq++;
u_free(rc->pause_buf); rc->pause_buf = NULL; rc->pause_len = 0;
}
rc->read_id = uasync_add_socket_t(rc->ua, rc->sock, rp_sock_read_cb, rp_sock_write_cb, rp_sock_error_cb, rc);
rc->pause_timer = NULL;
}
// ==================================================================== // ====================================================================
// Socket callbacks // Socket callbacks
// ==================================================================== // ====================================================================
@ -73,11 +161,20 @@ static void rp_sock_read_cb(socket_t sock, void* arg) {
if (rc->cli_closed || rc->error) return; if (rc->cli_closed || rc->error) return;
struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:RECV fd=%d sid=%08x — inst is NULL, drop", (int)rc->sock, rc->stream_id); return; } if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:RECV fd=%d sid=%08x — inst is NULL, drop", (int)rc->sock, rc->stream_id); return; }
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:RECV fd=%d sid=%08x len=%zd total=%d", (int)rc->sock, rc->stream_id, n, rp_conn_total(rc)); int qcnt = etcp_router_input_q_count(inst, rc->peer_node_id);
if (!rc->error) { if (qcnt > RP_NORMALIZER_Q_THRESHOLD) {
rp_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->send_seq, buf, (size_t)n); DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:PAUSE fd=%d sid=%08x len=%zd qcnt=%d — normalizer переполнен, пауза", (int)rc->sock, rc->stream_id, n, qcnt);
rc->send_seq++; rc->pause_buf = u_malloc(n);
if (rc->pause_buf) { memcpy(rc->pause_buf, buf, n); rc->pause_len = n; }
else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:PAUSE malloc(%zd) failed sid=%08x", n, rc->stream_id);
if (rc->read_id) { uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL; }
rc->pause_backoff = 50;
rc->pause_timer = uasync_set_timeout(rc->ua, rc->pause_backoff, rc, rp_sock_pause_cb, "rp_pause");
return;
} }
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:RECV fd=%d sid=%08x len=%zd total=%d", (int)rc->sock, rc->stream_id, n, rp_conn_total(rc));
rp_send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->send_seq, buf, (size_t)n);
rc->send_seq++;
} else if (n == 0) { } else if (n == 0) {
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:EOF fd=%d sid=%08x total=%d cli_closed=%d sock_closed=%d", DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:EOF fd=%d sid=%08x total=%d cli_closed=%d sock_closed=%d",
(int)rc->sock, rc->stream_id, rp_conn_total(rc), rc->cli_closed, rc->sock_closed); (int)rc->sock, rc->stream_id, rp_conn_total(rc), rc->cli_closed, rc->sock_closed);
@ -113,6 +210,11 @@ static void rp_sock_write_cb(socket_t sock, void* arg) {
rp_conn_total(rc)); rp_conn_total(rc));
struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
if (inst) rp_send_connected(inst, rc->peer_node_id, rc->stream_id, local_port, TCP_PROXY_CONNECTED_OK); if (inst) rp_send_connected(inst, rc->peer_node_id, rc->stream_id, local_port, TCP_PROXY_CONNECTED_OK);
if (rc->pending_buf && rc->pending_len > 0) {
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:FLUSH fd=%d sid=%08x len=%zu", (int)rc->sock, rc->stream_id, rc->pending_len);
rp_sock_send(rc, rc->pending_buf, rc->pending_len);
u_free(rc->pending_buf); rc->pending_buf = NULL; rc->pending_len = 0;
}
} else { } else {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:FAIL fd=%d sid=%08x err=%d %s", DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "SOCK:FAIL fd=%d sid=%08x err=%d %s",
(int)rc->sock, rc->stream_id, err, err ? strerror(err) : "unknown"); (int)rc->sock, rc->stream_id, err, err ? strerror(err) : "unknown");
@ -146,6 +248,11 @@ void rp_conn_free(struct remote_proxy_conn* rc) {
if (rc->read_id) { uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL; } if (rc->read_id) { uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL; }
socket_close_wrapper(rc->sock); rc->sock = SOCKET_INVALID; socket_close_wrapper(rc->sock); rc->sock = SOCKET_INVALID;
} }
if (rc->pending_buf) { u_free(rc->pending_buf); rc->pending_buf = NULL; }
if (rc->out_timer) { uasync_cancel_timeout(rc->ua, rc->out_timer); rc->out_timer = NULL; }
if (rc->out_buf) { u_free(rc->out_buf); rc->out_buf = NULL; }
if (rc->pause_timer) { uasync_cancel_timeout(rc->ua, rc->pause_timer); rc->pause_timer = NULL; }
if (rc->pause_buf) { u_free(rc->pause_buf); rc->pause_buf = NULL; }
u_free(rc); u_free(rc);
} }
@ -201,8 +308,8 @@ int remote_proxy_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn,
if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return -1; } if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return -1; }
struct remote_proxy_ctx* ctx = &inst->remote_proxy; struct remote_proxy_ctx* ctx = &inst->remote_proxy;
struct remote_proxy_conn* rc = remote_proxy_find_conn(ctx, stream_id); struct remote_proxy_conn* rc = remote_proxy_find_conn(ctx, stream_id);
if (!rc || rc->sock == SOCKET_INVALID || rc->connected != 1) { if (!rc || rc->sock == SOCKET_INVALID) {
DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "RP handle_data drop: нет активного соединения для sid=%08x, шлём CLOSE", stream_id); DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "RP handle_data drop: нет соединения для sid=%08x, шлём CLOSE", stream_id);
if (conn) rp_send_msg(inst, conn->peer_node_id, TCP_PROXY_SUBCMD_CLOSE, stream_id, 0, NULL, 0); if (conn) rp_send_msg(inst, conn->peer_node_id, TCP_PROXY_SUBCMD_CLOSE, stream_id, 0, NULL, 0);
queue_dgram_free(entry); queue_entry_free(entry); return -1; queue_dgram_free(entry); queue_entry_free(entry); return -1;
} }
@ -227,14 +334,18 @@ int remote_proxy_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn,
rc->recv_last_seq = seq; rc->recv_last_seq = seq;
} }
size_t data_len = entry->len - TCP_PROXY_HDR_SIZE; size_t data_len = entry->len - TCP_PROXY_HDR_SIZE;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:SEND fd=%d sid=%08x len=%zu total=%d", if (rc->connected) {
(int)rc->sock, stream_id, data_len, rp_conn_total(rc)); DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:SEND fd=%d sid=%08x len=%zu total=%d",
if (data_len > 0) { (int)rc->sock, stream_id, data_len, rp_conn_total(rc));
uint8_t* data = entry->dgram + TCP_PROXY_HDR_SIZE; if (data_len > 0) rp_sock_send(rc, entry->dgram + TCP_PROXY_HDR_SIZE, data_len);
ssize_t n = send(rc->sock, data, data_len, MSG_NOSIGNAL); } else {
if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) { size_t new_len = rc->pending_len + data_len;
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "remote_proxy: send error %s", strerror(errno)); uint8_t* new_buf = u_realloc(rc->pending_buf, new_len);
} if (new_buf) {
if (data_len > 0) memcpy(new_buf + rc->pending_len, entry->dgram + TCP_PROXY_HDR_SIZE, data_len);
rc->pending_buf = new_buf; rc->pending_len = new_len;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "RP buffer sid=%08x added=%zu total=%zu", stream_id, data_len, new_len);
} else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "RP buffer realloc(%zu) failed sid=%08x", new_len, stream_id);
} }
queue_dgram_free(entry); queue_entry_free(entry); queue_dgram_free(entry); queue_entry_free(entry);
return 0; return 0;

14
src/remote_proxy.h
Unescape View File

@ -43,6 +43,20 @@ struct remote_proxy_conn {
uint8_t sock_closed; // socket EOF received uint8_t sock_closed; // socket EOF received
uint8_t error; uint8_t error;
uint8_t connected; // connect() completed successfully uint8_t connected; // connect() completed successfully
uint8_t* pending_buf; // буфер данных пока connect() не завершён
size_t pending_len;
uint8_t* out_buf; // буфер отправки в сокет (неблокирующий send)
size_t out_len;
size_t out_off;
void* out_timer; // таймер повтора send
int out_backoff; // backoff: 10..5000 tb (1ms..500ms)
uint8_t* pause_buf; // буфер при паузе чтения (normalizer переполнен)
size_t pause_len;
void* pause_timer; // таймер паузы чтения
int pause_backoff; // backoff: 50..5000 tb (5ms..500ms)
}; };
struct remote_proxy_ctx { struct remote_proxy_ctx {

52
src/tcp_proxy.c
Unescape View File

@ -88,7 +88,6 @@ static int send_connect(struct proxy_conn* pc) {
} }
static int send_data(struct proxy_conn* pc, const uint8_t* data, uint16_t len) { static int send_data(struct proxy_conn* pc, const uint8_t* data, uint16_t len) {
if (!pc->connected) return -1;
DEBUG_INFO(DEBUG_CATEGORY_TRAFFIC, "PROXY SEND sid=%08x seq=%u len=%u", pc->stream_id, pc->send_seq, len); DEBUG_INFO(DEBUG_CATEGORY_TRAFFIC, "PROXY SEND sid=%08x seq=%u len=%u", pc->stream_id, pc->send_seq, len);
int ret = proxy_send_msg(pc->proxy->inst, pc->proxy->via_node_id, int ret = proxy_send_msg(pc->proxy->inst, pc->proxy->via_node_id,
TCP_PROXY_SUBCMD_DATA, pc->stream_id, pc->send_seq, data, len); TCP_PROXY_SUBCMD_DATA, pc->stream_id, pc->send_seq, data, len);
@ -236,17 +235,13 @@ 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) {
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY FIN sid=%08x pcb_state=%u sndbuf=%u cwnd=%u unsent=%p unacked=%p send_buf_len=%u close_sent=%d", DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY FIN sid=%08x pcb_state=%u sndbuf=%u cwnd=%u unsent=%p unacked=%p close_sent=%d",
pc->stream_id, pcb->state, pcb->snd_buf, pcb->cwnd, pc->stream_id, pcb->state, pcb->snd_buf, pcb->cwnd,
(void*)pcb->unsent, (void*)pcb->unacked, pc->send_len, pc->close_sent); (void*)pcb->unsent, (void*)pcb->unacked, pc->close_sent);
{ uint16_t wnd_gap = TCP_WND_MAX(pcb) - pcb->rcv_wnd; { uint16_t wnd_gap = TCP_WND_MAX(pcb) - pcb->rcv_wnd;
if (wnd_gap > 0) tcp_recved(pcb, wnd_gap); } if (wnd_gap > 0) tcp_recved(pcb, wnd_gap); }
pc->tun_closed = 1; pc->tun_closed = 1;
if (!pc->close_sent) { if (!pc->close_sent) {
if (pc->send_buf && pc->send_len > 0) {
if (send_data(pc, pc->send_buf, pc->send_len) < 0) DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY FIN send_data failed sid=%08x", pc->stream_id);
u_free(pc->send_buf); pc->send_buf = NULL; pc->send_len = 0;
}
if (send_close(pc) < 0) DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY FIN send_close failed sid=%08x", pc->stream_id); if (send_close(pc) < 0) DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY FIN send_close failed sid=%08x", pc->stream_id);
pc->close_sent = 1; pc->close_sent = 1;
} }
@ -258,22 +253,13 @@ static err_t proxy_recv_cb(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err_t
tcp_recved(pcb, len); pbuf_free(p); return LERR_OK; tcp_recved(pcb, len); pbuf_free(p); return LERR_OK;
} }
if (pc->connected) { uint8_t *data = u_malloc(len);
uint8_t *data = u_malloc(len); if (data) {
if (data) { pbuf_copy_partial(p, data, len, 0);
pbuf_copy_partial(p, data, len, 0); if (send_data(pc, data, len) < 0) DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY recv send_data failed sid=%08x len=%u", pc->stream_id, len);
if (send_data(pc, data, len) < 0) DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY recv send_data failed sid=%08x len=%u", pc->stream_id, len); u_free(data);
u_free(data); } else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY recv malloc(%u) failed sid=%08x", len, pc->stream_id);
} else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY recv malloc(%u) failed sid=%08x", len, pc->stream_id); DEBUG_INFO(DEBUG_CATEGORY_TRAFFIC, "PROXY RECV client->exit sid=%08x len=%u", pc->stream_id, len);
DEBUG_INFO(DEBUG_CATEGORY_TRAFFIC, "PROXY RECV client->exit sid=%08x len=%u connected=%d", pc->stream_id, len, pc->connected);
} else {
uint16_t new_len = pc->send_len + len;
uint8_t *new_buf = u_realloc(pc->send_buf, new_len);
if (new_buf) {
pbuf_copy_partial(p, new_buf + pc->send_len, len, 0);
pc->send_buf = new_buf; pc->send_len = new_len;
} else DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY recv realloc(%u) failed sid=%08x", new_len, pc->stream_id);
}
tcp_recved(pcb, len); pbuf_free(p); tcp_recved(pcb, len); pbuf_free(p);
return LERR_OK; return LERR_OK;
} }
@ -289,8 +275,8 @@ static err_t proxy_sent_cb(void *arg, struct tcp_pcb *pcb, uint16_t len) {
static void proxy_err_cb(void *arg, err_t err) { 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) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY proxy_err_cb: pc=NULL err=%d", err); return; } if (!pc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY proxy_err_cb: pc=NULL err=%d", err); return; }
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: error %d sid=%08x pcb_state=%u tun_closed=%d rem_closed=%d connected=%d send_buf_len=%u", DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: error %d sid=%08x pcb_state=%u tun_closed=%d rem_closed=%d connected=%d",
err, pc->stream_id, pc->pcb ? pc->pcb->state : 0, pc->tun_closed, pc->rem_closed, pc->connected, pc->send_len); err, pc->stream_id, pc->pcb ? pc->pcb->state : 0, pc->tun_closed, pc->rem_closed, pc->connected);
pc->error = 1; pc->error = 1;
if (send_error(pc) < 0) DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY send_error failed sid=%08x", pc->stream_id); if (send_error(pc) < 0) DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "PROXY send_error failed sid=%08x", pc->stream_id);
pc->close_sent = 1; pc->close_sent = 1;
@ -382,11 +368,10 @@ static void proxy_conn_free(struct proxy_conn *pc) {
if (!pc) return; if (!pc) return;
struct tcp_proxy *p = pc->proxy; struct tcp_proxy *p = pc->proxy;
uint32_t pending = pc->to_lwip ? queue_entry_count(pc->to_lwip) : 0; uint32_t pending = pc->to_lwip ? queue_entry_count(pc->to_lwip) : 0;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY FREE sid=%08x total_conns=%d send_buf=%u to_lwip_q=%u", DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY FREE sid=%08x total_conns=%d to_lwip_q=%u",
pc->stream_id, p->conn_count, pc->send_len, pending); pc->stream_id, p->conn_count, pending);
struct proxy_conn **prev = &p->conns; struct proxy_conn **prev = &p->conns;
while (*prev) { if (*prev == pc) { *prev = pc->next; p->conn_count--; break; } prev = &(*prev)->next; } while (*prev) { if (*prev == pc) { *prev = pc->next; p->conn_count--; break; } prev = &(*prev)->next; }
if (pc->send_buf) { u_free(pc->send_buf); pc->send_buf = NULL; }
if (pc->to_lwip) { if (pc->to_lwip) {
struct ll_entry *e; struct ll_entry *e;
while ((e = queue_data_get(pc->to_lwip))) { queue_dgram_free(e); queue_entry_free(e); } while ((e = queue_data_get(pc->to_lwip))) { queue_dgram_free(e); queue_entry_free(e); }
@ -412,11 +397,7 @@ static void handle_connected(struct tcp_proxy* p, uint32_t stream_id, struct ll_
if (status == TCP_PROXY_CONNECTED_OK) { if (status == TCP_PROXY_CONNECTED_OK) {
pc->connected = 1; pc->connected = 1;
uint16_t local_port = 0; memcpy(&local_port, entry->dgram + TCP_PROXY_HDR_SIZE, 2); uint16_t local_port = 0; memcpy(&local_port, entry->dgram + TCP_PROXY_HDR_SIZE, 2);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY CONNECTED sid=%08x exit_port=%u buffered_data=%u", stream_id, ntohs(local_port), pc->send_len); DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY CONNECTED sid=%08x exit_port=%u", stream_id, ntohs(local_port));
if (pc->send_buf && pc->send_len > 0) {
send_data(pc, pc->send_buf, pc->send_len);
u_free(pc->send_buf); pc->send_buf = NULL; pc->send_len = 0;
}
} else { } else {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: remote refused sid=%08x", stream_id); DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: remote refused sid=%08x", stream_id);
pc->rem_closed = 1; pc->rem_closed = 1;
@ -465,8 +446,8 @@ static void handle_data(struct tcp_proxy* p, struct ETCP_CONN* conn, uint32_t st
static void handle_close(struct tcp_proxy* p, uint32_t stream_id) { static void handle_close(struct tcp_proxy* p, uint32_t stream_id) {
struct proxy_conn* pc = find_pc_by_stream(p, stream_id); struct proxy_conn* pc = find_pc_by_stream(p, stream_id);
if (pc) { if (pc) {
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY REM_CLOSED sid=%08x tun_closed=%d connected=%d send_buf_len=%u", DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY REM_CLOSED sid=%08x tun_closed=%d connected=%d",
stream_id, pc->tun_closed, pc->connected, pc->send_len); stream_id, pc->tun_closed, pc->connected);
pc->rem_closed = 1; pc->rem_closed = 1;
} else DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "PROXY CLOSE sid=%08x — no conn", stream_id); } else DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "PROXY CLOSE sid=%08x — no conn", stream_id);
} }
@ -588,7 +569,6 @@ void tcp_proxy_destroy(struct tcp_proxy* p) {
while (pc) { while (pc) {
struct proxy_conn* next = pc->next; struct proxy_conn* next = pc->next;
if (pc->pcb) { tcp_arg(pc->pcb, NULL); tcp_abort(pc->pcb); pc->pcb = NULL; } if (pc->pcb) { tcp_arg(pc->pcb, NULL); tcp_abort(pc->pcb); pc->pcb = NULL; }
if (pc->send_buf) { u_free(pc->send_buf); pc->send_buf = NULL; }
if (pc->to_lwip) { if (pc->to_lwip) {
struct ll_entry *e; struct ll_entry *e;
while ((e = queue_data_get(pc->to_lwip))) { queue_dgram_free(e); queue_entry_free(e); } while ((e = queue_data_get(pc->to_lwip))) { queue_dgram_free(e); queue_entry_free(e); }

2
src/tcp_proxy.h
Unescape View File

@ -30,8 +30,6 @@ struct proxy_conn {
uint8_t recv_init; // первый DATA принят uint8_t recv_init; // первый DATA принят
struct ll_queue* to_lwip; // DATA от exit → lwIP (управление потоком) struct ll_queue* to_lwip; // DATA от exit → lwIP (управление потоком)
uint8_t* send_buf; // данные lwIP→exit до CONNECTED
uint16_t send_len;
uint8_t tun_closed; // сторона lwIP отправила FIN uint8_t tun_closed; // сторона lwIP отправила FIN
uint8_t rem_closed; // exit отправил CLOSE uint8_t rem_closed; // exit отправил CLOSE

Write Preview
Loading…
Cancel
Save