Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * parallel_slot.c
4 : : * Parallel support for front-end parallel database connections
5 : : *
6 : : *
7 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
8 : : * Portions Copyright (c) 1994, Regents of the University of California
9 : : *
10 : : * src/fe_utils/parallel_slot.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #if defined(WIN32) && FD_SETSIZE < 1024
16 : : #error FD_SETSIZE needs to have been increased
17 : : #endif
18 : :
19 : : #include "postgres_fe.h"
20 : :
21 : : #include <sys/select.h>
22 : :
23 : : #include "common/logging.h"
24 : : #include "fe_utils/cancel.h"
25 : : #include "fe_utils/parallel_slot.h"
26 : : #include "fe_utils/query_utils.h"
27 : :
28 : : #define ERRCODE_UNDEFINED_TABLE "42P01"
29 : :
30 : : static int select_loop(int maxFd, fd_set *workerset);
31 : : static bool processQueryResult(ParallelSlot *slot, PGresult *result);
32 : :
33 : : /*
34 : : * Process (and delete) a query result. Returns true if there's no problem,
35 : : * false otherwise. It's up to the handler to decide what constitutes a
36 : : * problem.
37 : : */
38 : : static bool
39 : 14627 : processQueryResult(ParallelSlot *slot, PGresult *result)
40 : : {
41 : : Assert(slot->handler != NULL);
42 : :
43 : : /* On failure, the handler should return NULL after freeing the result */
44 [ + + ]: 14627 : if (!slot->handler(result, slot->connection, slot->handler_context))
45 : 6 : return false;
46 : :
47 : : /* Ok, we have to free it ourself */
48 : 14621 : PQclear(result);
49 : 14621 : return true;
50 : : }
51 : :
52 : : /*
53 : : * Consume all the results generated for the given connection until
54 : : * nothing remains. If at least one error is encountered, return false.
55 : : * Note that this will block if the connection is busy.
56 : : */
57 : : static bool
58 : 267 : consumeQueryResult(ParallelSlot *slot)
59 : : {
60 : 267 : bool ok = true;
61 : : PGresult *result;
62 : :
63 : 267 : SetCancelConn(slot->connection);
64 [ + + ]: 533 : while ((result = PQgetResult(slot->connection)) != NULL)
65 : : {
66 [ + + ]: 266 : if (!processQueryResult(slot, result))
67 : 6 : ok = false;
68 : : }
69 : 267 : ResetCancelConn();
70 : 267 : return ok;
71 : : }
72 : :
73 : : /*
74 : : * Wait until a file descriptor from the given set becomes readable.
75 : : *
76 : : * Returns the number of ready descriptors, or -1 on failure (including
77 : : * getting a cancel request).
78 : : */
79 : : static int
80 : 14391 : select_loop(int maxFd, fd_set *workerset)
81 : : {
82 : : int i;
83 : 14391 : fd_set saveSet = *workerset;
84 : :
85 [ - + ]: 14391 : if (CancelRequested)
86 : 0 : return -1;
87 : :
88 : : for (;;)
89 : 0 : {
90 : : /*
91 : : * On Windows, we need to check once in a while for cancel requests;
92 : : * on other platforms we rely on select() returning when interrupted.
93 : : */
94 : : struct timeval *tvp;
95 : : #ifdef WIN32
96 : : struct timeval tv = {0, 1000000};
97 : :
98 : : tvp = &tv;
99 : : #else
100 : 14391 : tvp = NULL;
101 : : #endif
102 : :
103 : 14391 : *workerset = saveSet;
104 : 14391 : i = select(maxFd + 1, workerset, NULL, NULL, tvp);
105 : :
106 : : #ifdef WIN32
107 : : if (i == SOCKET_ERROR)
108 : : {
109 : : i = -1;
110 : :
111 : : if (WSAGetLastError() == WSAEINTR)
112 : : errno = EINTR;
113 : : }
114 : : #endif
115 : :
116 [ - + - - ]: 14391 : if (i < 0 && errno == EINTR)
117 : 0 : continue; /* ignore this */
118 [ + - - + ]: 14391 : if (i < 0 || CancelRequested)
119 : 0 : return -1; /* but not this */
120 [ - + ]: 14391 : if (i == 0)
121 : 0 : continue; /* timeout (Win32 only) */
122 : 14391 : break;
123 : : }
124 : :
125 : 14391 : return i;
126 : : }
127 : :
128 : : /*
129 : : * Return the offset of a suitable idle slot, or -1 if none are available. If
130 : : * the given dbname is not null, only idle slots connected to the given
131 : : * database are considered suitable, otherwise all idle connected slots are
132 : : * considered suitable.
133 : : */
134 : : static int
135 : 29018 : find_matching_idle_slot(const ParallelSlotArray *sa, const char *dbname)
136 : : {
137 : : int i;
138 : :
139 [ + + ]: 43531 : for (i = 0; i < sa->numslots; i++)
140 : : {
141 [ + + ]: 29119 : if (sa->slots[i].inUse)
142 : 14492 : continue;
143 : :
144 [ + + ]: 14627 : if (sa->slots[i].connection == NULL)
145 : 7 : continue;
146 : :
147 [ + + ]: 14620 : if (dbname == NULL ||
148 [ + + ]: 8199 : strcmp(PQdb(sa->slots[i].connection), dbname) == 0)
149 : 14606 : return i;
150 : : }
151 : 14412 : return -1;
152 : : }
153 : :
154 : : /*
155 : : * Return the offset of the first slot without a database connection, or -1 if
156 : : * all slots are connected.
157 : : */
158 : : static int
159 : 14599 : find_unconnected_slot(const ParallelSlotArray *sa)
160 : : {
161 : : int i;
162 : :
163 [ + + ]: 29073 : for (i = 0; i < sa->numslots; i++)
164 : : {
165 [ + + ]: 14668 : if (sa->slots[i].inUse)
166 : 14460 : continue;
167 : :
168 [ + + ]: 208 : if (sa->slots[i].connection == NULL)
169 : 194 : return i;
170 : : }
171 : :
172 : 14405 : return -1;
173 : : }
174 : :
175 : : /*
176 : : * Return the offset of the first idle slot, or -1 if all slots are busy.
177 : : */
178 : : static int
179 : 14405 : find_any_idle_slot(const ParallelSlotArray *sa)
180 : : {
181 : : int i;
182 : :
183 [ + + ]: 28861 : for (i = 0; i < sa->numslots; i++)
184 [ + + ]: 14470 : if (!sa->slots[i].inUse)
185 : 14 : return i;
186 : :
187 : 14391 : return -1;
188 : : }
189 : :
190 : : /*
191 : : * Wait for any slot's connection to have query results, consume the results,
192 : : * and update the slot's status as appropriate. Returns true on success,
193 : : * false on cancellation, on error, or if no slots are connected.
194 : : */
195 : : static bool
196 : 14391 : wait_on_slots(ParallelSlotArray *sa)
197 : : {
198 : : int i;
199 : : fd_set slotset;
200 : 14391 : int maxFd = 0;
201 : 14391 : PGconn *cancelconn = NULL;
202 : :
203 : : /* We must reconstruct the fd_set for each call to select_loop */
204 [ + + ]: 244647 : FD_ZERO(&slotset);
205 : :
206 [ + + ]: 28847 : for (i = 0; i < sa->numslots; i++)
207 : : {
208 : : int sock;
209 : :
210 : : /* We shouldn't get here if we still have slots without connections */
211 : : Assert(sa->slots[i].connection != NULL);
212 : :
213 : 14456 : sock = PQsocket(sa->slots[i].connection);
214 : :
215 : : /*
216 : : * We don't really expect any connections to lose their sockets after
217 : : * startup, but just in case, cope by ignoring them.
218 : : */
219 [ - + ]: 14456 : if (sock < 0)
220 : 0 : continue;
221 : :
222 : : /* Keep track of the first valid connection we see. */
223 [ + + ]: 14456 : if (cancelconn == NULL)
224 : 14391 : cancelconn = sa->slots[i].connection;
225 : :
226 : 14456 : FD_SET(sock, &slotset);
227 [ + - ]: 14456 : if (sock > maxFd)
228 : 14456 : maxFd = sock;
229 : : }
230 : :
231 : : /*
232 : : * If we get this far with no valid connections, processing cannot
233 : : * continue.
234 : : */
235 [ - + ]: 14391 : if (cancelconn == NULL)
236 : 0 : return false;
237 : :
238 : 14391 : SetCancelConn(cancelconn);
239 : 14391 : i = select_loop(maxFd, &slotset);
240 : 14391 : ResetCancelConn();
241 : :
242 : : /* failure? */
243 [ - + ]: 14391 : if (i < 0)
244 : 0 : return false;
245 : :
246 [ + + ]: 28847 : for (i = 0; i < sa->numslots; i++)
247 : : {
248 : : int sock;
249 : :
250 : 14456 : sock = PQsocket(sa->slots[i].connection);
251 : :
252 [ + - + + ]: 14456 : if (sock >= 0 && FD_ISSET(sock, &slotset))
253 : : {
254 : : /* select() says input is available, so consume it */
255 : 14403 : PQconsumeInput(sa->slots[i].connection);
256 : : }
257 : :
258 : : /* Collect result(s) as long as any are available */
259 [ + + ]: 28817 : while (!PQisBusy(sa->slots[i].connection))
260 : : {
261 : 28722 : PGresult *result = PQgetResult(sa->slots[i].connection);
262 : :
263 [ + + ]: 28722 : if (result != NULL)
264 : : {
265 : : /* Handle and discard the command result */
266 [ - + ]: 14361 : if (!processQueryResult(&sa->slots[i], result))
267 : 0 : return false;
268 : : }
269 : : else
270 : : {
271 : : /* This connection has become idle */
272 : 14361 : ParallelSlotSetIdle(&sa->slots[i]);
273 : 14361 : break;
274 : : }
275 : : }
276 : : }
277 : 14391 : return true;
278 : : }
279 : :
280 : : /*
281 : : * Open a new database connection using the stored connection parameters and
282 : : * optionally a given dbname if not null, execute the stored initial command if
283 : : * any, and associate the new connection with the given slot.
284 : : */
285 : : static void
286 : 21 : connect_slot(ParallelSlotArray *sa, int slotno, const char *dbname)
287 : : {
288 : : const char *old_override;
289 : 21 : ParallelSlot *slot = &sa->slots[slotno];
290 : :
291 : 21 : old_override = sa->cparams->override_dbname;
292 [ + + ]: 21 : if (dbname)
293 : 17 : sa->cparams->override_dbname = dbname;
294 : 21 : slot->connection = connectDatabase(sa->cparams, sa->progname, sa->echo, false, true);
295 : 21 : sa->cparams->override_dbname = old_override;
296 : :
297 : : /*
298 : : * POSIX defines FD_SETSIZE as the highest file descriptor acceptable to
299 : : * FD_SET() and allied macros. Windows defines it as a ceiling on the
300 : : * count of file descriptors in the set, not a ceiling on the value of
301 : : * each file descriptor; see
302 : : * https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-select
303 : : * and
304 : : * https://learn.microsoft.com/en-us/windows/win32/api/winsock/ns-winsock-fd_set.
305 : : * We can't ignore that, because Windows starts file descriptors at a
306 : : * higher value, delays reuse, and skips values. With less than ten
307 : : * concurrent file descriptors, opened and closed rapidly, one can reach
308 : : * file descriptor 1024.
309 : : *
310 : : * Doing a hard exit here is a bit grotty, but it doesn't seem worth
311 : : * complicating the API to make it less grotty.
312 : : */
313 : : #ifdef WIN32
314 : : if (slotno >= FD_SETSIZE)
315 : : {
316 : : pg_log_error("too many jobs for this platform: %d", slotno);
317 : : exit(1);
318 : : }
319 : : #else
320 : : {
321 : 21 : int fd = PQsocket(slot->connection);
322 : :
323 [ - + ]: 21 : if (fd >= FD_SETSIZE)
324 : : {
325 : 0 : pg_log_error("socket file descriptor out of range for select(): %d",
326 : : fd);
327 : 0 : pg_log_error_hint("Try fewer jobs.");
328 : 0 : exit(1);
329 : : }
330 : : }
331 : : #endif
332 : :
333 : : /* Setup the connection using the supplied command, if any. */
334 [ - + ]: 21 : if (sa->initcmd)
335 : 0 : executeCommand(slot->connection, sa->initcmd, sa->echo);
336 : 21 : }
337 : :
338 : : /*
339 : : * ParallelSlotsGetIdle
340 : : * Return a connection slot that is ready to execute a command.
341 : : *
342 : : * The slot returned is chosen as follows:
343 : : *
344 : : * If any idle slot already has an open connection, and if either dbname is
345 : : * null or the existing connection is to the given database, that slot will be
346 : : * returned allowing the connection to be reused.
347 : : *
348 : : * Otherwise, if any idle slot is not yet connected to any database, the slot
349 : : * will be returned with its connection opened using the stored cparams and
350 : : * optionally the given dbname if not null.
351 : : *
352 : : * Otherwise, if any idle slot exists, an idle slot will be chosen and returned
353 : : * after having its connection disconnected and reconnected using the stored
354 : : * cparams and optionally the given dbname if not null.
355 : : *
356 : : * Otherwise, if any slots have connections that are busy, we loop on select()
357 : : * until one socket becomes available. When this happens, we read the whole
358 : : * set and mark as free all sockets that become available. We then select a
359 : : * slot using the same rules as above.
360 : : *
361 : : * Otherwise, we cannot return a slot, which is an error, and NULL is returned.
362 : : *
363 : : * For any connection created, if the stored initcmd is not null, it will be
364 : : * executed as a command on the newly formed connection before the slot is
365 : : * returned.
366 : : *
367 : : * If an error occurs, NULL is returned.
368 : : */
369 : : ParallelSlot *
370 : 14627 : ParallelSlotsGetIdle(ParallelSlotArray *sa, const char *dbname)
371 : : {
372 : : int offset;
373 : :
374 : : Assert(sa);
375 : : Assert(sa->numslots > 0);
376 : :
377 : : while (1)
378 : : {
379 : : /* First choice: a slot already connected to the desired database. */
380 : 29018 : offset = find_matching_idle_slot(sa, dbname);
381 [ + + ]: 29018 : if (offset >= 0)
382 : : {
383 : 14606 : sa->slots[offset].inUse = true;
384 : 14606 : return &sa->slots[offset];
385 : : }
386 : :
387 : : /* Second choice: a slot not connected to any database. */
388 : 14412 : offset = find_unconnected_slot(sa);
389 [ + + ]: 14412 : if (offset >= 0)
390 : : {
391 : 7 : connect_slot(sa, offset, dbname);
392 : 7 : sa->slots[offset].inUse = true;
393 : 7 : return &sa->slots[offset];
394 : : }
395 : :
396 : : /* Third choice: a slot connected to the wrong database. */
397 : 14405 : offset = find_any_idle_slot(sa);
398 [ + + ]: 14405 : if (offset >= 0)
399 : : {
400 : 14 : disconnectDatabase(sa->slots[offset].connection);
401 : 14 : sa->slots[offset].connection = NULL;
402 : 14 : connect_slot(sa, offset, dbname);
403 : 14 : sa->slots[offset].inUse = true;
404 : 14 : return &sa->slots[offset];
405 : : }
406 : :
407 : : /*
408 : : * Fourth choice: block until one or more slots become available. If
409 : : * any slots hit a fatal error, we'll find out about that here and
410 : : * return NULL.
411 : : */
412 [ - + ]: 14391 : if (!wait_on_slots(sa))
413 : 0 : return NULL;
414 : : }
415 : : }
416 : :
417 : : /*
418 : : * ParallelSlotsSetup
419 : : * Prepare a set of parallel slots but do not connect to any database.
420 : : *
421 : : * This creates and initializes a set of slots, marking all parallel slots as
422 : : * free and ready to use. Establishing connections is delayed until requesting
423 : : * a free slot. The cparams, progname, echo, and initcmd are stored for later
424 : : * use and must remain valid for the lifetime of the returned array.
425 : : */
426 : : ParallelSlotArray *
427 : 190 : ParallelSlotsSetup(int numslots, ConnParams *cparams, const char *progname,
428 : : bool echo, const char *initcmd)
429 : : {
430 : : ParallelSlotArray *sa;
431 : :
432 : : Assert(numslots > 0);
433 : : Assert(cparams != NULL);
434 : : Assert(progname != NULL);
435 : :
436 : 190 : sa = (ParallelSlotArray *) palloc0(offsetof(ParallelSlotArray, slots) +
437 : 190 : numslots * sizeof(ParallelSlot));
438 : :
439 : 190 : sa->numslots = numslots;
440 : 190 : sa->cparams = cparams;
441 : 190 : sa->progname = progname;
442 : 190 : sa->echo = echo;
443 : 190 : sa->initcmd = initcmd;
444 : :
445 : 190 : return sa;
446 : : }
447 : :
448 : : /*
449 : : * ParallelSlotsAdoptConn
450 : : * Assign an open connection to the slots array for reuse.
451 : : *
452 : : * This turns over ownership of an open connection to a slots array. The
453 : : * caller should not further use or close the connection. All the connection's
454 : : * parameters (user, host, port, etc.) except possibly dbname should match
455 : : * those of the slots array's cparams, as given in ParallelSlotsSetup. If
456 : : * these parameters differ, subsequent behavior is undefined.
457 : : */
458 : : void
459 : 187 : ParallelSlotsAdoptConn(ParallelSlotArray *sa, PGconn *conn)
460 : : {
461 : : int offset;
462 : :
463 : 187 : offset = find_unconnected_slot(sa);
464 [ + - ]: 187 : if (offset >= 0)
465 : 187 : sa->slots[offset].connection = conn;
466 : : else
467 : 0 : disconnectDatabase(conn);
468 : 187 : }
469 : :
470 : : /*
471 : : * ParallelSlotsTerminate
472 : : * Clean up a set of parallel slots
473 : : *
474 : : * Iterate through all connections in a given set of ParallelSlots and
475 : : * terminate all connections.
476 : : */
477 : : void
478 : 190 : ParallelSlotsTerminate(ParallelSlotArray *sa)
479 : : {
480 : : int i;
481 : :
482 [ + + ]: 384 : for (i = 0; i < sa->numslots; i++)
483 : : {
484 : 194 : PGconn *conn = sa->slots[i].connection;
485 : :
486 [ - + ]: 194 : if (conn == NULL)
487 : 0 : continue;
488 : :
489 : 194 : disconnectDatabase(conn);
490 : : }
491 : 190 : }
492 : :
493 : : /*
494 : : * ParallelSlotsWaitCompletion
495 : : *
496 : : * Wait for all connections to finish, returning false if at least one
497 : : * error has been found on the way.
498 : : */
499 : : bool
500 : 262 : ParallelSlotsWaitCompletion(ParallelSlotArray *sa)
501 : : {
502 : : int i;
503 : :
504 [ + + ]: 523 : for (i = 0; i < sa->numslots; i++)
505 : : {
506 [ - + ]: 267 : if (sa->slots[i].connection == NULL)
507 : 0 : continue;
508 [ + + ]: 267 : if (!consumeQueryResult(&sa->slots[i]))
509 : 6 : return false;
510 : : /* Mark connection as idle */
511 : 261 : ParallelSlotSetIdle(&sa->slots[i]);
512 : : }
513 : :
514 : 256 : return true;
515 : : }
516 : :
517 : : /*
518 : : * TableCommandResultHandler
519 : : *
520 : : * ParallelSlotResultHandler for results of commands (not queries) against
521 : : * tables.
522 : : *
523 : : * Requires that the result status is either PGRES_COMMAND_OK or an error about
524 : : * a missing table. This is useful for utilities that compile a list of tables
525 : : * to process and then run commands (vacuum, reindex, or whatever) against
526 : : * those tables, as there is a race condition between the time the list is
527 : : * compiled and the time the command attempts to open the table.
528 : : *
529 : : * For missing tables, logs an error but allows processing to continue.
530 : : *
531 : : * For all other errors, logs an error and terminates further processing.
532 : : *
533 : : * res: PGresult from the query executed on the slot's connection
534 : : * conn: connection belonging to the slot
535 : : * context: unused
536 : : */
537 : : bool
538 : 6425 : TableCommandResultHandler(PGresult *res, PGconn *conn, void *context)
539 : : {
540 : : Assert(res != NULL);
541 : : Assert(conn != NULL);
542 : :
543 : : /*
544 : : * If it's an error, report it. Errors about a missing table are harmless
545 : : * so we continue processing; but die for other errors.
546 : : */
547 [ + + ]: 6425 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
548 : : {
549 : 6 : char *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
550 : :
551 : 6 : pg_log_error("processing of database \"%s\" failed: %s",
552 : : PQdb(conn), PQerrorMessage(conn));
553 : :
554 [ + - + - ]: 6 : if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
555 : : {
556 : 6 : PQclear(res);
557 : 6 : return false;
558 : : }
559 : : }
560 : :
561 : 6419 : return true;
562 : : }
|
