Generalize parallel slot result handling.

Instead of having a hard-coded behavior that we ignore missing
tables and report all other errors, let the caller decide what
to do by setting a callback.

Mark Dilger, reviewed and somewhat revised by me. The larger patch
series of which this is a part has also had review from Peter
Geoghegan, Andres Freund, Álvaro Herrera, Michael Paquier, and Amul
Sul, but I don't know whether any of them have reviewed this bit
specifically.

Discussion: http://postgr.es/m/12ED3DA8-25F0-4B68-937D-D907CFBF08E7@enterprisedb.com
Discussion: http://postgr.es/m/5F743835-3399-419C-8324-2D424237E999@enterprisedb.com
Discussion: http://postgr.es/m/70655DF3-33CE-4527-9A4D-DDEB582B6BA0@enterprisedb.com

diff --git a/src/bin/scripts/reindexdb.c b/src/bin/scripts/reindexdb.c
index 7781fb1151ab4942ced4f1a5b7466556a10f21f7..9f072ac49aea95f854e28510b44661a4f51c5adf 100644 (file)
--- a/src/bin/scripts/reindexdb.c
+++ b/src/bin/scripts/reindexdb.c
@@ -466,6 +466,7 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
                        goto finish;
                }
 
+               ParallelSlotSetHandler(free_slot, TableCommandResultHandler, NULL);
                run_reindex_command(free_slot->connection, process_type, objname,
                                                        echo, verbose, concurrently, true);
 

diff --git a/src/bin/scripts/vacuumdb.c b/src/bin/scripts/vacuumdb.c
index ed320817bc4feac6e90cac8f2bdaff6572d91810..9dc8aca29f351fb08a2cec8dcfd8a35d8055bc0e 100644 (file)
--- a/src/bin/scripts/vacuumdb.c
+++ b/src/bin/scripts/vacuumdb.c
@@ -713,6 +713,7 @@ vacuum_one_database(const ConnParams *cparams,
                 * Execute the vacuum.  All errors are handled in processQueryResult
                 * through ParallelSlotsGetIdle.
                 */
+               ParallelSlotSetHandler(free_slot, TableCommandResultHandler, NULL);
                run_vacuum_command(free_slot->connection, sql.data,
                                                   echo, tabname);
 

diff --git a/src/fe_utils/parallel_slot.c b/src/fe_utils/parallel_slot.c
index 3987a4702b5255a6258d1b4ade4a86cf810b7c23..b625deb2545a12fec6e22e2b913ebe4ce44878f8 100644 (file)
--- a/src/fe_utils/parallel_slot.c
+++ b/src/fe_utils/parallel_slot.c
@@ -30,7 +30,7 @@
 
 static void init_slot(ParallelSlot *slot, PGconn *conn);
 static int     select_loop(int maxFd, fd_set *workerset);
-static bool processQueryResult(PGconn *conn, PGresult *result);
+static bool processQueryResult(ParallelSlot *slot, PGresult *result);
 
 static void
 init_slot(ParallelSlot *slot, PGconn *conn)
@@ -38,34 +38,24 @@ init_slot(ParallelSlot *slot, PGconn *conn)
        slot->connection = conn;
        /* Initially assume connection is idle */
        slot->isFree = true;
+       ParallelSlotClearHandler(slot);
 }
 
 /*
- * Process (and delete) a query result.  Returns true if there's no error,
- * false otherwise -- but errors about trying to work on a missing relation
- * are reported and subsequently ignored.
+ * Process (and delete) a query result.  Returns true if there's no problem,
+ * false otherwise. It's up to the handler to decide what cosntitutes a
+ * problem.
  */
 static bool
-processQueryResult(PGconn *conn, PGresult *result)
+processQueryResult(ParallelSlot *slot, PGresult *result)
 {
-       /*
-        * If it's an error, report it.  Errors about a missing table are harmless
-        * so we continue processing; but die for other errors.
-        */
-       if (PQresultStatus(result) != PGRES_COMMAND_OK)
-       {
-               char       *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
+       Assert(slot->handler != NULL);
 
-               pg_log_error("processing of database \"%s\" failed: %s",
-                                        PQdb(conn), PQerrorMessage(conn));
-
-               if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
-               {
-                       PQclear(result);
-                       return false;
-               }
-       }
+       /* On failure, the handler should return NULL after freeing the result */
+       if (!slot->handler(result, slot->connection, slot->handler_context))
+               return false;
 
+       /* Ok, we have to free it ourself */
        PQclear(result);
        return true;
 }
@@ -76,15 +66,15 @@ processQueryResult(PGconn *conn, PGresult *result)
  * Note that this will block if the connection is busy.
  */
 static bool
-consumeQueryResult(PGconn *conn)
+consumeQueryResult(ParallelSlot *slot)
 {
        bool            ok = true;
        PGresult   *result;
 
-       SetCancelConn(conn);
-       while ((result = PQgetResult(conn)) != NULL)
+       SetCancelConn(slot->connection);
+       while ((result = PQgetResult(slot->connection)) != NULL)
        {
-               if (!processQueryResult(conn, result))
+               if (!processQueryResult(slot, result))
                        ok = false;
        }
        ResetCancelConn();
@@ -227,14 +217,15 @@ ParallelSlotsGetIdle(ParallelSlot *slots, int numslots)
 
                                if (result != NULL)
                                {
-                                       /* Check and discard the command result */
-                                       if (!processQueryResult(slots[i].connection, result))
+                                       /* Handle and discard the command result */
+                                       if (!processQueryResult(slots + i, result))
                                                return NULL;
                                }
                                else
                                {
                                        /* This connection has become idle */
                                        slots[i].isFree = true;
+                                       ParallelSlotClearHandler(slots + i);
                                        if (firstFree < 0)
                                                firstFree = i;
                                        break;
@@ -329,8 +320,52 @@ ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots)
 
        for (i = 0; i < numslots; i++)
        {
-               if (!consumeQueryResult((slots + i)->connection))
+               if (!consumeQueryResult(slots + i))
+                       return false;
+       }
+
+       return true;
+}
+
+/*
+ * TableCommandResultHandler
+ *
+ * ParallelSlotResultHandler for results of commands (not queries) against
+ * tables.
+ *
+ * Requires that the result status is either PGRES_COMMAND_OK or an error about
+ * a missing table.  This is useful for utilities that compile a list of tables
+ * to process and then run commands (vacuum, reindex, or whatever) against
+ * those tables, as there is a race condition between the time the list is
+ * compiled and the time the command attempts to open the table.
+ *
+ * For missing tables, logs an error but allows processing to continue.
+ *
+ * For all other errors, logs an error and terminates further processing.
+ *
+ * res: PGresult from the query executed on the slot's connection
+ * conn: connection belonging to the slot
+ * context: unused
+ */
+bool
+TableCommandResultHandler(PGresult *res, PGconn *conn, void *context)
+{
+       /*
+        * If it's an error, report it.  Errors about a missing table are harmless
+        * so we continue processing; but die for other errors.
+        */
+       if (PQresultStatus(res) != PGRES_COMMAND_OK)
+       {
+               char       *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
+
+               pg_log_error("processing of database \"%s\" failed: %s",
+                                        PQdb(conn), PQerrorMessage(conn));
+
+               if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
+               {
+                       PQclear(res);
                        return false;
+               }
        }
 
        return true;

diff --git a/src/include/fe_utils/parallel_slot.h b/src/include/fe_utils/parallel_slot.h
index 99eeb3328d60879cc9185f39de38c689e2ad9e65..8902f8d4f4816a13346a37bbd233d0d6f609e158 100644 (file)
--- a/src/include/fe_utils/parallel_slot.h
+++ b/src/include/fe_utils/parallel_slot.h
@@ -15,12 +15,39 @@
 #include "fe_utils/connect_utils.h"
 #include "libpq-fe.h"
 
+typedef bool (*ParallelSlotResultHandler) (PGresult *res, PGconn *conn,
+                                                                                  void *context);
+
 typedef struct ParallelSlot
 {
        PGconn     *connection;         /* One connection */
        bool            isFree;                 /* Is it known to be idle? */
+
+       /*
+        * Prior to issuing a command or query on 'connection', a handler callback
+        * function may optionally be registered to be invoked to process the
+        * results, and context information may optionally be registered for use
+        * by the handler.  If unset, these fields should be NULL.
+        */
+       ParallelSlotResultHandler handler;
+       void       *handler_context;
 } ParallelSlot;
 
+static inline void
+ParallelSlotSetHandler(ParallelSlot *slot, ParallelSlotResultHandler handler,
+                                          void *context)
+{
+       slot->handler = handler;
+       slot->handler_context = context;
+}
+
+static inline void
+ParallelSlotClearHandler(ParallelSlot *slot)
+{
+       slot->handler = NULL;
+       slot->handler_context = NULL;
+}
+
 extern ParallelSlot *ParallelSlotsGetIdle(ParallelSlot *slots, int numslots);
 
 extern ParallelSlot *ParallelSlotsSetup(const ConnParams *cparams,
@@ -31,5 +58,7 @@ extern void ParallelSlotsTerminate(ParallelSlot *slots, int numslots);
 
 extern bool ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots);
 
+extern bool TableCommandResultHandler(PGresult *res, PGconn *conn,
+                                                                         void *context);
 
 #endif                                                 /* PARALLEL_SLOT_H */