From 54c8740a66d671ca018392b7823b562921b39569 Mon Sep 17 00:00:00 2001 From: otegami Date: Mon, 1 Jun 2026 17:29:58 +0800 Subject: [PATCH] feat: route the non-Ruby-thread dispatch path through an optional proxy Split the dispatcher so the worker-thread path (Case 3) can route through a per-worker proxy. rbduckdb_function_executor_dispatch_via_proxy is the new entry point: when proxy is non-NULL it hands the callback to that proxy (Case 3a), otherwise it falls back to the global executor (Case 3b). dispatch_callback_to_proxy is the sender half of the proxy hand-off -- the counterpart of the receiver loop that landed with the primitive. rbduckdb_function_executor_dispatch keeps its signature and now simply delegates with NULL, so behavior is unchanged -- the only in-tree caller passes NULL and no proxy is ever exercised yet. Full suite unchanged (1147). The proxy path goes live once scalar/table integration lands. Also document rbduckdb_worker_proxy_create's raise contract: the executor runs callbacks unprotected, so a wrapper that dispatches create through it must rb_protect the call. The wrapper itself arrives with the first consumer (scalar integration), which is where the rb_protect lands. --- ext/duckdb/function_executor.c | 47 ++++++++++++++++++++++++++++++++-- ext/duckdb/function_executor.h | 12 +++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/ext/duckdb/function_executor.c b/ext/duckdb/function_executor.c index c6d1a564..b41ad76a 100644 --- a/ext/duckdb/function_executor.c +++ b/ext/duckdb/function_executor.c @@ -440,6 +440,42 @@ struct worker_proxy *rbduckdb_worker_proxy_create(void) { return proxy; } +/* + * Hand a callback to a proxy and block until it completes. + * Called from the DuckDB worker thread (non-Ruby thread) that owns this proxy. + */ +static void dispatch_callback_to_proxy(struct worker_proxy *proxy, rbduckdb_function_callback_t cb, void *user_data) { +#ifdef _MSC_VER + EnterCriticalSection(&proxy->lock); + proxy->cb = cb; + proxy->user_data = user_data; + proxy->request_done = 0; + proxy->has_request = 1; + WakeConditionVariable(&proxy->request_cond); + LeaveCriticalSection(&proxy->lock); + + EnterCriticalSection(&proxy->lock); + while (!proxy->request_done) { + SleepConditionVariableCS(&proxy->request_done_cond, &proxy->lock, INFINITE); + } + LeaveCriticalSection(&proxy->lock); +#else + pthread_mutex_lock(&proxy->lock); + proxy->cb = cb; + proxy->user_data = user_data; + proxy->request_done = 0; + proxy->has_request = 1; + pthread_cond_signal(&proxy->request_cond); + pthread_mutex_unlock(&proxy->lock); + + pthread_mutex_lock(&proxy->lock); + while (!proxy->request_done) { + pthread_cond_wait(&proxy->request_done_cond, &proxy->lock); + } + pthread_mutex_unlock(&proxy->lock); +#endif +} + /* Blocks until the proxy thread has fully exited. Runs without the GVL. */ static void *proxy_join_func(void *data) { struct worker_proxy *proxy = (struct worker_proxy *)data; @@ -505,7 +541,7 @@ void rbduckdb_worker_proxy_destroy(void *data) { free(proxy); } -void rbduckdb_function_executor_dispatch(rbduckdb_function_callback_t cb, void *user_data) { +void rbduckdb_function_executor_dispatch_via_proxy(rbduckdb_function_callback_t cb, void *user_data, struct worker_proxy *proxy) { if (ruby_native_thread_p()) { if (ruby_thread_has_gvl_p()) { /* Case 1: Ruby thread with GVL - call directly */ @@ -517,8 +553,15 @@ void rbduckdb_function_executor_dispatch(rbduckdb_function_callback_t cb, void * arg.user_data = user_data; rb_thread_call_with_gvl(callback_with_gvl, &arg); } + } else if (proxy != NULL) { + /* Case 3a: Non-Ruby thread with a per-worker proxy */ + dispatch_callback_to_proxy(proxy, cb, user_data); } else { - /* Case 3: Non-Ruby thread - dispatch to executor */ + /* Case 3b: Non-Ruby thread - dispatch to the global executor */ dispatch_callback_to_executor(cb, user_data); } } + +void rbduckdb_function_executor_dispatch(rbduckdb_function_callback_t cb, void *user_data) { + rbduckdb_function_executor_dispatch_via_proxy(cb, user_data, NULL); +} diff --git a/ext/duckdb/function_executor.h b/ext/duckdb/function_executor.h index 095e8b21..ba71c04e 100644 --- a/ext/duckdb/function_executor.h +++ b/ext/duckdb/function_executor.h @@ -65,6 +65,11 @@ struct worker_proxy; * Create a per-worker proxy thread. Must be called with the GVL held * (typically by dispatching this through the global executor from a per-worker * init callback, which itself runs on a non-Ruby thread). + * + * May raise (NoMemError, Thread.new failure). The executor runs callbacks + * unprotected, so a wrapper dispatched to it must rb_protect this call — + * otherwise a raise longjmps past the executor's done-signaling and the + * waiting DuckDB worker blocks forever. */ struct worker_proxy *rbduckdb_worker_proxy_create(void); @@ -75,4 +80,11 @@ struct worker_proxy *rbduckdb_worker_proxy_create(void); */ void rbduckdb_worker_proxy_destroy(void *proxy); +/* + * Like rbduckdb_function_executor_dispatch, but on the non-Ruby-thread path + * (Case 3) it routes through the given per-worker proxy when non-NULL, falling + * back to the global executor when NULL. Cases 1 and 2 are unchanged. + */ +void rbduckdb_function_executor_dispatch_via_proxy(rbduckdb_function_callback_t cb, void *user_data, struct worker_proxy *proxy); + #endif