commit 98fcb66a8816d6fffde95ec586be37db849751f0
| author | 斟酌 鵬兄 <tgckpg@gmail.com> |
|---|
| date | 2026-05-25T16:57:58Z |
|---|
| subject | Prime our code structure for pthread support in the future |
|---|
commit 98fcb66a8816d6fffde95ec586be37db849751f0
Author: 斟酌 鵬兄 <tgckpg@gmail.com>
Date: 2026-05-25T16:57:58Z
Prime our code structure for pthread support in the future
---
Makefile | 4 +-
file_conf.c | 42 ++++++-------
proxy_proto_v2.c | 4 +-
proxy_proto_v2.h | 10 +--
route.h | 41 +++++++++----
tcp_route.c | 173 ++++++++++++++++++++++++++++++----------------------
tcp_route.h | 6 +-
tests/test_proxy.py | 3 +-
tinyproxy.c | 7 ++-
9 files changed, 171 insertions(+), 119 deletions(-)
diff --git a/Makefile b/Makefile
index 94761c3..ce78eaf 100644
--- a/Makefile
+++ b/Makefile
@@ -27,8 +27,10 @@ LDFLAGS += -L$(LIBEVENT_PREFIX)/lib
ifeq ($(shell uname),Darwin)
LDFLAGS += -Wl,-dead_strip
+TEST_FLAGS := CONCURRENCY=1000 TOTAL=1000 FD_LIMIT=2560
else
LDFLAGS += -Wl,--gc-sections
+TEST_FLAGS :=
endif
LDLIBS += -levent_core
@@ -59,7 +61,7 @@ $(LIBEVENT_CORE_A):
AR=$(AR) RANLIB=$(RANLIB) $(MAKE) -C $(LIBEVENT_SRC) install
test: $(BIN)
- python3 tests/test_proxy.py $(BIN)
+ $(TEST_FLAGS) python3 tests/test_proxy.py $(BIN)
clean:
rm -rf $(BIN_DIR)
diff --git a/file_conf.c b/file_conf.c
index 2ef5ab1..a1f1ac3 100644
--- a/file_conf.c
+++ b/file_conf.c
@@ -44,34 +44,34 @@ static int parse_port(const char *s, uint16_t *out)
}
static int split_host_port(const char *input,
- char *host, size_t host_len,
- uint16_t *port)
+ char *host, size_t host_len,
+ uint16_t *port)
{
- const char *colon = strrchr(input, ':');
- if (!colon) {
- return -1;
- }
+ const char *colon = strrchr(input, ':');
+ if (!colon) {
+ return -1;
+ }
- size_t hlen = (size_t)(colon - input);
- const char *port_s = colon + 1;
+ size_t hlen = (size_t)(colon - input);
+ const char *port_s = colon + 1;
- if (hlen >= host_len || *port_s == '\0') {
- return -1;
- }
+ if (hlen >= host_len || *port_s == '\0') {
+ return -1;
+ }
- memcpy(host, input, hlen);
- host[hlen] = '\0';
+ memcpy(host, input, hlen);
+ host[hlen] = '\0';
- if (parse_port(port_s, port) != 0) {
- return -1;
- }
+ if (parse_port(port_s, port) != 0) {
+ return -1;
+ }
- // Allow ":80" as shorthand for all interfaces.
- if (host[0] == '\0') {
- snprintf(host, host_len, "0.0.0.0");
- }
+ // Allow ":80" as shorthand for all interfaces.
+ if (host[0] == '\0') {
+ snprintf(host, host_len, "0.0.0.0");
+ }
- return 0;
+ return 0;
}
static int parse_proto(const char *s, enum proto *out)
diff --git a/proxy_proto_v2.c b/proxy_proto_v2.c
index 8be5bdc..168afe5 100644
--- a/proxy_proto_v2.c
+++ b/proxy_proto_v2.c
@@ -47,7 +47,7 @@ int proxy_v2_build(
switch (src->sa_family) {
case AF_INET:
if (src_len < sizeof(struct sockaddr_in) ||
- dst_len < sizeof(struct sockaddr_in)) {
+ dst_len < sizeof(struct sockaddr_in)) {
return -EINVAL;
}
@@ -84,7 +84,7 @@ int proxy_v2_build(
case AF_INET6:
if (src_len < sizeof(struct sockaddr_in6) ||
- dst_len < sizeof(struct sockaddr_in6)) {
+ dst_len < sizeof(struct sockaddr_in6)) {
return -EINVAL;
}
diff --git a/proxy_proto_v2.h b/proxy_proto_v2.h
index 0a3ed47..03e3435 100644
--- a/proxy_proto_v2.h
+++ b/proxy_proto_v2.h
@@ -5,12 +5,12 @@
#define PP2_VERSION_CMD_PROXY 0x21
-#define PP2_FAM_INET 0x10
-#define PP2_FAM_INET6 0x20
-#define PP2_FAM_UNIX 0x30
+#define PP2_FAM_INET 0x10
+#define PP2_FAM_INET6 0x20
+#define PP2_FAM_UNIX 0x30
-#define PP2_TRANS_STREAM 0x01
-#define PP2_TRANS_DGRAM 0x02
+#define PP2_TRANS_STREAM 0x01
+#define PP2_TRANS_DGRAM 0x02
int proxy_v2_build(
unsigned char *buf,
diff --git a/route.h b/route.h
index 90ecea0..0dfbb0d 100644
--- a/route.h
+++ b/route.h
@@ -5,30 +5,49 @@
#include <stdbool.h>
enum proto {
- PROTO_TCP,
- PROTO_UDP,
+ PROTO_TCP,
+ PROTO_UDP,
};
struct route {
- char listen_host[64];
+ char listen_host[64];
uint16_t listen_port;
- char upstream_host[64];
+ char upstream_host[64];
uint16_t upstream_port;
- enum proto proto;
- bool send_proxy_v2;
+ enum proto proto;
+ bool send_proxy_v2;
+};
+
+struct worker {
+ struct event_base *base;
+ size_t id;
+};
+
+struct accepted_client {
+ evutil_socket_t fd;
+ struct sockaddr_storage peer_addr;
+ socklen_t peer_addr_len;
+ const struct route *route;
};
typedef struct conn_s {
- struct bufferevent *client;
- struct bufferevent *upstream;
+ struct worker *owner;
+ const struct route *route;
+
+ struct bufferevent *client;
+ struct bufferevent *upstream;
+
+ struct sockaddr_storage peer_addr;
+ socklen_t peer_addr_len;
} conn_t;
struct listener_ctx {
- struct event_base *base;
- const struct route *route;
- struct evconnlistener *listener;
+ struct event_base *accept_base;
+ struct worker *worker;
+ const struct route *route;
+ struct evconnlistener *listener;
};
#endif
diff --git a/tcp_route.c b/tcp_route.c
index c09abaf..a57ce61 100644
--- a/tcp_route.c
+++ b/tcp_route.c
@@ -73,27 +73,23 @@ static void event_cb(struct bufferevent *bev, short events, void *arg) {
(void)bev;
}
-static void accept_cb(
- struct evconnlistener *listener,
- evutil_socket_t client_fd,
- struct sockaddr *addr,
- int socklen,
- void *arg
-) {
- (void)listener;
-
- struct listener_ctx *ctx = arg;
- const struct route *r = ctx->route;
- struct event_base *base = ctx->base;
-
+static void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac) {
conn_t *conn = calloc(1, sizeof(*conn));
if (conn == NULL) {
- evutil_closesocket(client_fd);
+ evutil_closesocket(ac->fd);
return;
}
- conn->client = bufferevent_socket_new(base, client_fd, BEV_OPT_CLOSE_ON_FREE);
- conn->upstream = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);
+ const struct route *r = ac->route;
+
+ conn->owner = w;
+ conn->route = r;
+
+ conn->peer_addr = ac->peer_addr;
+ conn->peer_addr_len = ac->peer_addr_len;
+
+ conn->client = bufferevent_socket_new(w->base, ac->fd, BEV_OPT_CLOSE_ON_FREE);
+ conn->upstream = bufferevent_socket_new(w->base, -1, BEV_OPT_CLOSE_ON_FREE);
if (conn->client == NULL || conn->upstream == NULL) {
free_conn(conn);
@@ -151,20 +147,20 @@ static void accept_cb(
memset(&local_addr, 0, sizeof(local_addr));
- if (getsockname(client_fd, (struct sockaddr *)&local_addr, &local_len) < 0) {
+ if (getsockname(ac->fd, (struct sockaddr *)&local_addr, &local_len) < 0) {
perror("getsockname");
free_conn(conn);
return;
}
- if (addr == NULL || socklen < (int)sizeof(struct sockaddr_in)) {
+ if (0 < ac->peer_addr_len) {
fprintf(stderr, "invalid client address\n");
free_conn(conn);
return;
}
- if (((struct sockaddr *)addr)->sa_family != AF_INET ||
- local_addr.sin_family != AF_INET) {
+ if (ac->peer_addr.ss_family != AF_INET ||
+ local_addr.sin_family != AF_INET) {
fprintf(stderr, "PROXY v2 currently only supports IPv4 TCP\n");
free_conn(conn);
return;
@@ -172,8 +168,8 @@ static void accept_cb(
if (proxy_v2_write_bufferevent(
conn->upstream,
- addr,
- (socklen_t)socklen,
+ (const struct sockaddr *)&ac->peer_addr,
+ ac->peer_addr_len,
(struct sockaddr *)&local_addr,
local_len,
SOCK_STREAM
@@ -188,6 +184,34 @@ static void accept_cb(
bufferevent_enable(conn->upstream, EV_READ | EV_WRITE);
}
+static void dispatch_client_fd(struct worker *w, struct accepted_client *ac) {
+ worker_adopt_client_fd(w, ac);
+}
+
+static void accept_cb(
+ struct evconnlistener *listener,
+ evutil_socket_t client_fd,
+ struct sockaddr *addr,
+ int socklen,
+ void *arg
+) {
+ (void)listener;
+
+ struct listener_ctx *ctx = arg;
+ struct accepted_client ac = {
+ .fd = client_fd,
+ .route = ctx->route,
+ };
+
+ if(addr && 0 < socklen && (size_t)socklen <= sizeof(ac.peer_addr)) {
+ memcpy(&ac.peer_addr, addr, (size_t)socklen);
+ } else {
+ ac.peer_addr_len = 0;
+ }
+
+ dispatch_client_fd(ctx->worker, &ac);
+}
+
static void accept_error_cb(struct evconnlistener *listener, void *arg) {
struct event_base *base = arg;
int err = EVUTIL_SOCKET_ERROR();
@@ -200,63 +224,64 @@ static void accept_error_cb(struct evconnlistener *listener, void *arg) {
}
int start_tcp_route(
- struct event_base *base,
- const struct route *r,
- struct listener_ctx **out)
+ struct worker *w,
+ const struct route *r,
+ struct listener_ctx **out)
{
- struct sockaddr_in listen_addr;
- memset(&listen_addr, 0, sizeof(listen_addr));
- listen_addr.sin_family = AF_INET;
- listen_addr.sin_port = htons(r->listen_port);
-
- if (inet_pton(AF_INET, r->listen_host, &listen_addr.sin_addr) != 1) {
- fprintf(stderr, "invalid listen address: %s\n", r->listen_host);
- return -EINVAL;
- }
-
- struct listener_ctx *ctx = calloc(1, sizeof(*ctx));
- if (ctx == NULL) {
- return -ENOMEM;
- }
-
- ctx->base = base;
- ctx->route = r;
-
- ctx->listener = evconnlistener_new_bind(
- base,
- accept_cb,
- ctx,
- LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE,
- 128,
- (struct sockaddr *)&listen_addr,
- sizeof(listen_addr)
- );
-
- if (ctx->listener == NULL) {
- fprintf(stderr, "evconnlistener_new_bind failed\n");
- free(ctx);
- return -EADDRINUSE;
- }
-
- evconnlistener_set_error_cb(ctx->listener, accept_error_cb);
-
- fprintf(stderr, "listening on %s:%u, forwarding to %s:%u\n",
- r->listen_host, r->listen_port,
- r->upstream_host, r->upstream_port);
-
- *out = ctx;
- return 0;
+ struct sockaddr_in listen_addr;
+ memset(&listen_addr, 0, sizeof(listen_addr));
+ listen_addr.sin_family = AF_INET;
+ listen_addr.sin_port = htons(r->listen_port);
+
+ if (inet_pton(AF_INET, r->listen_host, &listen_addr.sin_addr) != 1) {
+ fprintf(stderr, "invalid listen address: %s\n", r->listen_host);
+ return -EINVAL;
+ }
+
+ struct listener_ctx *ctx = calloc(1, sizeof(*ctx));
+ if (ctx == NULL) {
+ return -ENOMEM;
+ }
+
+ ctx->accept_base = w->base;
+ ctx->worker = w;
+ ctx->route = r;
+
+ ctx->listener = evconnlistener_new_bind(
+ ctx->accept_base,
+ accept_cb,
+ ctx,
+ LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE,
+ 128,
+ (struct sockaddr *)&listen_addr,
+ sizeof(listen_addr)
+ );
+
+ if (ctx->listener == NULL) {
+ fprintf(stderr, "evconnlistener_new_bind failed\n");
+ free(ctx);
+ return -EADDRINUSE;
+ }
+
+ evconnlistener_set_error_cb(ctx->listener, accept_error_cb);
+
+ fprintf(stderr, "listening on %s:%u, forwarding to %s:%u\n",
+ r->listen_host, r->listen_port,
+ r->upstream_host, r->upstream_port);
+
+ *out = ctx;
+ return 0;
}
void free_tcp_route(struct listener_ctx *ctx)
{
- if (ctx == NULL) {
- return;
- }
+ if (ctx == NULL) {
+ return;
+ }
- if (ctx->listener != NULL) {
- evconnlistener_free(ctx->listener);
- }
+ if (ctx->listener != NULL) {
+ evconnlistener_free(ctx->listener);
+ }
- free(ctx);
+ free(ctx);
}
diff --git a/tcp_route.h b/tcp_route.h
index 99e2cc5..32b5a0b 100644
--- a/tcp_route.h
+++ b/tcp_route.h
@@ -4,9 +4,9 @@
#include "route.h"
int start_tcp_route(
- struct event_base *base,
- const struct route *r,
- struct listener_ctx **out);
+ struct worker *w,
+ const struct route *r,
+ struct listener_ctx **out);
void free_tcp_route(struct listener_ctx *ctx);
#endif
diff --git a/tests/test_proxy.py b/tests/test_proxy.py
index 5a9e5cd..148c660 100755
--- a/tests/test_proxy.py
+++ b/tests/test_proxy.py
@@ -157,7 +157,8 @@ async def test_concurrent_connections(
async def main_async(proxy_bin: str) -> int:
- raise_fd_limit(65535)
+ fd_limit = int(os.environ.get("FD_LIMIT", "65535"))
+ raise_fd_limit(fd_limit)
backend_server = await asyncio.start_server(
echo_handler,
diff --git a/tinyproxy.c b/tinyproxy.c
index c4b9416..6eb89ca 100644
--- a/tinyproxy.c
+++ b/tinyproxy.c
@@ -86,12 +86,17 @@ int main(int argc, char **argv)
return 1;
}
+ struct worker w = {
+ .base = base,
+ .id = 0,
+ };
+
for (size_t i = 0; i < route_count; i++) {
const struct route *r = &routes[i];
switch (r->proto) {
case PROTO_TCP:
- rc = start_tcp_route(base, r, &tcp_ctxs[tcp_ctx_count]);
+ rc = start_tcp_route(&w, r, &tcp_ctxs[tcp_ctx_count]);
if (rc == 0) {
tcp_ctx_count++;
}