Skip to content

Commit 387da18

Browse files
anarazelanarazel
committed
Use a nonblocking socket for FE/BE communication and block using latches.
This allows to introduce more elaborate handling of interrupts while reading from a socket. Currently some interrupt handlers have to do significant work from inside signal handlers, and it's very hard to correctly write code to do so. Generic signal handler limitations, combined with the fact that we can't safely jump out of a signal handler while reading from the client have prohibited implementation of features like timeouts for idle-in-transaction. Additionally we use the latch code to wait in a couple places where we previously only had waiting code on windows as other platforms just busy looped. This can increase the number of systemcalls happening during FE/BE communication. Benchmarks so far indicate that the impact isn't very high, and there's room for optimization in the latch code. The chance of cleaning up the usage of latches gives us, seem to outweigh the risk of small performance regressions. This commit theoretically can't used without the next patch in the series, as WaitLatchOrSocket is not defined to be fully signal safe. As we already do that in some cases though, it seems better to keep the commits separate, so they're easier to understand. Author: Andres Freund Reviewed-By: Heikki Linnakangas
1 parent 778d498 commit 387da18

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
int r;
340342
int err;
343+
int waitfor;
341344

342345
Assert(!port->ssl);
343346
Assert(!port->peer);
@@ -371,12 +374,15 @@ be_tls_open_server(Port *port)
371374
{
372375
case SSL_ERROR_WANT_READ:
373376
case SSL_ERROR_WANT_WRITE:
374-
#ifdef WIN32
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
goto aloop;
381387
case SSL_ERROR_SYSCALL:
382388
if (r < 0)
@@ -504,6 +510,7 @@ be_tls_read(Port *port, void *ptr, size_t len)
504510
{
505511
ssize_t n;
506512
int err;
513+
int waitfor;
507514

508515
rloop:
509516
errno = 0;
@@ -516,18 +523,20 @@ be_tls_read(Port *port, void *ptr, size_t len)
516523
break;
517524
case SSL_ERROR_WANT_READ:
518525
case SSL_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-
#ifdef WIN32
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
goto rloop;
532541
case SSL_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_t n;
569578
int err;
579+
int waitfor;
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
case SSL_ERROR_WANT_READ:
632642
case SSL_ERROR_WANT_WRITE:
633-
#ifdef WIN32
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
goto wloop;
640651
case SSL_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+
#ifdef WIN32
158+
pgwin32_noblock = true;
159+
#endif
150160
n = recv(port->sock, ptr, len, 0);
161+
#ifdef WIN32
162+
pgwin32_noblock = false;
163+
#endif
164+
165+
if (n < 0 && !port->noblock && (errno == EWOULDBLOCK || errno == EAGAIN))
166+
{
167+
int w;
168+
int save_errno = errno;
169+
170+
w = WaitLatchOrSocket(MyLatch,
171+
WL_SOCKET_READABLE,
172+
port->sock, 0);
173+
174+
if (w & WL_SOCKET_READABLE)
175+
{
176+
goto rloop;
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
return n;
176212
}
177213

178214
ssize_t
179215
secure_raw_write(Port *port, const void *ptr, size_t len)
180216
{
181-
return send(port->sock, ptr, len, 0);
217+
ssize_t n;
218+
219+
wloop:
220+
221+
#ifdef WIN32
222+
pgwin32_noblock = true;
223+
#endif
224+
n = send(port->sock, ptr, len, 0);
225+
#ifdef WIN32
226+
pgwin32_noblock = false;
227+
#endif
228+
229+
if (n < 0 && !port->noblock && (errno == EWOULDBLOCK || errno == EAGAIN))
230+
{
231+
int w;
232+
int save_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+
goto wloop;
247+
}
248+
249+
/*
250+
* Restore errno, clobbered by WaitLatchOrSocket, so the caller can
251+
* react properly.
252+
*/
253+
errno = save_errno;
254+
}
255+
256+
return n;
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+
#ifndef WIN32
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-
#ifdef WIN32
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)