PostgreSQL Source Code git master
task.c File Reference
#include "postgres_fe.h"
#include "common/connect.h"
#include "fe_utils/string_utils.h"
#include "pg_upgrade.h"
Include dependency graph for task.c:

Go to the source code of this file.

Data Structures

struct  UpgradeTaskStep
 
struct  UpgradeTask
 
struct  UpgradeTaskSlot
 

Typedefs

typedef struct UpgradeTaskStep UpgradeTaskStep
 
typedef enum UpgradeTaskSlotState UpgradeTaskSlotState
 
typedef struct UpgradeTaskSlot UpgradeTaskSlot
 

Enumerations

enum  UpgradeTaskSlotState { FREE , CONNECTING , RUNNING_QUERIES }
 

Functions

UpgradeTaskupgrade_task_create (void)
 
void upgrade_task_free (UpgradeTask *task)
 
void upgrade_task_add_step (UpgradeTask *task, const char *query, UpgradeTaskProcessCB process_cb, bool free_result, void *arg)
 
static void start_conn (const ClusterInfo *cluster, UpgradeTaskSlot *slot)
 
static void process_query_result (const ClusterInfo *cluster, UpgradeTaskSlot *slot, const UpgradeTask *task)
 
static void process_slot (const ClusterInfo *cluster, UpgradeTaskSlot *slot, const UpgradeTask *task)
 
static int select_loop (int maxFd, fd_set *input, fd_set *output)
 
static void wait_on_slots (UpgradeTaskSlot *slots, int numslots)
 
void upgrade_task_run (const UpgradeTask *task, const ClusterInfo *cluster)
 

Variables

static int dbs_complete
 
static int dbs_processing
 

Typedef Documentation

◆ UpgradeTaskSlot

◆ UpgradeTaskSlotState

◆ UpgradeTaskStep

Enumeration Type Documentation

◆ UpgradeTaskSlotState

Enumerator
FREE 
CONNECTING 
RUNNING_QUERIES 

Definition at line 92 of file task.c.

93{
94 FREE, /* slot available for use in a new database */
95 CONNECTING, /* waiting for connection to be established */
96 RUNNING_QUERIES, /* running/processing queries in the task */
UpgradeTaskSlotState
Definition: task.c:93
@ CONNECTING
Definition: task.c:95
@ RUNNING_QUERIES
Definition: task.c:96
@ FREE
Definition: task.c:94

Function Documentation

◆ process_query_result()

static void process_query_result ( const ClusterInfo cluster,
UpgradeTaskSlot slot,
const UpgradeTask task 
)
static

Definition at line 206 of file task.c.

208{
209 UpgradeTaskStep *steps = &task->steps[slot->step_idx];
210 UpgradeTaskProcessCB process_cb = steps->process_cb;
211 DbInfo *dbinfo = &cluster->dbarr.dbs[slot->db_idx];
212 PGresult *res = PQgetResult(slot->conn);
213
214 if (PQstatus(slot->conn) == CONNECTION_BAD ||
217 pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
218
219 /*
220 * We assume that a NULL process_cb callback function means there's
221 * nothing to process. This is primarily intended for the initial step in
222 * every task that sets a safe search_path.
223 */
224 if (process_cb)
225 (*process_cb) (dbinfo, res, steps->arg);
226
227 if (steps->free_result)
228 PQclear(res);
229}
void cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel)
Definition: cluster.c:107
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:7355
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:7418
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:2062
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3411
@ CONNECTION_BAD
Definition: libpq-fe.h:82
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:120
@ PGRES_TUPLES_OK
Definition: libpq-fe.h:123
#define pg_fatal(...)
static struct LogicalRepInfo * dbinfo
void(* UpgradeTaskProcessCB)(DbInfo *dbinfo, PGresult *res, void *arg)
Definition: pg_upgrade.h:500
int step_idx
Definition: task.c:106
int db_idx
Definition: task.c:105
PGconn * conn
Definition: task.c:107
bool free_result
Definition: task.c:74
UpgradeTaskProcessCB process_cb
Definition: task.c:73
void * arg
Definition: task.c:75
UpgradeTaskStep * steps
Definition: task.c:84

References UpgradeTaskStep::arg, cluster(), UpgradeTaskSlot::conn, CONNECTION_BAD, UpgradeTaskSlot::db_idx, dbinfo, UpgradeTaskStep::free_result, pg_fatal, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetResult(), PQresultStatus(), PQstatus(), UpgradeTaskStep::process_cb, res, UpgradeTaskSlot::step_idx, and UpgradeTask::steps.

Referenced by process_slot().

◆ process_slot()

static void process_slot ( const ClusterInfo cluster,
UpgradeTaskSlot slot,
const UpgradeTask task 
)
static

Definition at line 235 of file task.c.

236{
238
239 if (!slot->ready)
240 return;
241
242 switch (slot->state)
243 {
244 case FREE:
245
246 /*
247 * If all of the databases in the cluster have been processed or
248 * are currently being processed by other slots, we are done.
249 */
250 if (dbs_processing >= cluster->dbarr.ndbs)
251 return;
252
253 /*
254 * Claim the next database in the cluster's array and initiate a
255 * new connection.
256 */
257 slot->db_idx = dbs_processing++;
258 slot->state = CONNECTING;
259 start_conn(cluster, slot);
260
261 return;
262
263 case CONNECTING:
264
265 /* Check for connection failure. */
266 status = PQconnectPoll(slot->conn);
267 if (status == PGRES_POLLING_FAILED)
268 pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
269
270 /* Check whether the connection is still establishing. */
271 if (status != PGRES_POLLING_OK)
272 {
273 slot->select_mode = (status == PGRES_POLLING_READING);
274 return;
275 }
276
277 /*
278 * Move on to running/processing the queries in the task.
279 */
280 slot->state = RUNNING_QUERIES;
281 slot->select_mode = true; /* wait until ready for reading */
282 if (!PQsendQuery(slot->conn, task->queries->data))
283 pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
284
285 return;
286
287 case RUNNING_QUERIES:
288
289 /*
290 * Consume any available data and clear the read-ready indicator
291 * for the connection.
292 */
293 if (!PQconsumeInput(slot->conn))
294 pg_fatal("connection failure: %s", PQerrorMessage(slot->conn));
295
296 /*
297 * Process any results that are ready so that we can free up this
298 * slot for another database as soon as possible.
299 */
300 for (; slot->step_idx < task->num_steps; slot->step_idx++)
301 {
302 /* If no more results are available yet, move on. */
303 if (PQisBusy(slot->conn))
304 return;
305
306 process_query_result(cluster, slot, task);
307 }
308
309 /*
310 * If we just finished processing the result of the last step in
311 * the task, free the slot. We recursively call this function on
312 * the newly-freed slot so that we can start initiating the next
313 * connection immediately instead of waiting for the next loop
314 * through the slots.
315 */
316 dbs_complete++;
317 PQfinish(slot->conn);
318 memset(slot, 0, sizeof(UpgradeTaskSlot));
319 slot->ready = true;
320
321 process_slot(cluster, slot, task);
322
323 return;
324 }
325}
PostgresPollingStatusType PQconnectPoll(PGconn *conn)
Definition: fe-connect.c:2805
void PQfinish(PGconn *conn)
Definition: fe-connect.c:5089
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1984
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1416
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:2031
PostgresPollingStatusType
Definition: libpq-fe.h:109
@ PGRES_POLLING_OK
Definition: libpq-fe.h:113
@ PGRES_POLLING_READING
Definition: libpq-fe.h:111
@ PGRES_POLLING_FAILED
Definition: libpq-fe.h:110
bool select_mode
Definition: task.c:109
UpgradeTaskSlotState state
Definition: task.c:104
bool ready
Definition: task.c:108
PQExpBuffer queries
Definition: task.c:86
int num_steps
Definition: task.c:85
static int dbs_processing
Definition: task.c:63
static int dbs_complete
Definition: task.c:56
static void process_slot(const ClusterInfo *cluster, UpgradeTaskSlot *slot, const UpgradeTask *task)
Definition: task.c:235
static void start_conn(const ClusterInfo *cluster, UpgradeTaskSlot *slot)
Definition: task.c:174
static void process_query_result(const ClusterInfo *cluster, UpgradeTaskSlot *slot, const UpgradeTask *task)
Definition: task.c:206

References cluster(), UpgradeTaskSlot::conn, CONNECTING, PQExpBufferData::data, UpgradeTaskSlot::db_idx, dbs_complete, dbs_processing, FREE, UpgradeTask::num_steps, pg_fatal, PGRES_POLLING_FAILED, PGRES_POLLING_OK, PGRES_POLLING_READING, PQconnectPoll(), PQconsumeInput(), PQerrorMessage(), PQfinish(), PQisBusy(), PQsendQuery(), process_query_result(), process_slot(), UpgradeTask::queries, UpgradeTaskSlot::ready, RUNNING_QUERIES, UpgradeTaskSlot::select_mode, start_conn(), UpgradeTaskSlot::state, and UpgradeTaskSlot::step_idx.

Referenced by process_slot(), and upgrade_task_run().

◆ select_loop()

static int select_loop ( int  maxFd,
fd_set *  input,
fd_set *  output 
)
static

Definition at line 331 of file task.c.

332{
333 fd_set save_input = *input;
334 fd_set save_output = *output;
335
336 if (maxFd == 0)
337 return 0;
338
339 for (;;)
340 {
341 int i;
342
343 *input = save_input;
344 *output = save_output;
345
346 i = select(maxFd + 1, input, output, NULL, NULL);
347
348#ifndef WIN32
349 if (i < 0 && errno == EINTR)
350 continue;
351#else
352 if (i == SOCKET_ERROR && WSAGetLastError() == WSAEINTR)
353 continue;
354#endif
355 return i;
356 }
357}
FILE * input
FILE * output
int i
Definition: isn.c:72
#define EINTR
Definition: win32_port.h:364
#define select(n, r, w, e, timeout)
Definition: win32_port.h:503

References EINTR, i, input, output, and select.

Referenced by wait_on_slots().

◆ start_conn()

static void start_conn ( const ClusterInfo cluster,
UpgradeTaskSlot slot 
)
static

Definition at line 174 of file task.c.

175{
176 PQExpBufferData conn_opts;
177 DbInfo *dbinfo = &cluster->dbarr.dbs[slot->db_idx];
178
179 /* Build connection string with proper quoting */
180 initPQExpBuffer(&conn_opts);
181 appendPQExpBufferStr(&conn_opts, "dbname=");
182 appendConnStrVal(&conn_opts, dbinfo->db_name);
183 appendPQExpBufferStr(&conn_opts, " user=");
184 appendConnStrVal(&conn_opts, os_info.user);
185 appendPQExpBuffer(&conn_opts, " port=%d", cluster->port);
186 if (cluster->sockdir)
187 {
188 appendPQExpBufferStr(&conn_opts, " host=");
189 appendConnStrVal(&conn_opts, cluster->sockdir);
190 }
191
192 slot->conn = PQconnectStart(conn_opts.data);
193
194 if (!slot->conn)
195 pg_fatal("failed to create connection with connection string: \"%s\"",
196 conn_opts.data);
197
198 termPQExpBuffer(&conn_opts);
199}
PGconn * PQconnectStart(const char *conninfo)
Definition: fe-connect.c:887
OSInfo os_info
Definition: pg_upgrade.c:69
void initPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:90
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
Definition: pqexpbuffer.c:265
void appendPQExpBufferStr(PQExpBuffer str, const char *data)
Definition: pqexpbuffer.c:367
void termPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:129
void appendConnStrVal(PQExpBuffer buf, const char *str)
Definition: string_utils.c:545
char * user
Definition: pg_upgrade.h:344

References appendConnStrVal(), appendPQExpBuffer(), appendPQExpBufferStr(), cluster(), UpgradeTaskSlot::conn, PQExpBufferData::data, UpgradeTaskSlot::db_idx, dbinfo, initPQExpBuffer(), os_info, pg_fatal, PQconnectStart(), termPQExpBuffer(), and OSInfo::user.

Referenced by process_slot().

◆ upgrade_task_add_step()

void upgrade_task_add_step ( UpgradeTask task,
const char *  query,
UpgradeTaskProcessCB  process_cb,
bool  free_result,
void *  arg 
)

◆ upgrade_task_create()

UpgradeTask * upgrade_task_create ( void  )

Definition at line 117 of file task.c.

118{
119 UpgradeTask *task = pg_malloc0(sizeof(UpgradeTask));
120
121 task->queries = createPQExpBuffer();
122
123 /* All tasks must first set a secure search_path. */
125
126 return task;
127}
#define ALWAYS_SECURE_SEARCH_PATH_SQL
Definition: connect.h:25
void * pg_malloc0(size_t size)
Definition: fe_memutils.c:53
PQExpBuffer createPQExpBuffer(void)
Definition: pqexpbuffer.c:72
void upgrade_task_add_step(UpgradeTask *task, const char *query, UpgradeTaskProcessCB process_cb, bool free_result, void *arg)
Definition: task.c:151

References ALWAYS_SECURE_SEARCH_PATH_SQL, createPQExpBuffer(), pg_malloc0(), UpgradeTask::queries, and upgrade_task_add_step().

Referenced by check_for_data_types_usage(), check_for_incompatible_polymorphics(), check_for_isn_and_int8_passing_mismatch(), check_for_tables_with_oids(), check_for_user_defined_encoding_conversions(), check_for_user_defined_postfix_ops(), check_old_cluster_subscription_state(), get_db_rel_and_slot_infos(), get_loadable_libraries(), and report_extension_updates().

◆ upgrade_task_free()

◆ upgrade_task_run()

void upgrade_task_run ( const UpgradeTask task,
const ClusterInfo cluster 
)

Definition at line 420 of file task.c.

421{
422 int jobs = Max(1, user_opts.jobs);
423 UpgradeTaskSlot *slots = pg_malloc0(sizeof(UpgradeTaskSlot) * jobs);
424
425 dbs_complete = 0;
426 dbs_processing = 0;
427
428 /*
429 * Process every slot the first time round.
430 */
431 for (int i = 0; i < jobs; i++)
432 slots[i].ready = true;
433
434 while (dbs_complete < cluster->dbarr.ndbs)
435 {
436 for (int i = 0; i < jobs; i++)
437 process_slot(cluster, &slots[i], task);
438
439 wait_on_slots(slots, jobs);
440 }
441
442 pg_free(slots);
443}
#define Max(x, y)
Definition: c.h:955
UserOpts user_opts
Definition: option.c:30
int jobs
Definition: pg_upgrade.h:327
static void wait_on_slots(UpgradeTaskSlot *slots, int numslots)
Definition: task.c:364

References cluster(), dbs_complete, dbs_processing, i, UserOpts::jobs, Max, pg_free(), pg_malloc0(), process_slot(), user_opts, and wait_on_slots().

Referenced by check_for_data_types_usage(), check_for_incompatible_polymorphics(), check_for_isn_and_int8_passing_mismatch(), check_for_tables_with_oids(), check_for_user_defined_encoding_conversions(), check_for_user_defined_postfix_ops(), check_old_cluster_subscription_state(), get_db_rel_and_slot_infos(), get_loadable_libraries(), and report_extension_updates().

◆ wait_on_slots()

static void wait_on_slots ( UpgradeTaskSlot slots,
int  numslots 
)
static

Definition at line 364 of file task.c.

365{
366 fd_set input;
367 fd_set output;
368 int maxFd = 0;
369
370 FD_ZERO(&input);
371 FD_ZERO(&output);
372
373 for (int i = 0; i < numslots; i++)
374 {
375 /*
376 * We assume the previous call to process_slot() handled everything
377 * that was marked ready in the previous call to wait_on_slots(), if
378 * any.
379 */
380 slots[i].ready = false;
381
382 /*
383 * This function should only ever see free slots as we are finishing
384 * processing the last few databases, at which point we don't have any
385 * databases left for them to process. We'll never use these slots
386 * again, so we can safely ignore them.
387 */
388 if (slots[i].state == FREE)
389 continue;
390
391 /*
392 * Add the socket to the set.
393 */
394 slots[i].sock = PQsocket(slots[i].conn);
395 if (slots[i].sock < 0)
396 pg_fatal("invalid socket");
397 FD_SET(slots[i].sock, slots[i].select_mode ? &input : &output);
398 maxFd = Max(maxFd, slots[i].sock);
399 }
400
401 /*
402 * If we found socket(s) to wait on, wait.
403 */
404 if (select_loop(maxFd, &input, &output) == -1)
405 pg_fatal("select() failed: %m");
406
407 /*
408 * Mark which sockets appear to be ready.
409 */
410 for (int i = 0; i < numslots; i++)
411 slots[i].ready |= (FD_ISSET(slots[i].sock, &input) ||
412 FD_ISSET(slots[i].sock, &output));
413}
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:7444
PGconn * conn
Definition: streamutil.c:53
Definition: regguts.h:323
static int select_loop(int maxFd, fd_set *input, fd_set *output)
Definition: task.c:331

References conn, FREE, i, input, Max, output, pg_fatal, PQsocket(), UpgradeTaskSlot::ready, select_loop(), and UpgradeTaskSlot::sock.

Referenced by upgrade_task_run().

Variable Documentation

◆ dbs_complete

int dbs_complete
static

Definition at line 56 of file task.c.

Referenced by process_slot(), and upgrade_task_run().

◆ dbs_processing

int dbs_processing
static

Definition at line 63 of file task.c.

Referenced by process_slot(), and upgrade_task_run().