Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit387da18

Browse files
committed
Use a nonblocking socket for FE/BE communication and block using latches.
This allows to introduce more elaborate handling of interrupts whilereading from a socket. Currently some interrupt handlers have to dosignificant work from inside signal handlers, and it's very hard tocorrectly write code to do so. Generic signal handler limitations,combined with the fact that we can't safely jump out of a signalhandler while reading from the client have prohibited implementationof features like timeouts for idle-in-transaction.Additionally we use the latch code to wait in a couple places where wepreviously only had waiting code on windows as other platforms justbusy looped.This can increase the number of systemcalls happening during FE/BEcommunication. Benchmarks so far indicate that the impact isn't veryhigh, and there's room for optimization in the latch code. The chanceof cleaning up the usage of latches gives us, seem to outweigh therisk of small performance regressions.This commit theoretically can't used without the next patch in theseries, as WaitLatchOrSocket is not defined to be fully signalsafe. As we already do that in some cases though, it seems better tokeep the commits separate, so they're easier to understand.Author: Andres FreundReviewed-By: Heikki Linnakangas
1 parent778d498 commit387da18

File tree

3 files changed

+124
-46
lines changed

3 files changed

+124
-46
lines changed

‎src/backend/libpq/be-secure-openssl.c

Lines changed: 32 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,8 @@
7171
#endif
7272

7373
#include"libpq/libpq.h"
74+
#include"miscadmin.h"
75+
#include"storage/latch.h"
7476
#include"tcop/tcopprot.h"
7577
#include"utils/memutils.h"
7678

@@ -338,6 +340,7 @@ be_tls_open_server(Port *port)
338340
{
339341
intr;
340342
interr;
343+
intwaitfor;
341344

342345
Assert(!port->ssl);
343346
Assert(!port->peer);
@@ -371,12 +374,15 @@ be_tls_open_server(Port *port)
371374
{
372375
caseSSL_ERROR_WANT_READ:
373376
caseSSL_ERROR_WANT_WRITE:
374-
#ifdefWIN32
375-
pgwin32_waitforsinglesocket(SSL_get_fd(port->ssl),
376-
(err==SSL_ERROR_WANT_READ) ?
377-
FD_READ |FD_CLOSE |FD_ACCEPT :FD_WRITE |FD_CLOSE,
378-
INFINITE);
379-
#endif
377+
/* not allowed during connection establishment */
378+
Assert(!port->noblock);
379+
380+
if (err==SSL_ERROR_WANT_READ)
381+
waitfor=WL_SOCKET_READABLE;
382+
else
383+
waitfor=WL_SOCKET_WRITEABLE;
384+
385+
WaitLatchOrSocket(MyLatch,waitfor,port->sock,0);
380386
gotoaloop;
381387
caseSSL_ERROR_SYSCALL:
382388
if (r<0)
@@ -504,6 +510,7 @@ be_tls_read(Port *port, void *ptr, size_t len)
504510
{
505511
ssize_tn;
506512
interr;
513+
intwaitfor;
507514

508515
rloop:
509516
errno=0;
@@ -516,18 +523,20 @@ be_tls_read(Port *port, void *ptr, size_t len)
516523
break;
517524
caseSSL_ERROR_WANT_READ:
518525
caseSSL_ERROR_WANT_WRITE:
526+
/* Don't retry if the socket is in nonblocking mode. */
519527
if (port->noblock)
520528
{
521529
errno=EWOULDBLOCK;
522530
n=-1;
523531
break;
524532
}
525-
#ifdefWIN32
526-
pgwin32_waitforsinglesocket(SSL_get_fd(port->ssl),
527-
(err==SSL_ERROR_WANT_READ) ?
528-
FD_READ |FD_CLOSE :FD_WRITE |FD_CLOSE,
529-
INFINITE);
530-
#endif
533+
534+
if (err==SSL_ERROR_WANT_READ)
535+
waitfor=WL_SOCKET_READABLE;
536+
else
537+
waitfor=WL_SOCKET_WRITEABLE;
538+
539+
WaitLatchOrSocket(MyLatch,waitfor,port->sock,0);
531540
gotorloop;
532541
caseSSL_ERROR_SYSCALL:
533542
/* leave it to caller to ereport the value of errno */
@@ -567,6 +576,7 @@ be_tls_write(Port *port, void *ptr, size_t len)
567576
{
568577
ssize_tn;
569578
interr;
579+
intwaitfor;
570580

571581
/*
572582
* If SSL renegotiations are enabled and we're getting close to the
@@ -630,12 +640,13 @@ be_tls_write(Port *port, void *ptr, size_t len)
630640
break;
631641
caseSSL_ERROR_WANT_READ:
632642
caseSSL_ERROR_WANT_WRITE:
633-
#ifdefWIN32
634-
pgwin32_waitforsinglesocket(SSL_get_fd(port->ssl),
635-
(err==SSL_ERROR_WANT_READ) ?
636-
FD_READ |FD_CLOSE :FD_WRITE |FD_CLOSE,
637-
INFINITE);
638-
#endif
643+
644+
if (err==SSL_ERROR_WANT_READ)
645+
waitfor=WL_SOCKET_READABLE;
646+
else
647+
waitfor=WL_SOCKET_WRITEABLE;
648+
649+
WaitLatchOrSocket(MyLatch,waitfor,port->sock,0);
639650
gotowloop;
640651
caseSSL_ERROR_SYSCALL:
641652
/* leave it to caller to ereport the value of errno */
@@ -722,7 +733,7 @@ my_sock_read(BIO *h, char *buf, int size)
722733
if (res <=0)
723734
{
724735
/* If we were interrupted, tell caller to retry */
725-
if (errno==EINTR)
736+
if (errno==EINTR||errno==EWOULDBLOCK||errno==EAGAIN)
726737
{
727738
BIO_set_retry_read(h);
728739
}
@@ -741,7 +752,8 @@ my_sock_write(BIO *h, const char *buf, int size)
741752
BIO_clear_retry_flags(h);
742753
if (res <=0)
743754
{
744-
if (errno==EINTR)
755+
/* If we were interrupted, tell caller to retry */
756+
if (errno==EINTR||errno==EWOULDBLOCK||errno==EAGAIN)
745757
{
746758
BIO_set_retry_write(h);
747759
}

‎src/backend/libpq/be-secure.c

Lines changed: 76 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,10 @@
3232
#endif
3333

3434
#include"libpq/libpq.h"
35+
#include"miscadmin.h"
3536
#include"tcop/tcopprot.h"
3637
#include"utils/memutils.h"
38+
#include"storage/proc.h"
3739

3840

3941
char*ssl_cert_file;
@@ -147,7 +149,39 @@ secure_raw_read(Port *port, void *ptr, size_t len)
147149

148150
prepare_for_client_read();
149151

152+
/*
153+
* Try to read from the socket without blocking. If it succeeds we're
154+
* done, otherwise we'll wait for the socket using the latch mechanism.
155+
*/
156+
rloop:
157+
#ifdefWIN32
158+
pgwin32_noblock= true;
159+
#endif
150160
n=recv(port->sock,ptr,len,0);
161+
#ifdefWIN32
162+
pgwin32_noblock= false;
163+
#endif
164+
165+
if (n<0&& !port->noblock&& (errno==EWOULDBLOCK||errno==EAGAIN))
166+
{
167+
intw;
168+
intsave_errno=errno;
169+
170+
w=WaitLatchOrSocket(MyLatch,
171+
WL_SOCKET_READABLE,
172+
port->sock,0);
173+
174+
if (w&WL_SOCKET_READABLE)
175+
{
176+
gotorloop;
177+
}
178+
179+
/*
180+
* Restore errno, clobbered by WaitLatchOrSocket, so the caller can
181+
* react properly.
182+
*/
183+
errno=save_errno;
184+
}
151185

152186
client_read_ended();
153187

@@ -170,13 +204,54 @@ secure_write(Port *port, void *ptr, size_t len)
170204
}
171205
else
172206
#endif
207+
{
173208
n=secure_raw_write(port,ptr,len);
209+
}
174210

175211
returnn;
176212
}
177213

178214
ssize_t
179215
secure_raw_write(Port*port,constvoid*ptr,size_tlen)
180216
{
181-
returnsend(port->sock,ptr,len,0);
217+
ssize_tn;
218+
219+
wloop:
220+
221+
#ifdefWIN32
222+
pgwin32_noblock= true;
223+
#endif
224+
n=send(port->sock,ptr,len,0);
225+
#ifdefWIN32
226+
pgwin32_noblock= false;
227+
#endif
228+
229+
if (n<0&& !port->noblock&& (errno==EWOULDBLOCK||errno==EAGAIN))
230+
{
231+
intw;
232+
intsave_errno=errno;
233+
234+
/*
235+
* We probably want to check for latches being set at some point
236+
* here. That'd allow us to handle interrupts while blocked on
237+
* writes. If set we'd not retry directly, but return. That way we
238+
* don't do anything while (possibly) inside a ssl library.
239+
*/
240+
w=WaitLatchOrSocket(MyLatch,
241+
WL_SOCKET_WRITEABLE,
242+
port->sock,0);
243+
244+
if (w&WL_SOCKET_WRITEABLE)
245+
{
246+
gotowloop;
247+
}
248+
249+
/*
250+
* Restore errno, clobbered by WaitLatchOrSocket, so the caller can
251+
* react properly.
252+
*/
253+
errno=save_errno;
254+
}
255+
256+
returnn;
182257
}

‎src/backend/libpq/pqcomm.c

Lines changed: 16 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -181,6 +181,22 @@ pq_init(void)
181181
PqCommReadingMsg= false;
182182
DoingCopyOut= false;
183183
on_proc_exit(socket_close,0);
184+
185+
/*
186+
* In backends (as soon as forked) we operate the underlying socket in
187+
* nonblocking mode and use latches to implement blocking semantics if
188+
* needed. That allows us to provide safely interruptible reads.
189+
*
190+
* Use COMMERROR on failure, because ERROR would try to send the error to
191+
* the client, which might require changing the mode again, leading to
192+
* infinite recursion.
193+
*/
194+
#ifndefWIN32
195+
if (!pg_set_noblock(MyProcPort->sock))
196+
ereport(COMMERROR,
197+
(errmsg("could not set socket to nonblocking mode: %m")));
198+
#endif
199+
184200
}
185201

186202
/* --------------------------------
@@ -820,31 +836,6 @@ socket_set_nonblocking(bool nonblocking)
820836
(errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
821837
errmsg("there is no client connection")));
822838

823-
if (MyProcPort->noblock==nonblocking)
824-
return;
825-
826-
#ifdefWIN32
827-
pgwin32_noblock=nonblocking ?1 :0;
828-
#else
829-
830-
/*
831-
* Use COMMERROR on failure, because ERROR would try to send the error to
832-
* the client, which might require changing the mode again, leading to
833-
* infinite recursion.
834-
*/
835-
if (nonblocking)
836-
{
837-
if (!pg_set_noblock(MyProcPort->sock))
838-
ereport(COMMERROR,
839-
(errmsg("could not set socket to nonblocking mode: %m")));
840-
}
841-
else
842-
{
843-
if (!pg_set_block(MyProcPort->sock))
844-
ereport(COMMERROR,
845-
(errmsg("could not set socket to blocking mode: %m")));
846-
}
847-
#endif
848839
MyProcPort->noblock=nonblocking;
849840
}
850841

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp