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            result = 0;
 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                         appendPQExpBuffer(&buf, "%s",
 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         }
 416
 417         /*
 418          * Setup the database connections. We reuse the connection we already have
 419          * for the first slot.  If not in parallel mode, the first slot in the
 420          * array contains the connection.
 421          */
 422         slots = (ParallelSlot *) pg_malloc(sizeof(ParallelSlot) * concurrentCons);
 423         init_slot(slots, conn);
 424         if (parallel)
 425         {
 426                 for (i = 1; i < concurrentCons; i++)
 427                 {
 428                         conn = connectDatabase(dbname, host, port, username, prompt_password,
 429                                                                    progname, false);
 430                         init_slot(slots + i, conn);
 431                 }
 432         }
 433
 434         /*
 435          * Prepare all the connections to run the appropriate analyze stage, if
 436          * caller requested that mode.
 437          */
 438         if (stage != ANALYZE_NO_STAGE)
 439         {
 440                 int                     j;
 441
 442                 /* We already emitted the message above */
 443
 444                 for (j = 0; j < concurrentCons; j++)
 445                         executeCommand((slots + j)->connection,
 446                                                    stage_commands[stage], progname, echo);
 447         }
 448
 449         cell = tables ? tables->head : NULL;
 450         do
 451         {
 452                 ParallelSlot *free_slot;
 453                 const char *tabname = cell ? cell->val : NULL;
 454
 455                 prepare_vacuum_command(&sql, conn, vacopts, tabname);
 456
 457                 if (CancelRequested)
 458                 {
 459                         result = -1;
 460                         goto finish;
 461                 }
 462
 463                 /*
 464                  * Get the connection slot to use.  If in parallel mode, here we wait
 465                  * for one connection to become available if none already is.  In
 466                  * non-parallel mode we simply use the only slot we have, which we
 467                  * know to be free.
 468                  */
 469                 if (parallel)
 470                 {
 471                         /*
 472                          * Get a free slot, waiting until one becomes free if none
 473                          * currently is.
 474                          */
 475                         free_slot = GetIdleSlot(slots, concurrentCons, dbname, progname);
 476                         if (!free_slot)
 477                         {
 478                                 result = -1;
 479                                 goto finish;
 480                         }
 481
 482                         free_slot->isFree = false;
 483                 }
 484                 else
 485                         free_slot = slots;
 486
 487                 run_vacuum_command(free_slot->connection, sql.data,
 488                                                    echo, dbname, tabname, progname, parallel);
 489
 490                 if (cell)
 491                         cell = cell->next;
 492         } while (cell != NULL);
 493
 494         if (parallel)
 495         {
 496                 int                     j;
 497
 498                 for (j = 0; j < concurrentCons; j++)
 499                 {
 500                         /* wait for all connection to return the results */
 501                         if (!GetQueryResult((slots + j)->connection, dbname, progname))
 502                                 goto finish;
 503
 504                         (slots + j)->isFree = true;
 505                 }
 506         }
 507
 508 finish:
 509         for (i = 0; i < concurrentCons; i++)
 510                 DisconnectDatabase(slots + i);
 511         pfree(slots);
 512
 513         termPQExpBuffer(&sql);
 514
 515         if (result == -1)
 516                 exit(1);
 517 }
 518
 519 /*
 520  * Vacuum/analyze all connectable databases.
 521  *
 522  * In analyze-in-stages mode, we process all databases in one stage before
 523  * moving on to the next stage.  That ensure minimal stats are available
 524  * quickly everywhere before generating more detailed ones.
 525  */
 526 static void
 527 vacuum_all_databases(vacuumingOptions *vacopts,
 528                                          bool analyze_in_stages,
 529                                          const char *maintenance_db, const char *host,
 530                                          const char *port, const char *username,
 531                                          enum trivalue prompt_password,
 532                                          int concurrentCons,
 533                                          const char *progname, bool echo, bool quiet)
 534 {
 535         PGconn     *conn;
 536         PGresult   *result;
 537         int                     stage;
 538         int                     i;
 539
 540         conn = connectMaintenanceDatabase(maintenance_db, host, port,
 541                                                                           username, prompt_password, progname);
 542         result = executeQuery(conn,
 543                         "SELECT datname FROM pg_database WHERE datallowconn ORDER BY 1;",
 544                                                   progname, echo);
 545         PQfinish(conn);
 546
 547         if (analyze_in_stages)
 548         {
 549                 /*
 550                  * When analyzing all databases in stages, we analyze them all in the
 551                  * fastest stage first, so that initial statistics become available
 552                  * for all of them as soon as possible.
 553                  *
 554                  * This means we establish several times as many connections, but
 555                  * that's a secondary consideration.
 556                  */
 557                 for (stage = 0; stage < ANALYZE_NUM_STAGES; stage++)
 558                 {
 559                         for (i = 0; i < PQntuples(result); i++)
 560                         {
 561                                 const char *dbname;
 562
 563                                 dbname = PQgetvalue(result, i, 0);
 564                                 vacuum_one_database(dbname, vacopts,
 565                                                                         stage,
 566                                                                         NULL,
 567                                                                         host, port, username, prompt_password,
 568                                                                         concurrentCons,
 569                                                                         progname, echo, quiet);
 570                         }
 571                 }
 572         }
 573         else
 574         {
 575                 for (i = 0; i < PQntuples(result); i++)
 576                 {
 577                         const char *dbname;
 578
 579                         dbname = PQgetvalue(result, i, 0);
 580                         vacuum_one_database(dbname, vacopts,
 581                                                                 ANALYZE_NO_STAGE,
 582                                                                 NULL,
 583                                                                 host, port, username, prompt_password,
 584                                                                 concurrentCons,
 585                                                                 progname, echo, quiet);
 586                 }
 587         }
 588
 589         PQclear(result);
 590 }
 591
 592 /*
 593  * Construct a vacuum/analyze command to run based on the given options, in the
 594  * given string buffer, which may contain previous garbage.
 595  *
 596  * An optional table name can be passed; this must be already be properly
 597  * quoted.  The command is semicolon-terminated.
 598  */
 599 static void
 600 prepare_vacuum_command(PQExpBuffer sql, PGconn *conn, vacuumingOptions *vacopts,
 601                                            const char *table)
 602 {
 603         resetPQExpBuffer(sql);
 604
 605         if (vacopts->analyze_only)
 606         {
 607                 appendPQExpBufferStr(sql, "ANALYZE");
 608                 if (vacopts->verbose)
 609                         appendPQExpBufferStr(sql, " VERBOSE");
 610         }
 611         else
 612         {
 613                 appendPQExpBufferStr(sql, "VACUUM");
 614                 if (PQserverVersion(conn) >= 90000)
 615                 {
 616                         const char *paren = " (";
 617                         const char *comma = ", ";
 618                         const char *sep = paren;
 619
 620                         if (vacopts->full)
 621                         {
 622                                 appendPQExpBuffer(sql, "%sFULL", sep);
 623                                 sep = comma;
 624                         }
 625                         if (vacopts->freeze)
 626                         {
 627                                 appendPQExpBuffer(sql, "%sFREEZE", sep);
 628                                 sep = comma;
 629                         }
 630                         if (vacopts->verbose)
 631                         {
 632                                 appendPQExpBuffer(sql, "%sVERBOSE", sep);
 633                                 sep = comma;
 634                         }
 635                         if (vacopts->and_analyze)
 636                         {
 637                                 appendPQExpBuffer(sql, "%sANALYZE", sep);
 638                                 sep = comma;
 639                         }
 640                         if (sep != paren)
 641                                 appendPQExpBufferStr(sql, ")");
 642                 }
 643                 else
 644                 {
 645                         if (vacopts->full)
 646                                 appendPQExpBufferStr(sql, " FULL");
 647                         if (vacopts->freeze)
 648                                 appendPQExpBufferStr(sql, " FREEZE");
 649                         if (vacopts->verbose)
 650                                 appendPQExpBufferStr(sql, " VERBOSE");
 651                         if (vacopts->and_analyze)
 652                                 appendPQExpBufferStr(sql, " ANALYZE");
 653                 }
 654         }
 655
 656         if (table)
 657                 appendPQExpBuffer(sql, " %s", table);
 658         appendPQExpBufferChar(sql, ';');
 659 }
 660
 661 /*
 662  * Execute a vacuum/analyze command to the server.
 663  *
 664  * Result status is checked only if 'async' is false.
 665  */
 666 static void
 667 run_vacuum_command(PGconn *conn, const char *sql, bool echo,
 668                                    const char *dbname, const char *table,
 669                                    const char *progname, bool async)
 670 {
 671         if (async)
 672         {
 673                 if (echo)
 674                         printf("%s\n", sql);
 675
 676                 PQsendQuery(conn, sql);
 677         }
 678         else if (!executeMaintenanceCommand(conn, sql, echo))
 679         {
 680                 if (table)
 681                         fprintf(stderr,
 682                         _("%s: vacuuming of table \"%s\" in database \"%s\" failed: %s"),
 683                                         progname, table, dbname, PQerrorMessage(conn));
 684                 else
 685                         fprintf(stderr, _("%s: vacuuming of database \"%s\" failed: %s"),
 686                                         progname, dbname, PQerrorMessage(conn));
 687                 PQfinish(conn);
 688                 exit(1);
 689         }
 690 }
 691
 692 /*
 693  * GetIdleSlot
 694  *              Return a connection slot that is ready to execute a command.
 695  *
 696  * We return the first slot we find that is marked isFree, if one is;
 697  * otherwise, we loop on select() until one socket becomes available.  When
 698  * this happens, we read the whole set and mark as free all sockets that become
 699  * available.
 700  *
 701  * Process the slot list, if any free slot is available then return the slotid
 702  * else perform the select on all the socket's and wait until at least one slot
 703  * becomes available.
 704  *
 705  * If an error occurs, NULL is returned.
 706  */
 707 static ParallelSlot *
 708 GetIdleSlot(ParallelSlot slots[], int numslots, const char *dbname,
 709                         const char *progname)
 710 {
 711         int                     i;
 712         int                     firstFree = -1;
 713         fd_set          slotset;
 714         pgsocket        maxFd;
 715
 716         for (i = 0; i < numslots; i++)
 717                 if ((slots + i)->isFree)
 718                         return slots + i;
 719
 720         FD_ZERO(&slotset);
 721
 722         maxFd = slots->sock;
 723         for (i = 0; i < numslots; i++)
 724         {
 725                 FD_SET((slots + i)->sock, &slotset);
 726                 if ((slots + i)->sock > maxFd)
 727                         maxFd = (slots + i)->sock;
 728         }
 729
 730         /*
 731          * No free slot found, so wait until one of the connections has finished
 732          * its task and return the available slot.
 733          */
 734         for (firstFree = -1; firstFree < 0;)
 735         {
 736                 bool            aborting;
 737
 738                 SetCancelConn(slots->connection);
 739                 i = select_loop(maxFd, &slotset, &aborting);
 740                 ResetCancelConn();
 741
 742                 if (aborting)
 743                 {
 744                         /*
 745                          * We set the cancel-receiving connection to the one in the zeroth
 746                          * slot above, so fetch the error from there.
 747                          */
 748                         GetQueryResult(slots->connection, dbname, progname);
 749                         return NULL;
 750                 }
 751                 Assert(i != 0);
 752
 753                 for (i = 0; i < numslots; i++)
 754                 {
 755                         if (!FD_ISSET((slots + i)->sock, &slotset))
 756                                 continue;
 757
 758                         PQconsumeInput((slots + i)->connection);
 759                         if (PQisBusy((slots + i)->connection))
 760                                 continue;
 761
 762                         (slots + i)->isFree = true;
 763
 764                         if (!GetQueryResult((slots + i)->connection, dbname, progname))
 765                                 return NULL;
 766
 767                         if (firstFree < 0)
 768                                 firstFree = i;
 769                 }
 770         }
 771
 772         return slots + firstFree;
 773 }
 774
 775 /*
 776  * GetQueryResult
 777  *
 778  * Process the query result.  Returns true if there's no error, false
 779  * otherwise -- but errors about trying to vacuum a missing relation are
 780  * reported and subsequently ignored.
 781  */
 782 static bool
 783 GetQueryResult(PGconn *conn, const char *dbname, const char *progname)
 784 {
 785         PGresult   *result;
 786
 787         SetCancelConn(conn);
 788         while ((result = PQgetResult(conn)) != NULL)
 789         {
 790                 /*
 791                  * If errors are found, report them.  Errors about a missing table are
 792                  * harmless so we continue processing; but die for other errors.
 793                  */
 794                 if (PQresultStatus(result) != PGRES_COMMAND_OK)
 795                 {
 796                         char       *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
 797
 798                         fprintf(stderr, _("%s: vacuuming of database \"%s\" failed: %s"),
 799                                         progname, dbname, PQerrorMessage(conn));
 800
 801                         if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
 802                         {
 803                                 PQclear(result);
 804                                 return false;
 805                         }
 806                 }
 807
 808                 PQclear(result);
 809         }
 810         ResetCancelConn();
 811
 812         return true;
 813 }
 814
 815 /*
 816  * DisconnectDatabase
 817  *              Disconnect the connection associated with the given slot
 818  */
 819 static void
 820 DisconnectDatabase(ParallelSlot *slot)
 821 {
 822         char            errbuf[256];
 823
 824         if (!slot->connection)
 825                 return;
 826
 827         if (PQtransactionStatus(slot->connection) == PQTRANS_ACTIVE)
 828         {
 829                 PGcancel   *cancel;
 830
 831                 if ((cancel = PQgetCancel(slot->connection)))
 832                 {
 833                         PQcancel(cancel, errbuf, sizeof(errbuf));
 834                         PQfreeCancel(cancel);
 835                 }
 836         }
 837
 838         PQfinish(slot->connection);
 839         slot->connection = NULL;
 840 }
 841
 842 /*
 843  * Loop on select() until a descriptor from the given set becomes readable.
 844  *
 845  * If we get a cancel request while we're waiting, we forego all further
 846  * processing and set the *aborting flag to true.  The return value must be
 847  * ignored in this case.  Otherwise, *aborting is set to false.
 848  */
 849 static int
 850 select_loop(int maxFd, fd_set *workerset, bool *aborting)
 851 {
 852         int                     i;
 853         fd_set          saveSet = *workerset;
 854
 855         if (CancelRequested)
 856         {
 857                 *aborting = true;
 858                 return -1;
 859         }
 860         else
 861                 *aborting = false;
 862
 863         for (;;)
 864         {
 865                 /*
 866                  * On Windows, we need to check once in a while for cancel requests;
 867                  * on other platforms we rely on select() returning when interrupted.
 868                  */
 869                 struct timeval *tvp;
 870 #ifdef WIN32
 871                 struct timeval tv = {0, 1000000};
 872
 873                 tvp = &tv;
 874 #else
 875                 tvp = NULL;
 876 #endif
 877
 878                 *workerset = saveSet;
 879                 i = select(maxFd + 1, workerset, NULL, NULL, tvp);
 880
 881 #ifdef WIN32
 882                 if (i == SOCKET_ERROR)
 883                 {
 884                         i = -1;
 885
 886                         if (WSAGetLastError() == WSAEINTR)
 887                                 errno == EINTR;
 888                 }
 889 #endif
 890
 891                 if (i < 0 && errno == EINTR)
 892                         continue;                       /* ignore this */
 893                 if (i < 0 || CancelRequested)
 894                         *aborting = true;       /* but not this */
 895                 if (i == 0)
 896                         continue;                       /* timeout (Win32 only) */
 897                 break;
 898         }
 899
 900         return i;
 901 }
 902
 903 static void
 904 init_slot(ParallelSlot *slot, PGconn *conn)
 905 {
 906         slot->connection = conn;
 907         slot->isFree = true;
 908         slot->sock = PQsocket(conn);
 909 }
 910
 911 static void
 912 help(const char *progname)
 913 {
 914         printf(_("%s cleans and analyzes a PostgreSQL database.\n\n"), progname);
 915         printf(_("Usage:\n"));
 916         printf(_("  %s [OPTION]... [DBNAME]\n"), progname);
 917         printf(_("\nOptions:\n"));
 918         printf(_("  -a, --all                       vacuum all databases\n"));
 919         printf(_("  -d, --dbname=DBNAME             database to vacuum\n"));
 920         printf(_("  -e, --echo                      show the commands being sent to the server\n"));
 921         printf(_("  -f, --full                      do full vacuuming\n"));
 922         printf(_("  -F, --freeze                    freeze row transaction information\n"));
 923         printf(_("  -q, --quiet                     don't write any messages\n"));
 924         printf(_("  -t, --table='TABLE[(COLUMNS)]'  vacuum specific table(s) only\n"));
 925         printf(_("  -v, --verbose                   write a lot of output\n"));
 926         printf(_("  -V, --version                   output version information, then exit\n"));
 927         printf(_("  -z, --analyze                   update optimizer statistics\n"));
 928         printf(_("  -Z, --analyze-only              only update optimizer statistics\n"));
 929         printf(_("  -j, --jobs=NUM                  use this many concurrent connections to vacuum\n"));
 930         printf(_("      --analyze-in-stages         only update optimizer statistics, in multiple\n"
 931                    "                                  stages for faster results\n"));
 932         printf(_("  -?, --help                      show this help, then exit\n"));
 933         printf(_("\nConnection options:\n"));
 934         printf(_("  -h, --host=HOSTNAME       database server host or socket directory\n"));
 935         printf(_("  -p, --port=PORT           database server port\n"));
 936         printf(_("  -U, --username=USERNAME   user name to connect as\n"));
 937         printf(_("  -w, --no-password         never prompt for password\n"));
 938         printf(_("  -W, --password            force password prompt\n"));
 939         printf(_("  --maintenance-db=DBNAME   alternate maintenance database\n"));
 940         printf(_("\nRead the description of the SQL command VACUUM for details.\n"));
 941         printf(_("\nReport bugs to <pgsql-bugs@postgresql.org>.\n"));
 942 }