Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Remove old code for 'blocking: true' on Ruby before 2.0 #683

Merged
merged 1 commit into from Apr 12, 2019
Merged

Remove old code for 'blocking: true' on Ruby before 2.0 #683

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
27 changes: 3 additions & 24 deletions ext/ffi_c/Call.c
View file
Open in desktop
Expand Up @@ -43,10 +43,8 @@
#endif
#include <errno.h>
#include <ruby.h>
#if defined(HAVE_RUBY_THREAD_H)
#include <ruby/thread.h>
#endif
#if defined(HAVE_NATIVETHREAD) && (defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL)) && !defined(_WIN32)
#if defined(HAVE_NATIVETHREAD) && !defined(_WIN32)
# include <signal.h>
# include <pthread.h>
#endif
Expand Down Expand Up @@ -359,7 +357,7 @@ call_blocking_function(void* data)
VALUE
rbffi_do_blocking_call(void *data)
{
rbffi_thread_blocking_region(call_blocking_function, data, (void *) -1, NULL);
rb_thread_call_without_gvl(call_blocking_function, data, (void *) -1, NULL);

return Qnil;
}
Expand All @@ -386,22 +384,11 @@ rbffi_CallFunction(int argc, VALUE* argv, void* function, FunctionType* fnInfo)
if (unlikely(fnInfo->blocking)) {
rbffi_blocking_call_t* bc;

/*
* due to the way thread switching works on older ruby variants, we
* cannot allocate anything passed to the blocking function on the stack
*/
#if defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL)
/* allocate information passed to the blocking function on the stack */
ffiValues = ALLOCA_N(void *, fnInfo->parameterCount);
params = ALLOCA_N(FFIStorage, fnInfo->parameterCount);
bc = ALLOCA_N(rbffi_blocking_call_t, 1);
bc->retval = retval;
#else
ffiValues = ALLOC_N(void *, fnInfo->parameterCount);
params = ALLOC_N(FFIStorage, fnInfo->parameterCount);
bc = ALLOC_N(rbffi_blocking_call_t, 1);
bc->retval = xmalloc(MAX(fnInfo->ffi_cif.rtype->size, FFI_SIZEOF_ARG));
bc->stkretval = retval;
#endif
bc->cif = fnInfo->ffi_cif;
bc->function = function;
bc->ffiValues = ffiValues;
Expand All @@ -416,14 +403,6 @@ rbffi_CallFunction(int argc, VALUE* argv, void* function, FunctionType* fnInfo)
rb_rescue2(rbffi_do_blocking_call, (VALUE) bc, rbffi_save_frame_exception, (VALUE) &frame, rb_eException, (VALUE) 0);
rbffi_frame_pop(&frame);

#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
memcpy(bc->stkretval, bc->retval, MAX(bc->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);
Expand Down
3 changes: 0 additions & 3 deletions ext/ffi_c/Call.h
View file
Open in desktop
Expand Up @@ -94,9 +94,6 @@ typedef struct rbffi_blocking_call {
void **ffiValues;
void* retval;
void* params;
#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
void* stkretval;
#endif
} rbffi_blocking_call_t;

VALUE rbffi_do_blocking_call(void* data);
Expand Down
105 changes: 0 additions & 105 deletions ext/ffi_c/Function.c
View file
Open in desktop
Expand Up @@ -47,9 +47,7 @@
# endif
#endif
#include <ruby.h>
#if defined(HAVE_RUBY_THREAD_H)
#include <ruby/thread.h>
#endif

#include <ffi.h>
#if defined(HAVE_NATIVETHREAD) && !defined(_WIN32)
Expand Down Expand Up @@ -101,10 +99,6 @@ static VALUE async_cb_event(void *);
static VALUE async_cb_call(void *);
#endif

#ifdef HAVE_RB_THREAD_CALL_WITH_GVL
extern void *rb_thread_call_with_gvl(void *(*func)(void *), void *data1);
#endif

#ifdef HAVE_RUBY_THREAD_HAS_GVL_P
extern int ruby_thread_has_gvl_p(void);
#define rbffi_thread_has_gvl_p(frame) ruby_thread_has_gvl_p()
Expand Down Expand Up @@ -157,15 +151,9 @@ static struct gvl_callback* async_cb_list = NULL;
# ifndef _WIN32
static pthread_mutex_t async_cb_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t async_cb_cond = PTHREAD_COND_INITIALIZER;
# if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
static int async_cb_pipe[2];
# endif
# else
static HANDLE async_cb_cond;
static CRITICAL_SECTION async_cb_lock;
# if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
static int async_cb_pipe[2];
# endif
# endif
#endif

Expand Down Expand Up @@ -344,16 +332,8 @@ function_init(VALUE self, VALUE rbFunctionInfo, VALUE rbProc)

#if defined(DEFER_ASYNC_CALLBACK)
if (async_cb_thread == Qnil) {
#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL)) && defined(_WIN32)
_pipe(async_cb_pipe, 1024, O_BINARY);
#elif !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
pipe(async_cb_pipe);
fcntl(async_cb_pipe[0], F_SETFL, fcntl(async_cb_pipe[0], F_GETFL) | O_NONBLOCK);
fcntl(async_cb_pipe[1], F_SETFL, fcntl(async_cb_pipe[1], F_GETFL) | O_NONBLOCK);
#endif
async_cb_thread = rb_thread_create(async_cb_event, NULL);
}

#endif

fn->closure = rbffi_Closure_Alloc(fn->info->closurePool);
Expand Down Expand Up @@ -515,17 +495,8 @@ callback_invoke(ffi_cif* cif, void* retval, void** parameters, void* user_data)
cb.next = async_cb_list;
async_cb_list = &cb;

#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
pthread_mutex_unlock(&async_cb_mutex);
/* Only signal if the list was empty */
if (empty) {
char c;
write(async_cb_pipe[1], &c, 1);
}
#else
pthread_cond_signal(&async_cb_cond);
pthread_mutex_unlock(&async_cb_mutex);
#endif

/* Wait for the thread executing the ruby callback to signal it is done */
pthread_mutex_lock(&cb.async_mutex);
Expand All @@ -549,15 +520,7 @@ callback_invoke(ffi_cif* cif, void* retval, void** parameters, void* user_data)
async_cb_list = &cb;
LeaveCriticalSection(&async_cb_lock);

#if !(defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL))
/* Only signal if the list was empty */
if (empty) {
char c;
write(async_cb_pipe[1], &c, 1);
}
#else
SetEvent(async_cb_cond);
#endif

/* Wait for the thread executing the ruby callback to signal it is done */
WaitForSingleObject(cb.async_event, INFINITE);
Expand All @@ -575,19 +538,14 @@ struct async_wait {
static void * async_cb_wait(void *);
static void async_cb_stop(void *);

#if defined(HAVE_RB_THREAD_BLOCKING_REGION) || defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL)
static VALUE
async_cb_event(void* unused)
{
struct async_wait w = { 0 };

w.stop = false;
while (!w.stop) {
#if defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL)
rb_thread_call_without_gvl(async_cb_wait, &w, async_cb_stop, &w);
#else
rb_thread_blocking_region(async_cb_wait, &w, async_cb_stop, &w);
#endif
if (w.cb != NULL) {
/* Start up a new ruby thread to run the ruby callback */
rb_thread_create(async_cb_call, w.cb);
Expand All @@ -597,69 +555,6 @@ async_cb_event(void* unused)
return Qnil;
}

#elif defined(_WIN32)
static VALUE
async_cb_event(void* unused)
{
while (true) {
struct gvl_callback* cb;
char buf[64];
fd_set rfds;

FD_ZERO(&rfds);
FD_SET(async_cb_pipe[0], &rfds);
rb_thread_select(async_cb_pipe[0] + 1, &rfds, NULL, NULL, NULL);
read(async_cb_pipe[0], buf, sizeof(buf));

EnterCriticalSection(&async_cb_lock);
cb = async_cb_list;
async_cb_list = NULL;
LeaveCriticalSection(&async_cb_lock);

while (cb != NULL) {
struct gvl_callback* next = cb->next;
/* Start up a new ruby thread to run the ruby callback */
rb_thread_create(async_cb_call, cb);
cb = next;
}
}

return Qnil;
}
#else
static VALUE
async_cb_event(void* unused)
{
while (true) {
struct gvl_callback* cb;
char buf[64];

if (read(async_cb_pipe[0], buf, sizeof(buf)) < 0) {
rb_thread_wait_fd(async_cb_pipe[0]);
while (read(async_cb_pipe[0], buf, sizeof (buf)) < 0) {
if (rb_io_wait_readable(async_cb_pipe[0]) != Qtrue) {
return Qfalse;
}
}
}

pthread_mutex_lock(&async_cb_mutex);
cb = async_cb_list;
async_cb_list = NULL;
pthread_mutex_unlock(&async_cb_mutex);

while (cb != NULL) {
struct gvl_callback* next = cb->next;
/* Start up a new ruby thread to run the ruby callback */
rb_thread_create(async_cb_call, cb);
cb = next;
}
}

return Qnil;
}
#endif

#ifdef _WIN32
static void *
async_cb_wait(void *data)
Expand Down