commit 923d73f9fa4678ef5aa7404e2fc2c98f788e00f2
| author | 斟酌 鵬兄 <tgckpg@gmail.com> |
|---|
| date | 2026-05-30T06:24:54Z |
|---|
| subject | Wiring pthread |
|---|
commit 923d73f9fa4678ef5aa7404e2fc2c98f788e00f2
Author: 斟酌 鵬兄 <tgckpg@gmail.com>
Date: 2026-05-30T06:24:54Z
Wiring pthread
---
mk/common.mk | 2 +
src/compat_thread.c | 151 ++++++++++++++++++++
src/compat_thread.h | 41 ++++++
src/datagram_client.h | 12 ++
src/datagram_listener.c | 163 +++++++++++++---------
src/datagram_route.c | 14 +-
src/datagram_route.h | 10 +-
src/proxy_proto_v2.c | 4 -
src/route_runtime.c | 10 +-
src/route_runtime.h | 7 +-
src/stream_conn.c | 13 +-
src/stream_conn.h | 4 +-
src/stream_listener.c | 12 +-
src/stream_route.c | 11 +-
src/stream_route.h | 17 ++-
src/tinyproxy.c | 37 ++++-
src/worker.c | 364 ++++++++++++++++++++++++++++++++++++++++++++++++
src/worker.h | 41 +++++-
src/worker_pool.c | 121 ++++++++++++++++
src/worker_pool.h | 22 +++
20 files changed, 952 insertions(+), 104 deletions(-)
diff --git a/mk/common.mk b/mk/common.mk
index 2231f27..7d1f3f9 100644
--- a/mk/common.mk
+++ b/mk/common.mk
@@ -26,6 +26,7 @@ UNAME_S := $(shell uname)
ifeq ($(UNAME_S),Darwin)
STATIC ?= 0
+CFLAGS += -pthread
LDFLAGS += -Wl,-dead_strip
TEST_FLAGS := CONCURRENCY=1000 TOTAL=1000 FD_LIMIT=2560
else ifeq ($(OS),Windows_NT)
@@ -34,6 +35,7 @@ LDFLAGS += -Wl,--gc-sections
TEST_FLAGS := CONCURRENCY=100 TOTAL=100 FD_LIMIT=512
else
STATIC ?= 1
+CFLAGS += -pthread
LDFLAGS += -Wl,--gc-sections
TEST_FLAGS :=
endif
diff --git a/src/compat_thread.c b/src/compat_thread.c
new file mode 100644
index 0000000..f5e7e25
--- /dev/null
+++ b/src/compat_thread.c
@@ -0,0 +1,151 @@
+#include "compat_thread.h"
+
+#ifdef _WIN32
+
+#include <errno.h>
+#include <stdlib.h>
+
+struct compat_thread_start {
+ compat_thread_fn fn;
+ void *arg;
+};
+
+static DWORD WINAPI compat_thread_trampoline(LPVOID raw)
+{
+ struct compat_thread_start *start = raw;
+ compat_thread_fn fn = start->fn;
+ void *arg = start->arg;
+
+ free(start);
+
+ fn(arg);
+
+ return 0;
+}
+
+int compat_thread_create(
+ compat_thread_t *thread,
+ compat_thread_fn fn,
+ void *arg
+) {
+ struct compat_thread_start *start;
+
+ if (!thread || !fn) {
+ return EINVAL;
+ }
+
+ start = malloc(sizeof(*start));
+ if (!start) {
+ return ENOMEM;
+ }
+
+ start->fn = fn;
+ start->arg = arg;
+
+ *thread = CreateThread(
+ NULL,
+ 0,
+ compat_thread_trampoline,
+ start,
+ 0,
+ NULL
+ );
+
+ if (!*thread) {
+ free(start);
+ return (int)GetLastError();
+ }
+
+ return 0;
+}
+
+int compat_thread_join(compat_thread_t thread)
+{
+ DWORD rc;
+
+ if (!thread) {
+ return EINVAL;
+ }
+
+ rc = WaitForSingleObject(thread, INFINITE);
+ if (rc != WAIT_OBJECT_0) {
+ return (int)GetLastError();
+ }
+
+ CloseHandle(thread);
+
+ return 0;
+}
+
+int compat_mutex_init(compat_mutex_t *mutex)
+{
+ if (!mutex) {
+ return EINVAL;
+ }
+
+ InitializeCriticalSection(mutex);
+
+ return 0;
+}
+
+void compat_mutex_lock(compat_mutex_t *mutex)
+{
+ EnterCriticalSection(mutex);
+}
+
+void compat_mutex_unlock(compat_mutex_t *mutex)
+{
+ LeaveCriticalSection(mutex);
+}
+
+void compat_mutex_destroy(compat_mutex_t *mutex)
+{
+ DeleteCriticalSection(mutex);
+}
+
+#else
+
+#include <errno.h>
+
+int compat_thread_create(
+ compat_thread_t *thread,
+ compat_thread_fn fn,
+ void *arg
+) {
+ if (!thread || !fn) {
+ return EINVAL;
+ }
+
+ return pthread_create(thread, NULL, fn, arg);
+}
+
+int compat_thread_join(compat_thread_t thread)
+{
+ return pthread_join(thread, NULL);
+}
+
+int compat_mutex_init(compat_mutex_t *mutex)
+{
+ if (!mutex) {
+ return EINVAL;
+ }
+
+ return pthread_mutex_init(mutex, NULL);
+}
+
+void compat_mutex_lock(compat_mutex_t *mutex)
+{
+ pthread_mutex_lock(mutex);
+}
+
+void compat_mutex_unlock(compat_mutex_t *mutex)
+{
+ pthread_mutex_unlock(mutex);
+}
+
+void compat_mutex_destroy(compat_mutex_t *mutex)
+{
+ pthread_mutex_destroy(mutex);
+}
+
+#endif
diff --git a/src/compat_thread.h b/src/compat_thread.h
new file mode 100644
index 0000000..4326322
--- /dev/null
+++ b/src/compat_thread.h
@@ -0,0 +1,41 @@
+#ifndef TINYPROXY_COMPAT_THREAD_H
+#define TINYPROXY_COMPAT_THREAD_H
+
+#ifdef _WIN32
+#include <windows.h>
+#else
+#include <pthread.h>
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef void *(*compat_thread_fn)(void *arg);
+
+#ifdef _WIN32
+typedef HANDLE compat_thread_t;
+typedef CRITICAL_SECTION compat_mutex_t;
+#else
+typedef pthread_t compat_thread_t;
+typedef pthread_mutex_t compat_mutex_t;
+#endif
+
+int compat_thread_create(
+ compat_thread_t *thread,
+ compat_thread_fn fn,
+ void *arg
+);
+
+int compat_thread_join(compat_thread_t thread);
+
+int compat_mutex_init(compat_mutex_t *mutex);
+void compat_mutex_lock(compat_mutex_t *mutex);
+void compat_mutex_unlock(compat_mutex_t *mutex);
+void compat_mutex_destroy(compat_mutex_t *mutex);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/src/datagram_client.h b/src/datagram_client.h
index 2320ec3..a3f238d 100644
--- a/src/datagram_client.h
+++ b/src/datagram_client.h
@@ -19,6 +19,18 @@ struct datagram_client {
struct datagram_client *next;
};
+struct datagram_packet {
+ const struct route *route;
+
+ evutil_socket_t listen_fd;
+
+ struct sockaddr_storage peer_addr;
+ socklen_t peer_addr_len;
+
+ unsigned char data[65535];
+ size_t data_len;
+};
+
void cleanup_idle_datagram_clients(struct datagram_route_ctx *ctx);
void free_datagram_client(struct datagram_client *c);
diff --git a/src/datagram_listener.c b/src/datagram_listener.c
index 4fe0260..40d9603 100644
--- a/src/datagram_listener.c
+++ b/src/datagram_listener.c
@@ -8,86 +8,121 @@
#include "datagram_client.h"
#include "datagram_builtin.h"
-static void listen_read_cb(evutil_socket_t fd, short events, void *arg)
+static int handle_datagram_builtin_packet(
+ struct datagram_route_ctx *ctx,
+ const struct datagram_packet *pkt)
{
- (void)events;
-
- struct datagram_route_ctx *ctx = arg;
+ struct datagram_client tmp;
+ memset(&tmp, 0, sizeof(tmp));
+
+ tmp.ctx = ctx;
+ tmp.fd = EVUTIL_INVALID_SOCKET;
+ tmp.client_addr = pkt->peer_addr;
+ tmp.client_addr_len = pkt->peer_addr_len;
+ tmp.last_seen = time(NULL);
+
+ int rc = start_datagram_builtin(&tmp);
+ if (rc < 0) {
+ LOG_ERROR("udp builtin failed", "err", _LOGV(strerror(-rc)));
+ }
+ return rc;
+}
- cleanup_idle_datagram_clients(ctx);
+static struct datagram_client* datagram_route_get_or_create_client(
+ struct datagram_route_ctx *ctx,
+ const struct datagram_packet *pkt)
+{
+ struct datagram_client *c = find_datagram_client(ctx, &pkt->peer_addr, pkt->peer_addr_len);
+ if (c == NULL) {
+ c = create_datagram_client(ctx, &pkt->peer_addr, pkt->peer_addr_len);
+ if (c == NULL) {
+ return c;
+ }
+ }
+ c->last_seen = time(NULL);
+ return c;
+}
- for (;;) {
- unsigned char buf[UDP_MAX_PACKET];
+static int datagram_route_handle_packet(
+ struct datagram_route_ctx *ctx,
+ const struct datagram_packet *pkt)
+{
+ struct datagram_client *c;
+ int rc;
- struct sockaddr_storage client_addr;
- socklen_t client_addr_len = sizeof(client_addr);
+ if (pkt->route->upstream.kind == ENDPOINT_BUILTIN) {
+ return handle_datagram_builtin_packet(ctx, pkt);
+ }
- memset(&client_addr, 0, sizeof(client_addr));
+ c = datagram_route_get_or_create_client(ctx, pkt);
+ if (!c) {
+ return -ENOMEM;
+ }
- ssize_t n = recvfrom(
- fd,
- (char *)buf,
- sizeof(buf),
- 0,
- (struct sockaddr *)&client_addr,
- &client_addr_len
- );
+ rc = send_datagram_payload_to_upstream(c, pkt->data, pkt->data_len);
+ if (rc != 0) {
+ LOG_WARN("failed to send datagram payload upstream", "err", _LOGV(strerror(-rc)));
+ return rc;
+ }
- if (n < 0) {
- int err = EVUTIL_SOCKET_ERROR();
+ return 0;
+}
- if (socket_err_is_retriable(err)) {
- return;
- }
+static int dispatch_datagram_packet(
+ struct datagram_route_ctx *ctx,
+ const struct datagram_packet *pkt
+) {
+ /*
+ * Datagram routes are pinned to ctx->base for now.
+ * Later this function can choose a worker/shard.
+ */
+ return datagram_route_handle_packet(ctx, pkt);
+}
- LOG_ERROR("udp listen recv failed",
- "err", _LOGV(evutil_socket_error_to_string(err))
- );
- return;
- }
+static void listen_read_cb(evutil_socket_t fd, short events, void *arg)
+{
+ struct datagram_route_ctx *ctx = arg;
+ struct datagram_packet pkt;
+ ssize_t n;
- if (n == 0) {
- continue;
- }
+ (void)events;
- if (ctx->route->upstream.kind == ENDPOINT_BUILTIN) {
- struct datagram_client tmp;
-
- memset(&tmp, 0, sizeof(tmp));
- tmp.ctx = ctx;
- tmp.fd = -1;
- tmp.client_addr = client_addr;
- tmp.client_addr_len = client_addr_len;
- tmp.last_seen = time(NULL);
-
- int rc = start_datagram_builtin(&tmp);
- if (rc < 0) {
- LOG_ERROR("udp builtin failed",
- "err", _LOGV(evutil_socket_error_to_string(-rc))
- );
- return;
- }
-
- continue;
- }
+ cleanup_idle_datagram_clients(ctx);
- struct datagram_client *c = find_datagram_client(ctx, &client_addr, client_addr_len);
- if (c == NULL) {
- c = create_datagram_client(ctx, &client_addr, client_addr_len);
- if (c == NULL) {
- return;
- }
- }
+ memset(&pkt, 0, sizeof(pkt));
+ pkt.route = ctx->route;
+ pkt.listen_fd = fd;
+ pkt.peer_addr_len = sizeof(pkt.peer_addr);
+
+ n = recvfrom(
+ fd,
+ (char *)pkt.data,
+ sizeof(pkt.data),
+ 0,
+ (struct sockaddr *)&pkt.peer_addr,
+ &pkt.peer_addr_len
+ );
- c->last_seen = time(NULL);
+ if (n < 0) {
+ int err = EVUTIL_SOCKET_ERROR();
- int rc = send_datagram_payload_to_upstream(c, buf, (size_t)n);
- if (rc < 0) {
- LOG_ERROR("udp send to upstream failed",
- "err", _LOGV(evutil_socket_error_to_string(-rc))
- );
+ if (socket_err_is_retriable(err)) {
return;
}
+
+ LOG_ERROR("udp listen recv failed",
+ "err", _LOGV(evutil_socket_error_to_string(err))
+ );
+ return;
+ }
+
+ pkt.data_len = (size_t)n;
+
+ int rc = dispatch_datagram_packet(ctx, &pkt);
+ if (rc != 0) {
+ LOG_WARN("failed to dispatch datagram packet",
+ "err", _LOGV(rc)
+ );
}
}
diff --git a/src/datagram_route.c b/src/datagram_route.c
index 7ba2311..fcf2377 100644
--- a/src/datagram_route.c
+++ b/src/datagram_route.c
@@ -5,7 +5,6 @@
#include "klog.h"
#include "env.h"
-#include "worker.h"
#include "route.h"
#include "datagram_route.h"
#include "datagram_listener.h"
@@ -31,10 +30,11 @@ static int prepare_datagram_route(struct datagram_route_ctx *ctx)
}
int start_datagram_route(
- struct worker *w,
- const struct route *r,
- struct datagram_route_ctx *ctx
-) {
+ struct event_base *accept_base,
+ struct worker_pool *wpool,
+ const struct route *r,
+ struct datagram_route_ctx *ctx)
+{
int rc;
char opts[128];
@@ -44,8 +44,8 @@ int start_datagram_route(
memset(ctx, 0, sizeof(*ctx));
- ctx->base = w->base;
- ctx->worker = w;
+ ctx->base = accept_base;
+ ctx->worker_pool = wpool;
ctx->route = r;
ctx->listen_fd = -1;
diff --git a/src/datagram_route.h b/src/datagram_route.h
index 9b56c4d..0dec487 100644
--- a/src/datagram_route.h
+++ b/src/datagram_route.h
@@ -4,6 +4,7 @@
#include <event2/util.h>
#include "compat.h"
+#include "worker_pool.h"
struct event;
struct event_base;
@@ -13,7 +14,7 @@ struct datagram_client;
struct datagram_route_ctx {
struct event_base *base;
- struct worker *worker;
+ struct worker_pool *worker_pool;
const struct route *route;
evutil_socket_t listen_fd;
@@ -25,8 +26,11 @@ struct datagram_route_ctx {
struct datagram_client *clients;
};
-int start_datagram_route(struct worker *w, const struct route *r,
- struct datagram_route_ctx *ctx);
+int start_datagram_route(
+ struct event_base *accept_base,
+ struct worker_pool *wpool,
+ const struct route *r,
+ struct datagram_route_ctx *ctx);
void stop_datagram_route(struct datagram_route_ctx *ctx);
diff --git a/src/proxy_proto_v2.c b/src/proxy_proto_v2.c
index e584239..0046db7 100644
--- a/src/proxy_proto_v2.c
+++ b/src/proxy_proto_v2.c
@@ -29,10 +29,6 @@ int proxy_v2_build(
return -EINVAL;
}
- if (src_len < 0 || dst_len < 0) {
- return -EINVAL;
- }
-
if (src->sa_family != dst->sa_family) {
return -EAFNOSUPPORT;
}
diff --git a/src/route_runtime.c b/src/route_runtime.c
index 87b0244..39dfc67 100644
--- a/src/route_runtime.c
+++ b/src/route_runtime.c
@@ -50,7 +50,11 @@ int validate_route(const struct route *r)
return -EINVAL;
}
-int start_route(struct worker *w, const struct route *r, struct route_ctx *ctx)
+int start_route(
+ struct event_base *accept_base,
+ struct worker_pool *wp,
+ const struct route *r,
+ struct route_ctx *ctx)
{
int rc;
@@ -61,7 +65,7 @@ int start_route(struct worker *w, const struct route *r, struct route_ctx *ctx)
if (endpoint_is_stream(&r->listen)) {
ctx->kind = ROUTE_CTX_STREAM;
- rc = start_stream_route(w, r, &ctx->u.stream);
+ rc = start_stream_route(accept_base, wp, r, &ctx->u.stream);
if (rc != 0) {
ctx->kind = ROUTE_CTX_NONE;
memset(&ctx->u.stream, 0, sizeof(ctx->u.stream));
@@ -73,7 +77,7 @@ int start_route(struct worker *w, const struct route *r, struct route_ctx *ctx)
if (endpoint_is_datagram(&r->listen)) {
ctx->kind = ROUTE_CTX_DATAGRAM;
- rc = start_datagram_route(w, r, &ctx->u.datagram);
+ rc = start_datagram_route(accept_base, wp, r, &ctx->u.datagram);
if (rc != 0) {
ctx->kind = ROUTE_CTX_NONE;
memset(&ctx->u.datagram, 0, sizeof(ctx->u.datagram));
diff --git a/src/route_runtime.h b/src/route_runtime.h
index a192f36..22ad642 100644
--- a/src/route_runtime.h
+++ b/src/route_runtime.h
@@ -23,7 +23,12 @@ struct route_ctx {
int validate_route(const struct route *r);
-int start_route(struct worker *w, const struct route *r, struct route_ctx *ctx);
+int start_route(
+ struct event_base *accept_base,
+ struct worker_pool *wp,
+ const struct route *r,
+ struct route_ctx *ctx);
+
void stop_route(struct route_ctx *ctx);
#endif
diff --git a/src/stream_conn.c b/src/stream_conn.c
index 78d9328..2cd3c46 100644
--- a/src/stream_conn.c
+++ b/src/stream_conn.c
@@ -265,7 +265,7 @@ static int connect_upstream(struct bufferevent *bev, const struct endpoint *ep)
}
}
-static void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac) {
+void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac) {
conn_t *conn = calloc(1, sizeof(*conn));
if (conn == NULL) {
evutil_closesocket(ac->fd);
@@ -407,8 +407,15 @@ static void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac)
bufferevent_enable(conn->upstream, EV_READ | EV_WRITE);
}
-void dispatch_client_fd(struct worker *w, struct accepted_client *ac) {
- worker_adopt_client_fd(w, ac);
+int dispatch_client_fd(struct worker *w, struct accepted_client *ac)
+{
+ return worker_enqueue_client_fd(
+ w,
+ ac->route,
+ ac->fd,
+ (const struct sockaddr *)&ac->peer_addr,
+ ac->peer_addr_len
+ );
}
#ifdef FUZZ
diff --git a/src/stream_conn.h b/src/stream_conn.h
index bbac010..f4ef594 100644
--- a/src/stream_conn.h
+++ b/src/stream_conn.h
@@ -3,7 +3,7 @@
#include "stream_route.h"
-void dispatch_client_fd(struct worker *w, struct accepted_client *ac);
+int dispatch_client_fd(struct worker *w, struct accepted_client *ac);
void free_conn(conn_t *conn);
void set_client_idle_timeout(conn_t *conn, const struct route *r);
@@ -11,6 +11,8 @@ void set_client_idle_timeout(conn_t *conn, const struct route *r);
void stream_client_event_cb(struct bufferevent *bev, short events, void *arg);
void stream_upstream_event_cb(struct bufferevent *bev, short events, void *arg);
+void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac);
+
#ifdef FUZZ
int stream_route_adopt_client_for_fuzz(
struct event_base *base,
diff --git a/src/stream_listener.c b/src/stream_listener.c
index 3813fa4..1aa4255 100644
--- a/src/stream_listener.c
+++ b/src/stream_listener.c
@@ -5,6 +5,7 @@
#include "klog.h"
#include "route.h"
+#include "worker_pool.h"
#include "stream_listener.h"
#include "stream_conn.h"
@@ -32,7 +33,16 @@ static void accept_cb(
return;
}
- dispatch_client_fd(ctx->worker, &ac);
+ struct worker *w = worker_pool_next(ctx->worker_pool);
+ if (!w) {
+ LOG_ERROR("no worker available", "socklen", _LOGV(socklen));
+ evutil_closesocket(client_fd);
+ return;
+ }
+
+ if (dispatch_client_fd(w, &ac) != 0) {
+ evutil_closesocket(client_fd);
+ }
}
void accept_error_cb(struct evconnlistener *listener, void *arg)
diff --git a/src/stream_route.c b/src/stream_route.c
index f80ac48..fb553ad 100644
--- a/src/stream_route.c
+++ b/src/stream_route.c
@@ -9,16 +9,19 @@
#include "route.h"
#include "stream_listener.h"
-int start_stream_route(struct worker *w, const struct route *r,
- struct stream_route_ctx *ctx)
+int start_stream_route(
+ struct event_base *accept_base,
+ struct worker_pool *wpool,
+ const struct route *r,
+ struct stream_route_ctx *ctx)
{
int rc;
char opts[128];
memset(ctx, 0, sizeof(*ctx));
- ctx->accept_base = w->base;
- ctx->worker = w;
+ ctx->accept_base = accept_base;
+ ctx->worker_pool = wpool;
ctx->route = r;
rc = bind_stream_listener(ctx);
diff --git a/src/stream_route.h b/src/stream_route.h
index 45b38dc..f9d809e 100644
--- a/src/stream_route.h
+++ b/src/stream_route.h
@@ -14,7 +14,7 @@ struct evconnlistener;
struct stream_route_ctx {
struct event_base *accept_base;
- struct worker *worker;
+ struct worker_pool *worker_pool;
const struct route *route;
struct evconnlistener *listener;
};
@@ -34,14 +34,23 @@ typedef struct conn_s {
} conn_t;
struct accepted_client {
+ const struct route *route;
evutil_socket_t fd;
+
struct sockaddr_storage peer_addr;
socklen_t peer_addr_len;
- const struct route *route;
};
-int start_stream_route(struct worker *w, const struct route *r,
- struct stream_route_ctx *ctx);
+struct accepted_client_node {
+ struct accepted_client client;
+ struct accepted_client_node *next;
+};
+
+int start_stream_route(
+ struct event_base *accept_base,
+ struct worker_pool *wpool,
+ const struct route *r,
+ struct stream_route_ctx *ctx);
void stop_stream_route(struct stream_route_ctx *ctx);
diff --git a/src/tinyproxy.c b/src/tinyproxy.c
index b11e00f..7066cca 100644
--- a/src/tinyproxy.c
+++ b/src/tinyproxy.c
@@ -10,7 +10,7 @@
#include "klog.h"
#include "signals.h"
#include "file_conf.h"
-#include "worker.h"
+#include "worker_pool.h"
#include "route.h"
#include "route_runtime.h"
@@ -53,6 +53,8 @@ int main(int argc, char **argv)
const char **inline_routes = NULL;
size_t inline_route_count = 0;
size_t inline_route_cap = 0;
+ bool worker_pool_ready = false;
+ bool worker_pool_started = false;
#ifdef _WIN32
int wsa_started = 0;
@@ -112,7 +114,7 @@ int main(int argc, char **argv)
return 0;
case 'v':
- fprintf(stdout, "tinyproxy v0.1.4\n");
+ fprintf(stdout, "tinyproxy v0.2.0\n");
free(inline_routes);
return 0;
@@ -216,10 +218,22 @@ int main(int argc, char **argv)
}
signals_started = 1;
- struct worker w = {
- .base = base,
- .id = 0,
- };
+ struct worker_pool wpool;
+ size_t worker_count = 1;
+
+ rc = worker_pool_init(&wpool, worker_count);
+ if (rc != 0) {
+ LOG_ERROR("failed to init worker pool", "err", _LOGV(rc));
+ goto out;
+ }
+ worker_pool_ready = true;
+
+ rc = worker_pool_start(&wpool);
+ if (rc != 0) {
+ LOG_ERROR("failed to start worker pool", "err", _LOGV(rc));
+ goto out;
+ }
+ worker_pool_started = true;
for (size_t i = 0; i < route_count; i++) {
const struct route *r = &routes[i];
@@ -234,7 +248,7 @@ int main(int argc, char **argv)
goto out;
}
- rc = start_route(&w, r, &route_ctxs[route_ctx_count]);
+ rc = start_route(base, &wpool, r, &route_ctxs[route_ctx_count]);
if (rc == 0) {
route_ctx_count++;
}
@@ -262,6 +276,15 @@ out:
stop_route(&route_ctxs[i]);
}
+ if (worker_pool_started) {
+ worker_pool_stop(&wpool);
+ worker_pool_join(&wpool);
+ }
+
+ if (worker_pool_ready) {
+ worker_pool_free(&wpool);
+ }
+
if (signals_started) {
free_signal_handlers(&signals);
}
diff --git a/src/worker.c b/src/worker.c
new file mode 100644
index 0000000..ce9c753
--- /dev/null
+++ b/src/worker.c
@@ -0,0 +1,364 @@
+#include <event2/event.h>
+#include <event2/util.h>
+
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "klog.h"
+#include "stream_conn.h"
+#include "worker.h"
+
+static void worker_notify_cb(evutil_socket_t fd, short events, void *arg);
+
+static int notify_worker(struct worker *w)
+{
+ char byte = 1;
+ int rc;
+
+ (void)w;
+
+#ifdef _WIN32
+ rc = send(w->notify_send_fd, &byte, 1, 0);
+ if (rc == SOCKET_ERROR) {
+ return WSAGetLastError();
+ }
+#else
+ rc = (int)send(w->notify_send_fd, &byte, 1, 0);
+ if (rc < 0) {
+ return errno;
+ }
+#endif
+
+ return 0;
+}
+
+static void drain_notify_fd(evutil_socket_t fd)
+{
+ char buf[128];
+
+ for (;;) {
+#ifdef _WIN32
+ int n = recv(fd, buf, sizeof(buf), 0);
+ if (n == SOCKET_ERROR) {
+ int err = WSAGetLastError();
+ if (err == WSAEWOULDBLOCK) {
+ return;
+ }
+ return;
+ }
+#else
+ ssize_t n = recv(fd, buf, sizeof(buf), 0);
+ if (n < 0) {
+ if (errno == EAGAIN || errno == EWOULDBLOCK) {
+ return;
+ }
+ return;
+ }
+#endif
+
+ if (n == 0) {
+ return;
+ }
+
+ if ((size_t)n < sizeof(buf)) {
+ return;
+ }
+ }
+}
+
+static struct accepted_client_node *worker_take_pending(struct worker *w)
+{
+ struct accepted_client_node *head;
+
+ compat_mutex_lock(&w->pending_mu);
+
+ head = w->pending_head;
+ w->pending_head = NULL;
+ w->pending_tail = NULL;
+
+ compat_mutex_unlock(&w->pending_mu);
+
+ return head;
+}
+
+static void worker_process_pending(struct worker *w)
+{
+ struct accepted_client_node *node = worker_take_pending(w);
+
+ while (node) {
+ struct accepted_client_node *next = node->next;
+
+ worker_adopt_client_fd(w, &node->client);
+
+ free(node);
+ node = next;
+ }
+}
+
+static void *worker_main(void *arg)
+{
+ struct worker *w = arg;
+
+ event_base_dispatch(w->base);
+
+ return NULL;
+}
+
+int worker_init(struct worker *w, unsigned int id)
+{
+ evutil_socket_t fds[2] = {
+ EVUTIL_INVALID_SOCKET,
+ EVUTIL_INVALID_SOCKET,
+ };
+
+ if (!w) {
+ return EINVAL;
+ }
+
+ memset(w, 0, sizeof(*w));
+
+ w->id = id;
+ w->notify_recv_fd = EVUTIL_INVALID_SOCKET;
+ w->notify_send_fd = EVUTIL_INVALID_SOCKET;
+
+ w->base = event_base_new();
+ if (!w->base) {
+ return ENOMEM;
+ }
+
+ if (compat_mutex_init(&w->pending_mu) != 0) {
+ event_base_free(w->base);
+ w->base = NULL;
+ return errno ? errno : EINVAL;
+ }
+
+ if (evutil_socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0) {
+ compat_mutex_destroy(&w->pending_mu);
+ event_base_free(w->base);
+ w->base = NULL;
+ return EVUTIL_SOCKET_ERROR();
+ }
+
+ w->notify_recv_fd = fds[0];
+ w->notify_send_fd = fds[1];
+
+ evutil_make_socket_nonblocking(w->notify_recv_fd);
+ evutil_make_socket_nonblocking(w->notify_send_fd);
+ evutil_make_socket_closeonexec(w->notify_recv_fd);
+ evutil_make_socket_closeonexec(w->notify_send_fd);
+
+ w->notify_event = event_new(
+ w->base,
+ w->notify_recv_fd,
+ EV_READ | EV_PERSIST,
+ worker_notify_cb,
+ w
+ );
+ if (!w->notify_event) {
+ evutil_closesocket(w->notify_recv_fd);
+ evutil_closesocket(w->notify_send_fd);
+ w->notify_recv_fd = EVUTIL_INVALID_SOCKET;
+ w->notify_send_fd = EVUTIL_INVALID_SOCKET;
+
+ compat_mutex_destroy(&w->pending_mu);
+ event_base_free(w->base);
+ w->base = NULL;
+
+ return ENOMEM;
+ }
+
+ if (event_add(w->notify_event, NULL) < 0) {
+ event_free(w->notify_event);
+ w->notify_event = NULL;
+
+ evutil_closesocket(w->notify_recv_fd);
+ evutil_closesocket(w->notify_send_fd);
+ w->notify_recv_fd = EVUTIL_INVALID_SOCKET;
+ w->notify_send_fd = EVUTIL_INVALID_SOCKET;
+
+ compat_mutex_destroy(&w->pending_mu);
+ event_base_free(w->base);
+ w->base = NULL;
+
+ return EINVAL;
+ }
+
+ return 0;
+}
+
+int worker_start(struct worker *w)
+{
+ int rc;
+
+ if (!w || !w->base) {
+ return EINVAL;
+ }
+
+ if (w->started) {
+ return 0;
+ }
+
+ rc = compat_thread_create(&w->thread, worker_main, w);
+ if (rc != 0) {
+ return rc;
+ }
+
+ w->started = true;
+
+ return 0;
+}
+
+void worker_stop(struct worker *w)
+{
+ if (!w) {
+ return;
+ }
+
+ compat_mutex_lock(&w->pending_mu);
+ w->stopping = true;
+ compat_mutex_unlock(&w->pending_mu);
+
+ if (w->started) {
+ (void)notify_worker(w);
+ }
+}
+
+void worker_join(struct worker *w)
+{
+ if (!w || !w->started) {
+ return;
+ }
+
+ (void)compat_thread_join(w->thread);
+ w->started = false;
+}
+
+void worker_free(struct worker *w)
+{
+ struct accepted_client_node *node;
+
+ if (!w) {
+ return;
+ }
+
+ worker_stop(w);
+ worker_join(w);
+
+ node = worker_take_pending(w);
+ while (node) {
+ struct accepted_client_node *next = node->next;
+
+ if (node->client.fd != EVUTIL_INVALID_SOCKET) {
+ evutil_closesocket(node->client.fd);
+ }
+
+ free(node);
+ node = next;
+ }
+
+ if (w->notify_event) {
+ event_free(w->notify_event);
+ w->notify_event = NULL;
+ }
+
+ if (w->notify_recv_fd != EVUTIL_INVALID_SOCKET) {
+ evutil_closesocket(w->notify_recv_fd);
+ w->notify_recv_fd = EVUTIL_INVALID_SOCKET;
+ }
+
+ if (w->notify_send_fd != EVUTIL_INVALID_SOCKET) {
+ evutil_closesocket(w->notify_send_fd);
+ w->notify_send_fd = EVUTIL_INVALID_SOCKET;
+ }
+
+ compat_mutex_destroy(&w->pending_mu);
+
+ if (w->base) {
+ event_base_free(w->base);
+ w->base = NULL;
+ }
+}
+
+int worker_enqueue_client_fd(
+ struct worker *w,
+ const struct route *route,
+ evutil_socket_t fd,
+ const struct sockaddr *addr,
+ socklen_t addr_len
+) {
+ struct accepted_client_node *node;
+ int rc;
+
+ if (!w || !route || fd == EVUTIL_INVALID_SOCKET) {
+ return EINVAL;
+ }
+
+ if (addr_len > sizeof(node->client.peer_addr)) {
+ return EINVAL;
+ }
+
+ node = calloc(1, sizeof(*node));
+ if (!node) {
+ return ENOMEM;
+ }
+
+ node->client.route = route;
+ node->client.fd = fd;
+
+ if (addr && addr_len > 0) {
+ memcpy(&node->client.peer_addr, addr, addr_len);
+ node->client.peer_addr_len = addr_len;
+ }
+
+ compat_mutex_lock(&w->pending_mu);
+
+ if (w->stopping) {
+ compat_mutex_unlock(&w->pending_mu);
+ free(node);
+ return ECANCELED;
+ }
+
+ if (w->pending_tail) {
+ w->pending_tail->next = node;
+ } else {
+ w->pending_head = node;
+ }
+
+ w->pending_tail = node;
+
+ compat_mutex_unlock(&w->pending_mu);
+
+ rc = notify_worker(w);
+ if (rc != 0) {
+ /*
+ * The fd is already queued and owned by the worker.
+ * Do not close it here.
+ */
+ LOG_WARN("failed to notify worker",
+ "worker", _LOGV(w->id),
+ "err", _LOGV(rc)
+ );
+ }
+
+ return 0;
+}
+
+static void worker_notify_cb(evutil_socket_t fd, short events, void *arg)
+{
+ struct worker *w = arg;
+
+ (void)events;
+
+ drain_notify_fd(fd);
+
+ worker_process_pending(w);
+
+ compat_mutex_lock(&w->pending_mu);
+ bool stopping = w->stopping;
+ compat_mutex_unlock(&w->pending_mu);
+
+ if (stopping) {
+ event_base_loopbreak(w->base);
+ }
+}
diff --git a/src/worker.h b/src/worker.h
index 53c28e6..021b90d 100644
--- a/src/worker.h
+++ b/src/worker.h
@@ -1,11 +1,48 @@
#ifndef WORKER_H
#define WORKER_H
-struct event_base;
+#include <stdbool.h>
+#include <stddef.h>
+
+#include <event2/event.h>
+#include <event2/util.h>
+
+#include "compat_thread.h"
+#include "compat.h"
+#include "route.h"
+
+struct accept_client_node;
struct worker {
+ unsigned int id;
+
struct event_base *base;
- size_t id;
+
+ compat_thread_t thread;
+ bool started;
+ bool stopping;
+
+ evutil_socket_t notify_recv_fd;
+ evutil_socket_t notify_send_fd;
+ struct event *notify_event;
+
+ struct accepted_client_node *pending_head;
+ struct accepted_client_node *pending_tail;
+ compat_mutex_t pending_mu;
};
+int worker_init(struct worker *w, unsigned int id);
+int worker_start(struct worker *w);
+void worker_stop(struct worker *w);
+void worker_join(struct worker *w);
+void worker_free(struct worker *w);
+
+int worker_enqueue_client_fd(
+ struct worker *w,
+ const struct route *route,
+ evutil_socket_t fd,
+ const struct sockaddr *addr,
+ socklen_t addr_len
+);
+
#endif
diff --git a/src/worker_pool.c b/src/worker_pool.c
new file mode 100644
index 0000000..410a3b3
--- /dev/null
+++ b/src/worker_pool.c
@@ -0,0 +1,121 @@
+#include "worker_pool.h"
+
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+
+int worker_pool_init(struct worker_pool *pool, size_t count)
+{
+ size_t i;
+
+ if (!pool || count == 0) {
+ return EINVAL;
+ }
+
+ memset(pool, 0, sizeof(*pool));
+
+ pool->workers = calloc(count, sizeof(pool->workers[0]));
+ if (!pool->workers) {
+ return ENOMEM;
+ }
+
+ pool->count = count;
+
+ for (i = 0; i < count; i++) {
+ int rc = worker_init(&pool->workers[i], (unsigned int)i);
+ if (rc != 0) {
+ while (i > 0) {
+ i--;
+ worker_free(&pool->workers[i]);
+ }
+
+ free(pool->workers);
+ pool->workers = NULL;
+ pool->count = 0;
+
+ return rc;
+ }
+ }
+
+ return 0;
+}
+
+int worker_pool_start(struct worker_pool *pool)
+{
+ size_t i;
+
+ if (!pool || !pool->workers) {
+ return EINVAL;
+ }
+
+ for (i = 0; i < pool->count; i++) {
+ int rc = worker_start(&pool->workers[i]);
+ if (rc != 0) {
+ while (i > 0) {
+ i--;
+ worker_stop(&pool->workers[i]);
+ worker_join(&pool->workers[i]);
+ }
+
+ return rc;
+ }
+ }
+
+ return 0;
+}
+
+void worker_pool_stop(struct worker_pool *pool)
+{
+ size_t i;
+
+ if (!pool || !pool->workers) {
+ return;
+ }
+
+ for (i = 0; i < pool->count; i++) {
+ worker_stop(&pool->workers[i]);
+ }
+}
+
+void worker_pool_join(struct worker_pool *pool)
+{
+ size_t i;
+
+ if (!pool || !pool->workers) {
+ return;
+ }
+
+ for (i = 0; i < pool->count; i++) {
+ worker_join(&pool->workers[i]);
+ }
+}
+
+void worker_pool_free(struct worker_pool *pool)
+{
+ size_t i;
+
+ if (!pool || !pool->workers) {
+ return;
+ }
+
+ for (i = 0; i < pool->count; i++) {
+ worker_free(&pool->workers[i]);
+ }
+
+ free(pool->workers);
+ memset(pool, 0, sizeof(*pool));
+}
+
+struct worker *worker_pool_next(struct worker_pool *pool)
+{
+ struct worker *w;
+
+ if (!pool || !pool->workers || pool->count == 0) {
+ return NULL;
+ }
+
+ w = &pool->workers[pool->next % pool->count];
+ pool->next++;
+
+ return w;
+}
\ No newline at end of file
diff --git a/src/worker_pool.h b/src/worker_pool.h
new file mode 100644
index 0000000..7f8bd05
--- /dev/null
+++ b/src/worker_pool.h
@@ -0,0 +1,22 @@
+#ifndef WORKER_POOL_H
+#define WORKER_POOL_H
+
+#include <stddef.h>
+
+#include "worker.h"
+
+struct worker_pool {
+ struct worker *workers;
+ size_t count;
+ size_t next;
+};
+
+int worker_pool_init(struct worker_pool *pool, size_t count);
+int worker_pool_start(struct worker_pool *pool);
+void worker_pool_stop(struct worker_pool *pool);
+void worker_pool_join(struct worker_pool *pool);
+void worker_pool_free(struct worker_pool *pool);
+
+struct worker *worker_pool_next(struct worker_pool *pool);
+
+#endif