From 58eb8195a6a49b32a71e231ebce33a0a42590a9b Mon Sep 17 00:00:00 2001 From: Lars Kanis Date: Sun, 7 Jun 2026 22:18:54 +0200 Subject: [PATCH 1/4] Remove GLV unlocking at all functions which process data modifiable in a second thread This removes possible VM crashs when data to be sent is modified/cleared in a second thread. It works by keeping the GVL lock for libpq functions that don't immediately process all the data and don't make a copy of it. These are the `PQsend*`, `PQexec*` and some related functions. Since pg-1.3 all the blocking functions or states are avoided by using the non-blocking API of libpq. Therefore holding the GVL somewhat longer shouldn't matter that much. Having some libpq function with and without unlocked GVL, results in `rb_thread_call_with_gvl()` sometimes needed and sometimes not to process callbacks. Therefore `ruby_thread_has_gvl_p()` is used to check if it's needed on ruby<4.0. In ruby-4.0+ `rb_thread_call_with_gvl()` doesn't care about whether GVL is already locked or not, so that it can be called in both cases. Fixes #721 --- ext/gvl_wrappers.h | 152 ++++++++----------------------------- ext/pg.h | 2 + ext/pg_connection.c | 52 ++++++------- spec/pg/connection_spec.rb | 6 +- 4 files changed, 65 insertions(+), 147 deletions(-) diff --git a/ext/gvl_wrappers.h b/ext/gvl_wrappers.h index f048d7055..81eb646ea 100644 --- a/ext/gvl_wrappers.h +++ b/ext/gvl_wrappers.h @@ -15,12 +15,17 @@ #ifndef __gvl_wrappers_h #define __gvl_wrappers_h +#include #include #ifdef RUBY_EXTCONF_H # include RUBY_EXTCONF_H #endif +#if RUBY_API_VERSION_MAJOR < 4 +extern int ruby_thread_has_gvl_p(void); +#endif + #ifndef LIBPQ_HAS_CHUNK_MODE typedef struct pg_cancel_conn PGcancelConn; #endif @@ -83,20 +88,35 @@ typedef struct pg_cancel_conn PGcancelConn; } #ifdef ENABLE_GVL_UNLOCK -#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, ¶ms); \ - when_non_void( return params.retval; ) \ - } + #if RUBY_API_VERSION_MAJOR >= 4 || defined(TRUFFLERUBY) + #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, ¶ms); \ + 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){ \ + struct gvl_wrapper_##name##_params params = { \ + {FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST1) lastparamname}, when_non_void((rettype)0) \ + }; \ + if (ruby_thread_has_gvl_p()) { \ + gvl_##name##_skeleton(¶ms); \ + } else { \ + rb_thread_call_with_gvl(gvl_##name##_skeleton, ¶ms); \ + } \ + when_non_void( return params.retval; ) \ + } + #endif #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){ \ - when_non_void( return ) \ - name( FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST1) lastparamname ); \ - } + #define DEFINE_GVLCB_STUB(name, when_non_void, rettype, lastparamtype, lastparamname) \ + rettype gvl_##name(FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST3) lastparamtype lastparamname){ \ + when_non_void( return ) \ + name( FOR_EACH_PARAM_OF_##name(DEFINE_PARAM_LIST1) lastparamname ); \ + } #endif #define GVL_TYPE_VOID(string) @@ -121,50 +141,8 @@ typedef struct pg_cancel_conn PGcancelConn; #define FOR_EACH_PARAM_OF_PQping(param) -#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_PQclosePrepared(param) \ - param(PGconn *, conn) - -#define FOR_EACH_PARAM_OF_PQclosePortal(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) @@ -174,48 +152,8 @@ typedef struct pg_cancel_conn PGcancelConn; #define FOR_EACH_PARAM_OF_PQnotifies(param) -#define FOR_EACH_PARAM_OF_PQsendQuery(param) \ - param(PGconn *, conn) - -#define FOR_EACH_PARAM_OF_PQsendQueryParams(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_PQsendPrepare(param) \ - param(PGconn *, conn) \ - param(const char *, stmtName) \ - param(const char *, query) \ - param(int, nParams) - -#define FOR_EACH_PARAM_OF_PQsendQueryPrepared(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_PQsendDescribePrepared(param) \ - param(PGconn *, conn) - -#define FOR_EACH_PARAM_OF_PQsendDescribePortal(param) \ - param(PGconn *, conn) - -#define FOR_EACH_PARAM_OF_PQsendClosePrepared(param) \ - param(PGconn *, conn) - -#define FOR_EACH_PARAM_OF_PQsendClosePortal(param) \ - param(PGconn *, conn) - #define FOR_EACH_PARAM_OF_PQpipelineSync(param) -#define FOR_EACH_PARAM_OF_PQsendPipelineSync(param) - #define FOR_EACH_PARAM_OF_PQsetClientEncoding(param) \ param(PGconn *, conn) @@ -225,11 +163,6 @@ typedef struct pg_cancel_conn PGcancelConn; #define FOR_EACH_PARAM_OF_PQcancelStart(param) #define FOR_EACH_PARAM_OF_PQcancelPoll(param) -#define FOR_EACH_PARAM_OF_PQencryptPasswordConn(param) \ - param(PGconn *, conn) \ - param(const char *, passwd) \ - param(const char *, user) - #define FOR_EACH_PARAM_OF_PQcancel(param) \ param(PGcancel *, cancel) \ param(char *, errbuf) @@ -243,35 +176,16 @@ typedef struct pg_cancel_conn PGcancelConn; function(PQresetStart, GVL_TYPE_NONVOID, int, PGconn *, conn) \ function(PQresetPoll, GVL_TYPE_NONVOID, PostgresPollingStatusType, PGconn *, conn) \ function(PQping, GVL_TYPE_NONVOID, PGPing, const char *, conninfo) \ - 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(PQclosePrepared, GVL_TYPE_NONVOID, PGresult *, const char *, stmtName) \ - function(PQclosePortal, 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) \ function(PQnotifies, GVL_TYPE_NONVOID, PGnotify *, PGconn *, conn) \ - function(PQsendQuery, GVL_TYPE_NONVOID, int, const char *, query) \ - function(PQsendQueryParams, GVL_TYPE_NONVOID, int, int, resultFormat) \ - function(PQsendPrepare, GVL_TYPE_NONVOID, int, const Oid *, paramTypes) \ - function(PQsendQueryPrepared, GVL_TYPE_NONVOID, int, int, resultFormat) \ - function(PQsendDescribePrepared, GVL_TYPE_NONVOID, int, const char *, stmt) \ - function(PQsendDescribePortal, GVL_TYPE_NONVOID, int, const char *, portal) \ - function(PQsendClosePrepared, GVL_TYPE_NONVOID, int, const char *, stmt) \ - function(PQsendClosePortal, GVL_TYPE_NONVOID, int, const char *, portal) \ function(PQpipelineSync, GVL_TYPE_NONVOID, int, PGconn *, conn) \ - function(PQsendPipelineSync, GVL_TYPE_NONVOID, int, PGconn *, conn) \ function(PQsetClientEncoding, GVL_TYPE_NONVOID, int, const char *, encoding) \ function(PQisBusy, GVL_TYPE_NONVOID, int, PGconn *, conn) \ function(PQcancelBlocking, GVL_TYPE_NONVOID, int, PGcancelConn *, conn) \ function(PQcancelStart, GVL_TYPE_NONVOID, int, PGcancelConn *, conn) \ function(PQcancelPoll, GVL_TYPE_NONVOID, PostgresPollingStatusType, PGcancelConn *, conn) \ - function(PQencryptPasswordConn, GVL_TYPE_NONVOID, char *, const char *, algorithm) \ function(PQcancel, GVL_TYPE_NONVOID, int, int, errbufsize); FOR_EACH_BLOCKING_FUNCTION( DEFINE_GVL_STUB_DECL ); diff --git a/ext/pg.h b/ext/pg.h index 006eed48f..eb1dc5096 100644 --- a/ext/pg.h +++ b/ext/pg.h @@ -82,6 +82,8 @@ typedef long suseconds_t; #define pg_gc_location(x) x = rb_gc_location(x) +extern int ruby_native_thread_p(void); + /* For compatibility with ruby < 3.0 */ #ifndef RUBY_TYPED_FROZEN_SHAREABLE #define PG_RUBY_TYPED_FROZEN_SHAREABLE 0 diff --git a/ext/pg_connection.c b/ext/pg_connection.c index cf42782ca..de9fce048 100644 --- a/ext/pg_connection.c +++ b/ext/pg_connection.c @@ -439,7 +439,7 @@ pgconn_sync_encrypt_password(int argc, VALUE *argv, VALUE self) Check_Type(password, T_STRING); Check_Type(username, T_STRING); - encrypted = gvl_PQencryptPasswordConn(conn, StringValueCStr(password), StringValueCStr(username), RTEST(algorithm) ? StringValueCStr(algorithm) : NULL); + encrypted = PQencryptPasswordConn(conn, StringValueCStr(password), StringValueCStr(username), RTEST(algorithm) ? StringValueCStr(algorithm) : NULL); if ( encrypted ) { rval = rb_str_new2( encrypted ); PQfreemem( encrypted ); @@ -1134,11 +1134,11 @@ static VALUE pgconn_sync_exec_params( int, VALUE *, VALUE ); * This function has the same behavior as #async_exec, but is implemented using the synchronous command processing API of libpq. * It's not recommended to use explicit sync or async variants but #exec instead, unless you have a good reason to do so. * - * Both #sync_exec and #async_exec release the GVL while waiting for server response, so that concurrent threads will get executed. - * However #async_exec has two advantages: + * However #async_exec has some advantages: * - * 1. #async_exec can be aborted by signals (like Ctrl-C), while #exec blocks signal processing until the query is answered. - * 2. Ruby VM gets notified about IO blocked operations and can pass them through Fiber.scheduler. + * 1. Only #async_exec allows concurrent threads to run, while waiting for a server response. + * 2. #async_exec can be aborted by signals (like Ctrl-C), while #exec blocks signal processing until the query is answered. + * 3. Ruby VM gets notified about IO blocked operations and can pass them through Fiber.scheduler. * So only async_* methods are compatible to event based schedulers like the async gem. */ static VALUE @@ -1153,7 +1153,7 @@ pgconn_sync_exec(int argc, VALUE *argv, VALUE self) VALUE query_str = argv[0]; VALUE transcoded_str; - result = gvl_PQexec(this->pgconn, pg_cstr_enc(query_str, this->enc_idx, &transcoded_str)); + result = PQexec(this->pgconn, pg_cstr_enc(query_str, this->enc_idx, &transcoded_str)); RB_GC_GUARD(transcoded_str); rb_pgresult = pg_new_result(result, self); pg_result_check(rb_pgresult); @@ -1473,7 +1473,7 @@ pgconn_sync_exec_params( int argc, VALUE *argv, VALUE self ) resultFormat = NIL_P(in_res_fmt) ? 0 : NUM2INT(in_res_fmt); nParams = alloc_query_params( ¶msData ); - result = gvl_PQexecParams(this->pgconn, pg_cstr_enc(command, paramsData.enc_idx, &transcoded_str), nParams, paramsData.types, + result = PQexecParams(this->pgconn, pg_cstr_enc(command, paramsData.enc_idx, &transcoded_str), nParams, paramsData.types, (const char * const *)paramsData.values, paramsData.lengths, paramsData.formats, resultFormat); RB_GC_GUARD(transcoded_str); @@ -1529,7 +1529,7 @@ pgconn_sync_prepare(int argc, VALUE *argv, VALUE self) paramTypes[i] = NUM2UINT(param); } } - result = gvl_PQprepare(this->pgconn, name_cstr, command_cstr, nParams, paramTypes); + result = PQprepare(this->pgconn, name_cstr, command_cstr, nParams, paramTypes); RB_GC_GUARD(transcoded_str1); RB_GC_GUARD(transcoded_str2); @@ -1572,7 +1572,7 @@ pgconn_sync_exec_prepared(int argc, VALUE *argv, VALUE self) resultFormat = NIL_P(in_res_fmt) ? 0 : NUM2INT(in_res_fmt); nParams = alloc_query_params( ¶msData ); - result = gvl_PQexecPrepared(this->pgconn, pg_cstr_enc(name, paramsData.enc_idx, &transcoded_str), nParams, + result = PQexecPrepared(this->pgconn, pg_cstr_enc(name, paramsData.enc_idx, &transcoded_str), nParams, (const char * const *)paramsData.values, paramsData.lengths, paramsData.formats, resultFormat); @@ -1616,7 +1616,7 @@ pgconn_sync_describe_close_prepared_portal(VALUE self, VALUE name, PGresult *(*f static VALUE pgconn_sync_describe_prepared(VALUE self, VALUE stmt_name) { - return pgconn_sync_describe_close_prepared_portal(self, stmt_name, gvl_PQdescribePrepared); + return pgconn_sync_describe_close_prepared_portal(self, stmt_name, PQdescribePrepared); } @@ -1631,7 +1631,7 @@ pgconn_sync_describe_prepared(VALUE self, VALUE stmt_name) static VALUE pgconn_sync_describe_portal(VALUE self, VALUE stmt_name) { - return pgconn_sync_describe_close_prepared_portal(self, stmt_name, gvl_PQdescribePortal); + return pgconn_sync_describe_close_prepared_portal(self, stmt_name, PQdescribePortal); } @@ -1649,7 +1649,7 @@ pgconn_sync_describe_portal(VALUE self, VALUE stmt_name) static VALUE pgconn_sync_close_prepared(VALUE self, VALUE stmt_name) { - return pgconn_sync_describe_close_prepared_portal(self, stmt_name, gvl_PQclosePrepared); + return pgconn_sync_describe_close_prepared_portal(self, stmt_name, PQclosePrepared); } /* @@ -1665,7 +1665,7 @@ pgconn_sync_close_prepared(VALUE self, VALUE stmt_name) static VALUE pgconn_sync_close_portal(VALUE self, VALUE stmt_name) { - return pgconn_sync_describe_close_prepared_portal(self, stmt_name, gvl_PQclosePortal); + return pgconn_sync_describe_close_prepared_portal(self, stmt_name, PQclosePortal); } #endif @@ -2001,7 +2001,7 @@ pgconn_send_query(int argc, VALUE *argv, VALUE self) /* If called with no or nil parameters, use PQexec for compatibility */ if ( argc == 1 || (argc >= 2 && argc <= 4 && NIL_P(argv[1]) )) { - if(gvl_PQsendQuery(this->pgconn, pg_cstr_enc(argv[0], this->enc_idx, &transcoded_str)) == 0) + if(PQsendQuery(this->pgconn, pg_cstr_enc(argv[0], this->enc_idx, &transcoded_str)) == 0) pg_raise_conn_error( rb_eUnableToSend, self, "PQsendQuery %s", PQerrorMessage(this->pgconn)); RB_GC_GUARD(transcoded_str); @@ -2073,7 +2073,7 @@ pgconn_send_query_params(int argc, VALUE *argv, VALUE self) resultFormat = NIL_P(in_res_fmt) ? 0 : NUM2INT(in_res_fmt); nParams = alloc_query_params( ¶msData ); - result = gvl_PQsendQueryParams(this->pgconn, pg_cstr_enc(command, paramsData.enc_idx, &transcoded_str), nParams, paramsData.types, + result = PQsendQueryParams(this->pgconn, pg_cstr_enc(command, paramsData.enc_idx, &transcoded_str), nParams, paramsData.types, (const char * const *)paramsData.values, paramsData.lengths, paramsData.formats, resultFormat); RB_GC_GUARD(transcoded_str); @@ -2137,7 +2137,7 @@ pgconn_send_prepare(int argc, VALUE *argv, VALUE self) paramTypes[i] = NUM2UINT(param); } } - result = gvl_PQsendPrepare(this->pgconn, name_cstr, command_cstr, nParams, paramTypes); + result = PQsendPrepare(this->pgconn, name_cstr, command_cstr, nParams, paramTypes); RB_GC_GUARD(transcoded_str1); RB_GC_GUARD(transcoded_str2); @@ -2204,7 +2204,7 @@ pgconn_send_query_prepared(int argc, VALUE *argv, VALUE self) resultFormat = NIL_P(in_res_fmt) ? 0 : NUM2INT(in_res_fmt); nParams = alloc_query_params( ¶msData ); - result = gvl_PQsendQueryPrepared(this->pgconn, pg_cstr_enc(name, paramsData.enc_idx, &transcoded_str), nParams, + result = PQsendQueryPrepared(this->pgconn, pg_cstr_enc(name, paramsData.enc_idx, &transcoded_str), nParams, (const char * const *)paramsData.values, paramsData.lengths, paramsData.formats, resultFormat); @@ -2245,7 +2245,7 @@ static VALUE pgconn_send_describe_prepared(VALUE self, VALUE stmt_name) { return pgconn_send_describe_close_prepared_portal( - self, stmt_name, gvl_PQsendDescribePrepared, + self, stmt_name, PQsendDescribePrepared, "PQsendDescribePrepared"); } @@ -2261,7 +2261,7 @@ static VALUE pgconn_send_describe_portal(VALUE self, VALUE portal) { return pgconn_send_describe_close_prepared_portal( - self, portal, gvl_PQsendDescribePortal, + self, portal, PQsendDescribePortal, "PQsendDescribePortal"); } @@ -2279,7 +2279,7 @@ static VALUE pgconn_send_close_prepared(VALUE self, VALUE stmt_name) { return pgconn_send_describe_close_prepared_portal( - self, stmt_name, gvl_PQsendClosePrepared, + self, stmt_name, PQsendClosePrepared, "PQsendClosePrepared"); } @@ -2297,7 +2297,7 @@ static VALUE pgconn_send_close_portal(VALUE self, VALUE portal) { return pgconn_send_describe_close_prepared_portal( - self, portal, gvl_PQsendClosePortal, + self, portal, PQsendClosePortal, "PQsendClosePortal"); } #endif @@ -2793,7 +2793,7 @@ pgconn_sync_put_copy_data(int argc, VALUE *argv, VALUE self) Check_Type(buffer, T_STRING); - ret = gvl_PQputCopyData(this->pgconn, RSTRING_PTR(buffer), RSTRING_LENINT(buffer)); + ret = PQputCopyData(this->pgconn, RSTRING_PTR(buffer), RSTRING_LENINT(buffer)); if(ret == -1) pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(this->pgconn)); @@ -3456,12 +3456,12 @@ pgconn_discard_results(VALUE self) * and the PG::Result object will automatically be cleared when the block terminates. * In this instance, conn.exec returns the value of the block. * - * #exec is an alias for #async_exec which is almost identical to #sync_exec . + * #exec is an alias for #async_exec which is functional identical to #sync_exec . * #sync_exec is implemented on the simpler synchronous command processing API of libpq, whereas * #async_exec is implemented on the asynchronous API and on ruby's IO mechanisms. * Only #async_exec is compatible to Fiber.scheduler based asynchronous IO processing introduced in ruby-3.0. - * Both methods ensure that other threads can process while waiting for the server to - * complete the request, but #sync_exec blocks all signals to be processed until the query is finished. + * Only #async_exec ensures that other threads can process while waiting for the server to complete the request. + * In contrast #sync_exec blocks all threads and signals to be processed until the query is finished. * This is most notably visible by a delayed reaction to Control+C. * It's not recommended to use explicit sync or async variants but #exec instead, unless you have a good reason to do so. * @@ -3923,7 +3923,7 @@ static VALUE pgconn_send_pipeline_sync(VALUE self) { PGconn *conn = pg_get_pgconn(self); - int res = gvl_PQsendPipelineSync(conn); + int res = PQsendPipelineSync(conn); if( res != 1 ) pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn)); diff --git a/spec/pg/connection_spec.rb b/spec/pg/connection_spec.rb index 6ed2c5112..6ba963e55 100644 --- a/spec/pg/connection_spec.rb +++ b/spec/pg/connection_spec.rb @@ -805,11 +805,13 @@ conn.setnonblocking(true) res = nil - conn.exec <<-EOSQL + # use async_exec since sync_exec is no longer thread compatible, necessary for run_with_gate + conn.async_exec <<-EOSQL CREATE TEMP TABLE copytable (col1 TEXT); EOSQL - conn.exec( "COPY copytable FROM STDOUT CSV" ) + conn.async_exec( "COPY copytable FROM STDOUT CSV" ) + gate.stop data = "x" * 1000 * 1000 From b70c65a212422f2b1caa83bd4d87e9e134b2df1b Mon Sep 17 00:00:00 2001 From: Lars Kanis Date: Tue, 18 Aug 2026 15:20:54 +0200 Subject: [PATCH 2/4] Add a warning comment to the list of GVL-released functions --- ext/gvl_wrappers.h | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/ext/gvl_wrappers.h b/ext/gvl_wrappers.h index 81eb646ea..2f0922027 100644 --- a/ext/gvl_wrappers.h +++ b/ext/gvl_wrappers.h @@ -125,6 +125,12 @@ typedef struct pg_cancel_conn PGcancelConn; /* * Definitions of blocking functions and their parameters + * + * ATTENTION: + * Do not GVL-release functions that take pointers to ruby objects. + * If the ruby object is relocated by `GC.compact` from a second thread, before the pointer is used, it is no longer valid. + * That can lead to a crash or to wrong data. + * This issue has been raised in https://github.com/ged/ruby-pg/issues/738 and https://github.com/ged/ruby-pg/issues/721 . */ #define FOR_EACH_PARAM_OF_PQconnectdb(param) From f6e9fe84944b42bdcb70b53518d7d8932df81316 Mon Sep 17 00:00:00 2001 From: Lars Kanis Date: Tue, 18 Aug 2026 18:32:59 +0200 Subject: [PATCH 3/4] Shrink GVL-released libpq functions to necessary only All changed functions have a non-blocking implementation in default mode `PG::Connection.async_api=true`. So there's no need to release GVL for them. The intention is to make the list of GVL-releasing functions more consistent. The only remaining functions are connection esteblishing functions, now. They are known to block in some cases (GSSAPI auth, LDAP lookup), even if used in non-blocking/default mode. So these functions should still release the GVL. These remaining functions which take a connection string as ruby object shouldn't be an issue, since this string is created in `parse_connect_args` immediately before the call. They are stored as local variables only, which are not relocated and can not be changed by other threads. `PG::Connection.async_api=false` is significant less usable, since it blocks other ruby threads at any waiting. --- ext/gvl_wrappers.h | 40 --------------- ext/pg_cancel_connection.c | 4 +- ext/pg_connection.c | 38 +++++++------- ext/pg_result.c | 4 +- spec/pg/connection_async_spec.rb | 85 ++++++++++++++++++++++++++++++++ spec/pg/connection_spec.rb | 80 ------------------------------ 6 files changed, 108 insertions(+), 143 deletions(-) diff --git a/ext/gvl_wrappers.h b/ext/gvl_wrappers.h index 2f0922027..16d063c0e 100644 --- a/ext/gvl_wrappers.h +++ b/ext/gvl_wrappers.h @@ -133,66 +133,26 @@ typedef struct pg_cancel_conn PGcancelConn; * This issue has been raised in https://github.com/ged/ruby-pg/issues/738 and https://github.com/ged/ruby-pg/issues/721 . */ -#define FOR_EACH_PARAM_OF_PQconnectdb(param) - #define FOR_EACH_PARAM_OF_PQconnectStart(param) - #define FOR_EACH_PARAM_OF_PQconnectPoll(param) -#define FOR_EACH_PARAM_OF_PQreset(param) - #define FOR_EACH_PARAM_OF_PQresetStart(param) - #define FOR_EACH_PARAM_OF_PQresetPoll(param) #define FOR_EACH_PARAM_OF_PQping(param) -#define FOR_EACH_PARAM_OF_PQgetResult(param) - -#define FOR_EACH_PARAM_OF_PQputCopyEnd(param) \ - param(PGconn *, conn) - -#define FOR_EACH_PARAM_OF_PQgetCopyData(param) \ - param(PGconn *, conn) \ - param(char **, buffer) - -#define FOR_EACH_PARAM_OF_PQnotifies(param) - -#define FOR_EACH_PARAM_OF_PQpipelineSync(param) - -#define FOR_EACH_PARAM_OF_PQsetClientEncoding(param) \ - param(PGconn *, conn) - -#define FOR_EACH_PARAM_OF_PQisBusy(param) - -#define FOR_EACH_PARAM_OF_PQcancelBlocking(param) #define FOR_EACH_PARAM_OF_PQcancelStart(param) #define FOR_EACH_PARAM_OF_PQcancelPoll(param) -#define FOR_EACH_PARAM_OF_PQcancel(param) \ - param(PGcancel *, cancel) \ - param(char *, errbuf) - /* function( name, void_or_nonvoid, returntype, lastparamtype, lastparamname ) */ #define FOR_EACH_BLOCKING_FUNCTION(function) \ - function(PQconnectdb, GVL_TYPE_NONVOID, PGconn *, const char *, conninfo) \ function(PQconnectStart, GVL_TYPE_NONVOID, PGconn *, const char *, conninfo) \ function(PQconnectPoll, GVL_TYPE_NONVOID, PostgresPollingStatusType, PGconn *, conn) \ - function(PQreset, GVL_TYPE_VOID, void, PGconn *, conn) \ function(PQresetStart, GVL_TYPE_NONVOID, int, PGconn *, conn) \ function(PQresetPoll, GVL_TYPE_NONVOID, PostgresPollingStatusType, PGconn *, conn) \ function(PQping, GVL_TYPE_NONVOID, PGPing, const char *, conninfo) \ - function(PQgetResult, GVL_TYPE_NONVOID, PGresult *, PGconn *, conn) \ - function(PQputCopyEnd, GVL_TYPE_NONVOID, int, const char *, errormsg) \ - function(PQgetCopyData, GVL_TYPE_NONVOID, int, int, async) \ - function(PQnotifies, GVL_TYPE_NONVOID, PGnotify *, PGconn *, conn) \ - function(PQpipelineSync, GVL_TYPE_NONVOID, int, PGconn *, conn) \ - function(PQsetClientEncoding, GVL_TYPE_NONVOID, int, const char *, encoding) \ - function(PQisBusy, GVL_TYPE_NONVOID, int, PGconn *, conn) \ - function(PQcancelBlocking, GVL_TYPE_NONVOID, int, PGcancelConn *, conn) \ function(PQcancelStart, GVL_TYPE_NONVOID, int, PGcancelConn *, conn) \ function(PQcancelPoll, GVL_TYPE_NONVOID, PostgresPollingStatusType, PGcancelConn *, conn) \ - function(PQcancel, GVL_TYPE_NONVOID, int, int, errbufsize); FOR_EACH_BLOCKING_FUNCTION( DEFINE_GVL_STUB_DECL ); diff --git a/ext/pg_cancel_connection.c b/ext/pg_cancel_connection.c index 8adb1b19c..3a2bf7c39 100644 --- a/ext/pg_cancel_connection.c +++ b/ext/pg_cancel_connection.c @@ -155,7 +155,7 @@ pg_cancon_initialize(VALUE self, VALUE rb_conn) * * Requests that the server abandons processing of the current command in a blocking manner. * - * This method directly calls +PQcancelBlocking+ of libpq, so that it doesn't respond to ruby interrupts and doesn't trigger the +Thread.scheduler+ . + * This method directly calls +PQcancelBlocking+ of libpq, so that it doesn't allow parallel threads, doesn't respond to ruby interrupts and doesn't trigger the +Thread.scheduler+ . * It is threrfore recommended to call #cancel instead. * */ @@ -165,7 +165,7 @@ pg_cancon_sync_cancel(VALUE self) PGcancelConn *conn = pg_cancon_get_conn(self); pg_cancon_close_socket_io( self ); - if(gvl_PQcancelBlocking(conn) == 0) + if(PQcancelBlocking(conn) == 0) pg_raise_conn_error( rb_eConnectionBad, self, "PQcancelBlocking %s", PQcancelErrorMessage(conn)); return Qnil; } diff --git a/ext/pg_connection.c b/ext/pg_connection.c index de9fce048..3ed61ef20 100644 --- a/ext/pg_connection.c +++ b/ext/pg_connection.c @@ -282,7 +282,7 @@ pgconn_s_sync_connect(int argc, VALUE *argv, VALUE klass) this = pg_get_connection( self ); conninfo = rb_funcall2( rb_cPGconn, rb_intern("parse_connect_args"), argc, argv ); - this->pgconn = gvl_PQconnectdb(StringValueCStr(conninfo)); + this->pgconn = PQconnectdb(StringValueCStr(conninfo)); RB_GC_GUARD(conninfo); if(this->pgconn == NULL) @@ -565,7 +565,7 @@ static VALUE pgconn_sync_reset( VALUE self ) { pgconn_close_socket_io( self ); - gvl_PQreset( pg_get_pgconn(self) ); + PQreset( pg_get_pgconn(self) ); return self; } @@ -2309,7 +2309,7 @@ pgconn_sync_get_result(VALUE self) PGresult *result; VALUE rb_pgresult; - result = gvl_PQgetResult(conn); + result = PQgetResult(conn); if(result == NULL) return Qnil; rb_pgresult = pg_new_result(result, self); @@ -2351,7 +2351,7 @@ pgconn_consume_input(VALUE self) static VALUE pgconn_is_busy(VALUE self) { - return gvl_PQisBusy(pg_get_pgconn(self)) ? Qtrue : Qfalse; + return PQisBusy(pg_get_pgconn(self)) ? Qtrue : Qfalse; } static VALUE @@ -2404,7 +2404,7 @@ pgconn_sync_cancel(VALUE self) if(cancel == NULL) pg_raise_conn_error( rb_ePGerror, self, "Invalid connection!"); - ret = gvl_PQcancel(cancel, errbuf, sizeof(errbuf)); + ret = PQcancel(cancel, errbuf, sizeof(errbuf)); if(ret == 1) retval = Qnil; else @@ -2436,7 +2436,7 @@ pgconn_notifies(VALUE self) sym_be_pid = ID2SYM(rb_intern("be_pid")); sym_extra = ID2SYM(rb_intern("extra")); - notification = gvl_PQnotifies(this->pgconn); + notification = PQnotifies(this->pgconn); if (notification == NULL) { return Qnil; } @@ -2694,7 +2694,7 @@ pgconn_flush_data_set( VALUE self, VALUE enabled ){ static void * notify_readable(PGconn *conn) { - return (void*)gvl_PQnotifies(conn); + return (void*)PQnotifies(conn); } /* @@ -2816,7 +2816,7 @@ pgconn_sync_put_copy_end(int argc, VALUE *argv, VALUE self) else error_message = pg_cstr_enc(str, this->enc_idx, &transcoded_str); - ret = gvl_PQputCopyEnd(this->pgconn, error_message); + ret = PQputCopyEnd(this->pgconn, error_message); if(ret == -1) pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(this->pgconn)); @@ -2846,7 +2846,7 @@ pgconn_sync_get_copy_data(int argc, VALUE *argv, VALUE self ) TypedData_Get_Struct(decoder, t_pg_coder, &pg_coder_type, p_coder); } - ret = gvl_PQgetCopyData(this->pgconn, &buffer, RTEST(async_in)); + ret = PQgetCopyData(this->pgconn, &buffer, RTEST(async_in)); if(ret == -2){ /* error */ pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(this->pgconn)); } @@ -3152,7 +3152,7 @@ pgconn_sync_set_client_encoding(VALUE self, VALUE str) rb_check_frozen(self); Check_Type(str, T_STRING); - if ( (gvl_PQsetClientEncoding(conn, StringValueCStr(str))) == -1 ) + if ( (PQsetClientEncoding(conn, StringValueCStr(str))) == -1 ) pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn)); pgconn_set_internal_encoding_index( self ); @@ -3217,7 +3217,7 @@ pgconn_s_quote_ident(VALUE self, VALUE str_or_array) static void * get_result_readable(PGconn *conn) { - return gvl_PQisBusy(conn) ? NULL : (void*)1; + return PQisBusy(conn) ? NULL : (void*)1; } @@ -3275,7 +3275,7 @@ pgconn_sync_get_last_result(VALUE self) cur = prev = NULL; - while ((cur = gvl_PQgetResult(conn)) != NULL) { + while ((cur = PQgetResult(conn)) != NULL) { int status; if (prev) PQclear(prev); @@ -3327,7 +3327,7 @@ pgconn_async_get_last_result(VALUE self) */ wait_socket_readable(self, NULL, get_result_readable); - cur = gvl_PQgetResult(conn); + cur = PQgetResult(conn); if (cur == NULL) break; @@ -3384,7 +3384,7 @@ pgconn_discard_results(VALUE self) /* pgconn_block() raises an exception in case of errors. * To avoid this call pg_rb_io_wait() and PQconsumeInput() without rb_raise(). */ - while( gvl_PQisBusy(conn) ){ + while( PQisBusy(conn) ){ int events; switch( PQflush(conn) ) { @@ -3403,20 +3403,20 @@ pgconn_discard_results(VALUE self) } } - cur = gvl_PQgetResult(conn); + cur = PQgetResult(conn); if( cur == NULL) break; status = PQresultStatus(cur); PQclear(cur); if (status == PGRES_COPY_IN){ - while( gvl_PQputCopyEnd(conn, "COPY terminated by new query or discard_results") == 0 ){ + while( PQputCopyEnd(conn, "COPY terminated by new query or discard_results") == 0 ){ pgconn_async_flush(self); } } if (status == PGRES_COPY_OUT){ for(;;) { char *buffer = NULL; - int st = gvl_PQgetCopyData(conn, &buffer, 1); + int st = PQgetCopyData(conn, &buffer, 1); if( st == 0 ) { /* would block -> wait for readable data */ pg_rb_io_wait(socket_io, RB_INT2NUM(PG_RUBY_IO_READABLE), Qnil); @@ -3898,7 +3898,7 @@ static VALUE pgconn_sync_pipeline_sync(VALUE self) { PGconn *conn = pg_get_pgconn(self); - int res = gvl_PQpipelineSync(conn); + int res = PQpipelineSync(conn); if( res != 1 ) pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn)); @@ -4364,7 +4364,7 @@ pgconn_internal_encoding_set(VALUE self, VALUE enc) rb_encoding *rbenc = rb_to_encoding( enc ); const char *name = pg_get_rb_encoding_as_pg_encoding( rbenc ); - if ( gvl_PQsetClientEncoding(pg_get_pgconn( self ), name) == -1 ) { + if ( PQsetClientEncoding(pg_get_pgconn( self ), name) == -1 ) { VALUE server_encoding = pgconn_external_encoding( self ); rb_raise( rb_eEncCompatError, "incompatible character encodings: %s and %s", rb_enc_name(rb_to_encoding(server_encoding)), name ); diff --git a/ext/pg_result.c b/ext/pg_result.c index 28420d176..73725a6c1 100644 --- a/ext/pg_result.c +++ b/ext/pg_result.c @@ -1618,12 +1618,12 @@ pgresult_stream_any(VALUE self, int (*yielder)(VALUE, int, int, void*), void* da pgresult_clear( this ); } - if( gvl_PQisBusy(pgconn) ){ + if( PQisBusy(pgconn) ){ /* wait for input (without blocking) before reading each result */ pgconn_block( 0, NULL, this->connection ); } - pgresult = gvl_PQgetResult(pgconn); + pgresult = PQgetResult(pgconn); if( pgresult == NULL ) rb_raise( rb_eNoResultError, "no result received - possibly an intersection with another query"); diff --git a/spec/pg/connection_async_spec.rb b/spec/pg/connection_async_spec.rb index da7174e67..9ffdc821e 100644 --- a/spec/pg/connection_async_spec.rb +++ b/spec/pg/connection_async_spec.rb @@ -135,6 +135,31 @@ def interrupt_thread(exc=nil) end end + it "connects without port and then retrieves the default port" do + gate = Helpers::TcpGateSwitcher.new( + external_host: 'localhost', + external_port: ENV['PGPORT'].to_i, + internal_host: "127.0.0.1", + internal_port: PG::DEF_PGPORT, + debug: ENV['PG_DEBUG']=='1') + + PG.connect(host: "localhost", + port: "", + dbname: "test") do |conn| + expect( conn.port ).to eq( PG::DEF_PGPORT ) + end + + PG.connect(hostaddr: "127.0.0.1", + port: nil, + dbname: "test") do |conn| + expect( conn.port ).to eq( PG::DEF_PGPORT ) + end + + gate.finish + rescue Errno::EADDRINUSE, Errno::EACCES => err + skip err.to_s + end + it "doesn't duplicate hosts in conn.reset", :without_transaction, :ipv6, :postgresql_12 do set_etc_hosts "::1", "rubypg_test2 rubypg_test_ipv6" set_etc_hosts "127.0.0.1", "rubypg_test2 rubypg_test_ipv4" @@ -156,4 +181,64 @@ def interrupt_thread(exc=nil) expect( conn.hostaddr ).to eq( "::1" ) expect( conn.port ).to eq( @port ) end + + context "in nonblocking mode" do + after :each do + @conn.setnonblocking(false) + end + + it "rejects to send lots of COPY data" do + unless RUBY_PLATFORM =~ /i386-mingw|x86_64-darwin|x86_64-linux$/ + skip "this spec depends on out-of-memory condition in put_copy_data, which is not reliable on all platforms" + end + + run_with_gate(200) do |conn, gate| + conn.setnonblocking(true) + + res = nil + conn.exec <<-EOSQL + CREATE TEMP TABLE copytable (col1 TEXT); + EOSQL + + conn.exec( "COPY copytable FROM STDOUT CSV" ) + + gate.stop + + data = "x" * 1000 * 1000 + data << "\n" + 20000.times do |idx| + res = conn.put_copy_data(data) + break if res == false + end + expect( res ).to be_falsey + + gate.start + conn.cancel + conn.discard_results + end + end + + it "needs to flush data after send_query" do + run_with_gate(200) do |conn, gate| + conn.setnonblocking(true) + + gate.stop + data = "x" * 1000 * 1000 * 30 + res = conn.send_query_params("SELECT LENGTH($1)", [data]) + expect( res ).to be_nil + + res = conn.flush + expect( res ).to be_falsey + + gate.start + until conn.flush + IO.select(nil, [conn.socket_io], [conn.socket_io], 10) + end + expect( conn.flush ).to be_truthy + + res = conn.get_last_result + expect( res.values ).to eq( [[data.length.to_s]] ) + end + end + end end diff --git a/spec/pg/connection_spec.rb b/spec/pg/connection_spec.rb index 6ba963e55..aad9066f1 100644 --- a/spec/pg/connection_spec.rb +++ b/spec/pg/connection_spec.rb @@ -796,61 +796,6 @@ expect( res.values ).to eq([[data.length.to_s]]) end - it "rejects to send lots of COPY data" do - unless RUBY_PLATFORM =~ /i386-mingw|x86_64-darwin|x86_64-linux$/ - skip "this spec depends on out-of-memory condition in put_copy_data, which is not reliable on all platforms" - end - - run_with_gate(200) do |conn, gate| - conn.setnonblocking(true) - - res = nil - # use async_exec since sync_exec is no longer thread compatible, necessary for run_with_gate - conn.async_exec <<-EOSQL - CREATE TEMP TABLE copytable (col1 TEXT); - EOSQL - - conn.async_exec( "COPY copytable FROM STDOUT CSV" ) - - gate.stop - - data = "x" * 1000 * 1000 - data << "\n" - 20000.times do |idx| - res = conn.put_copy_data(data) - break if res == false - end - expect( res ).to be_falsey - - gate.start - conn.cancel - conn.discard_results - end - end - - it "needs to flush data after send_query" do - run_with_gate(200) do |conn, gate| - conn.setnonblocking(true) - - gate.stop - data = "x" * 1000 * 1000 * 30 - res = conn.send_query_params("SELECT LENGTH($1)", [data]) - expect( res ).to be_nil - - res = conn.flush - expect( res ).to be_falsey - - gate.start - until conn.flush - IO.select(nil, [conn.socket_io], [conn.socket_io], 10) - end - expect( conn.flush ).to be_truthy - - res = conn.get_last_result - expect( res.values ).to eq( [[data.length.to_s]] ) - end - end - it "returns immediately from get_copy_data(nonblock=true)" do expect do @conn.copy_data( "COPY (SELECT generate_series(0,999), NULL UNION ALL SELECT 1000, pg_sleep(10)) TO STDOUT" ) do |res| @@ -899,31 +844,6 @@ expect( @conn.options ).to eq( "" ) end - it "connects without port and then retrieves the default port" do - gate = Helpers::TcpGateSwitcher.new( - external_host: 'localhost', - external_port: ENV['PGPORT'].to_i, - internal_host: "127.0.0.1", - internal_port: PG::DEF_PGPORT, - debug: ENV['PG_DEBUG']=='1') - - PG.connect(host: "localhost", - port: "", - dbname: "test") do |conn| - expect( conn.port ).to eq( PG::DEF_PGPORT ) - end - - PG.connect(hostaddr: "127.0.0.1", - port: nil, - dbname: "test") do |conn| - expect( conn.port ).to eq( PG::DEF_PGPORT ) - end - - gate.finish - rescue Errno::EADDRINUSE, Errno::EACCES => err - skip err.to_s - end - it "can retrieve hostaddr for the established connection", :postgresql_12 do expect( @conn.hostaddr ).to match( /^127\.0\.0\.1$|^::1$/ ) end From c6f87d3aef89468087bb9f82699e9d5110909350 Mon Sep 17 00:00:00 2001 From: Lars Kanis Date: Sat, 22 Aug 2026 22:08:18 +0200 Subject: [PATCH 4/4] Move the specs from the previous commit back back ... so that it's executed with sync and async methods. This is possible by moving execution of TcpGateSwitcher from concurrent threads to a separate process. Call bind through DRb, so that Exceptions are passed to caller. --- Gemfile | 1 + spec/helpers.rb | 9 +-- spec/helpers/tcp_gate_switcher.rb | 2 + spec/helpers/tcp_gate_switcher_process.rb | 49 +++++++++++++ spec/pg/connection_async_spec.rb | 85 ----------------------- spec/pg/connection_spec.rb | 79 +++++++++++++++++++++ 6 files changed, 136 insertions(+), 89 deletions(-) create mode 100644 spec/helpers/tcp_gate_switcher_process.rb diff --git a/Gemfile b/Gemfile index c22c0988b..a0f069130 100644 --- a/Gemfile +++ b/Gemfile @@ -20,4 +20,5 @@ group :test do # With bigdecimal commented out here, corresponding tests are omitted on ruby-3.4+ but are executed on ruby < 3.4. # That way we can check both situations in CI. # gem "bigdecimal", "~> 3.0" + gem "drb" end diff --git a/spec/helpers.rb b/spec/helpers.rb index 015eccd07..cc6e0f017 100644 --- a/spec/helpers.rb +++ b/spec/helpers.rb @@ -7,9 +7,10 @@ require 'openssl' require 'fileutils' require 'objspace' -require_relative 'helpers/scheduler.rb' -require_relative 'helpers/tcp_gate_scheduler.rb' -require_relative 'helpers/tcp_gate_switcher.rb' +require_relative 'helpers/scheduler' +require_relative 'helpers/tcp_gate_scheduler' +require_relative 'helpers/tcp_gate_switcher' +require_relative 'helpers/tcp_gate_switcher_process' TEST_DIRECTORY = Pathname.new(ENV['RUBY_PG_TEST_DIR'] || Dir.pwd) DATA_OBJ_MEMSIZE = ObjectSpace.memsize_of(Object.new) @@ -616,7 +617,7 @@ def run_with_scheduler(timeout=10) def gate_setup # Run examples with gate - gate = Helpers::TcpGateSwitcher.new(external_host: 'localhost', external_port: ENV['PGPORT'].to_i, debug: ENV['PG_DEBUG']=='1') + gate = Helpers::TcpGateSwitcherProcess.new(external_host: 'localhost', external_port: ENV['PGPORT'].to_i, debug: ENV['PG_DEBUG']=='1') @conninfo_gate = @conninfo.gsub(/(^| )port=\d+/, " port=#{gate.internal_port}") # Run examples without gate diff --git a/spec/helpers/tcp_gate_switcher.rb b/spec/helpers/tcp_gate_switcher.rb index 2fa852e53..afa0a3c3c 100644 --- a/spec/helpers/tcp_gate_switcher.rb +++ b/spec/helpers/tcp_gate_switcher.rb @@ -1,5 +1,7 @@ # frozen_string_literal: true +require "socket" + # This is a transparent TCP proxy for testing blocking behaviour in a time insensitive way. # # It works as a gate between the client and the server, which is enabled or disabled by the spec. diff --git a/spec/helpers/tcp_gate_switcher_process.rb b/spec/helpers/tcp_gate_switcher_process.rb new file mode 100644 index 000000000..88cd6e197 --- /dev/null +++ b/spec/helpers/tcp_gate_switcher_process.rb @@ -0,0 +1,49 @@ +# frozen_string_literal: true + +require 'drb/drb' + +# This is a wrapper of TcpGateSwitcher running in a separate process to avoid the need of threads. +# It can therefore be used in conjunction with blocking GVL locking functions. + +module Helpers +class TcpGateSwitcherProcess + def initialize(**kwargs) + file = File.expand_path("tcp_gate_switcher", __dir__) + rbtext = <<~RBTEXT + require #{file.inspect} + require "drb/drb" + + switcher = Helpers::TcpGateSwitcher.allocate + def switcher.finish + super + DRb.stop_service + end + def switcher.init + initialize(**#{kwargs.inspect}) + self + end + DRb.start_service('druby://localhost:0', switcher) + puts DRb.uri + # Redirect STDOUT to STDERR, so that p prints to STDERR + STDOUT.reopen(STDERR) + + # Wait for the drb server thread to finish before exiting. + DRb.thread.join + RBTEXT + + io = IO.popen("ruby", "w+") + io.write rbtext + io.close_write + server_uri = io.gets.strip + @server = DRbObject.new_with_uri(server_uri) + # Call initialize through DRb, so that Exceptions are passed to caller + @server.init + end + + %i[finish internal_port start stop].each do |meth| + define_method(meth) do + @server.send(meth) + end + end +end +end diff --git a/spec/pg/connection_async_spec.rb b/spec/pg/connection_async_spec.rb index 9ffdc821e..da7174e67 100644 --- a/spec/pg/connection_async_spec.rb +++ b/spec/pg/connection_async_spec.rb @@ -135,31 +135,6 @@ def interrupt_thread(exc=nil) end end - it "connects without port and then retrieves the default port" do - gate = Helpers::TcpGateSwitcher.new( - external_host: 'localhost', - external_port: ENV['PGPORT'].to_i, - internal_host: "127.0.0.1", - internal_port: PG::DEF_PGPORT, - debug: ENV['PG_DEBUG']=='1') - - PG.connect(host: "localhost", - port: "", - dbname: "test") do |conn| - expect( conn.port ).to eq( PG::DEF_PGPORT ) - end - - PG.connect(hostaddr: "127.0.0.1", - port: nil, - dbname: "test") do |conn| - expect( conn.port ).to eq( PG::DEF_PGPORT ) - end - - gate.finish - rescue Errno::EADDRINUSE, Errno::EACCES => err - skip err.to_s - end - it "doesn't duplicate hosts in conn.reset", :without_transaction, :ipv6, :postgresql_12 do set_etc_hosts "::1", "rubypg_test2 rubypg_test_ipv6" set_etc_hosts "127.0.0.1", "rubypg_test2 rubypg_test_ipv4" @@ -181,64 +156,4 @@ def interrupt_thread(exc=nil) expect( conn.hostaddr ).to eq( "::1" ) expect( conn.port ).to eq( @port ) end - - context "in nonblocking mode" do - after :each do - @conn.setnonblocking(false) - end - - it "rejects to send lots of COPY data" do - unless RUBY_PLATFORM =~ /i386-mingw|x86_64-darwin|x86_64-linux$/ - skip "this spec depends on out-of-memory condition in put_copy_data, which is not reliable on all platforms" - end - - run_with_gate(200) do |conn, gate| - conn.setnonblocking(true) - - res = nil - conn.exec <<-EOSQL - CREATE TEMP TABLE copytable (col1 TEXT); - EOSQL - - conn.exec( "COPY copytable FROM STDOUT CSV" ) - - gate.stop - - data = "x" * 1000 * 1000 - data << "\n" - 20000.times do |idx| - res = conn.put_copy_data(data) - break if res == false - end - expect( res ).to be_falsey - - gate.start - conn.cancel - conn.discard_results - end - end - - it "needs to flush data after send_query" do - run_with_gate(200) do |conn, gate| - conn.setnonblocking(true) - - gate.stop - data = "x" * 1000 * 1000 * 30 - res = conn.send_query_params("SELECT LENGTH($1)", [data]) - expect( res ).to be_nil - - res = conn.flush - expect( res ).to be_falsey - - gate.start - until conn.flush - IO.select(nil, [conn.socket_io], [conn.socket_io], 10) - end - expect( conn.flush ).to be_truthy - - res = conn.get_last_result - expect( res.values ).to eq( [[data.length.to_s]] ) - end - end - end end diff --git a/spec/pg/connection_spec.rb b/spec/pg/connection_spec.rb index aad9066f1..24027903c 100644 --- a/spec/pg/connection_spec.rb +++ b/spec/pg/connection_spec.rb @@ -796,6 +796,60 @@ expect( res.values ).to eq([[data.length.to_s]]) end + it "rejects to send lots of COPY data" do + unless RUBY_PLATFORM =~ /i386-mingw|x86_64-darwin|x86_64-linux$/ + skip "this spec depends on out-of-memory condition in put_copy_data, which is not reliable on all platforms" + end + + run_with_gate(200) do |conn, gate| + conn.setnonblocking(true) + + res = nil + conn.exec <<-EOSQL + CREATE TEMP TABLE copytable (col1 TEXT); + EOSQL + + conn.exec( "COPY copytable FROM STDOUT CSV" ) + + gate.stop + + data = "x" * 1000 * 1000 + data << "\n" + 20000.times do |idx| + res = conn.put_copy_data(data) + break if res == false + end + expect( res ).to be_falsey + + gate.start + conn.cancel + conn.discard_results + end + end + + it "needs to flush data after send_query" do + run_with_gate(200) do |conn, gate| + conn.setnonblocking(true) + + gate.stop + data = "x" * 1000 * 1000 * 30 + res = conn.send_query_params("SELECT LENGTH($1)", [data]) + expect( res ).to be_nil + + res = conn.flush + expect( res ).to be_falsey + + gate.start + until conn.flush + IO.select(nil, [conn.socket_io], [conn.socket_io], 10) + end + expect( conn.flush ).to be_truthy + + res = conn.get_last_result + expect( res.values ).to eq( [[data.length.to_s]] ) + end + end + it "returns immediately from get_copy_data(nonblock=true)" do expect do @conn.copy_data( "COPY (SELECT generate_series(0,999), NULL UNION ALL SELECT 1000, pg_sleep(10)) TO STDOUT" ) do |res| @@ -844,6 +898,31 @@ expect( @conn.options ).to eq( "" ) end + it "connects without port and then retrieves the default port" do + gate = Helpers::TcpGateSwitcherProcess.new( + external_host: 'localhost', + external_port: ENV['PGPORT'].to_i, + internal_host: "127.0.0.1", + internal_port: PG::DEF_PGPORT, + debug: ENV['PG_DEBUG']=='1') + + PG.connect(host: "localhost", + port: "", + dbname: "test") do |conn| + expect( conn.port ).to eq( PG::DEF_PGPORT ) + end + + PG.connect(hostaddr: "127.0.0.1", + port: nil, + dbname: "test") do |conn| + expect( conn.port ).to eq( PG::DEF_PGPORT ) + end + + gate.finish + rescue Errno::EADDRINUSE, Errno::EACCES => err + skip err.to_s + end + it "can retrieve hostaddr for the established connection", :postgresql_12 do expect( @conn.hostaddr ).to match( /^127\.0\.0\.1$|^::1$/ ) end