PostgreSQL Source Code  git master
parallel_slot.h File Reference
#include "fe_utils/connect_utils.h"
#include "libpq-fe.h"
Include dependency graph for parallel_slot.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  ParallelSlot
 
struct  ParallelSlotArray
 

Typedefs

typedef bool(* ParallelSlotResultHandler) (PGresult *res, PGconn *conn, void *context)
 
typedef struct ParallelSlot ParallelSlot
 
typedef struct ParallelSlotArray ParallelSlotArray
 

Functions

static void ParallelSlotSetHandler (ParallelSlot *slot, ParallelSlotResultHandler handler, void *context)
 
static void ParallelSlotClearHandler (ParallelSlot *slot)
 
ParallelSlotParallelSlotsGetIdle (ParallelSlotArray *slots, const char *dbname)
 
ParallelSlotArrayParallelSlotsSetup (int numslots, ConnParams *cparams, const char *progname, bool echo, const char *initcmd)
 
void ParallelSlotsAdoptConn (ParallelSlotArray *sa, PGconn *conn)
 
void ParallelSlotsTerminate (ParallelSlotArray *sa)
 
bool ParallelSlotsWaitCompletion (ParallelSlotArray *sa)
 
bool TableCommandResultHandler (PGresult *res, PGconn *conn, void *context)
 

Typedef Documentation

◆ ParallelSlot

typedef struct ParallelSlot ParallelSlot

◆ ParallelSlotArray

◆ ParallelSlotResultHandler

typedef bool(* ParallelSlotResultHandler) (PGresult *res, PGconn *conn, void *context)

Definition at line 18 of file parallel_slot.h.

Function Documentation

◆ ParallelSlotClearHandler()

static void ParallelSlotClearHandler ( ParallelSlot slot)
inlinestatic

Definition at line 55 of file parallel_slot.h.

56 {
57  slot->handler = NULL;
58  slot->handler_context = NULL;
59 }
ParallelSlotResultHandler handler
Definition: parallel_slot.h:32
void * handler_context
Definition: parallel_slot.h:33

References ParallelSlot::handler, and ParallelSlot::handler_context.

Referenced by wait_on_slots().

◆ ParallelSlotsAdoptConn()

void ParallelSlotsAdoptConn ( ParallelSlotArray sa,
PGconn conn 
)

Definition at line 429 of file parallel_slot.c.

430 {
431  int offset;
432 
433  offset = find_unconnected_slot(sa);
434  if (offset >= 0)
435  sa->slots[offset].connection = conn;
436  else
438 }
void disconnectDatabase(PGconn *conn)
static int find_unconnected_slot(const ParallelSlotArray *sa)
PGconn * conn
Definition: streamutil.c:54

References conn, disconnectDatabase(), and find_unconnected_slot().

Referenced by main(), reindex_one_database(), and vacuum_one_database().

◆ ParallelSlotSetHandler()

static void ParallelSlotSetHandler ( ParallelSlot slot,
ParallelSlotResultHandler  handler,
void *  context 
)
inlinestatic

Definition at line 47 of file parallel_slot.h.

49 {
50  slot->handler = handler;
51  slot->handler_context = context;
52 }

References ParallelSlot::handler, and ParallelSlot::handler_context.

Referenced by main(), reindex_one_database(), and vacuum_one_database().

◆ ParallelSlotsGetIdle()

ParallelSlot* ParallelSlotsGetIdle ( ParallelSlotArray slots,
const char *  dbname 
)

Definition at line 340 of file parallel_slot.c.

341 {
342  int offset;
343 
344  Assert(sa);
345  Assert(sa->numslots > 0);
346 
347  while (1)
348  {
349  /* First choice: a slot already connected to the desired database. */
350  offset = find_matching_idle_slot(sa, dbname);
351  if (offset >= 0)
352  {
353  sa->slots[offset].inUse = true;
354  return &sa->slots[offset];
355  }
356 
357  /* Second choice: a slot not connected to any database. */
358  offset = find_unconnected_slot(sa);
359  if (offset >= 0)
360  {
361  connect_slot(sa, offset, dbname);
362  sa->slots[offset].inUse = true;
363  return &sa->slots[offset];
364  }
365 
366  /* Third choice: a slot connected to the wrong database. */
367  offset = find_any_idle_slot(sa);
368  if (offset >= 0)
369  {
370  disconnectDatabase(sa->slots[offset].connection);
371  sa->slots[offset].connection = NULL;
372  connect_slot(sa, offset, dbname);
373  sa->slots[offset].inUse = true;
374  return &sa->slots[offset];
375  }
376 
377  /*
378  * Fourth choice: block until one or more slots become available. If
379  * any slots hit a fatal error, we'll find out about that here and
380  * return NULL.
381  */
382  if (!wait_on_slots(sa))
383  return NULL;
384  }
385 }
Assert(fmt[strlen(fmt) - 1] !='\n')
static bool wait_on_slots(ParallelSlotArray *sa)
static int find_matching_idle_slot(const ParallelSlotArray *sa, const char *dbname)
static void connect_slot(ParallelSlotArray *sa, int slotno, const char *dbname)
static int find_any_idle_slot(const ParallelSlotArray *sa)
char * dbname
Definition: streamutil.c:51

References Assert(), connect_slot(), dbname, disconnectDatabase(), find_any_idle_slot(), find_matching_idle_slot(), find_unconnected_slot(), and wait_on_slots().

Referenced by main(), reindex_one_database(), and vacuum_one_database().

◆ ParallelSlotsSetup()

ParallelSlotArray* ParallelSlotsSetup ( int  numslots,
ConnParams cparams,
const char *  progname,
bool  echo,
const char *  initcmd 
)

Definition at line 397 of file parallel_slot.c.

399 {
401 
402  Assert(numslots > 0);
403  Assert(cparams != NULL);
404  Assert(progname != NULL);
405 
407  numslots * sizeof(ParallelSlot));
408 
409  sa->numslots = numslots;
410  sa->cparams = cparams;
411  sa->progname = progname;
412  sa->echo = echo;
413  sa->initcmd = initcmd;
414 
415  return sa;
416 }
struct ParallelSlot ParallelSlot
Definition: parallel.h:52
#define offsetof(type, field)
Definition: c.h:727
const char * progname
Definition: main.c:50
void * palloc0(Size size)
Definition: mcxt.c:1099

References Assert(), offsetof, palloc0(), and progname.

Referenced by main(), reindex_one_database(), and vacuum_one_database().

◆ ParallelSlotsTerminate()

void ParallelSlotsTerminate ( ParallelSlotArray sa)

Definition at line 448 of file parallel_slot.c.

449 {
450  int i;
451 
452  for (i = 0; i < sa->numslots; i++)
453  {
454  PGconn *conn = sa->slots[i].connection;
455 
456  if (conn == NULL)
457  continue;
458 
460  }
461 }
int i
Definition: isn.c:73

References conn, disconnectDatabase(), and i.

Referenced by main(), reindex_one_database(), and vacuum_one_database().

◆ ParallelSlotsWaitCompletion()

bool ParallelSlotsWaitCompletion ( ParallelSlotArray sa)

Definition at line 470 of file parallel_slot.c.

471 {
472  int i;
473 
474  for (i = 0; i < sa->numslots; i++)
475  {
476  if (sa->slots[i].connection == NULL)
477  continue;
478  if (!consumeQueryResult(&sa->slots[i]))
479  return false;
480  }
481 
482  return true;
483 }
static bool consumeQueryResult(ParallelSlot *slot)
Definition: parallel_slot.c:60

References consumeQueryResult(), and i.

Referenced by main(), reindex_one_database(), and vacuum_one_database().

◆ TableCommandResultHandler()

bool TableCommandResultHandler ( PGresult res,
PGconn conn,
void *  context 
)

Definition at line 506 of file parallel_slot.c.

507 {
508  Assert(res != NULL);
509  Assert(conn != NULL);
510 
511  /*
512  * If it's an error, report it. Errors about a missing table are harmless
513  * so we continue processing; but die for other errors.
514  */
516  {
517  char *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
518 
519  pg_log_error("processing of database \"%s\" failed: %s",
521 
522  if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
523  {
524  PQclear(res);
525  return false;
526  }
527  }
528 
529  return true;
530 }
char * PQdb(const PGconn *conn)
Definition: fe-connect.c:6754
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6908
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3270
void PQclear(PGresult *res)
Definition: fe-exec.c:718
char * PQresultErrorField(const PGresult *res, int fieldcode)
Definition: fe-exec.c:3325
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:97
#define pg_log_error(...)
Definition: logging.h:106
#define ERRCODE_UNDEFINED_TABLE
Definition: parallel_slot.c:30
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57

References Assert(), conn, ERRCODE_UNDEFINED_TABLE, PG_DIAG_SQLSTATE, pg_log_error, PGRES_COMMAND_OK, PQclear(), PQdb(), PQerrorMessage(), PQresultErrorField(), PQresultStatus(), and res.

Referenced by reindex_one_database(), and vacuum_one_database().