granicus.if.org Git - postgresql/blob - src/bin/scripts/vacuumdb.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * vacuumdb
   4  *
   5  * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
   6  * Portions Copyright (c) 1994, Regents of the University of California
   7  *
   8  * src/bin/scripts/vacuumdb.c
   9  *
  10  *-------------------------------------------------------------------------
  11  */
  12
  13 #include "postgres_fe.h"
  14
  15 #include "common.h"
  16 #include "dumputils.h"
  17
  18
  19 #define ERRCODE_UNDEFINED_TABLE  "42P01"
  20
  21 /* Parallel vacuuming stuff */
  22 typedef struct ParallelSlot
  23 {
  24         PGconn     *connection;
  25         pgsocket        sock;
  26         bool            isFree;
  27 } ParallelSlot;
  28
  29 /* vacuum options controlled by user flags */
  30 typedef struct vacuumingOptions
  31 {
  32         bool            analyze_only;
  33         bool            verbose;
  34         bool            and_analyze;
  35         bool            full;
  36         bool            freeze;
  37 } vacuumingOptions;
  38
  39
  40 static void vacuum_one_database(const char *dbname, vacuumingOptions *vacopts,
  41                                         int stage,
  42                                         SimpleStringList *tables,
  43                                         const char *host, const char *port,
  44                                         const char *username, enum trivalue prompt_password,
  45                                         int concurrentCons,
  46                                         const char *progname, bool echo, bool quiet);
  47
  48 static void vacuum_all_databases(vacuumingOptions *vacopts,
  49                                          bool analyze_in_stages,
  50                                          const char *maintenance_db,
  51                                          const char *host, const char *port,
  52                                          const char *username, enum trivalue prompt_password,
  53                                          int concurrentCons,
  54                                          const char *progname, bool echo, bool quiet);
  55
  56 static void prepare_vacuum_command(PQExpBuffer sql, PGconn *conn,
  57                                            vacuumingOptions *vacopts, const char *table);
  58
  59 static void run_vacuum_command(PGconn *conn, const char *sql, bool echo,
  60                                    const char *dbname, const char *table,
  61                                    const char *progname, bool async);
  62
  63 static ParallelSlot *GetIdleSlot(ParallelSlot slots[], int numslots,
  64                         const char *dbname, const char *progname);
  65
  66 static bool GetQueryResult(PGconn *conn, const char *dbname,
  67                            const char *progname);
  68
  69 static void DisconnectDatabase(ParallelSlot *slot);
  70
  71 static int      select_loop(int maxFd, fd_set *workerset, bool *aborting);
  72
  73 static void init_slot(ParallelSlot *slot, PGconn *conn);
  74
  75 static void help(const char *progname);
  76
  77 /* For analyze-in-stages mode */
  78 #define ANALYZE_NO_STAGE        -1
  79 #define ANALYZE_NUM_STAGES      3
  80
  81
  82 int
  83 main(int argc, char *argv[])
  84 {
  85         static struct option long_options[] = {
  86                 {"host", required_argument, NULL, 'h'},
  87                 {"port", required_argument, NULL, 'p'},
  88                 {"username", required_argument, NULL, 'U'},
  89                 {"no-password", no_argument, NULL, 'w'},
  90                 {"password", no_argument, NULL, 'W'},
  91                 {"echo", no_argument, NULL, 'e'},
  92                 {"quiet", no_argument, NULL, 'q'},
  93                 {"dbname", required_argument, NULL, 'd'},
  94                 {"analyze", no_argument, NULL, 'z'},
  95                 {"analyze-only", no_argument, NULL, 'Z'},
  96                 {"freeze", no_argument, NULL, 'F'},
  97                 {"all", no_argument, NULL, 'a'},
  98                 {"table", required_argument, NULL, 't'},
  99                 {"full", no_argument, NULL, 'f'},
 100                 {"verbose", no_argument, NULL, 'v'},
 101                 {"jobs", required_argument, NULL, 'j'},
 102                 {"maintenance-db", required_argument, NULL, 2},
 103                 {"analyze-in-stages", no_argument, NULL, 3},
 104                 {NULL, 0, NULL, 0}
 105         };
 106
 107         const char *progname;
 108         int                     optindex;
 109         int                     c;
 110         const char *dbname = NULL;
 111         const char *maintenance_db = NULL;
 112         char       *host = NULL;
 113         char       *port = NULL;
 114         char       *username = NULL;
 115         enum trivalue prompt_password = TRI_DEFAULT;
 116         bool            echo = false;
 117         bool            quiet = false;
 118         vacuumingOptions vacopts;
 119         bool            analyze_in_stages = false;
 120         bool            alldb = false;
 121         SimpleStringList tables = {NULL, NULL};
 122         int                     concurrentCons = 1;
 123         int                     tbl_count = 0;
 124
 125         /* initialize options to all false */
 126         memset(&vacopts, 0, sizeof(vacopts));
 127
 128         progname = get_progname(argv[0]);
 129
 130         set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pgscripts"));
 131
 132         handle_help_version_opts(argc, argv, "vacuumdb", help);
 133
 134         while ((c = getopt_long(argc, argv, "h:p:U:wWeqd:zZFat:fvj:", long_options, &optindex)) != -1)
 135         {
 136                 switch (c)
 137                 {
 138                         case 'h':
 139                                 host = pg_strdup(optarg);
 140                                 break;
 141                         case 'p':
 142                                 port = pg_strdup(optarg);
 143                                 break;
 144                         case 'U':
 145                                 username = pg_strdup(optarg);
 146                                 break;
 147                         case 'w':
 148                                 prompt_password = TRI_NO;
 149                                 break;
 150                         case 'W':
 151                                 prompt_password = TRI_YES;
 152                                 break;
 153                         case 'e':
 154                                 echo = true;
 155                                 break;
 156                         case 'q':
 157                                 quiet = true;
 158                                 break;
 159                         case 'd':
 160                                 dbname = pg_strdup(optarg);
 161                                 break;
 162                         case 'z':
 163                                 vacopts.and_analyze = true;
 164                                 break;
 165                         case 'Z':
 166                                 vacopts.analyze_only = true;
 167                                 break;
 168                         case 'F':
 169                                 vacopts.freeze = true;
 170                                 break;
 171                         case 'a':
 172                                 alldb = true;
 173                                 break;
 174                         case 't':
 175                                 {
 176                                         simple_string_list_append(&tables, optarg);
 177                                         tbl_count++;
 178                                         break;
 179                                 }
 180                         case 'f':
 181                                 vacopts.full = true;
 182                                 break;
 183                         case 'v':
 184                                 vacopts.verbose = true;
 185                                 break;
 186                         case 'j':
 187                                 concurrentCons = atoi(optarg);
 188                                 if (concurrentCons <= 0)
 189                                 {
 190                                         fprintf(stderr, _("%s: number of parallel \"jobs\" must be at least 1\n"),
 191                                                         progname);
 192                                         exit(1);
 193                                 }
 194                                 if (concurrentCons > FD_SETSIZE - 1)
 195                                 {
 196                                         fprintf(stderr, _("%s: too many parallel jobs requested (maximum: %d)\n"),
 197                                                         progname, FD_SETSIZE - 1);
 198                                         exit(1);
 199                                 }
 200                                 break;
 201                         case 2:
 202                                 maintenance_db = pg_strdup(optarg);
 203                                 break;
 204                         case 3:
 205                                 analyze_in_stages = vacopts.analyze_only = true;
 206                                 break;
 207                         default:
 208                                 fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
 209                                 exit(1);
 210                 }
 211         }
 212
 213         /*
 214          * Non-option argument specifies database name as long as it wasn't
 215          * already specified with -d / --dbname
 216          */
 217         if (optind < argc && dbname == NULL)
 218         {
 219                 dbname = argv[optind];
 220                 optind++;
 221         }
 222
 223         if (optind < argc)
 224         {
 225                 fprintf(stderr, _("%s: too many command-line arguments (first is \"%s\")\n"),
 226                                 progname, argv[optind]);
 227                 fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
 228                 exit(1);
 229         }
 230
 231         if (vacopts.analyze_only)
 232         {
 233                 if (vacopts.full)
 234                 {
 235                         fprintf(stderr, _("%s: cannot use the \"%s\" option when performing only analyze\n"),
 236                                         progname, "full");
 237                         exit(1);
 238                 }
 239                 if (vacopts.freeze)
 240                 {
 241                         fprintf(stderr, _("%s: cannot use the \"%s\" option when performing only analyze\n"),
 242                                         progname, "freeze");
 243                         exit(1);
 244                 }
 245                 /* allow 'and_analyze' with 'analyze_only' */
 246         }
 247
 248         setup_cancel_handler();
 249
 250         /* Avoid opening extra connections. */
 251         if (tbl_count && (concurrentCons > tbl_count))
 252                 concurrentCons = tbl_count;
 253
 254         if (alldb)
 255         {
 256                 if (dbname)
 257                 {
 258                         fprintf(stderr, _("%s: cannot vacuum all databases and a specific one at the same time\n"),
 259                                         progname);
 260                         exit(1);
 261                 }
 262                 if (tables.head != NULL)
 263                 {
 264                         fprintf(stderr, _("%s: cannot vacuum specific table(s) in all databases\n"),
 265                                         progname);
 266                         exit(1);
 267                 }
 268
 269                 vacuum_all_databases(&vacopts,
 270                                                          analyze_in_stages,
 271                                                          maintenance_db,
 272                                                          host, port, username, prompt_password,
 273                                                          concurrentCons,
 274                                                          progname, echo, quiet);
 275         }
 276         else
 277         {
 278                 if (dbname == NULL)
 279                 {
 280                         if (getenv("PGDATABASE"))
 281                                 dbname = getenv("PGDATABASE");
 282                         else if (getenv("PGUSER"))
 283                                 dbname = getenv("PGUSER");
 284                         else
 285                                 dbname = get_user_name_or_exit(progname);
 286                 }
 287
 288                 if (analyze_in_stages)
 289                 {
 290                         int                     stage;
 291
 292                         for (stage = 0; stage < ANALYZE_NUM_STAGES; stage++)
 293                         {
 294                                 vacuum_one_database(dbname, &vacopts,
 295                                                                         stage,
 296                                                                         &tables,
 297                                                                         host, port, username, prompt_password,
 298                                                                         concurrentCons,
 299                                                                         progname, echo, quiet);
 300                         }
 301                 }
 302                 else
 303                         vacuum_one_database(dbname, &vacopts,
 304                                                                 ANALYZE_NO_STAGE,
 305                                                                 &tables,
 306                                                                 host, port, username, prompt_password,
 307                                                                 concurrentCons,
 308                                                                 progname, echo, quiet);
 309         }
 310
 311         exit(0);
 312 }
 313
 314 /*
 315  * vacuum_one_database
 316  *
 317  * Process tables in the given database.  If the 'tables' list is empty,
 318  * process all tables in the database.
 319  *
 320  * Note that this function is only concerned with running exactly one stage
 321  * when in analyze-in-stages mode; caller must iterate on us if necessary.
 322  *
 323  * If concurrentCons is > 1, multiple connections are used to vacuum tables
 324  * in parallel.  In this case and if the table list is empty, we first obtain
 325  * a list of tables from the database.
 326  */
 327 static void
 328 vacuum_one_database(const char *dbname, vacuumingOptions *vacopts,
 329                                         int stage,
 330                                         SimpleStringList *tables,
 331                                         const char *host, const char *port,
 332                                         const char *username, enum trivalue prompt_password,
 333                                         int concurrentCons,
 334                                         const char *progname, bool echo, bool quiet)
 335 {
 336         PQExpBufferData sql;
 337         PGconn     *conn;
 338         SimpleStringListCell *cell;
 339         ParallelSlot *slots = NULL;
 340         SimpleStringList dbtables = {NULL, NULL};
 341         int                     i;
 342         bool            failed = false;
 343         bool            parallel = concurrentCons > 1;
 344         const char *stage_commands[] = {
 345                 "SET default_statistics_target=1; SET vacuum_cost_delay=0;",
 346                 "SET default_statistics_target=10; RESET vacuum_cost_delay;",
 347                 "RESET default_statistics_target;"
 348         };
 349         const char *stage_messages[] = {
 350                 gettext_noop("Generating minimal optimizer statistics (1 target)"),
 351                 gettext_noop("Generating medium optimizer statistics (10 targets)"),
 352                 gettext_noop("Generating default (full) optimizer statistics")
 353         };
 354
 355         Assert(stage == ANALYZE_NO_STAGE ||
 356                    (stage >= 0 && stage < ANALYZE_NUM_STAGES));
 357
 358         if (!quiet)
 359         {
 360                 if (stage != ANALYZE_NO_STAGE)
 361                         printf(_("%s: processing database \"%s\": %s\n"), progname, dbname,
 362                                    stage_messages[stage]);
 363                 else
 364                         printf(_("%s: vacuuming database \"%s\"\n"), progname, dbname);
 365                 fflush(stdout);
 366         }
 367
 368         conn = connectDatabase(dbname, host, port, username, prompt_password,
 369                                                    progname, false);
 370
 371         initPQExpBuffer(&sql);
 372
 373         /*
 374          * If a table list is not provided and we're using multiple connections,
 375          * prepare the list of tables by querying the catalogs.
 376          */
 377         if (parallel && (!tables || !tables->head))
 378         {
 379                 PQExpBufferData buf;
 380                 PGresult   *res;
 381                 int                     ntups;
 382                 int                     i;
 383
 384                 initPQExpBuffer(&buf);
 385
 386                 res = executeQuery(conn,
 387                         "SELECT c.relname, ns.nspname FROM pg_class c, pg_namespace ns\n"
 388                          " WHERE relkind IN (\'r\', \'m\') AND c.relnamespace = ns.oid\n"
 389                                                    " ORDER BY c.relpages DESC;",
 390                                                    progname, echo);
 391
 392                 ntups = PQntuples(res);
 393                 for (i = 0; i < ntups; i++)
 394                 {
 395                         appendPQExpBufferStr(&buf,
 396                                                                  fmtQualifiedId(PQserverVersion(conn),
 397                                                                                                 PQgetvalue(res, i, 1),
 398                                                                                                 PQgetvalue(res, i, 0)));
 399
 400                         simple_string_list_append(&dbtables, buf.data);
 401                         resetPQExpBuffer(&buf);
 402                 }
 403
 404                 termPQExpBuffer(&buf);
 405                 tables = &dbtables;
 406
 407                 /*
 408                  * If there are more connections than vacuumable relations, we don't
 409                  * need to use them all.
 410                  */
 411                 if (concurrentCons > ntups)
 412                         concurrentCons = ntups;
 413                 if (concurrentCons <= 1)
 414                         parallel = false;
 415                 PQclear(res);
 416         }
 417
 418         /*
 419          * Setup the database connections. We reuse the connection we already have
 420          * for the first slot.  If not in parallel mode, the first slot in the
 421          * array contains the connection.
 422          */
 423         slots = (ParallelSlot *) pg_malloc(sizeof(ParallelSlot) * concurrentCons);
 424         init_slot(slots, conn);
 425         if (parallel)
 426         {
 427                 for (i = 1; i < concurrentCons; i++)
 428                 {
 429                         conn = connectDatabase(dbname, host, port, username, prompt_password,
 430                                                                    progname, false);
 431                         init_slot(slots + i, conn);
 432                 }
 433         }
 434
 435         /*
 436          * Prepare all the connections to run the appropriate analyze stage, if
 437          * caller requested that mode.
 438          */
 439         if (stage != ANALYZE_NO_STAGE)
 440         {
 441                 int                     j;
 442
 443                 /* We already emitted the message above */
 444
 445                 for (j = 0; j < concurrentCons; j++)
 446                         executeCommand((slots + j)->connection,
 447                                                    stage_commands[stage], progname, echo);
 448         }
 449
 450         cell = tables ? tables->head : NULL;
 451         do
 452         {
 453                 ParallelSlot *free_slot;
 454                 const char *tabname = cell ? cell->val : NULL;
 455
 456                 prepare_vacuum_command(&sql, conn, vacopts, tabname);
 457
 458                 if (CancelRequested)
 459                 {
 460                         failed = true;
 461                         goto finish;
 462                 }
 463
 464                 /*
 465                  * Get the connection slot to use.  If in parallel mode, here we wait
 466                  * for one connection to become available if none already is.  In
 467                  * non-parallel mode we simply use the only slot we have, which we
 468                  * know to be free.
 469                  */
 470                 if (parallel)
 471                 {
 472                         /*
 473                          * Get a free slot, waiting until one becomes free if none
 474                          * currently is.
 475                          */
 476                         free_slot = GetIdleSlot(slots, concurrentCons, dbname, progname);
 477                         if (!free_slot)
 478                         {
 479                                 failed = true;
 480                                 goto finish;
 481                         }
 482
 483                         free_slot->isFree = false;
 484                 }
 485                 else
 486                         free_slot = slots;
 487
 488                 /*
 489                  * Execute the vacuum.  If not in parallel mode, this terminates the
 490                  * program in case of an error.  (The parallel case handles query
 491                  * errors in GetQueryResult through GetIdleSlot.)
 492                  */
 493                 run_vacuum_command(free_slot->connection, sql.data,
 494                                                    echo, dbname, tabname, progname, parallel);
 495
 496                 if (cell)
 497                         cell = cell->next;
 498         } while (cell != NULL);
 499
 500         if (parallel)
 501         {
 502                 int                     j;
 503
 504                 for (j = 0; j < concurrentCons; j++)
 505                 {
 506                         /* wait for all connection to return the results */
 507                         if (!GetQueryResult((slots + j)->connection, dbname, progname))
 508                                 goto finish;
 509
 510                         (slots + j)->isFree = true;
 511                 }
 512         }
 513
 514 finish:
 515         for (i = 0; i < concurrentCons; i++)
 516                 DisconnectDatabase(slots + i);
 517         pfree(slots);
 518
 519         termPQExpBuffer(&sql);
 520
 521         if (failed)
 522                 exit(1);
 523 }
 524
 525 /*
 526  * Vacuum/analyze all connectable databases.
 527  *
 528  * In analyze-in-stages mode, we process all databases in one stage before
 529  * moving on to the next stage.  That ensure minimal stats are available
 530  * quickly everywhere before generating more detailed ones.
 531  */
 532 static void
 533 vacuum_all_databases(vacuumingOptions *vacopts,
 534                                          bool analyze_in_stages,
 535                                          const char *maintenance_db, const char *host,
 536                                          const char *port, const char *username,
 537                                          enum trivalue prompt_password,
 538                                          int concurrentCons,
 539                                          const char *progname, bool echo, bool quiet)
 540 {
 541         PGconn     *conn;
 542         PGresult   *result;
 543         int                     stage;
 544         int                     i;
 545
 546         conn = connectMaintenanceDatabase(maintenance_db, host, port,
 547                                                                           username, prompt_password, progname);
 548         result = executeQuery(conn,
 549                         "SELECT datname FROM pg_database WHERE datallowconn ORDER BY 1;",
 550                                                   progname, echo);
 551         PQfinish(conn);
 552
 553         if (analyze_in_stages)
 554         {
 555                 /*
 556                  * When analyzing all databases in stages, we analyze them all in the
 557                  * fastest stage first, so that initial statistics become available
 558                  * for all of them as soon as possible.
 559                  *
 560                  * This means we establish several times as many connections, but
 561                  * that's a secondary consideration.
 562                  */
 563                 for (stage = 0; stage < ANALYZE_NUM_STAGES; stage++)
 564                 {
 565                         for (i = 0; i < PQntuples(result); i++)
 566                         {
 567                                 const char *dbname;
 568
 569                                 dbname = PQgetvalue(result, i, 0);
 570                                 vacuum_one_database(dbname, vacopts,
 571                                                                         stage,
 572                                                                         NULL,
 573                                                                         host, port, username, prompt_password,
 574                                                                         concurrentCons,
 575                                                                         progname, echo, quiet);
 576                         }
 577                 }
 578         }
 579         else
 580         {
 581                 for (i = 0; i < PQntuples(result); i++)
 582                 {
 583                         const char *dbname;
 584
 585                         dbname = PQgetvalue(result, i, 0);
 586                         vacuum_one_database(dbname, vacopts,
 587                                                                 ANALYZE_NO_STAGE,
 588                                                                 NULL,
 589                                                                 host, port, username, prompt_password,
 590                                                                 concurrentCons,
 591                                                                 progname, echo, quiet);
 592                 }
 593         }
 594
 595         PQclear(result);
 596 }
 597
 598 /*
 599  * Construct a vacuum/analyze command to run based on the given options, in the
 600  * given string buffer, which may contain previous garbage.
 601  *
 602  * An optional table name can be passed; this must be already be properly
 603  * quoted.  The command is semicolon-terminated.
 604  */
 605 static void
 606 prepare_vacuum_command(PQExpBuffer sql, PGconn *conn, vacuumingOptions *vacopts,
 607                                            const char *table)
 608 {
 609         resetPQExpBuffer(sql);
 610
 611         if (vacopts->analyze_only)
 612         {
 613                 appendPQExpBufferStr(sql, "ANALYZE");
 614                 if (vacopts->verbose)
 615                         appendPQExpBufferStr(sql, " VERBOSE");
 616         }
 617         else
 618         {
 619                 appendPQExpBufferStr(sql, "VACUUM");
 620                 if (PQserverVersion(conn) >= 90000)
 621                 {
 622                         const char *paren = " (";
 623                         const char *comma = ", ";
 624                         const char *sep = paren;
 625
 626                         if (vacopts->full)
 627                         {
 628                                 appendPQExpBuffer(sql, "%sFULL", sep);
 629                                 sep = comma;
 630                         }
 631                         if (vacopts->freeze)
 632                         {
 633                                 appendPQExpBuffer(sql, "%sFREEZE", sep);
 634                                 sep = comma;
 635                         }
 636                         if (vacopts->verbose)
 637                         {
 638                                 appendPQExpBuffer(sql, "%sVERBOSE", sep);
 639                                 sep = comma;
 640                         }
 641                         if (vacopts->and_analyze)
 642                         {
 643                                 appendPQExpBuffer(sql, "%sANALYZE", sep);
 644                                 sep = comma;
 645                         }
 646                         if (sep != paren)
 647                                 appendPQExpBufferChar(sql, ')');
 648                 }
 649                 else
 650                 {
 651                         if (vacopts->full)
 652                                 appendPQExpBufferStr(sql, " FULL");
 653                         if (vacopts->freeze)
 654                                 appendPQExpBufferStr(sql, " FREEZE");
 655                         if (vacopts->verbose)
 656                                 appendPQExpBufferStr(sql, " VERBOSE");
 657                         if (vacopts->and_analyze)
 658                                 appendPQExpBufferStr(sql, " ANALYZE");
 659                 }
 660         }
 661
 662         if (table)
 663                 appendPQExpBuffer(sql, " %s", table);
 664         appendPQExpBufferChar(sql, ';');
 665 }
 666
 667 /*
 668  * Execute a vacuum/analyze command to the server.
 669  *
 670  * Any errors during command execution are reported to stderr.  If async is
 671  * false, this function exits the program after reporting the error.
 672  */
 673 static void
 674 run_vacuum_command(PGconn *conn, const char *sql, bool echo,
 675                                    const char *dbname, const char *table,
 676                                    const char *progname, bool async)
 677 {
 678         bool            status;
 679
 680         if (async)
 681         {
 682                 if (echo)
 683                         printf("%s\n", sql);
 684
 685                 status = PQsendQuery(conn, sql) == 1;
 686         }
 687         else
 688                 status = executeMaintenanceCommand(conn, sql, echo);
 689
 690         if (!status)
 691         {
 692                 if (table)
 693                         fprintf(stderr,
 694                         _("%s: vacuuming of table \"%s\" in database \"%s\" failed: %s"),
 695                                         progname, table, dbname, PQerrorMessage(conn));
 696                 else
 697                         fprintf(stderr, _("%s: vacuuming of database \"%s\" failed: %s"),
 698                                         progname, dbname, PQerrorMessage(conn));
 699
 700                 if (!async)
 701                 {
 702                         PQfinish(conn);
 703                         exit(1);
 704                 }
 705         }
 706 }
 707
 708 /*
 709  * GetIdleSlot
 710  *              Return a connection slot that is ready to execute a command.
 711  *
 712  * We return the first slot we find that is marked isFree, if one is;
 713  * otherwise, we loop on select() until one socket becomes available.  When
 714  * this happens, we read the whole set and mark as free all sockets that become
 715  * available.
 716  *
 717  * Process the slot list, if any free slot is available then return the slotid
 718  * else perform the select on all the socket's and wait until at least one slot
 719  * becomes available.
 720  *
 721  * If an error occurs, NULL is returned.
 722  */
 723 static ParallelSlot *
 724 GetIdleSlot(ParallelSlot slots[], int numslots, const char *dbname,
 725                         const char *progname)
 726 {
 727         int                     i;
 728         int                     firstFree = -1;
 729         fd_set          slotset;
 730         pgsocket        maxFd;
 731
 732         for (i = 0; i < numslots; i++)
 733                 if ((slots + i)->isFree)
 734                         return slots + i;
 735
 736         FD_ZERO(&slotset);
 737
 738         maxFd = slots->sock;
 739         for (i = 0; i < numslots; i++)
 740         {
 741                 FD_SET((slots + i)->sock, &slotset);
 742                 if ((slots + i)->sock > maxFd)
 743                         maxFd = (slots + i)->sock;
 744         }
 745
 746         /*
 747          * No free slot found, so wait until one of the connections has finished
 748          * its task and return the available slot.
 749          */
 750         for (firstFree = -1; firstFree < 0;)
 751         {
 752                 bool            aborting;
 753
 754                 SetCancelConn(slots->connection);
 755                 i = select_loop(maxFd, &slotset, &aborting);
 756                 ResetCancelConn();
 757
 758                 if (aborting)
 759                 {
 760                         /*
 761                          * We set the cancel-receiving connection to the one in the zeroth
 762                          * slot above, so fetch the error from there.
 763                          */
 764                         GetQueryResult(slots->connection, dbname, progname);
 765                         return NULL;
 766                 }
 767                 Assert(i != 0);
 768
 769                 for (i = 0; i < numslots; i++)
 770                 {
 771                         if (!FD_ISSET((slots + i)->sock, &slotset))
 772                                 continue;
 773
 774                         PQconsumeInput((slots + i)->connection);
 775                         if (PQisBusy((slots + i)->connection))
 776                                 continue;
 777
 778                         (slots + i)->isFree = true;
 779
 780                         if (!GetQueryResult((slots + i)->connection, dbname, progname))
 781                                 return NULL;
 782
 783                         if (firstFree < 0)
 784                                 firstFree = i;
 785                 }
 786         }
 787
 788         return slots + firstFree;
 789 }
 790
 791 /*
 792  * GetQueryResult
 793  *
 794  * Process the query result.  Returns true if there's no error, false
 795  * otherwise -- but errors about trying to vacuum a missing relation are
 796  * reported and subsequently ignored.
 797  */
 798 static bool
 799 GetQueryResult(PGconn *conn, const char *dbname, const char *progname)
 800 {
 801         PGresult   *result;
 802
 803         SetCancelConn(conn);
 804         while ((result = PQgetResult(conn)) != NULL)
 805         {
 806                 /*
 807                  * If errors are found, report them.  Errors about a missing table are
 808                  * harmless so we continue processing; but die for other errors.
 809                  */
 810                 if (PQresultStatus(result) != PGRES_COMMAND_OK)
 811                 {
 812                         char       *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
 813
 814                         fprintf(stderr, _("%s: vacuuming of database \"%s\" failed: %s"),
 815                                         progname, dbname, PQerrorMessage(conn));
 816
 817                         if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
 818                         {
 819                                 PQclear(result);
 820                                 return false;
 821                         }
 822                 }
 823
 824                 PQclear(result);
 825         }
 826         ResetCancelConn();
 827
 828         return true;
 829 }
 830
 831 /*
 832  * DisconnectDatabase
 833  *              Disconnect the connection associated with the given slot
 834  */
 835 static void
 836 DisconnectDatabase(ParallelSlot *slot)
 837 {
 838         char            errbuf[256];
 839
 840         if (!slot->connection)
 841                 return;
 842
 843         if (PQtransactionStatus(slot->connection) == PQTRANS_ACTIVE)
 844         {
 845                 PGcancel   *cancel;
 846
 847                 if ((cancel = PQgetCancel(slot->connection)))
 848                 {
 849                         PQcancel(cancel, errbuf, sizeof(errbuf));
 850                         PQfreeCancel(cancel);
 851                 }
 852         }
 853
 854         PQfinish(slot->connection);
 855         slot->connection = NULL;
 856 }
 857
 858 /*
 859  * Loop on select() until a descriptor from the given set becomes readable.
 860  *
 861  * If we get a cancel request while we're waiting, we forego all further
 862  * processing and set the *aborting flag to true.  The return value must be
 863  * ignored in this case.  Otherwise, *aborting is set to false.
 864  */
 865 static int
 866 select_loop(int maxFd, fd_set *workerset, bool *aborting)
 867 {
 868         int                     i;
 869         fd_set          saveSet = *workerset;
 870
 871         if (CancelRequested)
 872         {
 873                 *aborting = true;
 874                 return -1;
 875         }
 876         else
 877                 *aborting = false;
 878
 879         for (;;)
 880         {
 881                 /*
 882                  * On Windows, we need to check once in a while for cancel requests;
 883                  * on other platforms we rely on select() returning when interrupted.
 884                  */
 885                 struct timeval *tvp;
 886 #ifdef WIN32
 887                 struct timeval tv = {0, 1000000};
 888
 889                 tvp = &tv;
 890 #else
 891                 tvp = NULL;
 892 #endif
 893
 894                 *workerset = saveSet;
 895                 i = select(maxFd + 1, workerset, NULL, NULL, tvp);
 896
 897 #ifdef WIN32
 898                 if (i == SOCKET_ERROR)
 899                 {
 900                         i = -1;
 901
 902                         if (WSAGetLastError() == WSAEINTR)
 903                                 errno = EINTR;
 904                 }
 905 #endif
 906
 907                 if (i < 0 && errno == EINTR)
 908                         continue;                       /* ignore this */
 909                 if (i < 0 || CancelRequested)
 910                         *aborting = true;       /* but not this */
 911                 if (i == 0)
 912                         continue;                       /* timeout (Win32 only) */
 913                 break;
 914         }
 915
 916         return i;
 917 }
 918
 919 static void
 920 init_slot(ParallelSlot *slot, PGconn *conn)
 921 {
 922         slot->connection = conn;
 923         slot->isFree = true;
 924         slot->sock = PQsocket(conn);
 925 }
 926
 927 static void
 928 help(const char *progname)
 929 {
 930         printf(_("%s cleans and analyzes a PostgreSQL database.\n\n"), progname);
 931         printf(_("Usage:\n"));
 932         printf(_("  %s [OPTION]... [DBNAME]\n"), progname);
 933         printf(_("\nOptions:\n"));
 934         printf(_("  -a, --all                       vacuum all databases\n"));
 935         printf(_("  -d, --dbname=DBNAME             database to vacuum\n"));
 936         printf(_("  -e, --echo                      show the commands being sent to the server\n"));
 937         printf(_("  -f, --full                      do full vacuuming\n"));
 938         printf(_("  -F, --freeze                    freeze row transaction information\n"));
 939         printf(_("  -q, --quiet                     don't write any messages\n"));
 940         printf(_("  -t, --table='TABLE[(COLUMNS)]'  vacuum specific table(s) only\n"));
 941         printf(_("  -v, --verbose                   write a lot of output\n"));
 942         printf(_("  -V, --version                   output version information, then exit\n"));
 943         printf(_("  -z, --analyze                   update optimizer statistics\n"));
 944         printf(_("  -Z, --analyze-only              only update optimizer statistics;  no vacuum\n"));
 945         printf(_("  -j, --jobs=NUM                  use this many concurrent connections to vacuum\n"));
 946         printf(_("      --analyze-in-stages         only update optimizer statistics, in multiple\n"
 947                          "                                  stages for faster results;  no vacuum\n"));
 948         printf(_("  -?, --help                      show this help, then exit\n"));
 949         printf(_("\nConnection options:\n"));
 950         printf(_("  -h, --host=HOSTNAME       database server host or socket directory\n"));
 951         printf(_("  -p, --port=PORT           database server port\n"));
 952         printf(_("  -U, --username=USERNAME   user name to connect as\n"));
 953         printf(_("  -w, --no-password         never prompt for password\n"));
 954         printf(_("  -W, --password            force password prompt\n"));
 955         printf(_("  --maintenance-db=DBNAME   alternate maintenance database\n"));
 956         printf(_("\nRead the description of the SQL command VACUUM for details.\n"));
 957         printf(_("\nReport bugs to <pgsql-bugs@postgresql.org>.\n"));
 958 }