Re-work exception saving & non-gil function callbacks

7 files changed, 147 insertions, 89 deletions

diff --git a/ext/ffi_c/Call.c b/ext/ffi_c/Call.c
index 7dcdd93..42c0128 100644
--- a/ext/ffi_c/Call.c
+++ b/ext/ffi_c/Call.c
@@ -266,6 +266,7 @@ rbffi_SetupCallParams(int argc, VALUE* argv, int paramCount, Type** paramTypes,
 
 
 typedef struct BlockingCall_ {
+    rbffi_frame_t* frame;
     void* function;
     FunctionType* info;
     void **ffiValues;
@@ -280,14 +281,13 @@ static VALUE
 call_blocking_function(void* data)
 {
     BlockingCall* b = (BlockingCall *) data;
-
+    b->frame->has_gvl = false;
     ffi_call(&b->info->ffi_cif, FFI_FN(b->function), b->retval, b->ffiValues);
+    b->frame->has_gvl = true;
 
     return Qnil;
 }
 
-#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
-
 static VALUE
 do_blocking_call(void *data)
 {
@@ -297,30 +297,22 @@ do_blocking_call(void *data)
 }
 
 static VALUE
-cleanup_blocking_call(void *data)
+save_frame_exception(void *data, VALUE exc)
 {
-    BlockingCall* bc = (BlockingCall *) data;
-
-    memcpy(bc->stkretval, bc->retval, MAX(bc->info->ffi_cif.rtype->size, FFI_SIZEOF_ARG));
-    xfree(bc->params);
-    xfree(bc->ffiValues);
-    xfree(bc->retval);
-    xfree(bc);
-
+    rbffi_frame_t* frame = (rbffi_frame_t *) data;
+    frame->exc = exc;
     return Qnil;
 }
 
-#endif /* HAVE_RB_THREAD_BLOCKING_REGION */
-
 VALUE
 rbffi_CallFunction(int argc, VALUE* argv, void* function, FunctionType* fnInfo)
 {
     void* retval;
     void** ffiValues;
     FFIStorage* params;
-    VALUE rbReturnValue, exc;    
-    rbffi_thread_t oldThread;
-
+    VALUE rbReturnValue;
+    rbffi_frame_t frame = { 0 };
+    
     retval = alloca(MAX(fnInfo->ffi_cif.rtype->size, FFI_SIZEOF_ARG));
     
     if (unlikely(fnInfo->blocking)) {
@@ -346,21 +338,24 @@ rbffi_CallFunction(int argc, VALUE* argv, void* function, FunctionType* fnInfo)
         bc->function = function;
         bc->ffiValues = ffiValues;
         bc->params = params;
+        bc->frame = &frame;
 
         rbffi_SetupCallParams(argc, argv,
             fnInfo->parameterCount, fnInfo->parameterTypes, params, ffiValues,
             fnInfo->callbackParameters, fnInfo->callbackCount, fnInfo->rbEnums);
 
-#if defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL) 
-        rb_thread_call_without_gvl(call_blocking_function, bc, (void *) -1, NULL);
+        rbffi_frame_push(&frame); 
+        rb_rescue2(do_blocking_call, (VALUE) bc, save_frame_exception, (VALUE) &frame, rb_eException, (VALUE) 0);
+        rbffi_frame_pop(&frame);
 
-#elif defined(HAVE_RB_THREAD_BLOCKING_REGION)
-        rb_thread_blocking_region(call_blocking_function, bc, (void *) -1, NULL);
-
-#else
-        rb_ensure(do_blocking_call, (VALUE) bc, cleanup_blocking_call, (VALUE) bc);
+#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
+        memcpy(bc->stkretval, bc->retval, MAX(bc->info->ffi_cif.rtype->size, FFI_SIZEOF_ARG));
+        xfree(bc->params);
+        xfree(bc->ffiValues);
+        xfree(bc->retval);
+        xfree(bc);
 #endif
-        
+    
     } else {
 
         ffiValues = ALLOCA_N(void *, fnInfo->parameterCount);
@@ -370,22 +365,19 @@ rbffi_CallFunction(int argc, VALUE* argv, void* function, FunctionType* fnInfo)
             fnInfo->parameterCount, fnInfo->parameterTypes, params, ffiValues,
             fnInfo->callbackParameters, fnInfo->callbackCount, fnInfo->rbEnums);
 
-        oldThread = rbffi_active_thread;
-        rbffi_active_thread = rbffi_thread_self();
-        
+        rbffi_frame_push(&frame);
         ffi_call(&fnInfo->ffi_cif, FFI_FN(function), retval, ffiValues);
-
-        exc = rbffi_active_thread.exc;
-        rbffi_active_thread = oldThread;
-        if (exc != Qnil) {
-            rb_exc_raise(exc);
-        }
+        rbffi_frame_pop(&frame);
     }
 
     if (unlikely(!fnInfo->ignoreErrno)) {
         rbffi_save_errno();
     }    
-    
+
+    if (RTEST(frame.exc) && frame.exc != Qnil) {
+        rb_exc_raise(frame.exc);
+    }
+
     RB_GC_GUARD(rbReturnValue) = rbffi_NativeValue_ToRuby(fnInfo->returnType, fnInfo->rbReturnType, retval);
     RB_GC_GUARD(fnInfo->rbReturnType);
     
diff --git a/ext/ffi_c/Function.c b/ext/ffi_c/Function.c
index 47fc568..86f38db 100644
--- a/ext/ffi_c/Function.c
+++ b/ext/ffi_c/Function.c
@@ -78,6 +78,7 @@ static VALUE function_init(VALUE self, VALUE rbFunctionInfo, VALUE rbProc);
 static void callback_invoke(ffi_cif* cif, void* retval, void** parameters, void* user_data);
 static bool callback_prep(void* ctx, void* code, Closure* closure, char* errmsg, size_t errmsgsize);
 static VALUE callback_with_gvl(void* data);
+static VALUE invoke_callback(void* data);
 static VALUE save_callback_exception(void* data, VALUE exc);
 
 #define DEFER_ASYNC_CALLBACK 1
@@ -105,6 +106,7 @@ struct gvl_callback {
     void*    retval;
     void**   parameters;
     bool done;
+    rbffi_frame_t *frame;
 #if defined(DEFER_ASYNC_CALLBACK)
     struct gvl_callback* next;
 # ifndef _WIN32
@@ -451,16 +453,22 @@ function_release(VALUE self)
 static void
 callback_invoke(ffi_cif* cif, void* retval, void** parameters, void* user_data)
 {
-    struct gvl_callback cb;
+    struct gvl_callback cb = { 0 };
+    
     cb.closure = (Closure *) user_data;
     cb.retval = retval;
     cb.parameters = parameters;
     cb.done = false;
-
-    if (rbffi_thread_has_gvl_p()) {
-        rbffi_active_thread.exc = Qnil;
-        rb_rescue2(callback_with_gvl, (VALUE) &cb, save_callback_exception, (VALUE) &cb, rb_eException, (VALUE) 0);
+    cb.frame = rbffi_frame_current();
     
+    if (cb.frame != NULL) cb.frame->exc = Qnil;
+    if (cb.frame != NULL && cb.frame->has_gvl) {
+        callback_with_gvl(&cb);
+
+#if defined(RB_THREAD_CALL_WITH_GVL)
+    } else if (cb.frame != NULL) {
+        rb_thread_call_with_gvl(callback_with_gvl, &cb);
+#endif
 #if defined(DEFER_ASYNC_CALLBACK) && !defined(_WIN32)
     } else {
         bool empty = false;
@@ -695,9 +703,9 @@ static VALUE
 async_cb_call(void *data)
 {
     struct gvl_callback* cb = (struct gvl_callback *) data;
-
-    callback_with_gvl(cb);
-
+    
+    callback_with_gvl(data);
+        
     /* Signal the original native thread that the ruby code has completed */
 #ifdef _WIN32
     SetEvent(cb->async_event);
@@ -713,10 +721,15 @@ async_cb_call(void *data)
 
 #endif
 
-
 static VALUE
 callback_with_gvl(void* data)
 {
+    rb_rescue2(invoke_callback, (VALUE) data, save_callback_exception, (VALUE) data, rb_eException, (VALUE) 0);
+}
+
+static VALUE
+invoke_callback(void* data)
+{
     struct gvl_callback* cb = (struct gvl_callback *) data;
 
     Function* fn = (Function *) cb->closure->info;
@@ -911,7 +924,7 @@ save_callback_exception(void* data, VALUE exc)
     struct gvl_callback* cb = (struct gvl_callback *) data;
     
     memset(cb->retval, 0, ((Function *) cb->closure->info)->info->returnType->ffiType->size);
-    rbffi_active_thread.exc = exc;
+    if (cb->frame != NULL) cb->frame->exc = exc;
     
     return Qnil;
 }
diff --git a/ext/ffi_c/Thread.c b/ext/ffi_c/Thread.c
index 7c23aeb..a67a6ed 100644
--- a/ext/ffi_c/Thread.c
+++ b/ext/ffi_c/Thread.c
@@ -38,46 +38,56 @@
 #include <fcntl.h>
 #include "Thread.h"
 
+#ifdef _WIN32
+static volatile DWORD frame_thread_key = TLS_OUT_OF_INDEXES;
+#else
+static pthread_key_t thread_data_key;
+struct thread_data {
+    rbffi_frame_t* frame;
+};
+static inline struct thread_data* thread_data_get(void);
 
-rbffi_thread_t rbffi_active_thread;
+#endif
 
-rbffi_thread_t
-rbffi_thread_self()
+rbffi_frame_t*
+rbffi_frame_current(void)
 {
-    rbffi_thread_t self;
 #ifdef _WIN32
-    self.id = GetCurrentThreadId();
+    return (rbffi_frame_t *) TlsGetValue(frame_thread_key);
 #else
-    self.id = pthread_self();
+    struct thread_data* td = (struct thread_data *) pthread_getspecific(thread_data_key);
+    return td != NULL ? td->frame : NULL;
 #endif
-    self.valid = true;
-    self.exc = Qnil;
-
-    return self;
 }
 
-bool
-rbffi_thread_equal(const rbffi_thread_t* lhs, const rbffi_thread_t* rhs)
+void 
+rbffi_frame_push(rbffi_frame_t* frame)
 {
-    return lhs->valid && rhs->valid && 
+    memset(frame, 0, sizeof(*frame));
+    frame->has_gvl = true;
+    frame->exc = Qnil;
+    
 #ifdef _WIN32
-            lhs->id == rhs->id;
+    frame->prev = TlsGetValue(frame_thread_key);
+    TlsSetValue(frame_thread_key, frame);
 #else
-            pthread_equal(lhs->id, rhs->id);
+    frame->td = thread_data_get();
+    frame->prev = frame->td->frame;
+    frame->td->frame = frame;
 #endif
 }
 
-bool
-rbffi_thread_has_gvl_p(void)
+void 
+rbffi_frame_pop(rbffi_frame_t* frame)
 {
 #ifdef _WIN32
-    return rbffi_active_thread.valid && rbffi_active_thread.id == GetCurrentThreadId();
+    TlsSetValue(frame_thread_key, frame->prev);
 #else
-    return rbffi_active_thread.valid && pthread_equal(rbffi_active_thread.id, pthread_self());
+    frame->td->frame = frame->prev;
 #endif
 }
 
-#ifndef HAVE_RB_THREAD_BLOCKING_REGION
+#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
 
 #if !defined(_WIN32)
 
@@ -295,3 +305,39 @@ rbffi_thread_blocking_region(VALUE (*func)(void *), void *data1, void (*ubf)(voi
 
 #endif /* HAVE_RB_THREAD_BLOCKING_REGION */
 
+#ifndef _WIN32
+static struct thread_data* thread_data_init(void);
+
+static inline struct thread_data*
+thread_data_get(void)
+{
+    struct thread_data* td = (struct thread_data *) pthread_getspecific(thread_data_key);
+    return td != NULL ? td : thread_data_init();
+}
+
+static struct thread_data*
+thread_data_init(void)
+{
+    struct thread_data* td = calloc(1, sizeof(struct thread_data));
+
+    pthread_setspecific(thread_data_key, td);
+
+    return td;
+}
+
+static void
+thread_data_free(void *ptr)
+{
+    free(ptr);
+}
+#endif
+
+void
+rbffi_Thread_Init(VALUE moduleFFI)
+{
+#ifdef _WIN32
+    frame_thread_key = TlsAlloc();
+#else
+    pthread_key_create(&thread_data_key, thread_data_free);    
+#endif
+}
diff --git a/ext/ffi_c/Thread.h b/ext/ffi_c/Thread.h
index 85050ba..aa7280b 100644
--- a/ext/ffi_c/Thread.h
+++ b/ext/ffi_c/Thread.h
@@ -35,11 +35,6 @@ extern "C" {
 #endif
 
 
-#ifdef HAVE_RUBY_THREAD_HAS_GVL_P
-    extern int ruby_thread_has_gvl_p(void);
-# define rbffi_thread_has_gvl_p ruby_thread_has_gvl_p
-#else
-
 #ifdef _WIN32
 # include <windows.h>
 #else
@@ -53,17 +48,29 @@ typedef struct {
     pthread_t id;
 #endif
     bool valid;
+    bool has_gvl;
     VALUE exc;
 } rbffi_thread_t;
 
-extern rbffi_thread_t rbffi_active_thread;
-rbffi_thread_t rbffi_thread_self();
-bool rbffi_thread_equal(const rbffi_thread_t* lhs, const rbffi_thread_t* rhs);
-bool rbffi_thread_has_gvl_p(void);
-
+typedef struct rbffi_frame {
+#ifndef _WIN32
+    struct thread_data* td;
 #endif
-#ifdef HAVE_RB_THREAD_BLOCKING_REGION
+    struct rbffi_frame* prev;
+    bool has_gvl;
+    VALUE exc;
+} rbffi_frame_t;
+
+rbffi_frame_t* rbffi_frame_current(void);
+void rbffi_frame_push(rbffi_frame_t* frame);
+void rbffi_frame_pop(rbffi_frame_t* frame);
+
+#ifdef HAVE_RB_THREAD_CALL_WITHOUT_GVL
+# define rbffi_thread_blocking_region rb_thread_call_without_gvl
+
+#elif defined(HAVE_RB_THREAD_BLOCKING_REGION)
 # define rbffi_thread_blocking_region rb_thread_blocking_region
+
 #else
 
 VALUE rbffi_thread_blocking_region(VALUE (*func)(void *), void *data1, void (*ubf)(void *), void *data2);
diff --git a/ext/ffi_c/Variadic.c b/ext/ffi_c/Variadic.c
index ca02e19..18cf9bb 100644
--- a/ext/ffi_c/Variadic.c
+++ b/ext/ffi_c/Variadic.c
@@ -162,8 +162,7 @@ variadic_invoke(VALUE self, VALUE parameterTypes, VALUE parameterValues)
     VALUE* argv;
     int paramCount = 0, i;
     ffi_status ffiStatus;
-    VALUE exc;
-    rbffi_thread_t oldThread;
+    rbffi_frame_t frame = { 0 };
 
     Check_Type(parameterTypes, T_ARRAY);
     Check_Type(parameterValues, T_ARRAY);
@@ -238,18 +237,16 @@ variadic_invoke(VALUE self, VALUE parameterTypes, VALUE parameterValues)
 
     rbffi_SetupCallParams(paramCount, argv, -1, paramTypes, params,
         ffiValues, NULL, 0, invoker->rbEnums);
-    oldThread = rbffi_active_thread;
-    rbffi_active_thread = rbffi_thread_self();
-
+    
+    rbffi_frame_push(&frame);
     ffi_call(&cif, FFI_FN(invoker->function), retval, ffiValues);
-
-    exc = rbffi_active_thread.exc;
-    rbffi_active_thread = oldThread;
-    if (exc != Qnil) {
-        rb_exc_raise(exc);
-    }
+    rbffi_frame_pop(&frame);
     
     rbffi_save_errno();
+    
+    if (RTEST(frame.exc) && frame.exc != Qnil) {
+        rb_exc_raise(frame.exc);
+    }
 
     return rbffi_NativeValue_ToRuby(invoker->returnType, invoker->rbReturnType, retval);
 }
diff --git a/ext/ffi_c/ffi.c b/ext/ffi_c/ffi.c
index 4490cc6..2554511 100644
--- a/ext/ffi_c/ffi.c
+++ b/ext/ffi_c/ffi.c
@@ -49,7 +49,8 @@ VALUE rbffi_FFIModule = Qnil;
 static VALUE moduleFFI = Qnil;
 
 void
-Init_ffi_c(void) {
+Init_ffi_c(void) 
+{
     /* 
      * Document-module: FFI
      *
@@ -58,7 +59,8 @@ Init_ffi_c(void) {
     rbffi_FFIModule = moduleFFI = rb_define_module("FFI");
     rb_global_variable(&rbffi_FFIModule);
 
-
+    rbffi_Thread_Init(rbffi_FFIModule);
+    
     /* FFI::Type needs to be initialized before most other classes */
     rbffi_Type_Init(moduleFFI);
 
diff --git a/ext/ffi_c/rbffi.h b/ext/ffi_c/rbffi.h
index 46605f3..a8d6e3e 100644
--- a/ext/ffi_c/rbffi.h
+++ b/ext/ffi_c/rbffi.h
@@ -38,6 +38,7 @@ extern void rbffi_Variadic_Init(VALUE ffiModule);
 extern void rbffi_DataConverter_Init(VALUE ffiModule);
 extern VALUE rbffi_AbstractMemoryClass, rbffi_InvokerClass;
 extern int rbffi_type_size(VALUE type);
+extern void rbffi_Thread_Init(VALUE moduleFFI);
 
 #ifdef	__cplusplus
 }