On Fri Mar 22, 2024 at 12:17 PM CDT, Robert Haas wrote:
On Fri, Mar 22, 2024 at 1:05 PM Tristan Partin <tris...@neon.tech> wrote:
> Sorry for taking a while to get back to y'all. I have taken your
> feedback into consideration for v9. This is my first time writing
> Postgres docs, so I'm ready for another round of editing :).
Yeah, that looks like it needs some wordsmithing yet. I can take a
crack at that at some point, but here are a few notes:
- "takes care of" and "working through the state machine" seem quite
vague to me.
- the meanings of forRead and forWrite don't seem to be documented.
- "retuns" is a typo.
> Robert, could you point out some places where comments would be useful
> in 0002? I did rename the function and moved it as suggested, thanks! In
> turn, I also realized I forgot a prototype, so also added it.
Well, for starters, I'd give the function a header comment.
Telling me that a 1 second timeout is a 1 second timeout is less
useful than telling me why we've picked a 1 second timeout. Presumably
the answer is "so we can respond to cancel interrupts in a timely
fashion", but I'd make that explicit.
It might be worth a comment saying that PQsocket() shouldn't be
hoisted out of the loop because it could be a multi-host connection
string and the socket might change under us. Unless that's not true,
in which case we should hoist the PQsocket() call out of the loop.
I think it would also be worth a comment saying that we don't need to
report errors here, as the caller will do that; we just need to wait
until the connection is known to have either succeeded or failed, or
until the user presses cancel.
This is good feedback, thanks. I have added comments where you
suggested. I reworded the PQsocketPoll docs to hopefully meet your
expectations. I am using the term "connection sequence" which is from
the PQconnectStartParams docs, so hopefully people will be able to make
that connection. I wrote documentation for "forRead" and "forWrite" as
well.
I had a question about parameter naming. Right now I have a mix of
camel-case and snake-case in the function signature since that is what
I inherited. Should I change that to be consistent? If so, which case
would you like?
Thanks for your continued reviews.
--
Tristan Partin
Neon (https://neon.tech)
From dc45e95ab443d973845dd840600d6719dcd4cae2 Mon Sep 17 00:00:00 2001
From: Tristan Partin <tris...@neon.tech>
Date: Tue, 30 Jan 2024 15:40:57 -0600
Subject: [PATCH v10 1/2] Expose PQsocketPoll for frontends
PQsocketPoll should help with state machine processing when connecting
to a database asynchronously via PQconnectStart().
---
doc/src/sgml/libpq.sgml | 43 +++++++++++++++++++++++++++++++-
src/interfaces/libpq/exports.txt | 1 +
src/interfaces/libpq/fe-misc.c | 7 +++---
src/interfaces/libpq/libpq-fe.h | 4 +++
4 files changed, 50 insertions(+), 5 deletions(-)
diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml
index d3e87056f2c..774b57ba70b 100644
--- a/doc/src/sgml/libpq.sgml
+++ b/doc/src/sgml/libpq.sgml
@@ -262,6 +262,44 @@ PGconn *PQsetdb(char *pghost,
</listitem>
</varlistentry>
+ <varlistentry id="libpq-PQsocketPoll">
+ <term><function>PQsocketPoll</function><indexterm><primary>PQsocketPoll</primary></indexterm></term>
+ <listitem>
+ <para>
+ <indexterm><primary>nonblocking connection</primary></indexterm>
+ Poll a connection's underlying socket descriptor retrieved with <xref linkend="libpq-PQsocket"/>.
+<synopsis>
+int PQsocketPoll(int sock, int forRead, int forWrite, time_t end_time);
+</synopsis>
+ </para>
+
+ <para>
+ This function sets up polling of a file descriptor. The underlying function is either
+ <function>poll(2)</function> or <function>select(2)</function>, depending on platform
+ support. The primary use of this function is iterating through the connection sequence
+ described in the documentation of <xref linkend="libpq-PQconnectStartParams"/>. If
+ <parameter>forRead</parameter> is specified, the function waits for the socket to be ready
+ for reading. If <parameter>forWrite</parameter> is specified, the function waits for the
+ socket to be ready for write. See <literal>POLLIN</literal> and <literal>POLLOUT</literal>
+ from <function>poll(2)</function>, or <parameter>readfds</parameter> and
+ <parameter>writefds</parameter> from <function>select(2)</function> for more information.
+ </para>
+
+ <para>
+ The function returns a value greater than <literal>0</literal> if the specified condition
+ is met, <literal>0</literal> if a timeout occurred, or <literal>-1</literal> if an error
+ or interrupt occurred. In the event <literal>forRead</literal> and
+ <literal>forWrite</literal> are not set, the function immediately returns a timeout
+ condition.
+ </para>
+
+ <para>
+ <literal>end_time</literal> is the time in the future in seconds starting from the UNIX
+ epoch in which you would like the function to return if the condition is not met.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry id="libpq-PQconnectStartParams">
<term><function>PQconnectStartParams</function><indexterm><primary>PQconnectStartParams</primary></indexterm></term>
<term><function>PQconnectStart</function><indexterm><primary>PQconnectStart</primary></indexterm></term>
@@ -358,7 +396,10 @@ PostgresPollingStatusType PQconnectPoll(PGconn *conn);
Loop thus: If <function>PQconnectPoll(conn)</function> last returned
<symbol>PGRES_POLLING_READING</symbol>, wait until the socket is ready to
read (as indicated by <function>select()</function>, <function>poll()</function>, or
- similar system function).
+ similar system function). Note that <function>PQsocketPoll</function>
+ can help reduce boilerplate by abstracting the setup of
+ <function>select(2)</function> or <function>poll(2)</function> if it is
+ available on your system.
Then call <function>PQconnectPoll(conn)</function> again.
Conversely, if <function>PQconnectPoll(conn)</function> last returned
<symbol>PGRES_POLLING_WRITING</symbol>, wait until the socket is ready
diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt
index 9fbd3d34074..1e48d37677d 100644
--- a/src/interfaces/libpq/exports.txt
+++ b/src/interfaces/libpq/exports.txt
@@ -202,3 +202,4 @@ PQcancelSocket 199
PQcancelErrorMessage 200
PQcancelReset 201
PQcancelFinish 202
+PQsocketPoll 203
diff --git a/src/interfaces/libpq/fe-misc.c b/src/interfaces/libpq/fe-misc.c
index f2fc78a481c..f562cd8d344 100644
--- a/src/interfaces/libpq/fe-misc.c
+++ b/src/interfaces/libpq/fe-misc.c
@@ -55,7 +55,6 @@ static int pqPutMsgBytes(const void *buf, size_t len, PGconn *conn);
static int pqSendSome(PGconn *conn, int len);
static int pqSocketCheck(PGconn *conn, int forRead, int forWrite,
time_t end_time);
-static int pqSocketPoll(int sock, int forRead, int forWrite, time_t end_time);
/*
* PQlibVersion: return the libpq version number
@@ -1059,7 +1058,7 @@ pqSocketCheck(PGconn *conn, int forRead, int forWrite, time_t end_time)
/* We will retry as long as we get EINTR */
do
- result = pqSocketPoll(conn->sock, forRead, forWrite, end_time);
+ result = PQsocketPoll(conn->sock, forRead, forWrite, end_time);
while (result < 0 && SOCK_ERRNO == EINTR);
if (result < 0)
@@ -1083,8 +1082,8 @@ pqSocketCheck(PGconn *conn, int forRead, int forWrite, time_t end_time)
* Timeout is infinite if end_time is -1. Timeout is immediate (no blocking)
* if end_time is 0 (or indeed, any time before now).
*/
-static int
-pqSocketPoll(int sock, int forRead, int forWrite, time_t end_time)
+int
+PQsocketPoll(int sock, int forRead, int forWrite, time_t end_time)
{
/* We use poll(2) if available, otherwise select(2) */
#ifdef HAVE_POLL
diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h
index 09b485bd2bc..8d3c5c6f662 100644
--- a/src/interfaces/libpq/libpq-fe.h
+++ b/src/interfaces/libpq/libpq-fe.h
@@ -21,6 +21,7 @@ extern "C"
#endif
#include <stdio.h>
+#include <time.h>
/*
* postgres_ext.h defines the backend's externally visible types,
@@ -670,6 +671,9 @@ extern int lo_export(PGconn *conn, Oid lobjId, const char *filename);
/* Get the version of the libpq library in use */
extern int PQlibVersion(void);
+/* Poll a socket for reading and/or writing with an optional timeout */
+extern int PQsocketPoll(int sock, int forRead, int forWrite, time_t end_time);
+
/* Determine length of multibyte encoded char at *s */
extern int PQmblen(const char *s, int encoding);
--
Tristan Partin
Neon (https://neon.tech)
From dc1eb3866975680b55451b79b40f109e3104eee8 Mon Sep 17 00:00:00 2001
From: Tristan Partin <tris...@neon.tech>
Date: Tue, 30 Jan 2024 15:53:10 -0600
Subject: [PATCH v10 2/2] Allow SIGINT to cancel psql database reconnections
After installing the SIGINT handler in psql, SIGINT can no longer cancel
database reconnections. For instance, if the user starts a reconnection
and then needs to do some form of interaction (ie psql is polling),
there is no way to cancel the reconnection process currently.
Use PQconnectStartParams() in order to insert a cancel_pressed check
into the polling loop.
---
src/bin/psql/command.c | 72 +++++++++++++++++++++++++++++++++++++++++-
1 file changed, 71 insertions(+), 1 deletion(-)
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index 9b0fa041f73..1e00b0d4869 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -159,6 +159,7 @@ static void discard_query_text(PsqlScanState scan_state, ConditionalStack cstack
static bool copy_previous_query(PQExpBuffer query_buf, PQExpBuffer previous_buf);
static bool do_connect(enum trivalue reuse_previous_specification,
char *dbname, char *user, char *host, char *port);
+static void wait_until_connected(PGconn *conn);
static bool do_edit(const char *filename_arg, PQExpBuffer query_buf,
int lineno, bool discard_on_quit, bool *edited);
static bool do_shell(const char *command);
@@ -3595,11 +3596,12 @@ do_connect(enum trivalue reuse_previous_specification,
values[paramnum] = NULL;
/* Note we do not want libpq to re-expand the dbname parameter */
- n_conn = PQconnectdbParams(keywords, values, false);
+ n_conn = PQconnectStartParams(keywords, values, false);
pg_free(keywords);
pg_free(values);
+ wait_until_connected(n_conn);
if (PQstatus(n_conn) == CONNECTION_OK)
break;
@@ -3748,6 +3750,74 @@ do_connect(enum trivalue reuse_previous_specification,
return true;
}
+/*
+ * Processes the connection sequence described by PQconnectStartParams(). Don't
+ * worry about reporting errors in this function. Our caller will check the
+ * connection's status, and report appropriately.
+ */
+static void
+wait_until_connected(PGconn *conn)
+{
+ bool forRead = false;
+
+ while (true)
+ {
+ int rc;
+ int sock;
+ time_t end_time;
+
+ /*
+ * On every iteration of the connection sequence, let's check if the
+ * user has requested a cancellation.
+ */
+ if (cancel_pressed)
+ break;
+
+ /*
+ * Do not assume that the socket remains the same across
+ * PQconnectPoll() calls.
+ */
+ sock = PQsocket(conn);
+ if (sock == -1)
+ break;
+
+ /*
+ * If the user sends SIGINT between the cancel_pressed check, and
+ * polling of the socket, it will not be recognized. Instead, we will
+ * just wait until the next step in the connection sequence or forever,
+ * which might require users to send SIGTERM or SIGQUIT.
+ *
+ * Some solutions would include the "self-pipe trick," using
+ * pselect(2) and ppoll(2), or using a timeout.
+ *
+ * The self-pipe trick requires a bit of code to setup. pselect(2) and
+ * ppoll(2) are not on all the platforms we support. The simplest
+ * solution happens to just be adding a timeout, so let's wait for 1
+ * second and check cancel_pressed again.
+ */
+ end_time = time(NULL) + 1;
+ rc = PQsocketPoll(sock, forRead, !forRead, end_time);
+ if (rc == -1)
+ return;
+
+ switch (PQconnectPoll(conn))
+ {
+ case PGRES_POLLING_OK:
+ case PGRES_POLLING_FAILED:
+ return;
+ case PGRES_POLLING_READING:
+ forRead = true;
+ continue;
+ case PGRES_POLLING_WRITING:
+ forRead = false;
+ continue;
+ case PGRES_POLLING_ACTIVE:
+ pg_unreachable();
+ }
+ }
+
+ pg_unreachable();
+}
void
connection_warnings(bool in_startup)
--
Tristan Partin
Neon (https://neon.tech)
