public inbox for [email protected]  
help / color / mirror / Atom feed
From: Jelte Fennema <[email protected]>
To: Tom Lane <[email protected]>
To: Fujii Masao <[email protected]>
Cc: Zhihong Yu <[email protected]>
Cc: [email protected] <[email protected]>
Subject: Re: [EXTERNAL] Re: PQcancel does not use tcp_user_timeout, connect_timeout and keepalive settings
Date: Tue, 28 Dec 2021 15:49:00 +0000
Message-ID: <AM5PR83MB0178DCFFC1BE77C4C390DA89F7439@AM5PR83MB0178.EURPRD83.prod.outlook.com> (raw)
In-Reply-To: <HE1PR83MB0186173FC957C2554D0A31B6F7939@HE1PR83MB0186.EURPRD83.prod.outlook.com>
References: <AM5PR83MB017870DE81FC84D5E21E9D1EF7AA9@AM5PR83MB0178.EURPRD83.prod.outlook.com>
	<CALNJ-vSbOyVm9nk3y1UwA3e+M8atwJtTg_3ugRfC5W+TyAbMZQ@mail.gmail.com>
	<AM5PR83MB01781C8E94AEC8073B646872F7B09@AM5PR83MB0178.EURPRD83.prod.outlook.com>
	<AM5PR83MB0178586E6B079DF8E7F0D855F7B09@AM5PR83MB0178.EURPRD83.prod.outlook.com>
	<[email protected]>
	<[email protected]>
	<HE1PR83MB0186173FC957C2554D0A31B6F7939@HE1PR83MB0186.EURPRD83.prod.outlook.com>

I was able to spend some time on this again. I attached two patches to this email:

The first patch is a cleaned up version of my previous patch. I think I addressed
all feedback on the previous version in that patch (e.g. removed windows code, 
fixed formatting).

The second patch is a new one, it implements honouring of the connect_timeout 
connection option in PQcancel. This patch requires the first patch to also be applied,
but since it seemed fairly separate and the code is not trivial I didn't want the first
patch to be blocked on this.

Finally, I would love it if once these fixes are merged the would also be backpatched to 
previous versions of libpq. Does that seem possible? As far as I can tell it would be fine, 
since it doesn't really change any of the public APIs. The only change is that the pg_cancel 
struct now has a few additional fields. But since that struct is defined in libpq-int.h, so that 
struct should not be used by users of libpq directly, right?.

Attachments:

  [application/octet-stream] 0001-Use-timeout-socket-options-for-cancel-connections.patch (8.4K, ../AM5PR83MB0178DCFFC1BE77C4C390DA89F7439@AM5PR83MB0178.EURPRD83.prod.outlook.com/2-0001-Use-timeout-socket-options-for-cancel-connections.patch)
  download | inline diff:
From afd1d9ec2fceda6fdac8a4c5e9885a6db115df97 Mon Sep 17 00:00:00 2001
From: Jelte Fennema <[email protected]>
Date: Tue, 28 Dec 2021 12:46:35 +0100
Subject: [PATCH 1/2] Use timeout socket options for cancel connections

PQcancel would not adhere to the timeout specified in the
tcp_user_timeout connection option. It would also not enable keepalives
on the socket. So no keepalives would be used by the on these
connections at all, not even the system configured ones.

This means a call to PQcancel could take much longer than expected,
possibly indefinitely. This can especially be an issue because there's
no non-blocking version of PQcancel.

This patch correctly sets both tcp_user_timeout and keepalive related
socket options on the connection that's used to cancel a query. It only
does so for non Windows systems, because the signal safety of the
related windows functions is not clear to me.
---
 src/interfaces/libpq/fe-connect.c | 159 +++++++++++++++++++++++++++---
 src/interfaces/libpq/libpq-int.h  |   7 ++
 2 files changed, 154 insertions(+), 12 deletions(-)

diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c
index 9b6a6939f0..ed29c64484 100644
--- a/src/interfaces/libpq/fe-connect.c
+++ b/src/interfaces/libpq/fe-connect.c
@@ -4385,13 +4385,60 @@ PQgetCancel(PGconn *conn)
 	if (conn->sock == PGINVALID_SOCKET)
 		return NULL;
 
-	cancel = malloc(sizeof(PGcancel));
+	cancel = calloc(1, sizeof(PGcancel));
 	if (cancel == NULL)
 		return NULL;
 
 	memcpy(&cancel->raddr, &conn->raddr, sizeof(SockAddr));
 	cancel->be_pid = conn->be_pid;
 	cancel->be_key = conn->be_key;
+	/* Set all the socket options to -1, so we can use that to see if the */
+	/* socket options are set or not. */
+	cancel->pgtcp_user_timeout = -1;
+	cancel->keepalives = -1;
+	cancel->keepalives_idle = -1;
+	cancel->keepalives_interval = -1;
+	cancel->keepalives_count = -1;
+	if (conn->pgtcp_user_timeout != NULL)
+	{
+		if (!parse_int_param(conn->pgtcp_user_timeout, &cancel->pgtcp_user_timeout, conn,
+							 "tcp_user_timeout"))
+		{
+			return NULL;
+		}
+	}
+	if (conn->keepalives != NULL)
+	{
+		if (!parse_int_param(conn->keepalives, &cancel->keepalives, conn,
+							 "keepalives"))
+		{
+			return NULL;
+		}
+	}
+	if (conn->keepalives_idle != NULL)
+	{
+		if (!parse_int_param(conn->keepalives_idle, &cancel->keepalives_idle, conn,
+							 "keepalives_idle"))
+		{
+			return NULL;
+		}
+	}
+	if (conn->keepalives_interval != NULL)
+	{
+		if (!parse_int_param(conn->keepalives_interval, &cancel->keepalives_interval, conn,
+							 "keepalives_interval"))
+		{
+			return NULL;
+		}
+	}
+	if (conn->keepalives_count != NULL)
+	{
+		if (!parse_int_param(conn->keepalives_count, &cancel->keepalives_count, conn,
+							 "keepalives_count"))
+		{
+			return NULL;
+		}
+	}
 
 	return cancel;
 }
@@ -4426,8 +4473,7 @@ PQfreeCancel(PGcancel *cancel)
  * between the two versions of the cancel function possible.
  */
 static int
-internal_cancel(SockAddr *raddr, int be_pid, int be_key,
-				char *errbuf, int errbufsize)
+internal_cancel(PGcancel *cancel, char *errbuf, int errbufsize)
 {
 	int			save_errno = SOCK_ERRNO;
 	pgsocket	tmpsock = PGINVALID_SOCKET;
@@ -4443,14 +4489,102 @@ internal_cancel(SockAddr *raddr, int be_pid, int be_key,
 	 * We need to open a temporary connection to the postmaster. Do this with
 	 * only kernel calls.
 	 */
-	if ((tmpsock = socket(raddr->addr.ss_family, SOCK_STREAM, 0)) == PGINVALID_SOCKET)
+	if ((tmpsock = socket(cancel->raddr.addr.ss_family, SOCK_STREAM, 0)) == PGINVALID_SOCKET)
 	{
 		strlcpy(errbuf, "PQcancel() -- socket() failed: ", errbufsize);
 		goto cancel_errReturn;
 	}
+
+	/*
+	 * Since this connection will only be used to send a tiny bit of data, we
+	 * don't need NODELAY. We also don't set the socket to nonblocking mode,
+	 * because the API definition of PQcancel currently requires the cancel to
+	 * be sent in a blocking way.
+	 *
+	 * We do set various other socket options related to keepalives and other
+	 * TCP timeouts. This is necessary to make sure that the call to this
+	 * function does not block indefinitly, when reasonable keepalive and
+	 * settings have been provided.
+	 *
+	 * NOTE: These socket options are currently not set for Windows. The
+	 * reason is that signal safety in this function is very important, and it
+	 * was not clear to if the functions required to set the socket options on
+	 * Windows were signal-safe.
+	 */
+#ifndef WIN32
+	if (!IS_AF_UNIX(cancel->raddr.addr.ss_family))
+	{
+#ifdef TCP_USER_TIMEOUT
+		if (cancel->pgtcp_user_timeout >= 0)
+		{
+			if (setsockopt(tmpsock, IPPROTO_TCP, TCP_USER_TIMEOUT,
+						   (char *) &cancel->pgtcp_user_timeout,
+						   sizeof(cancel->pgtcp_user_timeout)) < 0)
+			{
+				strlcpy(errbuf, "PQcancel() -- setsockopt(TCP_USER_TIMEOUT) failed: ", errbufsize);
+				goto cancel_errReturn;
+			}
+		}
+#endif
+
+		if (cancel->keepalives != 0)
+		{
+			int			on = 1;
+
+			if (setsockopt(tmpsock,
+						   SOL_SOCKET, SO_KEEPALIVE,
+						   (char *) &on, sizeof(on)) < 0)
+			{
+				strlcpy(errbuf, "PQcancel() -- setsockopt(SO_KEEPALIVE) failed: ", errbufsize);
+				goto cancel_errReturn;
+			}
+		}
+
+#ifdef PG_TCP_KEEPALIVE_IDLE
+		if (cancel->keepalives_idle >= 0)
+		{
+			if (setsockopt(tmpsock, IPPROTO_TCP, PG_TCP_KEEPALIVE_IDLE,
+						   (char *) &cancel->keepalives_idle,
+						   sizeof(cancel->keepalives_idle)) < 0)
+			{
+				strlcpy(errbuf, "PQcancel() -- setsockopt(PG_TPC_KEEPALIVE_IDLE) failed: ", errbufsize);
+				goto cancel_errReturn;
+			}
+		}
+#endif
+
+#ifdef TCP_KEEPINTVL
+		if (cancel->keepalives_interval >= 0)
+		{
+			if (setsockopt(tmpsock, IPPROTO_TCP, TCP_KEEPINTVL,
+						   (char *) &cancel->keepalives_interval,
+						   sizeof(cancel->keepalives_interval)) < 0)
+			{
+				strlcpy(errbuf, "PQcancel() -- setsockopt(TPC_KEEPINTVL) failed: ", errbufsize);
+				goto cancel_errReturn;
+			}
+		}
+#endif
+
+#ifdef TCP_KEEPCNT
+		if (cancel->keepalives_count >= 0)
+		{
+			if (setsockopt(tmpsock, IPPROTO_TCP, TCP_KEEPCNT,
+						   (char *) &cancel->keepalives_count,
+						   sizeof(cancel->keepalives_count)) < 0)
+			{
+				strlcpy(errbuf, "PQcancel() -- setsockopt(TPC_KEEPCNT) failed: ", errbufsize);
+				goto cancel_errReturn;
+			}
+		}
+#endif
+	}
+#endif
+
+
 retry3:
-	if (connect(tmpsock, (struct sockaddr *) &raddr->addr,
-				raddr->salen) < 0)
+	if (connect(tmpsock, (struct sockaddr *) &cancel->raddr.addr,
+				cancel->raddr.salen) < 0)
 	{
 		if (SOCK_ERRNO == EINTR)
 			/* Interrupted system call - we'll just try again */
@@ -4467,8 +4601,8 @@ retry3:
 
 	crp.packetlen = pg_hton32((uint32) sizeof(crp));
 	crp.cp.cancelRequestCode = (MsgType) pg_hton32(CANCEL_REQUEST_CODE);
-	crp.cp.backendPID = pg_hton32(be_pid);
-	crp.cp.cancelAuthCode = pg_hton32(be_key);
+	crp.cp.backendPID = pg_hton32(cancel->be_pid);
+	crp.cp.cancelAuthCode = pg_hton32(cancel->be_key);
 
 retry4:
 	if (send(tmpsock, (char *) &crp, sizeof(crp), 0) != (int) sizeof(crp))
@@ -4538,8 +4672,7 @@ PQcancel(PGcancel *cancel, char *errbuf, int errbufsize)
 		return false;
 	}
 
-	return internal_cancel(&cancel->raddr, cancel->be_pid, cancel->be_key,
-						   errbuf, errbufsize);
+	return internal_cancel(cancel, errbuf, errbufsize);
 }
 
 /*
@@ -4558,6 +4691,7 @@ int
 PQrequestCancel(PGconn *conn)
 {
 	int			r;
+	PGcancel   *cancel;
 
 	/* Check we have an open connection */
 	if (!conn)
@@ -4573,8 +4707,9 @@ PQrequestCancel(PGconn *conn)
 		return false;
 	}
 
-	r = internal_cancel(&conn->raddr, conn->be_pid, conn->be_key,
-						conn->errorMessage.data, conn->errorMessage.maxlen);
+	cancel = PQgetCancel(conn);
+	r = PQcancel(cancel, conn->errorMessage.data, conn->errorMessage.maxlen);
+	PQfreeCancel(cancel);
 
 	if (!r)
 		conn->errorMessage.len = strlen(conn->errorMessage.data);
diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h
index 490458adef..6dabb14451 100644
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -583,6 +583,13 @@ struct pg_cancel
 	SockAddr	raddr;			/* Remote address */
 	int			be_pid;			/* PID of backend --- needed for cancels */
 	int			be_key;			/* key of backend --- needed for cancels */
+	int			pgtcp_user_timeout; /* tcp user timeout */
+	int			keepalives;		/* use TCP keepalives? */
+	int			keepalives_idle;	/* time between TCP keepalives */
+	int			keepalives_interval;	/* time between TCP keepalive
+										 * retransmits */
+	int			keepalives_count;	/* maximum number of TCP keepalive
+									 * retransmits */
 };
 
 
-- 
2.17.1



  [application/octet-stream] 0002-Honor-connect_timeout-when-connecting-with-PQcancel.patch (6.4K, ../AM5PR83MB0178DCFFC1BE77C4C390DA89F7439@AM5PR83MB0178.EURPRD83.prod.outlook.com/3-0002-Honor-connect_timeout-when-connecting-with-PQcancel.patch)
  download | inline diff:
From 78b1e7fc2bcef3d9b1b85b88387e24918411b2e9 Mon Sep 17 00:00:00 2001
From: Jelte Fennema <[email protected]>
Date: Tue, 28 Dec 2021 14:29:51 +0100
Subject: [PATCH 2/2] Honor connect_timeout when connecting with PQcancel

PQcancel uses its own connection mechanism, because it needs to be
signal safe. This meant that it was not honoring connect_timeout
connection option. This change fixes that for non windows environments,
by using the timeout option of select() to ensure a connect_timeout.
---
 src/interfaces/libpq/fe-connect.c | 128 +++++++++++++++++++++++++++---
 src/interfaces/libpq/libpq-int.h  |   1 +
 2 files changed, 118 insertions(+), 11 deletions(-)

diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c
index ed29c64484..61185db329 100644
--- a/src/interfaces/libpq/fe-connect.c
+++ b/src/interfaces/libpq/fe-connect.c
@@ -4394,11 +4394,20 @@ PQgetCancel(PGconn *conn)
 	cancel->be_key = conn->be_key;
 	/* Set all the socket options to -1, so we can use that to see if the */
 	/* socket options are set or not. */
+	cancel->connect_timeout = -1;
 	cancel->pgtcp_user_timeout = -1;
 	cancel->keepalives = -1;
 	cancel->keepalives_idle = -1;
 	cancel->keepalives_interval = -1;
 	cancel->keepalives_count = -1;
+	if (conn->connect_timeout != NULL)
+	{
+		if (!parse_int_param(conn->connect_timeout, &cancel->connect_timeout, conn,
+							 "connect_timeout"))
+		{
+			return NULL;
+		}
+	}
 	if (conn->pgtcp_user_timeout != NULL)
 	{
 		if (!parse_int_param(conn->pgtcp_user_timeout, &cancel->pgtcp_user_timeout, conn,
@@ -4479,6 +4488,14 @@ internal_cancel(PGcancel *cancel, char *errbuf, int errbufsize)
 	pgsocket	tmpsock = PGINVALID_SOCKET;
 	char		sebuf[PG_STRERROR_R_BUFLEN];
 	int			maxlen;
+#ifndef WIN32
+	int			tmpsockflags = 0;
+	fd_set		fdset;
+	bool		nonblocking_connect = cancel->connect_timeout > 0;
+	struct timeval connect_timeout;
+#else
+	bool		nonblocking_connect = false;
+#endif
 	struct
 	{
 		uint32		packetlen;
@@ -4581,21 +4598,110 @@ internal_cancel(PGcancel *cancel, char *errbuf, int errbufsize)
 	}
 #endif
 
+#ifndef WIN32
+	if (cancel->connect_timeout > 0)
+	{
+		/*
+		 * The connect() system call does not allow a timeout to be specified.
+		 * So to honor the connect_timeout that the user specified, we connect
+		 * in non blocking mode. Then we provide the connecti select() to
+		 */
+		nonblocking_connect = true;
+		tmpsockflags = fcntl(tmpsock, F_GETFL);
+		if (tmpsockflags < 0)
+		{
+			strlcpy(errbuf, "PQcancel() -- fcntl(F_GETFL) failed: ", errbufsize);
+			goto cancel_errReturn;
+		}
+		if (fcntl(tmpsock, F_SETFL, (tmpsockflags | O_NONBLOCK)) == -1)
+		{
+			strlcpy(errbuf, "PQcancel() -- fcntl(F_SETFL, O_NONBLOCK) failed: ", errbufsize);
+			goto cancel_errReturn;
+		}
+	}
+#endif
 
 retry3:
 	if (connect(tmpsock, (struct sockaddr *) &cancel->raddr.addr,
 				cancel->raddr.salen) < 0)
 	{
-		if (SOCK_ERRNO == EINTR)
-			/* Interrupted system call - we'll just try again */
+		if (nonblocking_connect && (
+									SOCK_ERRNO == EINPROGRESS ||
+#ifdef WIN32
+									SOCK_ERRNO == EWOULDBLOCK ||
+#endif
+									SOCK_ERRNO == EINTR))
+		{
+			/*
+			 * This is fine - we're in non-blocking mode, and the connection
+			 * is in progress.
+			 */
+		}
+		else if (SOCK_ERRNO == EINTR)
+		{
+			/*
+			 * Interrupted system call while in blocking mode - we'll just try
+			 * again
+			 */
 			goto retry3;
-		strlcpy(errbuf, "PQcancel() -- connect() failed: ", errbufsize);
-		goto cancel_errReturn;
+		}
+		else
+		{
+			strlcpy(errbuf, "PQcancel() -- connect() failed: ", errbufsize);
+			goto cancel_errReturn;
+		}
 	}
 
-	/*
-	 * We needn't set nonblocking I/O or NODELAY options here.
-	 */
+
+#ifndef WIN32
+	if (cancel->connect_timeout > 0)
+	{
+		connect_timeout.tv_sec = cancel->connect_timeout;
+		connect_timeout.tv_usec = 0;
+
+		FD_ZERO(&fdset);
+		FD_SET(tmpsock, &fdset);
+retry4:
+		if (select(tmpsock + 1, NULL, &fdset, NULL, &connect_timeout) == 1)
+		{
+			int			so_error;
+			socklen_t	len = sizeof so_error;
+
+			if (SOCK_ERRNO == EINTR)
+				/* Interrupted system call - we'll just try again */
+				goto retry4;
+			if (getsockopt(tmpsock, SOL_SOCKET, SO_ERROR, &so_error, &len))
+			{
+				strlcpy(errbuf, "PQcancel() -- getsockopt(SO_ERROR) failed: ", errbufsize);
+				goto cancel_errReturn;
+			}
+			if (so_error != 0)
+			{
+				SOCK_ERRNO = so_error;
+				strlcpy(errbuf, "PQcancel() -- select() failed: ", errbufsize);
+				goto cancel_errReturn;
+			}
+		}
+		else
+		{
+			strlcpy(errbuf, "PQcancel() -- Connection timed out\n", errbufsize);
+			closesocket(tmpsock);
+			SOCK_ERRNO_SET(save_errno);
+			return false;
+		}
+
+		/*
+		 * Now that we're connected we want a blocking socket again. So we
+		 * reset the flags to the value they had before changing to non
+		 * blocking mode.
+		 */
+		if (fcntl(tmpsock, F_SETFL, tmpsockflags) == -1)
+		{
+			strlcpy(errbuf, "PQcancel() -- fcntl(F_SETFL, !O_NONBLOCK) failed: ", errbufsize);
+			goto cancel_errReturn;
+		}
+	}
+#endif
 
 	/* Create and send the cancel request packet. */
 
@@ -4604,12 +4710,12 @@ retry3:
 	crp.cp.backendPID = pg_hton32(cancel->be_pid);
 	crp.cp.cancelAuthCode = pg_hton32(cancel->be_key);
 
-retry4:
+retry5:
 	if (send(tmpsock, (char *) &crp, sizeof(crp), 0) != (int) sizeof(crp))
 	{
 		if (SOCK_ERRNO == EINTR)
 			/* Interrupted system call - we'll just try again */
-			goto retry4;
+			goto retry5;
 		strlcpy(errbuf, "PQcancel() -- send() failed: ", errbufsize);
 		goto cancel_errReturn;
 	}
@@ -4621,12 +4727,12 @@ retry4:
 	 * one we thought we were canceling.  Note we don't actually expect this
 	 * read to obtain any data, we are just waiting for EOF to be signaled.
 	 */
-retry5:
+retry6:
 	if (recv(tmpsock, (char *) &crp, 1, 0) < 0)
 	{
 		if (SOCK_ERRNO == EINTR)
 			/* Interrupted system call - we'll just try again */
-			goto retry5;
+			goto retry6;
 		/* we ignore other error conditions */
 	}
 
diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h
index 6dabb14451..c08ecbd06f 100644
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -583,6 +583,7 @@ struct pg_cancel
 	SockAddr	raddr;			/* Remote address */
 	int			be_pid;			/* PID of backend --- needed for cancels */
 	int			be_key;			/* key of backend --- needed for cancels */
+	int			connect_timeout;	/* connection timeout */
 	int			pgtcp_user_timeout; /* tcp user timeout */
 	int			keepalives;		/* use TCP keepalives? */
 	int			keepalives_idle;	/* time between TCP keepalives */
-- 
2.17.1



view thread (8+ messages)

reply

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Reply to all the recipients using the --to and --cc options:
  reply via email

  To: [email protected]
  Cc: [email protected], [email protected], [email protected], [email protected]
  Subject: Re: [EXTERNAL] Re: PQcancel does not use tcp_user_timeout, connect_timeout and keepalive settings
  In-Reply-To: <AM5PR83MB0178DCFFC1BE77C4C390DA89F7439@AM5PR83MB0178.EURPRD83.prod.outlook.com>

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox