Skip to content

Commit ff42e3b

Browse files
larskanislarskanis
authored
Merge pull request #498 from larskanis/fix-lo
Disable nonblocking mode while large object calls
2 parents af01d9d + dba5322 commit ff42e3b

File tree

2 files changed

+108
-20
lines changed

2 files changed

+108
-20
lines changed

‎ext/pg_connection.c

Lines changed: 67 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -3643,6 +3643,14 @@ pgconn_send_flush_request(VALUE self)
36433643
* LARGE OBJECT SUPPORT
36443644
**************************************************************************/
36453645

3646+
#define BLOCKING_BEGIN(conn) do { \
3647+
int old_nonblocking = PQisnonblocking(conn); \
3648+
PQsetnonblocking(conn, 0);
3649+
3650+
#define BLOCKING_END(th) \
3651+
PQsetnonblocking(conn, old_nonblocking); \
3652+
} while(0);
3653+
36463654
/*
36473655
* call-seq:
36483656
* conn.lo_creat( [mode] ) -> Integer
@@ -3663,7 +3671,10 @@ pgconn_locreat(int argc, VALUE *argv, VALUE self)
36633671
else
36643672
mode = NUM2INT(nmode);
36653673

3666-
lo_oid = lo_creat(conn, mode);
3674+
BLOCKING_BEGIN(conn)
3675+
lo_oid = lo_creat(conn, mode);
3676+
BLOCKING_END(conn)
3677+
36673678
if (lo_oid == 0)
36683679
pg_raise_conn_error( rb_ePGerror, self, "lo_creat failed");
36693680

@@ -3708,7 +3719,10 @@ pgconn_loimport(VALUE self, VALUE filename)
37083719

37093720
Check_Type(filename, T_STRING);
37103721

3711-
lo_oid = lo_import(conn, StringValueCStr(filename));
3722+
BLOCKING_BEGIN(conn)
3723+
lo_oid = lo_import(conn, StringValueCStr(filename));
3724+
BLOCKING_END(conn)
3725+
37123726
if (lo_oid == 0) {
37133727
pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn));
37143728
}
@@ -3726,11 +3740,16 @@ pgconn_loexport(VALUE self, VALUE lo_oid, VALUE filename)
37263740
{
37273741
PGconn *conn = pg_get_pgconn(self);
37283742
Oid oid;
3743+
int ret;
37293744
Check_Type(filename, T_STRING);
37303745

37313746
oid = NUM2UINT(lo_oid);
37323747

3733-
if (lo_export(conn, oid, StringValueCStr(filename)) < 0) {
3748+
BLOCKING_BEGIN(conn)
3749+
ret = lo_export(conn, oid, StringValueCStr(filename));
3750+
BLOCKING_END(conn)
3751+
3752+
if (ret < 0) {
37343753
pg_raise_conn_error( rb_ePGerror, self, "%s", PQerrorMessage(conn));
37353754
}
37363755
return Qnil;
@@ -3761,7 +3780,11 @@ pgconn_loopen(int argc, VALUE *argv, VALUE self)
37613780
else
37623781
mode = NUM2INT(nmode);
37633782

3764-
if((fd = lo_open(conn, lo_oid, mode)) < 0) {
3783+
BLOCKING_BEGIN(conn)
3784+
fd = lo_open(conn, lo_oid, mode);
3785+
BLOCKING_END(conn)
3786+
3787+
if(fd < 0) {
37653788
pg_raise_conn_error( rb_ePGerror, self, "can't open large object: %s", PQerrorMessage(conn));
37663789
}
37673790
return INT2FIX(fd);
@@ -3786,8 +3809,12 @@ pgconn_lowrite(VALUE self, VALUE in_lo_desc, VALUE buffer)
37863809
if( RSTRING_LEN(buffer) < 0) {
37873810
pg_raise_conn_error( rb_ePGerror, self, "write buffer zero string");
37883811
}
3789-
if((n = lo_write(conn, fd, StringValuePtr(buffer),
3790-
RSTRING_LEN(buffer))) < 0) {
3812+
BLOCKING_BEGIN(conn)
3813+
n = lo_write(conn, fd, StringValuePtr(buffer),
3814+
RSTRING_LEN(buffer));
3815+
BLOCKING_END(conn)
3816+
3817+
if(n < 0) {
37913818
pg_raise_conn_error( rb_ePGerror, self, "lo_write failed: %s", PQerrorMessage(conn));
37923819
}
37933820

@@ -3815,7 +3842,12 @@ pgconn_loread(VALUE self, VALUE in_lo_desc, VALUE in_len)
38153842
pg_raise_conn_error( rb_ePGerror, self, "negative length %d given", len);
38163843

38173844
buffer = ALLOC_N(char, len);
3818-
if((ret = lo_read(conn, lo_desc, buffer, len)) < 0)
3845+
3846+
BLOCKING_BEGIN(conn)
3847+
ret = lo_read(conn, lo_desc, buffer, len);
3848+
BLOCKING_END(conn)
3849+
3850+
if(ret < 0)
38193851
pg_raise_conn_error( rb_ePGerror, self, "lo_read failed");
38203852

38213853
if(ret == 0) {
@@ -3845,7 +3877,11 @@ pgconn_lolseek(VALUE self, VALUE in_lo_desc, VALUE offset, VALUE whence)
38453877
int lo_desc = NUM2INT(in_lo_desc);
38463878
int ret;
38473879

3848-
if((ret = lo_lseek(conn, lo_desc, NUM2INT(offset), NUM2INT(whence))) < 0) {
3880+
BLOCKING_BEGIN(conn)
3881+
ret = lo_lseek(conn, lo_desc, NUM2INT(offset), NUM2INT(whence));
3882+
BLOCKING_END(conn)
3883+
3884+
if(ret < 0) {
38493885
pg_raise_conn_error( rb_ePGerror, self, "lo_lseek failed");
38503886
}
38513887

@@ -3865,7 +3901,11 @@ pgconn_lotell(VALUE self, VALUE in_lo_desc)
38653901
PGconn *conn = pg_get_pgconn(self);
38663902
int lo_desc = NUM2INT(in_lo_desc);
38673903

3868-
if((position = lo_tell(conn, lo_desc)) < 0)
3904+
BLOCKING_BEGIN(conn)
3905+
position = lo_tell(conn, lo_desc);
3906+
BLOCKING_END(conn)
3907+
3908+
if(position < 0)
38693909
pg_raise_conn_error( rb_ePGerror, self, "lo_tell failed");
38703910

38713911
return INT2FIX(position);
@@ -3883,8 +3923,13 @@ pgconn_lotruncate(VALUE self, VALUE in_lo_desc, VALUE in_len)
38833923
PGconn *conn = pg_get_pgconn(self);
38843924
int lo_desc = NUM2INT(in_lo_desc);
38853925
size_t len = NUM2INT(in_len);
3926+
int ret;
3927+
3928+
BLOCKING_BEGIN(conn)
3929+
ret = lo_truncate(conn,lo_desc,len);
3930+
BLOCKING_END(conn)
38863931

3887-
if(lo_truncate(conn,lo_desc,len) < 0)
3932+
if(ret < 0)
38883933
pg_raise_conn_error( rb_ePGerror, self, "lo_truncate failed");
38893934

38903935
return Qnil;
@@ -3901,8 +3946,13 @@ pgconn_loclose(VALUE self, VALUE in_lo_desc)
39013946
{
39023947
PGconn *conn = pg_get_pgconn(self);
39033948
int lo_desc = NUM2INT(in_lo_desc);
3949+
int ret;
3950+
3951+
BLOCKING_BEGIN(conn)
3952+
ret = lo_close(conn,lo_desc);
3953+
BLOCKING_END(conn)
39043954

3905-
if(lo_close(conn,lo_desc) < 0)
3955+
if(ret < 0)
39063956
pg_raise_conn_error( rb_ePGerror, self, "lo_close failed");
39073957

39083958
return Qnil;
@@ -3919,8 +3969,13 @@ pgconn_lounlink(VALUE self, VALUE in_oid)
39193969
{
39203970
PGconn *conn = pg_get_pgconn(self);
39213971
Oid oid = NUM2UINT(in_oid);
3972+
int ret;
3973+
3974+
BLOCKING_BEGIN(conn)
3975+
ret = lo_unlink(conn,oid);
3976+
BLOCKING_END(conn)
39223977

3923-
if(lo_unlink(conn,oid) < 0)
3978+
if(ret < 0)
39243979
pg_raise_conn_error( rb_ePGerror, self, "lo_unlink failed");
39253980

39263981
return Qnil;

‎spec/pg/connection_spec.rb

Lines changed: 41 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -915,14 +915,47 @@
915915
end
916916
end
917917

918-
it "not read past the end of a large object" do
919-
@conn.transaction do
920-
oid = @conn.lo_create( 0 )
921-
fd = @conn.lo_open( oid, PG::INV_READ|PG::INV_WRITE )
922-
@conn.lo_write( fd, "foobar" )
923-
expect( @conn.lo_read( fd, 10 ) ).to be_nil()
924-
@conn.lo_lseek( fd, 0, PG::SEEK_SET )
925-
expect( @conn.lo_read( fd, 10 ) ).to eq( 'foobar' )
918+
describe "large objects" do
919+
920+
it "not read past the end of a large object" do
921+
@conn.transaction do
922+
oid = @conn.lo_create( 0 )
923+
fd = @conn.lo_open( oid, PG::INV_READ|PG::INV_WRITE )
924+
expect( @conn.lo_write( fd, "foobar" ) ).to eq( 6 )
925+
expect( @conn.lo_read( fd, 10 ) ).to be_nil()
926+
expect( @conn.lo_lseek( fd, 0, PG::SEEK_SET ) ).to eq( 0 )
927+
expect( @conn.lo_read( fd, 10 ) ).to eq( 'foobar' )
928+
expect( @conn.lo_close( fd ) ).to be_nil
929+
expect( @conn.lo_unlink( oid ) ).to be_nil
930+
end
931+
end
932+
933+
it "large object can handle big data", :unix_socket do
934+
# Using lo_write with > 300000 bytes on a UnixSocket connection in nonblocking mode results in the following error:
935+
# PG::UnableToSend: unexpected response from server; first received character was "V"
936+
# This is because the lo_write call doesn't wait for the response of the server function, but sends the next command early, so that results overlap.
937+
# Switching to blocking mode as part of lo_* calls fixes this issue and is tested here.
938+
939+
uri = "postgres://#{@unix_socket.gsub("/", "%2F")}:#{@port}/test"
940+
conn = described_class.connect( uri )
941+
942+
bytes = Random.urandom(512000)
943+
oid = conn.lo_creat
944+
conn.transaction do
945+
fd = conn.lo_open( oid, PG::INV_WRITE )
946+
conn.lo_write( fd, bytes )
947+
expect( conn.lo_close( fd ) ).to be_nil
948+
end
949+
950+
conn.transaction do
951+
fd = conn.lo_open( oid, PG::INV_READ )
952+
bytes2 = conn.lo_read( fd, bytes.bytesize )
953+
expect( bytes2 ).to eq( bytes )
954+
expect( conn.lo_close( fd ) ).to be_nil
955+
end
956+
expect( conn.lo_unlink( oid ) ).to be_nil
957+
ensure
958+
conn&.finish
926959
end
927960
end
928961

0 commit comments

Comments
 (0)