Skip to content
Permalink
Browse files

Add wrappers to blocking functions and callbacks in order to release

the GVL of ruby 1.9 as long as not in ruby code.

This allowes better concurrency in threaded applications.
  • Loading branch information
larskanis committed Jan 22, 2013
1 parent faf96fa commit 869043c5fb0bbc6cd272e5245b975a75f359f8a8
Showing with 234 additions and 20 deletions.
  1. +2 −0 ext/extconf.rb
  2. +13 −0 ext/gvl_wrappers.c
  3. +185 −0 ext/gvl_wrappers.h
  4. +3 −0 ext/pg.h
  5. +19 −20 ext/pg_connection.c
  6. +12 −0 spec/pg/connection_spec.rb
@@ -69,6 +69,8 @@

have_func 'rb_encdb_alias'
have_func 'rb_enc_alias'
have_func 'rb_thread_call_without_gvl'
have_func 'rb_thread_call_with_gvl'

have_const 'PGRES_COPY_BOTH', 'libpq-fe.h'
have_const 'PGRES_SINGLE_TUPLE', 'libpq-fe.h'
@@ -0,0 +1,13 @@
/*
* gvl_wrappers.c - Wrapper functions for locking/unlocking the Ruby GVL
*
*/

#include "pg.h"

FOR_EACH_BLOCKING_FUNCTION( DEFINE_GVL_WRAPPER_STRUCT );
FOR_EACH_BLOCKING_FUNCTION( DEFINE_GVL_SKELETON );
FOR_EACH_BLOCKING_FUNCTION( DEFINE_GVL_STUB );
FOR_EACH_CALLBACK_FUNCTION( DEFINE_GVL_WRAPPER_STRUCT );
FOR_EACH_CALLBACK_FUNCTION( DEFINE_GVLCB_SKELETON );
FOR_EACH_CALLBACK_FUNCTION( DEFINE_GVLCB_STUB );
@@ -0,0 +1,185 @@
/*
* gvl_wrappers.h - Wrapper functions for locking/unlocking the Ruby GVL
*
* These are some obscure preprocessor directives that allow to generate
* drop-in replacement wrapper functions in a declarative manner.
* These wrapper functions ensure that ruby's GVL is released on each
* function call and reacquired at the end of the call or in callbacks.
* This way blocking functions calls don't block concurrent ruby threads.
*
* The wrapper of each function is prefixed by "gvl_".
*
* Use "gcc -E" to retrieve the generated code.
*/

#ifndef __gvl_wrappers_h
#define __gvl_wrappers_h

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

#if defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL)
extern void *rb_thread_call_without_gvl(void *(*func)(void *), void *data1,
rb_unblock_function_t *ubf, void *data2);
#endif

#define DEFINE_PARAM_LIST1(type, name) \
name,

#define DEFINE_PARAM_LIST2(type, name) \
p->params.name,

#define DEFINE_PARAM_LIST3(type, name) \
type name,

#define DEFINE_PARAM_DECL(type, name) \
type name;

#define DEFINE_GVL_WRAPPER_STRUCT(name, when_non_void, rettype, lastparamtype, lastparamname) \
struct gvl_wrapper_##name##_params { \
struct { \
FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_DECL) \
lastparamtype lastparamname; \
} params; \
when_non_void( rettype retval; ) \
};

#define DEFINE_GVL_SKELETON(name, when_non_void, rettype, lastparamtype, lastparamname) \
static void * gvl_##name##_skeleton( void *data ){ \
struct gvl_wrapper_##name##_params *p = (struct gvl_wrapper_##name##_params*)data; \
when_non_void( p->retval = ) \
name( FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST2) p->params.lastparamname ); \
return NULL; \
}

#if defined(HAVE_RB_THREAD_CALL_WITHOUT_GVL)
#define DEFINE_GVL_STUB(name, when_non_void, rettype, lastparamtype, lastparamname) \
rettype gvl_##name(FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST3) lastparamtype lastparamname){ \
struct gvl_wrapper_##name##_params params = { \
{FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST1) lastparamname}, when_non_void((rettype)0) \
}; \
rb_thread_call_without_gvl(gvl_##name##_skeleton, &params, RUBY_UBF_IO, 0); \
when_non_void( return params.retval; ) \
}
#else
#define DEFINE_GVL_STUB(name, when_non_void, rettype, lastparamtype, lastparamname) \
rettype gvl_##name(FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST3) lastparamtype lastparamname){ \
return name( FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST1) lastparamname ); \
}
#endif

#define DEFINE_GVL_STUB_DECL(name, when_non_void, rettype, lastparamtype, lastparamname) \
rettype gvl_##name(FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST3) lastparamtype lastparamname);

#define DEFINE_GVLCB_SKELETON(name, when_non_void, rettype, lastparamtype, lastparamname) \
static void * gvl_##name##_skeleton( void *data ){ \
struct gvl_wrapper_##name##_params *p = (struct gvl_wrapper_##name##_params*)data; \
when_non_void( p->retval = ) \
name( FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST2) p->params.lastparamname ); \
return NULL; \
}

#if defined(HAVE_RB_THREAD_CALL_WITH_GVL)
#define DEFINE_GVLCB_STUB(name, when_non_void, rettype, lastparamtype, lastparamname) \
rettype gvl_##name(FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST3) lastparamtype lastparamname){ \
struct gvl_wrapper_##name##_params params = { \
{FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST1) lastparamname}, when_non_void((rettype)0) \
}; \
rb_thread_call_with_gvl(gvl_##name##_skeleton, &params); \
when_non_void( return params.retval; ) \
}
#else
#define DEFINE_GVLCB_STUB(name, when_non_void, rettype, lastparamtype, lastparamname) \
rettype gvl_##name(FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST3) lastparamtype lastparamname){ \
return name( FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST1) lastparamname ); \
}
#endif

#define GVL_TYPE_VOID(string)
#define GVL_TYPE_NONVOID(string) string


/*
* Definitions of blocking functions and their parameters
*/

#define FOR_EACH_PARAM_OF_PQexec(param) \
param(PGconn *, conn)

#define FOR_EACH_PARAM_OF_PQexecParams(param) \
param(PGconn *, conn) \
param(const char *, command) \
param(int, nParams) \
param(const Oid *, paramTypes) \
param(const char * const *, paramValues) \
param(const int *, paramLengths) \
param(const int *, paramFormats)

#define FOR_EACH_PARAM_OF_PQexecPrepared(param) \
param(PGconn *, conn) \
param(const char *, stmtName) \
param(int, nParams) \
param(const char * const *, paramValues) \
param(const int *, paramLengths) \
param(const int *, paramFormats)

#define FOR_EACH_PARAM_OF_PQprepare(param) \
param(PGconn *, conn) \
param(const char *, stmtName) \
param(const char *, query) \
param(int, nParams)

#define FOR_EACH_PARAM_OF_PQdescribePrepared(param) \
param(PGconn *, conn)

#define FOR_EACH_PARAM_OF_PQdescribePortal(param) \
param(PGconn *, conn)

#define FOR_EACH_PARAM_OF_PQgetResult(param)

#define FOR_EACH_PARAM_OF_PQputCopyData(param) \
param(PGconn *, conn) \
param(const char *, buffer)

#define FOR_EACH_PARAM_OF_PQputCopyEnd(param) \
param(PGconn *, conn)

#define FOR_EACH_PARAM_OF_PQgetCopyData(param) \
param(PGconn *, conn) \
param(char **, buffer)

/* function( name, void_or_nonvoid, returntype, lastparamtype, lastparamname ) */
#define FOR_EACH_BLOCKING_FUNCTION(function) \
function(PQexec, GVL_TYPE_NONVOID, PGresult *, const char *, command) \
function(PQexecParams, GVL_TYPE_NONVOID, PGresult *, int, resultFormat) \
function(PQexecPrepared, GVL_TYPE_NONVOID, PGresult *, int, resultFormat) \
function(PQprepare, GVL_TYPE_NONVOID, PGresult *, const Oid *, paramTypes) \
function(PQdescribePrepared, GVL_TYPE_NONVOID, PGresult *, const char *, stmtName) \
function(PQdescribePortal, GVL_TYPE_NONVOID, PGresult *, const char *, portalName) \
function(PQgetResult, GVL_TYPE_NONVOID, PGresult *, PGconn *, conn) \
function(PQputCopyData, GVL_TYPE_NONVOID, int, int, nbytes) \
function(PQputCopyEnd, GVL_TYPE_NONVOID, int, const char *, errormsg) \
function(PQgetCopyData, GVL_TYPE_NONVOID, int, int, async) \

FOR_EACH_BLOCKING_FUNCTION( DEFINE_GVL_STUB_DECL );


/*
* Definitions of callback functions and their parameters
*/

#define FOR_EACH_PARAM_OF_notice_processor_proxy(param) \
param(void *, arg)

#define FOR_EACH_PARAM_OF_notice_receiver_proxy(param) \
param(void *, arg)

/* function( name, void_or_nonvoid, returntype, lastparamtype, lastparamname ) */
#define FOR_EACH_CALLBACK_FUNCTION(function) \
function(notice_processor_proxy, GVL_TYPE_VOID, void, const char *, message) \
function(notice_receiver_proxy, GVL_TYPE_VOID, void, const PGresult *, result) \

FOR_EACH_CALLBACK_FUNCTION( DEFINE_GVL_STUB_DECL );

#endif /* end __gvl_wrappers_h */
@@ -77,6 +77,7 @@ __declspec(dllexport)
typedef long suseconds_t;
#endif

#include "gvl_wrappers.h"

/***************************************************************************
* Globals
@@ -119,5 +120,7 @@ int pg_enc_get_index _(( VALUE ));
rb_encoding *pg_conn_enc_get _(( PGconn * ));
#endif /* M17N_SUPPORTED */

void notice_receiver_proxy(void *arg, const PGresult *result);
void notice_processor_proxy(void *arg, const char *message);

#endif /* end __pg_h */

0 comments on commit 869043c

Please sign in to comment.
You can’t perform that action at this time.