granicus.if.org Git - postgresql/blob - src/backend/postmaster/pgstat.c

   1 /* ----------
   2  * pgstat.c
   3  *
   4  *      All the statistics collector stuff hacked up in one big, ugly file.
   5  *
   6  *      TODO:   - Separate collector, postmaster and backend stuff
   7  *                        into different files.
   8  *
   9  *                      - Add some automatic call for pgstat vacuuming.
  10  *
  11  *                      - Add a pgstat config column to pg_database, so this
  12  *                        entire thing can be enabled/disabled on a per db basis.
  13  *
  14  *      Copyright (c) 2001-2007, PostgreSQL Global Development Group
  15  *
  16  *      $PostgreSQL: pgsql/src/backend/postmaster/pgstat.c,v 1.144 2007/01/26 20:06:52 tgl Exp $
  17  * ----------
  18  */
  19 #include "postgres.h"
  20
  21 #include <unistd.h>
  22 #include <fcntl.h>
  23 #include <sys/param.h>
  24 #include <sys/time.h>
  25 #include <sys/socket.h>
  26 #include <netdb.h>
  27 #include <netinet/in.h>
  28 #include <arpa/inet.h>
  29 #include <signal.h>
  30 #include <time.h>
  31 #ifdef HAVE_POLL_H
  32 #include <poll.h>
  33 #endif
  34 #ifdef HAVE_SYS_POLL_H
  35 #include <sys/poll.h>
  36 #endif
  37
  38 #include "pgstat.h"
  39
  40 #include "access/heapam.h"
  41 #include "access/transam.h"
  42 #include "access/xact.h"
  43 #include "catalog/pg_database.h"
  44 #include "libpq/ip.h"
  45 #include "libpq/libpq.h"
  46 #include "libpq/pqsignal.h"
  47 #include "mb/pg_wchar.h"
  48 #include "miscadmin.h"
  49 #include "postmaster/autovacuum.h"
  50 #include "postmaster/fork_process.h"
  51 #include "postmaster/postmaster.h"
  52 #include "storage/backendid.h"
  53 #include "storage/fd.h"
  54 #include "storage/ipc.h"
  55 #include "storage/pg_shmem.h"
  56 #include "storage/pmsignal.h"
  57 #include "utils/memutils.h"
  58 #include "utils/ps_status.h"
  59
  60
  61 /* ----------
  62  * Paths for the statistics files (relative to installation's $PGDATA).
  63  * ----------
  64  */
  65 #define PGSTAT_STAT_FILENAME    "global/pgstat.stat"
  66 #define PGSTAT_STAT_TMPFILE             "global/pgstat.tmp"
  67
  68 /* ----------
  69  * Timer definitions.
  70  * ----------
  71  */
  72 #define PGSTAT_STAT_INTERVAL    500             /* How often to write the status file;
  73                                                                                  * in milliseconds. */
  74
  75 #define PGSTAT_RESTART_INTERVAL 60              /* How often to attempt to restart a
  76                                                                                  * failed statistics collector; in
  77                                                                                  * seconds. */
  78
  79 #define PGSTAT_SELECT_TIMEOUT   2               /* How often to check for postmaster
  80                                                                                  * death; in seconds. */
  81
  82
  83 /* ----------
  84  * The initial size hints for the hash tables used in the collector.
  85  * ----------
  86  */
  87 #define PGSTAT_DB_HASH_SIZE             16
  88 #define PGSTAT_TAB_HASH_SIZE    512
  89
  90
  91 /* ----------
  92  * GUC parameters
  93  * ----------
  94  */
  95 bool            pgstat_collect_startcollector = true;
  96 bool            pgstat_collect_resetonpmstart = false;
  97 bool            pgstat_collect_tuplelevel = false;
  98 bool            pgstat_collect_blocklevel = false;
  99 bool            pgstat_collect_querystring = false;
 100
 101 /* ----------
 102  * Local data
 103  * ----------
 104  */
 105 NON_EXEC_STATIC int pgStatSock = -1;
 106
 107 static struct sockaddr_storage pgStatAddr;
 108
 109 static time_t last_pgstat_start_time;
 110
 111 static bool pgStatRunningInCollector = false;
 112
 113 /*
 114  * Place where backends store per-table info to be sent to the collector.
 115  * We store shared relations separately from non-shared ones, to be able to
 116  * send them in separate messages.
 117  */
 118 typedef struct TabStatArray
 119 {
 120         int                     tsa_alloc;              /* num allocated */
 121         int                     tsa_used;               /* num actually used */
 122         PgStat_MsgTabstat **tsa_messages;       /* the array itself */
 123 } TabStatArray;
 124
 125 #define TABSTAT_QUANTUM         4       /* we alloc this many at a time */
 126
 127 static TabStatArray RegularTabStat = {0, 0, NULL};
 128 static TabStatArray SharedTabStat = {0, 0, NULL};
 129
 130 static int      pgStatXactCommit = 0;
 131 static int      pgStatXactRollback = 0;
 132
 133 static TransactionId pgStatDBHashXact = InvalidTransactionId;
 134 static HTAB *pgStatDBHash = NULL;
 135 static TransactionId pgStatLocalStatusXact = InvalidTransactionId;
 136 static PgBackendStatus *localBackendStatusTable = NULL;
 137 static int      localNumBackends = 0;
 138
 139 static volatile bool need_exit = false;
 140 static volatile bool need_statwrite = false;
 141
 142
 143 /* ----------
 144  * Local function forward declarations
 145  * ----------
 146  */
 147 #ifdef EXEC_BACKEND
 148 static pid_t pgstat_forkexec(void);
 149 #endif
 150
 151 NON_EXEC_STATIC void PgstatCollectorMain(int argc, char *argv[]);
 152 static void pgstat_exit(SIGNAL_ARGS);
 153 static void force_statwrite(SIGNAL_ARGS);
 154 static void pgstat_beshutdown_hook(int code, Datum arg);
 155
 156 static PgStat_StatDBEntry *pgstat_get_db_entry(Oid databaseid, bool create);
 157 static void pgstat_drop_database(Oid databaseid);
 158 static void pgstat_write_statsfile(void);
 159 static void pgstat_read_statsfile(HTAB **dbhash, Oid onlydb);
 160 static void backend_read_statsfile(void);
 161 static void pgstat_read_current_status(void);
 162 static HTAB *pgstat_collect_oids(Oid catalogid);
 163
 164 static void pgstat_setheader(PgStat_MsgHdr *hdr, StatMsgType mtype);
 165 static void pgstat_send(void *msg, int len);
 166
 167 static void pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len);
 168 static void pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len);
 169 static void pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len);
 170 static void pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len);
 171 static void pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len);
 172 static void pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len);
 173 static void pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len);
 174
 175
 176 /* ------------------------------------------------------------
 177  * Public functions called from postmaster follow
 178  * ------------------------------------------------------------
 179  */
 180
 181 /* ----------
 182  * pgstat_init() -
 183  *
 184  *      Called from postmaster at startup. Create the resources required
 185  *      by the statistics collector process.  If unable to do so, do not
 186  *      fail --- better to let the postmaster start with stats collection
 187  *      disabled.
 188  * ----------
 189  */
 190 void
 191 pgstat_init(void)
 192 {
 193         ACCEPT_TYPE_ARG3 alen;
 194         struct addrinfo *addrs = NULL,
 195                            *addr,
 196                                 hints;
 197         int                     ret;
 198         fd_set          rset;
 199         struct timeval tv;
 200         char            test_byte;
 201         int                     sel_res;
 202         int                     tries = 0;
 203
 204 #define TESTBYTEVAL ((char) 199)
 205
 206         /*
 207          * Force start of collector daemon if something to collect.  Note that
 208          * pgstat_collect_querystring is now an independent facility that does not
 209          * require the collector daemon.
 210          */
 211         if (pgstat_collect_tuplelevel ||
 212                 pgstat_collect_blocklevel)
 213                 pgstat_collect_startcollector = true;
 214
 215         /*
 216          * If we don't have to start a collector or should reset the collected
 217          * statistics on postmaster start, simply remove the stats file.
 218          */
 219         if (!pgstat_collect_startcollector || pgstat_collect_resetonpmstart)
 220                 pgstat_reset_all();
 221
 222         /*
 223          * Nothing else required if collector will not get started
 224          */
 225         if (!pgstat_collect_startcollector)
 226                 return;
 227
 228         /*
 229          * Create the UDP socket for sending and receiving statistic messages
 230          */
 231         hints.ai_flags = AI_PASSIVE;
 232         hints.ai_family = PF_UNSPEC;
 233         hints.ai_socktype = SOCK_DGRAM;
 234         hints.ai_protocol = 0;
 235         hints.ai_addrlen = 0;
 236         hints.ai_addr = NULL;
 237         hints.ai_canonname = NULL;
 238         hints.ai_next = NULL;
 239         ret = pg_getaddrinfo_all("localhost", NULL, &hints, &addrs);
 240         if (ret || !addrs)
 241         {
 242                 ereport(LOG,
 243                                 (errmsg("could not resolve \"localhost\": %s",
 244                                                 gai_strerror(ret))));
 245                 goto startup_failed;
 246         }
 247
 248         /*
 249          * On some platforms, pg_getaddrinfo_all() may return multiple addresses
 250          * only one of which will actually work (eg, both IPv6 and IPv4 addresses
 251          * when kernel will reject IPv6).  Worse, the failure may occur at the
 252          * bind() or perhaps even connect() stage.      So we must loop through the
 253          * results till we find a working combination. We will generate LOG
 254          * messages, but no error, for bogus combinations.
 255          */
 256         for (addr = addrs; addr; addr = addr->ai_next)
 257         {
 258 #ifdef HAVE_UNIX_SOCKETS
 259                 /* Ignore AF_UNIX sockets, if any are returned. */
 260                 if (addr->ai_family == AF_UNIX)
 261                         continue;
 262 #endif
 263
 264                 if (++tries > 1)
 265                         ereport(LOG,
 266                         (errmsg("trying another address for the statistics collector")));
 267
 268                 /*
 269                  * Create the socket.
 270                  */
 271                 if ((pgStatSock = socket(addr->ai_family, SOCK_DGRAM, 0)) < 0)
 272                 {
 273                         ereport(LOG,
 274                                         (errcode_for_socket_access(),
 275                         errmsg("could not create socket for statistics collector: %m")));
 276                         continue;
 277                 }
 278
 279                 /*
 280                  * Bind it to a kernel assigned port on localhost and get the assigned
 281                  * port via getsockname().
 282                  */
 283                 if (bind(pgStatSock, addr->ai_addr, addr->ai_addrlen) < 0)
 284                 {
 285                         ereport(LOG,
 286                                         (errcode_for_socket_access(),
 287                           errmsg("could not bind socket for statistics collector: %m")));
 288                         closesocket(pgStatSock);
 289                         pgStatSock = -1;
 290                         continue;
 291                 }
 292
 293                 alen = sizeof(pgStatAddr);
 294                 if (getsockname(pgStatSock, (struct sockaddr *) & pgStatAddr, &alen) < 0)
 295                 {
 296                         ereport(LOG,
 297                                         (errcode_for_socket_access(),
 298                                          errmsg("could not get address of socket for statistics collector: %m")));
 299                         closesocket(pgStatSock);
 300                         pgStatSock = -1;
 301                         continue;
 302                 }
 303
 304                 /*
 305                  * Connect the socket to its own address.  This saves a few cycles by
 306                  * not having to respecify the target address on every send. This also
 307                  * provides a kernel-level check that only packets from this same
 308                  * address will be received.
 309                  */
 310                 if (connect(pgStatSock, (struct sockaddr *) & pgStatAddr, alen) < 0)
 311                 {
 312                         ereport(LOG,
 313                                         (errcode_for_socket_access(),
 314                         errmsg("could not connect socket for statistics collector: %m")));
 315                         closesocket(pgStatSock);
 316                         pgStatSock = -1;
 317                         continue;
 318                 }
 319
 320                 /*
 321                  * Try to send and receive a one-byte test message on the socket. This
 322                  * is to catch situations where the socket can be created but will not
 323                  * actually pass data (for instance, because kernel packet filtering
 324                  * rules prevent it).
 325                  */
 326                 test_byte = TESTBYTEVAL;
 327
 328 retry1:
 329                 if (send(pgStatSock, &test_byte, 1, 0) != 1)
 330                 {
 331                         if (errno == EINTR)
 332                                 goto retry1;    /* if interrupted, just retry */
 333                         ereport(LOG,
 334                                         (errcode_for_socket_access(),
 335                                          errmsg("could not send test message on socket for statistics collector: %m")));
 336                         closesocket(pgStatSock);
 337                         pgStatSock = -1;
 338                         continue;
 339                 }
 340
 341                 /*
 342                  * There could possibly be a little delay before the message can be
 343                  * received.  We arbitrarily allow up to half a second before deciding
 344                  * it's broken.
 345                  */
 346                 for (;;)                                /* need a loop to handle EINTR */
 347                 {
 348                         FD_ZERO(&rset);
 349                         FD_SET(pgStatSock, &rset);
 350                         tv.tv_sec = 0;
 351                         tv.tv_usec = 500000;
 352                         sel_res = select(pgStatSock + 1, &rset, NULL, NULL, &tv);
 353                         if (sel_res >= 0 || errno != EINTR)
 354                                 break;
 355                 }
 356                 if (sel_res < 0)
 357                 {
 358                         ereport(LOG,
 359                                         (errcode_for_socket_access(),
 360                                          errmsg("select() failed in statistics collector: %m")));
 361                         closesocket(pgStatSock);
 362                         pgStatSock = -1;
 363                         continue;
 364                 }
 365                 if (sel_res == 0 || !FD_ISSET(pgStatSock, &rset))
 366                 {
 367                         /*
 368                          * This is the case we actually think is likely, so take pains to
 369                          * give a specific message for it.
 370                          *
 371                          * errno will not be set meaningfully here, so don't use it.
 372                          */
 373                         ereport(LOG,
 374                                         (errcode(ERRCODE_CONNECTION_FAILURE),
 375                                          errmsg("test message did not get through on socket for statistics collector")));
 376                         closesocket(pgStatSock);
 377                         pgStatSock = -1;
 378                         continue;
 379                 }
 380
 381                 test_byte++;                    /* just make sure variable is changed */
 382
 383 retry2:
 384                 if (recv(pgStatSock, &test_byte, 1, 0) != 1)
 385                 {
 386                         if (errno == EINTR)
 387                                 goto retry2;    /* if interrupted, just retry */
 388                         ereport(LOG,
 389                                         (errcode_for_socket_access(),
 390                                          errmsg("could not receive test message on socket for statistics collector: %m")));
 391                         closesocket(pgStatSock);
 392                         pgStatSock = -1;
 393                         continue;
 394                 }
 395
 396                 if (test_byte != TESTBYTEVAL)   /* strictly paranoia ... */
 397                 {
 398                         ereport(LOG,
 399                                         (errcode(ERRCODE_INTERNAL_ERROR),
 400                                          errmsg("incorrect test message transmission on socket for statistics collector")));
 401                         closesocket(pgStatSock);
 402                         pgStatSock = -1;
 403                         continue;
 404                 }
 405
 406                 /* If we get here, we have a working socket */
 407                 break;
 408         }
 409
 410         /* Did we find a working address? */
 411         if (!addr || pgStatSock < 0)
 412                 goto startup_failed;
 413
 414         /*
 415          * Set the socket to non-blocking IO.  This ensures that if the collector
 416          * falls behind, statistics messages will be discarded; backends won't
 417          * block waiting to send messages to the collector.
 418          */
 419         if (!pg_set_noblock(pgStatSock))
 420         {
 421                 ereport(LOG,
 422                                 (errcode_for_socket_access(),
 423                                  errmsg("could not set statistics collector socket to nonblocking mode: %m")));
 424                 goto startup_failed;
 425         }
 426
 427         pg_freeaddrinfo_all(hints.ai_family, addrs);
 428
 429         return;
 430
 431 startup_failed:
 432         ereport(LOG,
 433           (errmsg("disabling statistics collector for lack of working socket")));
 434
 435         if (addrs)
 436                 pg_freeaddrinfo_all(hints.ai_family, addrs);
 437
 438         if (pgStatSock >= 0)
 439                 closesocket(pgStatSock);
 440         pgStatSock = -1;
 441
 442         /* Adjust GUC variables to suppress useless activity */
 443         pgstat_collect_startcollector = false;
 444         pgstat_collect_tuplelevel = false;
 445         pgstat_collect_blocklevel = false;
 446 }
 447
 448 /*
 449  * pgstat_reset_all() -
 450  *
 451  * Remove the stats file.  This is used on server start if the
 452  * stats_reset_on_server_start feature is enabled, or if WAL
 453  * recovery is needed after a crash.
 454  */
 455 void
 456 pgstat_reset_all(void)
 457 {
 458         unlink(PGSTAT_STAT_FILENAME);
 459 }
 460
 461 #ifdef EXEC_BACKEND
 462
 463 /*
 464  * pgstat_forkexec() -
 465  *
 466  * Format up the arglist for, then fork and exec, statistics collector process
 467  */
 468 static pid_t
 469 pgstat_forkexec(void)
 470 {
 471         char       *av[10];
 472         int                     ac = 0;
 473
 474         av[ac++] = "postgres";
 475         av[ac++] = "--forkcol";
 476         av[ac++] = NULL;                        /* filled in by postmaster_forkexec */
 477
 478         av[ac] = NULL;
 479         Assert(ac < lengthof(av));
 480
 481         return postmaster_forkexec(ac, av);
 482 }
 483 #endif   /* EXEC_BACKEND */
 484
 485
 486 /* ----------
 487  * pgstat_start() -
 488  *
 489  *      Called from postmaster at startup or after an existing collector
 490  *      died.  Attempt to fire up a fresh statistics collector.
 491  *
 492  *      Returns PID of child process, or 0 if fail.
 493  *
 494  *      Note: if fail, we will be called again from the postmaster main loop.
 495  * ----------
 496  */
 497 int
 498 pgstat_start(void)
 499 {
 500         time_t          curtime;
 501         pid_t           pgStatPid;
 502
 503         /*
 504          * Do nothing if no collector needed
 505          */
 506         if (!pgstat_collect_startcollector)
 507                 return 0;
 508
 509         /*
 510          * Do nothing if too soon since last collector start.  This is a safety
 511          * valve to protect against continuous respawn attempts if the collector
 512          * is dying immediately at launch.      Note that since we will be re-called
 513          * from the postmaster main loop, we will get another chance later.
 514          */
 515         curtime = time(NULL);
 516         if ((unsigned int) (curtime - last_pgstat_start_time) <
 517                 (unsigned int) PGSTAT_RESTART_INTERVAL)
 518                 return 0;
 519         last_pgstat_start_time = curtime;
 520
 521         /*
 522          * Check that the socket is there, else pgstat_init failed.
 523          */
 524         if (pgStatSock < 0)
 525         {
 526                 ereport(LOG,
 527                                 (errmsg("statistics collector startup skipped")));
 528
 529                 /*
 530                  * We can only get here if someone tries to manually turn
 531                  * pgstat_collect_startcollector on after it had been off.
 532                  */
 533                 pgstat_collect_startcollector = false;
 534                 return 0;
 535         }
 536
 537         /*
 538          * Okay, fork off the collector.
 539          */
 540 #ifdef EXEC_BACKEND
 541         switch ((pgStatPid = pgstat_forkexec()))
 542 #else
 543         switch ((pgStatPid = fork_process()))
 544 #endif
 545         {
 546                 case -1:
 547                         ereport(LOG,
 548                                         (errmsg("could not fork statistics collector: %m")));
 549                         return 0;
 550
 551 #ifndef EXEC_BACKEND
 552                 case 0:
 553                         /* in postmaster child ... */
 554                         /* Close the postmaster's sockets */
 555                         ClosePostmasterPorts(false);
 556
 557                         /* Lose the postmaster's on-exit routines */
 558                         on_exit_reset();
 559
 560                         /* Drop our connection to postmaster's shared memory, as well */
 561                         PGSharedMemoryDetach();
 562
 563                         PgstatCollectorMain(0, NULL);
 564                         break;
 565 #endif
 566
 567                 default:
 568                         return (int) pgStatPid;
 569         }
 570
 571         /* shouldn't get here */
 572         return 0;
 573 }
 574
 575
 576 /* ------------------------------------------------------------
 577  * Public functions used by backends follow
 578  *------------------------------------------------------------
 579  */
 580
 581
 582 /* ----------
 583  * pgstat_report_tabstat() -
 584  *
 585  *      Called from tcop/postgres.c to send the so far collected
 586  *      per table access statistics to the collector.
 587  * ----------
 588  */
 589 void
 590 pgstat_report_tabstat(void)
 591 {
 592         int                     i;
 593
 594         if (pgStatSock < 0 ||
 595                 (!pgstat_collect_tuplelevel &&
 596                  !pgstat_collect_blocklevel))
 597         {
 598                 /* Not reporting stats, so just flush whatever we have */
 599                 RegularTabStat.tsa_used = 0;
 600                 SharedTabStat.tsa_used = 0;
 601                 return;
 602         }
 603
 604         /*
 605          * For each message buffer used during the last query set the header
 606          * fields and send it out.
 607          */
 608         for (i = 0; i < RegularTabStat.tsa_used; i++)
 609         {
 610                 PgStat_MsgTabstat *tsmsg = RegularTabStat.tsa_messages[i];
 611                 int                     n;
 612                 int                     len;
 613
 614                 n = tsmsg->m_nentries;
 615                 len = offsetof(PgStat_MsgTabstat, m_entry[0]) +
 616                         n * sizeof(PgStat_TableEntry);
 617
 618                 tsmsg->m_xact_commit = pgStatXactCommit;
 619                 tsmsg->m_xact_rollback = pgStatXactRollback;
 620                 pgStatXactCommit = 0;
 621                 pgStatXactRollback = 0;
 622
 623                 pgstat_setheader(&tsmsg->m_hdr, PGSTAT_MTYPE_TABSTAT);
 624                 tsmsg->m_databaseid = MyDatabaseId;
 625                 pgstat_send(tsmsg, len);
 626         }
 627         RegularTabStat.tsa_used = 0;
 628
 629         /* Ditto, for shared relations */
 630         for (i = 0; i < SharedTabStat.tsa_used; i++)
 631         {
 632                 PgStat_MsgTabstat *tsmsg = SharedTabStat.tsa_messages[i];
 633                 int                     n;
 634                 int                     len;
 635
 636                 n = tsmsg->m_nentries;
 637                 len = offsetof(PgStat_MsgTabstat, m_entry[0]) +
 638                         n * sizeof(PgStat_TableEntry);
 639
 640                 /* We don't report transaction commit/abort here */
 641                 tsmsg->m_xact_commit = 0;
 642                 tsmsg->m_xact_rollback = 0;
 643
 644                 pgstat_setheader(&tsmsg->m_hdr, PGSTAT_MTYPE_TABSTAT);
 645                 tsmsg->m_databaseid = InvalidOid;
 646                 pgstat_send(tsmsg, len);
 647         }
 648         SharedTabStat.tsa_used = 0;
 649 }
 650
 651
 652 /* ----------
 653  * pgstat_vacuum_tabstat() -
 654  *
 655  *      Will tell the collector about objects he can get rid of.
 656  * ----------
 657  */
 658 void
 659 pgstat_vacuum_tabstat(void)
 660 {
 661         HTAB       *htab;
 662         PgStat_MsgTabpurge msg;
 663         HASH_SEQ_STATUS hstat;
 664         PgStat_StatDBEntry *dbentry;
 665         PgStat_StatTabEntry *tabentry;
 666         int                     len;
 667
 668         if (pgStatSock < 0)
 669                 return;
 670
 671         /*
 672          * If not done for this transaction, read the statistics collector stats
 673          * file into some hash tables.
 674          */
 675         backend_read_statsfile();
 676
 677         /*
 678          * Read pg_database and make a list of OIDs of all existing databases
 679          */
 680         htab = pgstat_collect_oids(DatabaseRelationId);
 681
 682         /*
 683          * Search the database hash table for dead databases and tell the
 684          * collector to drop them.
 685          */
 686         hash_seq_init(&hstat, pgStatDBHash);
 687         while ((dbentry = (PgStat_StatDBEntry *) hash_seq_search(&hstat)) != NULL)
 688         {
 689                 Oid                     dbid = dbentry->databaseid;
 690
 691                 CHECK_FOR_INTERRUPTS();
 692
 693                 if (hash_search(htab, (void *) &dbid, HASH_FIND, NULL) == NULL)
 694                         pgstat_drop_database(dbid);
 695         }
 696
 697         /* Clean up */
 698         hash_destroy(htab);
 699
 700         /*
 701          * Lookup our own database entry; if not found, nothing more to do.
 702          */
 703         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
 704                                                                                                  (void *) &MyDatabaseId,
 705                                                                                                  HASH_FIND, NULL);
 706         if (dbentry == NULL || dbentry->tables == NULL)
 707                 return;
 708
 709         /*
 710          * Similarly to above, make a list of all known relations in this DB.
 711          */
 712         htab = pgstat_collect_oids(RelationRelationId);
 713
 714         /*
 715          * Initialize our messages table counter to zero
 716          */
 717         msg.m_nentries = 0;
 718
 719         /*
 720          * Check for all tables listed in stats hashtable if they still exist.
 721          */
 722         hash_seq_init(&hstat, dbentry->tables);
 723         while ((tabentry = (PgStat_StatTabEntry *) hash_seq_search(&hstat)) != NULL)
 724         {
 725                 Oid                     tabid = tabentry->tableid;
 726
 727                 CHECK_FOR_INTERRUPTS();
 728
 729                 if (hash_search(htab, (void *) &tabid, HASH_FIND, NULL) != NULL)
 730                         continue;
 731
 732                 /*
 733                  * Not there, so add this table's Oid to the message
 734                  */
 735                 msg.m_tableid[msg.m_nentries++] = tabid;
 736
 737                 /*
 738                  * If the message is full, send it out and reinitialize to empty
 739                  */
 740                 if (msg.m_nentries >= PGSTAT_NUM_TABPURGE)
 741                 {
 742                         len = offsetof(PgStat_MsgTabpurge, m_tableid[0])
 743                                 +msg.m_nentries * sizeof(Oid);
 744
 745                         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
 746                         msg.m_databaseid = MyDatabaseId;
 747                         pgstat_send(&msg, len);
 748
 749                         msg.m_nentries = 0;
 750                 }
 751         }
 752
 753         /*
 754          * Send the rest
 755          */
 756         if (msg.m_nentries > 0)
 757         {
 758                 len = offsetof(PgStat_MsgTabpurge, m_tableid[0])
 759                         +msg.m_nentries * sizeof(Oid);
 760
 761                 pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
 762                 msg.m_databaseid = MyDatabaseId;
 763                 pgstat_send(&msg, len);
 764         }
 765
 766         /* Clean up */
 767         hash_destroy(htab);
 768 }
 769
 770
 771 /* ----------
 772  * pgstat_collect_oids() -
 773  *
 774  *      Collect the OIDs of either all databases or all tables, according to
 775  *      the parameter, into a temporary hash table.  Caller should hash_destroy
 776  *      the result when done with it.
 777  * ----------
 778  */
 779 static HTAB *
 780 pgstat_collect_oids(Oid catalogid)
 781 {
 782         HTAB       *htab;
 783         HASHCTL         hash_ctl;
 784         Relation        rel;
 785         HeapScanDesc scan;
 786         HeapTuple       tup;
 787
 788         memset(&hash_ctl, 0, sizeof(hash_ctl));
 789         hash_ctl.keysize = sizeof(Oid);
 790         hash_ctl.entrysize = sizeof(Oid);
 791         hash_ctl.hash = oid_hash;
 792         htab = hash_create("Temporary table of OIDs",
 793                                            PGSTAT_TAB_HASH_SIZE,
 794                                            &hash_ctl,
 795                                            HASH_ELEM | HASH_FUNCTION);
 796
 797         rel = heap_open(catalogid, AccessShareLock);
 798         scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
 799         while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)
 800         {
 801                 Oid             thisoid = HeapTupleGetOid(tup);
 802
 803                 CHECK_FOR_INTERRUPTS();
 804
 805                 (void) hash_search(htab, (void *) &thisoid, HASH_ENTER, NULL);
 806         }
 807         heap_endscan(scan);
 808         heap_close(rel, AccessShareLock);
 809
 810         return htab;
 811 }
 812
 813
 814 /* ----------
 815  * pgstat_drop_database() -
 816  *
 817  *      Tell the collector that we just dropped a database.
 818  *      (If the message gets lost, we will still clean the dead DB eventually
 819  *      via future invocations of pgstat_vacuum_tabstat().)
 820  * ----------
 821  */
 822 static void
 823 pgstat_drop_database(Oid databaseid)
 824 {
 825         PgStat_MsgDropdb msg;
 826
 827         if (pgStatSock < 0)
 828                 return;
 829
 830         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_DROPDB);
 831         msg.m_databaseid = databaseid;
 832         pgstat_send(&msg, sizeof(msg));
 833 }
 834
 835
 836 /* ----------
 837  * pgstat_drop_relation() -
 838  *
 839  *      Tell the collector that we just dropped a relation.
 840  *      (If the message gets lost, we will still clean the dead entry eventually
 841  *      via future invocations of pgstat_vacuum_tabstat().)
 842  * ----------
 843  */
 844 void
 845 pgstat_drop_relation(Oid relid)
 846 {
 847         PgStat_MsgTabpurge msg;
 848         int                     len;
 849
 850         if (pgStatSock < 0)
 851                 return;
 852
 853         msg.m_tableid[0] = relid;
 854         msg.m_nentries = 1;
 855
 856         len = offsetof(PgStat_MsgTabpurge, m_tableid[0]) +sizeof(Oid);
 857
 858         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
 859         msg.m_databaseid = MyDatabaseId;
 860         pgstat_send(&msg, len);
 861 }
 862
 863
 864 /* ----------
 865  * pgstat_reset_counters() -
 866  *
 867  *      Tell the statistics collector to reset counters for our database.
 868  * ----------
 869  */
 870 void
 871 pgstat_reset_counters(void)
 872 {
 873         PgStat_MsgResetcounter msg;
 874
 875         if (pgStatSock < 0)
 876                 return;
 877
 878         if (!superuser())
 879                 ereport(ERROR,
 880                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 881                                  errmsg("must be superuser to reset statistics counters")));
 882
 883         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RESETCOUNTER);
 884         msg.m_databaseid = MyDatabaseId;
 885         pgstat_send(&msg, sizeof(msg));
 886 }
 887
 888
 889 /* ----------
 890  * pgstat_report_autovac() -
 891  *
 892  *      Called from autovacuum.c to report startup of an autovacuum process.
 893  *      We are called before InitPostgres is done, so can't rely on MyDatabaseId;
 894  *      the db OID must be passed in, instead.
 895  * ----------
 896  */
 897 void
 898 pgstat_report_autovac(Oid dboid)
 899 {
 900         PgStat_MsgAutovacStart msg;
 901
 902         if (pgStatSock < 0)
 903                 return;
 904
 905         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_AUTOVAC_START);
 906         msg.m_databaseid = dboid;
 907         msg.m_start_time = GetCurrentTimestamp();
 908
 909         pgstat_send(&msg, sizeof(msg));
 910 }
 911
 912
 913 /* ---------
 914  * pgstat_report_vacuum() -
 915  *
 916  *      Tell the collector about the table we just vacuumed.
 917  * ---------
 918  */
 919 void
 920 pgstat_report_vacuum(Oid tableoid, bool shared,
 921                                          bool analyze, PgStat_Counter tuples)
 922 {
 923         PgStat_MsgVacuum msg;
 924
 925         if (pgStatSock < 0 ||
 926                 !pgstat_collect_tuplelevel)
 927                 return;
 928
 929         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_VACUUM);
 930         msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
 931         msg.m_tableoid = tableoid;
 932         msg.m_analyze = analyze;
 933         msg.m_autovacuum = IsAutoVacuumProcess();       /* is this autovacuum? */
 934         msg.m_vacuumtime = GetCurrentTimestamp();
 935         msg.m_tuples = tuples;
 936         pgstat_send(&msg, sizeof(msg));
 937 }
 938
 939 /* --------
 940  * pgstat_report_analyze() -
 941  *
 942  *      Tell the collector about the table we just analyzed.
 943  * --------
 944  */
 945 void
 946 pgstat_report_analyze(Oid tableoid, bool shared, PgStat_Counter livetuples,
 947                                           PgStat_Counter deadtuples)
 948 {
 949         PgStat_MsgAnalyze msg;
 950
 951         if (pgStatSock < 0 ||
 952                 !pgstat_collect_tuplelevel)
 953                 return;
 954
 955         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_ANALYZE);
 956         msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
 957         msg.m_tableoid = tableoid;
 958         msg.m_autovacuum = IsAutoVacuumProcess();       /* is this autovacuum? */
 959         msg.m_analyzetime = GetCurrentTimestamp();
 960         msg.m_live_tuples = livetuples;
 961         msg.m_dead_tuples = deadtuples;
 962         pgstat_send(&msg, sizeof(msg));
 963 }
 964
 965
 966 /* ----------
 967  * pgstat_ping() -
 968  *
 969  *      Send some junk data to the collector to increase traffic.
 970  * ----------
 971  */
 972 void
 973 pgstat_ping(void)
 974 {
 975         PgStat_MsgDummy msg;
 976
 977         if (pgStatSock < 0)
 978                 return;
 979
 980         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_DUMMY);
 981         pgstat_send(&msg, sizeof(msg));
 982 }
 983
 984 /*
 985  * Enlarge a TabStatArray
 986  */
 987 static void
 988 more_tabstat_space(TabStatArray *tsarr)
 989 {
 990         PgStat_MsgTabstat *newMessages;
 991         PgStat_MsgTabstat **msgArray;
 992         int                     newAlloc;
 993         int                     i;
 994
 995         AssertArg(PointerIsValid(tsarr));
 996
 997         newAlloc = tsarr->tsa_alloc + TABSTAT_QUANTUM;
 998
 999         /* Create (another) quantum of message buffers */
1000         newMessages = (PgStat_MsgTabstat *)
1001                 MemoryContextAllocZero(TopMemoryContext,
1002                                                            sizeof(PgStat_MsgTabstat) * TABSTAT_QUANTUM);
1003
1004         /* Create or enlarge the pointer array */
1005         if (tsarr->tsa_messages == NULL)
1006                 msgArray = (PgStat_MsgTabstat **)
1007                         MemoryContextAlloc(TopMemoryContext,
1008                                                            sizeof(PgStat_MsgTabstat *) * newAlloc);
1009         else
1010                 msgArray = (PgStat_MsgTabstat **)
1011                         repalloc(tsarr->tsa_messages,
1012                                          sizeof(PgStat_MsgTabstat *) * newAlloc);
1013
1014         for (i = 0; i < TABSTAT_QUANTUM; i++)
1015                 msgArray[tsarr->tsa_alloc + i] = newMessages++;
1016         tsarr->tsa_messages = msgArray;
1017         tsarr->tsa_alloc = newAlloc;
1018
1019         Assert(tsarr->tsa_used < tsarr->tsa_alloc);
1020 }
1021
1022 /* ----------
1023  * pgstat_initstats() -
1024  *
1025  *      Called from various places usually dealing with initialization
1026  *      of Relation or Scan structures. The data placed into these
1027  *      structures from here tell where later to count for buffer reads,
1028  *      scans and tuples fetched.
1029  * ----------
1030  */
1031 void
1032 pgstat_initstats(PgStat_Info *stats, Relation rel)
1033 {
1034         Oid                     rel_id = rel->rd_id;
1035         PgStat_TableEntry *useent;
1036         TabStatArray *tsarr;
1037         PgStat_MsgTabstat *tsmsg;
1038         int                     mb;
1039         int                     i;
1040
1041         /*
1042          * Initialize data not to count at all.
1043          */
1044         stats->tabentry = NULL;
1045
1046         if (pgStatSock < 0 ||
1047                 !(pgstat_collect_tuplelevel ||
1048                   pgstat_collect_blocklevel))
1049                 return;
1050
1051         tsarr = rel->rd_rel->relisshared ? &SharedTabStat : &RegularTabStat;
1052
1053         /*
1054          * Search the already-used message slots for this relation.
1055          */
1056         for (mb = 0; mb < tsarr->tsa_used; mb++)
1057         {
1058                 tsmsg = tsarr->tsa_messages[mb];
1059
1060                 for (i = tsmsg->m_nentries; --i >= 0;)
1061                 {
1062                         if (tsmsg->m_entry[i].t_id == rel_id)
1063                         {
1064                                 stats->tabentry = (void *) &(tsmsg->m_entry[i]);
1065                                 return;
1066                         }
1067                 }
1068
1069                 if (tsmsg->m_nentries >= PGSTAT_NUM_TABENTRIES)
1070                         continue;
1071
1072                 /*
1073                  * Not found, but found a message buffer with an empty slot instead.
1074                  * Fine, let's use this one.
1075                  */
1076                 i = tsmsg->m_nentries++;
1077                 useent = &tsmsg->m_entry[i];
1078                 MemSet(useent, 0, sizeof(PgStat_TableEntry));
1079                 useent->t_id = rel_id;
1080                 stats->tabentry = (void *) useent;
1081                 return;
1082         }
1083
1084         /*
1085          * If we ran out of message buffers, we just allocate more.
1086          */
1087         if (tsarr->tsa_used >= tsarr->tsa_alloc)
1088                 more_tabstat_space(tsarr);
1089
1090         /*
1091          * Use the first entry of the next message buffer.
1092          */
1093         mb = tsarr->tsa_used++;
1094         tsmsg = tsarr->tsa_messages[mb];
1095         tsmsg->m_nentries = 1;
1096         useent = &tsmsg->m_entry[0];
1097         MemSet(useent, 0, sizeof(PgStat_TableEntry));
1098         useent->t_id = rel_id;
1099         stats->tabentry = (void *) useent;
1100 }
1101
1102
1103 /* ----------
1104  * pgstat_count_xact_commit() -
1105  *
1106  *      Called from access/transam/xact.c to count transaction commits.
1107  * ----------
1108  */
1109 void
1110 pgstat_count_xact_commit(void)
1111 {
1112         if (!pgstat_collect_tuplelevel &&
1113                 !pgstat_collect_blocklevel)
1114                 return;
1115
1116         pgStatXactCommit++;
1117
1118         /*
1119          * If there was no relation activity yet, just make one existing message
1120          * buffer used without slots, causing the next report to tell new
1121          * xact-counters.
1122          */
1123         if (RegularTabStat.tsa_alloc == 0)
1124                 more_tabstat_space(&RegularTabStat);
1125
1126         if (RegularTabStat.tsa_used == 0)
1127         {
1128                 RegularTabStat.tsa_used++;
1129                 RegularTabStat.tsa_messages[0]->m_nentries = 0;
1130         }
1131 }
1132
1133
1134 /* ----------
1135  * pgstat_count_xact_rollback() -
1136  *
1137  *      Called from access/transam/xact.c to count transaction rollbacks.
1138  * ----------
1139  */
1140 void
1141 pgstat_count_xact_rollback(void)
1142 {
1143         if (!pgstat_collect_tuplelevel &&
1144                 !pgstat_collect_blocklevel)
1145                 return;
1146
1147         pgStatXactRollback++;
1148
1149         /*
1150          * If there was no relation activity yet, just make one existing message
1151          * buffer used without slots, causing the next report to tell new
1152          * xact-counters.
1153          */
1154         if (RegularTabStat.tsa_alloc == 0)
1155                 more_tabstat_space(&RegularTabStat);
1156
1157         if (RegularTabStat.tsa_used == 0)
1158         {
1159                 RegularTabStat.tsa_used++;
1160                 RegularTabStat.tsa_messages[0]->m_nentries = 0;
1161         }
1162 }
1163
1164
1165 /* ----------
1166  * pgstat_fetch_stat_dbentry() -
1167  *
1168  *      Support function for the SQL-callable pgstat* functions. Returns
1169  *      the collected statistics for one database or NULL. NULL doesn't mean
1170  *      that the database doesn't exist, it is just not yet known by the
1171  *      collector, so the caller is better off to report ZERO instead.
1172  * ----------
1173  */
1174 PgStat_StatDBEntry *
1175 pgstat_fetch_stat_dbentry(Oid dbid)
1176 {
1177         /*
1178          * If not done for this transaction, read the statistics collector stats
1179          * file into some hash tables.
1180          */
1181         backend_read_statsfile();
1182
1183         /*
1184          * Lookup the requested database; return NULL if not found
1185          */
1186         return (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
1187                                                                                           (void *) &dbid,
1188                                                                                           HASH_FIND, NULL);
1189 }
1190
1191
1192 /* ----------
1193  * pgstat_fetch_stat_tabentry() -
1194  *
1195  *      Support function for the SQL-callable pgstat* functions. Returns
1196  *      the collected statistics for one table or NULL. NULL doesn't mean
1197  *      that the table doesn't exist, it is just not yet known by the
1198  *      collector, so the caller is better off to report ZERO instead.
1199  * ----------
1200  */
1201 PgStat_StatTabEntry *
1202 pgstat_fetch_stat_tabentry(Oid relid)
1203 {
1204         Oid                     dbid;
1205         PgStat_StatDBEntry *dbentry;
1206         PgStat_StatTabEntry *tabentry;
1207
1208         /*
1209          * If not done for this transaction, read the statistics collector stats
1210          * file into some hash tables.
1211          */
1212         backend_read_statsfile();
1213
1214         /*
1215          * Lookup our database, then look in its table hash table.
1216          */
1217         dbid = MyDatabaseId;
1218         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
1219                                                                                                  (void *) &dbid,
1220                                                                                                  HASH_FIND, NULL);
1221         if (dbentry != NULL && dbentry->tables != NULL)
1222         {
1223                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
1224                                                                                                            (void *) &relid,
1225                                                                                                            HASH_FIND, NULL);
1226                 if (tabentry)
1227                         return tabentry;
1228         }
1229
1230         /*
1231          * If we didn't find it, maybe it's a shared table.
1232          */
1233         dbid = InvalidOid;
1234         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
1235                                                                                                  (void *) &dbid,
1236                                                                                                  HASH_FIND, NULL);
1237         if (dbentry != NULL && dbentry->tables != NULL)
1238         {
1239                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
1240                                                                                                            (void *) &relid,
1241                                                                                                            HASH_FIND, NULL);
1242                 if (tabentry)
1243                         return tabentry;
1244         }
1245
1246         return NULL;
1247 }
1248
1249
1250 /* ----------
1251  * pgstat_fetch_stat_beentry() -
1252  *
1253  *      Support function for the SQL-callable pgstat* functions. Returns
1254  *      our local copy of the current-activity entry for one backend.
1255  *
1256  *      NB: caller is responsible for a check if the user is permitted to see
1257  *      this info (especially the querystring).
1258  * ----------
1259  */
1260 PgBackendStatus *
1261 pgstat_fetch_stat_beentry(int beid)
1262 {
1263         pgstat_read_current_status();
1264
1265         if (beid < 1 || beid > localNumBackends)
1266                 return NULL;
1267
1268         return &localBackendStatusTable[beid - 1];
1269 }
1270
1271
1272 /* ----------
1273  * pgstat_fetch_stat_numbackends() -
1274  *
1275  *      Support function for the SQL-callable pgstat* functions. Returns
1276  *      the maximum current backend id.
1277  * ----------
1278  */
1279 int
1280 pgstat_fetch_stat_numbackends(void)
1281 {
1282         pgstat_read_current_status();
1283
1284         return localNumBackends;
1285 }
1286
1287
1288 /* ------------------------------------------------------------
1289  * Functions for management of the shared-memory PgBackendStatus array
1290  * ------------------------------------------------------------
1291  */
1292
1293 static PgBackendStatus *BackendStatusArray = NULL;
1294 static PgBackendStatus *MyBEEntry = NULL;
1295
1296
1297 /*
1298  * Report shared-memory space needed by CreateSharedBackendStatus.
1299  */
1300 Size
1301 BackendStatusShmemSize(void)
1302 {
1303         Size            size;
1304
1305         size = mul_size(sizeof(PgBackendStatus), MaxBackends);
1306         return size;
1307 }
1308
1309 /*
1310  * Initialize the shared status array during postmaster startup.
1311  */
1312 void
1313 CreateSharedBackendStatus(void)
1314 {
1315         Size            size = BackendStatusShmemSize();
1316         bool            found;
1317
1318         /* Create or attach to the shared array */
1319         BackendStatusArray = (PgBackendStatus *)
1320                 ShmemInitStruct("Backend Status Array", size, &found);
1321
1322         if (!found)
1323         {
1324                 /*
1325                  * We're the first - initialize.
1326                  */
1327                 MemSet(BackendStatusArray, 0, size);
1328         }
1329 }
1330
1331
1332 /* ----------
1333  * pgstat_bestart() -
1334  *
1335  *      Initialize this backend's entry in the PgBackendStatus array,
1336  *      and set up an on-proc-exit hook that will clear it again.
1337  *      Called from InitPostgres.  MyBackendId and MyDatabaseId must be set.
1338  * ----------
1339  */
1340 void
1341 pgstat_bestart(void)
1342 {
1343         volatile PgBackendStatus *beentry;
1344         TimestampTz proc_start_timestamp;
1345         Oid                     userid;
1346         SockAddr        clientaddr;
1347
1348         Assert(MyBackendId >= 1 && MyBackendId <= MaxBackends);
1349         MyBEEntry = &BackendStatusArray[MyBackendId - 1];
1350
1351         /*
1352          * To minimize the time spent modifying the entry, fetch all the needed
1353          * data first.
1354          *
1355          * If we have a MyProcPort, use its session start time (for consistency,
1356          * and to save a kernel call).
1357          */
1358         if (MyProcPort)
1359                 proc_start_timestamp = MyProcPort->SessionStartTime;
1360         else
1361                 proc_start_timestamp = GetCurrentTimestamp();
1362         userid = GetSessionUserId();
1363
1364         /*
1365          * We may not have a MyProcPort (eg, if this is the autovacuum process).
1366          * If so, use all-zeroes client address, which is dealt with specially in
1367          * pg_stat_get_backend_client_addr and pg_stat_get_backend_client_port.
1368          */
1369         if (MyProcPort)
1370                 memcpy(&clientaddr, &MyProcPort->raddr, sizeof(clientaddr));
1371         else
1372                 MemSet(&clientaddr, 0, sizeof(clientaddr));
1373
1374         /*
1375          * Initialize my status entry, following the protocol of bumping
1376          * st_changecount before and after; and make sure it's even afterwards. We
1377          * use a volatile pointer here to ensure the compiler doesn't try to get
1378          * cute.
1379          */
1380         beentry = MyBEEntry;
1381         do
1382         {
1383                 beentry->st_changecount++;
1384         } while ((beentry->st_changecount & 1) == 0);
1385
1386         beentry->st_procpid = MyProcPid;
1387         beentry->st_proc_start_timestamp = proc_start_timestamp;
1388         beentry->st_activity_start_timestamp = 0;
1389         beentry->st_txn_start_timestamp = 0;
1390         beentry->st_databaseid = MyDatabaseId;
1391         beentry->st_userid = userid;
1392         beentry->st_clientaddr = clientaddr;
1393         beentry->st_waiting = false;
1394         beentry->st_activity[0] = '\0';
1395         /* Also make sure the last byte in the string area is always 0 */
1396         beentry->st_activity[PGBE_ACTIVITY_SIZE - 1] = '\0';
1397
1398         beentry->st_changecount++;
1399         Assert((beentry->st_changecount & 1) == 0);
1400
1401         /*
1402          * Set up a process-exit hook to clean up.
1403          */
1404         on_shmem_exit(pgstat_beshutdown_hook, 0);
1405 }
1406
1407 /*
1408  * Shut down a single backend's statistics reporting at process exit.
1409  *
1410  * Flush any remaining statistics counts out to the collector.
1411  * Without this, operations triggered during backend exit (such as
1412  * temp table deletions) won't be counted.
1413  *
1414  * Lastly, clear out our entry in the PgBackendStatus array.
1415  */
1416 static void
1417 pgstat_beshutdown_hook(int code, Datum arg)
1418 {
1419         volatile PgBackendStatus *beentry = MyBEEntry;
1420
1421         pgstat_report_tabstat();
1422
1423         /*
1424          * Clear my status entry, following the protocol of bumping st_changecount
1425          * before and after.  We use a volatile pointer here to ensure the
1426          * compiler doesn't try to get cute.
1427          */
1428         beentry->st_changecount++;
1429
1430         beentry->st_procpid = 0;        /* mark invalid */
1431
1432         beentry->st_changecount++;
1433         Assert((beentry->st_changecount & 1) == 0);
1434 }
1435
1436
1437 /* ----------
1438  * pgstat_report_activity() -
1439  *
1440  *      Called from tcop/postgres.c to report what the backend is actually doing
1441  *      (usually "<IDLE>" or the start of the query to be executed).
1442  * ----------
1443  */
1444 void
1445 pgstat_report_activity(const char *cmd_str)
1446 {
1447         volatile PgBackendStatus *beentry = MyBEEntry;
1448         TimestampTz start_timestamp;
1449         int                     len;
1450
1451         if (!pgstat_collect_querystring || !beentry)
1452                 return;
1453
1454         /*
1455          * To minimize the time spent modifying the entry, fetch all the needed
1456          * data first.
1457          */
1458         start_timestamp = GetCurrentStatementStartTimestamp();
1459
1460         len = strlen(cmd_str);
1461         len = pg_mbcliplen(cmd_str, len, PGBE_ACTIVITY_SIZE - 1);
1462
1463         /*
1464          * Update my status entry, following the protocol of bumping
1465          * st_changecount before and after.  We use a volatile pointer here to
1466          * ensure the compiler doesn't try to get cute.
1467          */
1468         beentry->st_changecount++;
1469
1470         beentry->st_activity_start_timestamp = start_timestamp;
1471         memcpy((char *) beentry->st_activity, cmd_str, len);
1472         beentry->st_activity[len] = '\0';
1473
1474         beentry->st_changecount++;
1475         Assert((beentry->st_changecount & 1) == 0);
1476 }
1477
1478 /*
1479  * Set the current transaction start timestamp to the specified
1480  * value. If there is no current active transaction, this is signified
1481  * by 0.
1482  */
1483 void
1484 pgstat_report_txn_timestamp(TimestampTz tstamp)
1485 {
1486         volatile PgBackendStatus *beentry = MyBEEntry;
1487
1488         if (!pgstat_collect_querystring || !beentry)
1489                 return;
1490
1491         /*
1492          * Update my status entry, following the protocol of bumping
1493          * st_changecount before and after.  We use a volatile pointer
1494          * here to ensure the compiler doesn't try to get cute.
1495          */
1496         beentry->st_changecount++;
1497         beentry->st_txn_start_timestamp = tstamp;
1498         beentry->st_changecount++;
1499         Assert((beentry->st_changecount & 1) == 0);
1500 }
1501
1502 /* ----------
1503  * pgstat_report_waiting() -
1504  *
1505  *      Called from lock manager to report beginning or end of a lock wait.
1506  *
1507  * NB: this *must* be able to survive being called before MyBEEntry has been
1508  * initialized.
1509  * ----------
1510  */
1511 void
1512 pgstat_report_waiting(bool waiting)
1513 {
1514         volatile PgBackendStatus *beentry = MyBEEntry;
1515
1516         if (!pgstat_collect_querystring || !beentry)
1517                 return;
1518
1519         /*
1520          * Since this is a single-byte field in a struct that only this process
1521          * may modify, there seems no need to bother with the st_changecount
1522          * protocol.  The update must appear atomic in any case.
1523          */
1524         beentry->st_waiting = waiting;
1525 }
1526
1527
1528 /* ----------
1529  * pgstat_read_current_status() -
1530  *
1531  *      Copy the current contents of the PgBackendStatus array to local memory,
1532  *      if not already done in this transaction.
1533  * ----------
1534  */
1535 static void
1536 pgstat_read_current_status(void)
1537 {
1538         TransactionId topXid = GetTopTransactionId();
1539         volatile PgBackendStatus *beentry;
1540         PgBackendStatus *localentry;
1541         int                     i;
1542
1543         Assert(!pgStatRunningInCollector);
1544         if (TransactionIdEquals(pgStatLocalStatusXact, topXid))
1545                 return;                                 /* already done */
1546
1547         localBackendStatusTable = (PgBackendStatus *)
1548                 MemoryContextAlloc(TopTransactionContext,
1549                                                    sizeof(PgBackendStatus) * MaxBackends);
1550         localNumBackends = 0;
1551
1552         beentry = BackendStatusArray;
1553         localentry = localBackendStatusTable;
1554         for (i = 1; i <= MaxBackends; i++)
1555         {
1556                 /*
1557                  * Follow the protocol of retrying if st_changecount changes while we
1558                  * copy the entry, or if it's odd.  (The check for odd is needed to
1559                  * cover the case where we are able to completely copy the entry while
1560                  * the source backend is between increment steps.)      We use a volatile
1561                  * pointer here to ensure the compiler doesn't try to get cute.
1562                  */
1563                 for (;;)
1564                 {
1565                         int                     save_changecount = beentry->st_changecount;
1566
1567                         /*
1568                          * XXX if PGBE_ACTIVITY_SIZE is really large, it might be best to
1569                          * use strcpy not memcpy for copying the activity string?
1570                          */
1571                         memcpy(localentry, (char *) beentry, sizeof(PgBackendStatus));
1572
1573                         if (save_changecount == beentry->st_changecount &&
1574                                 (save_changecount & 1) == 0)
1575                                 break;
1576
1577                         /* Make sure we can break out of loop if stuck... */
1578                         CHECK_FOR_INTERRUPTS();
1579                 }
1580
1581                 beentry++;
1582                 /* Only valid entries get included into the local array */
1583                 if (localentry->st_procpid > 0)
1584                 {
1585                         localentry++;
1586                         localNumBackends++;
1587                 }
1588         }
1589
1590         pgStatLocalStatusXact = topXid;
1591 }
1592
1593
1594 /* ------------------------------------------------------------
1595  * Local support functions follow
1596  * ------------------------------------------------------------
1597  */
1598
1599
1600 /* ----------
1601  * pgstat_setheader() -
1602  *
1603  *              Set common header fields in a statistics message
1604  * ----------
1605  */
1606 static void
1607 pgstat_setheader(PgStat_MsgHdr *hdr, StatMsgType mtype)
1608 {
1609         hdr->m_type = mtype;
1610 }
1611
1612
1613 /* ----------
1614  * pgstat_send() -
1615  *
1616  *              Send out one statistics message to the collector
1617  * ----------
1618  */
1619 static void
1620 pgstat_send(void *msg, int len)
1621 {
1622         int                     rc;
1623
1624         if (pgStatSock < 0)
1625                 return;
1626
1627         ((PgStat_MsgHdr *) msg)->m_size = len;
1628
1629         /* We'll retry after EINTR, but ignore all other failures */
1630         do
1631         {
1632                 rc = send(pgStatSock, msg, len, 0);
1633         } while (rc < 0 && errno == EINTR);
1634
1635 #ifdef USE_ASSERT_CHECKING
1636         /* In debug builds, log send failures ... */
1637         if (rc < 0)
1638                 elog(LOG, "could not send to statistics collector: %m");
1639 #endif
1640 }
1641
1642
1643 /* ----------
1644  * PgstatCollectorMain() -
1645  *
1646  *      Start up the statistics collector process.      This is the body of the
1647  *      postmaster child process.
1648  *
1649  *      The argc/argv parameters are valid only in EXEC_BACKEND case.
1650  * ----------
1651  */
1652 NON_EXEC_STATIC void
1653 PgstatCollectorMain(int argc, char *argv[])
1654 {
1655         struct itimerval write_timeout;
1656         bool            need_timer = false;
1657         int                     len;
1658         PgStat_Msg      msg;
1659
1660 #ifndef WIN32
1661 #ifdef HAVE_POLL
1662         struct pollfd input_fd;
1663 #else
1664         struct timeval sel_timeout;
1665         fd_set          rfds;
1666 #endif
1667 #endif
1668
1669         IsUnderPostmaster = true;       /* we are a postmaster subprocess now */
1670
1671         MyProcPid = getpid();           /* reset MyProcPid */
1672
1673         /*
1674          * If possible, make this process a group leader, so that the postmaster
1675          * can signal any child processes too.  (pgstat probably never has
1676          * any child processes, but for consistency we make all postmaster
1677          * child processes do this.)
1678          */
1679 #ifdef HAVE_SETSID
1680         if (setsid() < 0)
1681                 elog(FATAL, "setsid() failed: %m");
1682 #endif
1683
1684         /*
1685          * Ignore all signals usually bound to some action in the postmaster,
1686          * except SIGQUIT and SIGALRM.
1687          */
1688         pqsignal(SIGHUP, SIG_IGN);
1689         pqsignal(SIGINT, SIG_IGN);
1690         pqsignal(SIGTERM, SIG_IGN);
1691         pqsignal(SIGQUIT, pgstat_exit);
1692         pqsignal(SIGALRM, force_statwrite);
1693         pqsignal(SIGPIPE, SIG_IGN);
1694         pqsignal(SIGUSR1, SIG_IGN);
1695         pqsignal(SIGUSR2, SIG_IGN);
1696         pqsignal(SIGCHLD, SIG_DFL);
1697         pqsignal(SIGTTIN, SIG_DFL);
1698         pqsignal(SIGTTOU, SIG_DFL);
1699         pqsignal(SIGCONT, SIG_DFL);
1700         pqsignal(SIGWINCH, SIG_DFL);
1701         PG_SETMASK(&UnBlockSig);
1702
1703         /*
1704          * Identify myself via ps
1705          */
1706         init_ps_display("stats collector process", "", "", "");
1707
1708         /*
1709          * Arrange to write the initial status file right away
1710          */
1711         need_statwrite = true;
1712
1713         /* Preset the delay between status file writes */
1714         MemSet(&write_timeout, 0, sizeof(struct itimerval));
1715         write_timeout.it_value.tv_sec = PGSTAT_STAT_INTERVAL / 1000;
1716         write_timeout.it_value.tv_usec = PGSTAT_STAT_INTERVAL % 1000;
1717
1718         /*
1719          * Read in an existing statistics stats file or initialize the stats to
1720          * zero.
1721          */
1722         pgStatRunningInCollector = true;
1723         pgstat_read_statsfile(&pgStatDBHash, InvalidOid);
1724
1725         /*
1726          * Setup the descriptor set for select(2).      Since only one bit in the set
1727          * ever changes, we need not repeat FD_ZERO each time.
1728          */
1729 #if !defined(HAVE_POLL) && !defined(WIN32)
1730         FD_ZERO(&rfds);
1731 #endif
1732
1733         /*
1734          * Loop to process messages until we get SIGQUIT or detect ungraceful
1735          * death of our parent postmaster.
1736          *
1737          * For performance reasons, we don't want to do a PostmasterIsAlive() test
1738          * after every message; instead, do it at statwrite time and if
1739          * select()/poll() is interrupted by timeout.
1740          */
1741         for (;;)
1742         {
1743                 int                     got_data;
1744
1745                 /*
1746                  * Quit if we get SIGQUIT from the postmaster.
1747                  */
1748                 if (need_exit)
1749                         break;
1750
1751                 /*
1752                  * If time to write the stats file, do so.      Note that the alarm
1753                  * interrupt isn't re-enabled immediately, but only after we next
1754                  * receive a stats message; so no cycles are wasted when there is
1755                  * nothing going on.
1756                  */
1757                 if (need_statwrite)
1758                 {
1759                         /* Check for postmaster death; if so we'll write file below */
1760                         if (!PostmasterIsAlive(true))
1761                                 break;
1762
1763                         pgstat_write_statsfile();
1764                         need_statwrite = false;
1765                         need_timer = true;
1766                 }
1767
1768                 /*
1769                  * Wait for a message to arrive; but not for more than
1770                  * PGSTAT_SELECT_TIMEOUT seconds. (This determines how quickly we will
1771                  * shut down after an ungraceful postmaster termination; so it needn't
1772                  * be very fast.  However, on some systems SIGQUIT won't interrupt the
1773                  * poll/select call, so this also limits speed of response to SIGQUIT,
1774                  * which is more important.)
1775                  *
1776                  * We use poll(2) if available, otherwise select(2).
1777                  * Win32 has its own implementation.
1778                  */
1779 #ifndef WIN32
1780 #ifdef HAVE_POLL
1781                 input_fd.fd = pgStatSock;
1782                 input_fd.events = POLLIN | POLLERR;
1783                 input_fd.revents = 0;
1784
1785                 if (poll(&input_fd, 1, PGSTAT_SELECT_TIMEOUT * 1000) < 0)
1786                 {
1787                         if (errno == EINTR)
1788                                 continue;
1789                         ereport(ERROR,
1790                                         (errcode_for_socket_access(),
1791                                          errmsg("poll() failed in statistics collector: %m")));
1792                 }
1793
1794                 got_data = (input_fd.revents != 0);
1795 #else                                                   /* !HAVE_POLL */
1796
1797                 FD_SET(pgStatSock, &rfds);
1798
1799                 /*
1800                  * timeout struct is modified by select() on some operating systems,
1801                  * so re-fill it each time.
1802                  */
1803                 sel_timeout.tv_sec = PGSTAT_SELECT_TIMEOUT;
1804                 sel_timeout.tv_usec = 0;
1805
1806                 if (select(pgStatSock + 1, &rfds, NULL, NULL, &sel_timeout) < 0)
1807                 {
1808                         if (errno == EINTR)
1809                                 continue;
1810                         ereport(ERROR,
1811                                         (errcode_for_socket_access(),
1812                                          errmsg("select() failed in statistics collector: %m")));
1813                 }
1814
1815                 got_data = FD_ISSET(pgStatSock, &rfds);
1816 #endif   /* HAVE_POLL */
1817 #else /* WIN32 */
1818                 got_data = pgwin32_waitforsinglesocket(pgStatSock, FD_READ,
1819                                                                                            PGSTAT_SELECT_TIMEOUT*1000);
1820 #endif
1821
1822                 /*
1823                  * If there is a message on the socket, read it and check for
1824                  * validity.
1825                  */
1826                 if (got_data)
1827                 {
1828                         len = recv(pgStatSock, (char *) &msg,
1829                                            sizeof(PgStat_Msg), 0);
1830                         if (len < 0)
1831                         {
1832                                 if (errno == EINTR)
1833                                         continue;
1834                                 ereport(ERROR,
1835                                                 (errcode_for_socket_access(),
1836                                                  errmsg("could not read statistics message: %m")));
1837                         }
1838
1839                         /*
1840                          * We ignore messages that are smaller than our common header
1841                          */
1842                         if (len < sizeof(PgStat_MsgHdr))
1843                                 continue;
1844
1845                         /*
1846                          * The received length must match the length in the header
1847                          */
1848                         if (msg.msg_hdr.m_size != len)
1849                                 continue;
1850
1851                         /*
1852                          * O.K. - we accept this message.  Process it.
1853                          */
1854                         switch (msg.msg_hdr.m_type)
1855                         {
1856                                 case PGSTAT_MTYPE_DUMMY:
1857                                         break;
1858
1859                                 case PGSTAT_MTYPE_TABSTAT:
1860                                         pgstat_recv_tabstat((PgStat_MsgTabstat *) &msg, len);
1861                                         break;
1862
1863                                 case PGSTAT_MTYPE_TABPURGE:
1864                                         pgstat_recv_tabpurge((PgStat_MsgTabpurge *) &msg, len);
1865                                         break;
1866
1867                                 case PGSTAT_MTYPE_DROPDB:
1868                                         pgstat_recv_dropdb((PgStat_MsgDropdb *) &msg, len);
1869                                         break;
1870
1871                                 case PGSTAT_MTYPE_RESETCOUNTER:
1872                                         pgstat_recv_resetcounter((PgStat_MsgResetcounter *) &msg,
1873                                                                                          len);
1874                                         break;
1875
1876                                 case PGSTAT_MTYPE_AUTOVAC_START:
1877                                         pgstat_recv_autovac((PgStat_MsgAutovacStart *) &msg, len);
1878                                         break;
1879
1880                                 case PGSTAT_MTYPE_VACUUM:
1881                                         pgstat_recv_vacuum((PgStat_MsgVacuum *) &msg, len);
1882                                         break;
1883
1884                                 case PGSTAT_MTYPE_ANALYZE:
1885                                         pgstat_recv_analyze((PgStat_MsgAnalyze *) &msg, len);
1886                                         break;
1887
1888                                 default:
1889                                         break;
1890                         }
1891
1892                         /*
1893                          * If this is the first message after we wrote the stats file the
1894                          * last time, enable the alarm interrupt to make it be written
1895                          * again later.
1896                          */
1897                         if (need_timer)
1898                         {
1899                                 if (setitimer(ITIMER_REAL, &write_timeout, NULL))
1900                                         ereport(ERROR,
1901                                         (errmsg("could not set statistics collector timer: %m")));
1902                                 need_timer = false;
1903                         }
1904                 }
1905                 else
1906                 {
1907                         /*
1908                          * We can only get here if the select/poll timeout elapsed. Check
1909                          * for postmaster death.
1910                          */
1911                         if (!PostmasterIsAlive(true))
1912                                 break;
1913                 }
1914         }                                                       /* end of message-processing loop */
1915
1916         /*
1917          * Save the final stats to reuse at next startup.
1918          */
1919         pgstat_write_statsfile();
1920
1921         exit(0);
1922 }
1923
1924
1925 /* SIGQUIT signal handler for collector process */
1926 static void
1927 pgstat_exit(SIGNAL_ARGS)
1928 {
1929         need_exit = true;
1930 }
1931
1932 /* SIGALRM signal handler for collector process */
1933 static void
1934 force_statwrite(SIGNAL_ARGS)
1935 {
1936         need_statwrite = true;
1937 }
1938
1939
1940 /*
1941  * Lookup the hash table entry for the specified database. If no hash
1942  * table entry exists, initialize it, if the create parameter is true.
1943  * Else, return NULL.
1944  */
1945 static PgStat_StatDBEntry *
1946 pgstat_get_db_entry(Oid databaseid, bool create)
1947 {
1948         PgStat_StatDBEntry *result;
1949         bool            found;
1950         HASHACTION      action = (create ? HASH_ENTER : HASH_FIND);
1951
1952         /* Lookup or create the hash table entry for this database */
1953         result = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
1954                                                                                                 &databaseid,
1955                                                                                                 action, &found);
1956
1957         if (!create && !found)
1958                 return NULL;
1959
1960         /* If not found, initialize the new one. */
1961         if (!found)
1962         {
1963                 HASHCTL         hash_ctl;
1964
1965                 result->tables = NULL;
1966                 result->n_xact_commit = 0;
1967                 result->n_xact_rollback = 0;
1968                 result->n_blocks_fetched = 0;
1969                 result->n_blocks_hit = 0;
1970                 result->last_autovac_time = 0;
1971
1972                 memset(&hash_ctl, 0, sizeof(hash_ctl));
1973                 hash_ctl.keysize = sizeof(Oid);
1974                 hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
1975                 hash_ctl.hash = oid_hash;
1976                 result->tables = hash_create("Per-database table",
1977                                                                          PGSTAT_TAB_HASH_SIZE,
1978                                                                          &hash_ctl,
1979                                                                          HASH_ELEM | HASH_FUNCTION);
1980         }
1981
1982         return result;
1983 }
1984
1985
1986 /* ----------
1987  * pgstat_write_statsfile() -
1988  *
1989  *      Tell the news.
1990  * ----------
1991  */
1992 static void
1993 pgstat_write_statsfile(void)
1994 {
1995         HASH_SEQ_STATUS hstat;
1996         HASH_SEQ_STATUS tstat;
1997         PgStat_StatDBEntry *dbentry;
1998         PgStat_StatTabEntry *tabentry;
1999         FILE       *fpout;
2000         int32           format_id;
2001
2002         /*
2003          * Open the statistics temp file to write out the current values.
2004          */
2005         fpout = fopen(PGSTAT_STAT_TMPFILE, PG_BINARY_W);
2006         if (fpout == NULL)
2007         {
2008                 ereport(LOG,
2009                                 (errcode_for_file_access(),
2010                                  errmsg("could not open temporary statistics file \"%s\": %m",
2011                                                 PGSTAT_STAT_TMPFILE)));
2012                 return;
2013         }
2014
2015         /*
2016          * Write the file header --- currently just a format ID.
2017          */
2018         format_id = PGSTAT_FILE_FORMAT_ID;
2019         fwrite(&format_id, sizeof(format_id), 1, fpout);
2020
2021         /*
2022          * Walk through the database table.
2023          */
2024         hash_seq_init(&hstat, pgStatDBHash);
2025         while ((dbentry = (PgStat_StatDBEntry *) hash_seq_search(&hstat)) != NULL)
2026         {
2027                 /*
2028                  * Write out the DB entry including the number of live backends. We
2029                  * don't write the tables pointer since it's of no use to any other
2030                  * process.
2031                  */
2032                 fputc('D', fpout);
2033                 fwrite(dbentry, offsetof(PgStat_StatDBEntry, tables), 1, fpout);
2034
2035                 /*
2036                  * Walk through the database's access stats per table.
2037                  */
2038                 hash_seq_init(&tstat, dbentry->tables);
2039                 while ((tabentry = (PgStat_StatTabEntry *) hash_seq_search(&tstat)) != NULL)
2040                 {
2041                         fputc('T', fpout);
2042                         fwrite(tabentry, sizeof(PgStat_StatTabEntry), 1, fpout);
2043                 }
2044
2045                 /*
2046                  * Mark the end of this DB
2047                  */
2048                 fputc('d', fpout);
2049         }
2050
2051         /*
2052          * No more output to be done. Close the temp file and replace the old
2053          * pgstat.stat with it.  The ferror() check replaces testing for error
2054          * after each individual fputc or fwrite above.
2055          */
2056         fputc('E', fpout);
2057
2058         if (ferror(fpout))
2059         {
2060                 ereport(LOG,
2061                                 (errcode_for_file_access(),
2062                            errmsg("could not write temporary statistics file \"%s\": %m",
2063                                           PGSTAT_STAT_TMPFILE)));
2064                 fclose(fpout);
2065                 unlink(PGSTAT_STAT_TMPFILE);
2066         }
2067         else if (fclose(fpout) < 0)
2068         {
2069                 ereport(LOG,
2070                                 (errcode_for_file_access(),
2071                            errmsg("could not close temporary statistics file \"%s\": %m",
2072                                           PGSTAT_STAT_TMPFILE)));
2073                 unlink(PGSTAT_STAT_TMPFILE);
2074         }
2075         else if (rename(PGSTAT_STAT_TMPFILE, PGSTAT_STAT_FILENAME) < 0)
2076         {
2077                 ereport(LOG,
2078                                 (errcode_for_file_access(),
2079                                  errmsg("could not rename temporary statistics file \"%s\" to \"%s\": %m",
2080                                                 PGSTAT_STAT_TMPFILE, PGSTAT_STAT_FILENAME)));
2081                 unlink(PGSTAT_STAT_TMPFILE);
2082         }
2083 }
2084
2085
2086 /* ----------
2087  * pgstat_read_statsfile() -
2088  *
2089  *      Reads in an existing statistics collector file and initializes the
2090  *      databases' hash table (whose entries point to the tables' hash tables).
2091  * ----------
2092  */
2093 static void
2094 pgstat_read_statsfile(HTAB **dbhash, Oid onlydb)
2095 {
2096         PgStat_StatDBEntry *dbentry;
2097         PgStat_StatDBEntry dbbuf;
2098         PgStat_StatTabEntry *tabentry;
2099         PgStat_StatTabEntry tabbuf;
2100         HASHCTL         hash_ctl;
2101         HTAB       *tabhash = NULL;
2102         FILE       *fpin;
2103         int32           format_id;
2104         bool            found;
2105         MemoryContext use_mcxt;
2106         int                     mcxt_flags;
2107
2108         /*
2109          * If running in the collector or the autovacuum process, we use the
2110          * DynaHashCxt memory context.  If running in a backend, we use the
2111          * TopTransactionContext instead, so the caller must only know the last
2112          * XactId when this call happened to know if his tables are still valid or
2113          * already gone!
2114          */
2115         if (pgStatRunningInCollector || IsAutoVacuumProcess())
2116         {
2117                 use_mcxt = NULL;
2118                 mcxt_flags = 0;
2119         }
2120         else
2121         {
2122                 use_mcxt = TopTransactionContext;
2123                 mcxt_flags = HASH_CONTEXT;
2124         }
2125
2126         /*
2127          * Create the DB hashtable
2128          */
2129         memset(&hash_ctl, 0, sizeof(hash_ctl));
2130         hash_ctl.keysize = sizeof(Oid);
2131         hash_ctl.entrysize = sizeof(PgStat_StatDBEntry);
2132         hash_ctl.hash = oid_hash;
2133         hash_ctl.hcxt = use_mcxt;
2134         *dbhash = hash_create("Databases hash", PGSTAT_DB_HASH_SIZE, &hash_ctl,
2135                                                   HASH_ELEM | HASH_FUNCTION | mcxt_flags);
2136
2137         /*
2138          * Try to open the status file. If it doesn't exist, the backends simply
2139          * return zero for anything and the collector simply starts from scratch
2140          * with empty counters.
2141          */
2142         if ((fpin = AllocateFile(PGSTAT_STAT_FILENAME, PG_BINARY_R)) == NULL)
2143                 return;
2144
2145         /*
2146          * Verify it's of the expected format.
2147          */
2148         if (fread(&format_id, 1, sizeof(format_id), fpin) != sizeof(format_id)
2149                 || format_id != PGSTAT_FILE_FORMAT_ID)
2150         {
2151                 ereport(pgStatRunningInCollector ? LOG : WARNING,
2152                                 (errmsg("corrupted pgstat.stat file")));
2153                 goto done;
2154         }
2155
2156         /*
2157          * We found an existing collector stats file. Read it and put all the
2158          * hashtable entries into place.
2159          */
2160         for (;;)
2161         {
2162                 switch (fgetc(fpin))
2163                 {
2164                                 /*
2165                                  * 'D'  A PgStat_StatDBEntry struct describing a database
2166                                  * follows. Subsequently, zero to many 'T' entries will follow
2167                                  * until a 'd' is encountered.
2168                                  */
2169                         case 'D':
2170                                 if (fread(&dbbuf, 1, offsetof(PgStat_StatDBEntry, tables),
2171                                                   fpin) != offsetof(PgStat_StatDBEntry, tables))
2172                                 {
2173                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2174                                                         (errmsg("corrupted pgstat.stat file")));
2175                                         goto done;
2176                                 }
2177
2178                                 /*
2179                                  * Add to the DB hash
2180                                  */
2181                                 dbentry = (PgStat_StatDBEntry *) hash_search(*dbhash,
2182                                                                                                   (void *) &dbbuf.databaseid,
2183                                                                                                                          HASH_ENTER,
2184                                                                                                                          &found);
2185                                 if (found)
2186                                 {
2187                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2188                                                         (errmsg("corrupted pgstat.stat file")));
2189                                         goto done;
2190                                 }
2191
2192                                 memcpy(dbentry, &dbbuf, sizeof(PgStat_StatDBEntry));
2193                                 dbentry->tables = NULL;
2194
2195                                 /*
2196                                  * Don't collect tables if not the requested DB (or the
2197                                  * shared-table info)
2198                                  */
2199                                 if (onlydb != InvalidOid)
2200                                 {
2201                                         if (dbbuf.databaseid != onlydb &&
2202                                                 dbbuf.databaseid != InvalidOid)
2203                                                 break;
2204                                 }
2205
2206                                 memset(&hash_ctl, 0, sizeof(hash_ctl));
2207                                 hash_ctl.keysize = sizeof(Oid);
2208                                 hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
2209                                 hash_ctl.hash = oid_hash;
2210                                 hash_ctl.hcxt = use_mcxt;
2211                                 dbentry->tables = hash_create("Per-database table",
2212                                                                                           PGSTAT_TAB_HASH_SIZE,
2213                                                                                           &hash_ctl,
2214                                                                          HASH_ELEM | HASH_FUNCTION | mcxt_flags);
2215
2216                                 /*
2217                                  * Arrange that following 'T's add entries to this database's
2218                                  * tables hash table.
2219                                  */
2220                                 tabhash = dbentry->tables;
2221                                 break;
2222
2223                                 /*
2224                                  * 'd'  End of this database.
2225                                  */
2226                         case 'd':
2227                                 tabhash = NULL;
2228                                 break;
2229
2230                                 /*
2231                                  * 'T'  A PgStat_StatTabEntry follows.
2232                                  */
2233                         case 'T':
2234                                 if (fread(&tabbuf, 1, sizeof(PgStat_StatTabEntry),
2235                                                   fpin) != sizeof(PgStat_StatTabEntry))
2236                                 {
2237                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2238                                                         (errmsg("corrupted pgstat.stat file")));
2239                                         goto done;
2240                                 }
2241
2242                                 /*
2243                                  * Skip if table belongs to a not requested database.
2244                                  */
2245                                 if (tabhash == NULL)
2246                                         break;
2247
2248                                 tabentry = (PgStat_StatTabEntry *) hash_search(tabhash,
2249                                                                                                         (void *) &tabbuf.tableid,
2250                                                                                                                  HASH_ENTER, &found);
2251
2252                                 if (found)
2253                                 {
2254                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2255                                                         (errmsg("corrupted pgstat.stat file")));
2256                                         goto done;
2257                                 }
2258
2259                                 memcpy(tabentry, &tabbuf, sizeof(tabbuf));
2260                                 break;
2261
2262                                 /*
2263                                  * 'E'  The EOF marker of a complete stats file.
2264                                  */
2265                         case 'E':
2266                                 goto done;
2267
2268                         default:
2269                                 ereport(pgStatRunningInCollector ? LOG : WARNING,
2270                                                 (errmsg("corrupted pgstat.stat file")));
2271                                 goto done;
2272                 }
2273         }
2274
2275 done:
2276         FreeFile(fpin);
2277 }
2278
2279 /*
2280  * If not done for this transaction, read the statistics collector
2281  * stats file into some hash tables.
2282  *
2283  * Because we store the tables in TopTransactionContext, the result
2284  * is good for the entire current main transaction.
2285  *
2286  * Inside the autovacuum process, the statfile is assumed to be valid
2287  * "forever", that is one iteration, within one database.  This means
2288  * we only consider the statistics as they were when the autovacuum
2289  * iteration started.
2290  */
2291 static void
2292 backend_read_statsfile(void)
2293 {
2294         if (IsAutoVacuumProcess())
2295         {
2296                 /* already read it? */
2297                 if (pgStatDBHash)
2298                         return;
2299                 Assert(!pgStatRunningInCollector);
2300                 pgstat_read_statsfile(&pgStatDBHash, InvalidOid);
2301         }
2302         else
2303         {
2304                 TransactionId topXid = GetTopTransactionId();
2305
2306                 if (!TransactionIdEquals(pgStatDBHashXact, topXid))
2307                 {
2308                         Assert(!pgStatRunningInCollector);
2309                         pgstat_read_statsfile(&pgStatDBHash, MyDatabaseId);
2310                         pgStatDBHashXact = topXid;
2311                 }
2312         }
2313 }
2314
2315 /* ----------
2316  * pgstat_recv_tabstat() -
2317  *
2318  *      Count what the backend has done.
2319  * ----------
2320  */
2321 static void
2322 pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
2323 {
2324         PgStat_TableEntry *tabmsg = &(msg->m_entry[0]);
2325         PgStat_StatDBEntry *dbentry;
2326         PgStat_StatTabEntry *tabentry;
2327         int                     i;
2328         bool            found;
2329
2330         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
2331
2332         /*
2333          * Update database-wide stats.
2334          */
2335         dbentry->n_xact_commit += (PgStat_Counter) (msg->m_xact_commit);
2336         dbentry->n_xact_rollback += (PgStat_Counter) (msg->m_xact_rollback);
2337
2338         /*
2339          * Process all table entries in the message.
2340          */
2341         for (i = 0; i < msg->m_nentries; i++)
2342         {
2343                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
2344                                                                                                   (void *) &(tabmsg[i].t_id),
2345                                                                                                            HASH_ENTER, &found);
2346
2347                 if (!found)
2348                 {
2349                         /*
2350                          * If it's a new table entry, initialize counters to the values we
2351                          * just got.
2352                          */
2353                         tabentry->numscans = tabmsg[i].t_numscans;
2354                         tabentry->tuples_returned = tabmsg[i].t_tuples_returned;
2355                         tabentry->tuples_fetched = tabmsg[i].t_tuples_fetched;
2356                         tabentry->tuples_inserted = tabmsg[i].t_tuples_inserted;
2357                         tabentry->tuples_updated = tabmsg[i].t_tuples_updated;
2358                         tabentry->tuples_deleted = tabmsg[i].t_tuples_deleted;
2359
2360                         tabentry->n_live_tuples = tabmsg[i].t_tuples_inserted;
2361                         tabentry->n_dead_tuples = tabmsg[i].t_tuples_updated +
2362                                 tabmsg[i].t_tuples_deleted;
2363                         tabentry->last_anl_tuples = 0;
2364                         tabentry->vacuum_timestamp = 0;
2365                         tabentry->autovac_vacuum_timestamp = 0;
2366                         tabentry->analyze_timestamp = 0;
2367                         tabentry->autovac_analyze_timestamp = 0;
2368
2369                         tabentry->blocks_fetched = tabmsg[i].t_blocks_fetched;
2370                         tabentry->blocks_hit = tabmsg[i].t_blocks_hit;
2371                 }
2372                 else
2373                 {
2374                         /*
2375                          * Otherwise add the values to the existing entry.
2376                          */
2377                         tabentry->numscans += tabmsg[i].t_numscans;
2378                         tabentry->tuples_returned += tabmsg[i].t_tuples_returned;
2379                         tabentry->tuples_fetched += tabmsg[i].t_tuples_fetched;
2380                         tabentry->tuples_inserted += tabmsg[i].t_tuples_inserted;
2381                         tabentry->tuples_updated += tabmsg[i].t_tuples_updated;
2382                         tabentry->tuples_deleted += tabmsg[i].t_tuples_deleted;
2383
2384                         tabentry->n_live_tuples += tabmsg[i].t_tuples_inserted -
2385                                 tabmsg[i].t_tuples_deleted;
2386                         tabentry->n_dead_tuples += tabmsg[i].t_tuples_updated +
2387                                 tabmsg[i].t_tuples_deleted;
2388
2389                         tabentry->blocks_fetched += tabmsg[i].t_blocks_fetched;
2390                         tabentry->blocks_hit += tabmsg[i].t_blocks_hit;
2391                 }
2392
2393                 /*
2394                  * And add the block IO to the database entry.
2395                  */
2396                 dbentry->n_blocks_fetched += tabmsg[i].t_blocks_fetched;
2397                 dbentry->n_blocks_hit += tabmsg[i].t_blocks_hit;
2398         }
2399 }
2400
2401
2402 /* ----------
2403  * pgstat_recv_tabpurge() -
2404  *
2405  *      Arrange for dead table removal.
2406  * ----------
2407  */
2408 static void
2409 pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len)
2410 {
2411         PgStat_StatDBEntry *dbentry;
2412         int                     i;
2413
2414         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
2415
2416         /*
2417          * No need to purge if we don't even know the database.
2418          */
2419         if (!dbentry || !dbentry->tables)
2420                 return;
2421
2422         /*
2423          * Process all table entries in the message.
2424          */
2425         for (i = 0; i < msg->m_nentries; i++)
2426         {
2427                 /* Remove from hashtable if present; we don't care if it's not. */
2428                 (void) hash_search(dbentry->tables,
2429                                                    (void *) &(msg->m_tableid[i]),
2430                                                    HASH_REMOVE, NULL);
2431         }
2432 }
2433
2434
2435 /* ----------
2436  * pgstat_recv_dropdb() -
2437  *
2438  *      Arrange for dead database removal
2439  * ----------
2440  */
2441 static void
2442 pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len)
2443 {
2444         PgStat_StatDBEntry *dbentry;
2445
2446         /*
2447          * Lookup the database in the hashtable.
2448          */
2449         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
2450
2451         /*
2452          * If found, remove it.
2453          */
2454         if (dbentry)
2455         {
2456                 if (dbentry->tables != NULL)
2457                         hash_destroy(dbentry->tables);
2458
2459                 if (hash_search(pgStatDBHash,
2460                                                 (void *) &(dbentry->databaseid),
2461                                                 HASH_REMOVE, NULL) == NULL)
2462                         ereport(ERROR,
2463                                         (errmsg("database hash table corrupted "
2464                                                         "during cleanup --- abort")));
2465         }
2466 }
2467
2468
2469 /* ----------
2470  * pgstat_recv_resetcounter() -
2471  *
2472  *      Reset the statistics for the specified database.
2473  * ----------
2474  */
2475 static void
2476 pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len)
2477 {
2478         HASHCTL         hash_ctl;
2479         PgStat_StatDBEntry *dbentry;
2480
2481         /*
2482          * Lookup the database in the hashtable.  Nothing to do if not there.
2483          */
2484         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
2485
2486         if (!dbentry)
2487                 return;
2488
2489         /*
2490          * We simply throw away all the database's table entries by recreating a
2491          * new hash table for them.
2492          */
2493         if (dbentry->tables != NULL)
2494                 hash_destroy(dbentry->tables);
2495
2496         dbentry->tables = NULL;
2497         dbentry->n_xact_commit = 0;
2498         dbentry->n_xact_rollback = 0;
2499         dbentry->n_blocks_fetched = 0;
2500         dbentry->n_blocks_hit = 0;
2501
2502         memset(&hash_ctl, 0, sizeof(hash_ctl));
2503         hash_ctl.keysize = sizeof(Oid);
2504         hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
2505         hash_ctl.hash = oid_hash;
2506         dbentry->tables = hash_create("Per-database table",
2507                                                                   PGSTAT_TAB_HASH_SIZE,
2508                                                                   &hash_ctl,
2509                                                                   HASH_ELEM | HASH_FUNCTION);
2510 }
2511
2512 /* ----------
2513  * pgstat_recv_autovac() -
2514  *
2515  *      Process an autovacuum signalling message.
2516  * ----------
2517  */
2518 static void
2519 pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len)
2520 {
2521         PgStat_StatDBEntry *dbentry;
2522
2523         /*
2524          * Lookup the database in the hashtable.  Don't create the entry if it
2525          * doesn't exist, because autovacuum may be processing a template
2526          * database.  If this isn't the case, the database is most likely to have
2527          * an entry already.  (If it doesn't, not much harm is done anyway --
2528          * it'll get created as soon as somebody actually uses the database.)
2529          */
2530         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
2531         if (dbentry == NULL)
2532                 return;
2533
2534         /*
2535          * Store the last autovacuum time in the database entry.
2536          */
2537         dbentry->last_autovac_time = msg->m_start_time;
2538 }
2539
2540 /* ----------
2541  * pgstat_recv_vacuum() -
2542  *
2543  *      Process a VACUUM message.
2544  * ----------
2545  */
2546 static void
2547 pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len)
2548 {
2549         PgStat_StatDBEntry *dbentry;
2550         PgStat_StatTabEntry *tabentry;
2551
2552         /*
2553          * Don't create either the database or table entry if it doesn't already
2554          * exist.  This avoids bloating the stats with entries for stuff that is
2555          * only touched by vacuum and not by live operations.
2556          */
2557         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
2558         if (dbentry == NULL)
2559                 return;
2560
2561         tabentry = hash_search(dbentry->tables, &(msg->m_tableoid),
2562                                                    HASH_FIND, NULL);
2563         if (tabentry == NULL)
2564                 return;
2565
2566         if (msg->m_autovacuum)
2567                 tabentry->autovac_vacuum_timestamp = msg->m_vacuumtime;
2568         else
2569                 tabentry->vacuum_timestamp = msg->m_vacuumtime;
2570         tabentry->n_live_tuples = msg->m_tuples;
2571         tabentry->n_dead_tuples = 0;
2572         if (msg->m_analyze)
2573         {
2574                 tabentry->last_anl_tuples = msg->m_tuples;
2575                 if (msg->m_autovacuum)
2576                         tabentry->autovac_analyze_timestamp = msg->m_vacuumtime;
2577                 else
2578                         tabentry->analyze_timestamp = msg->m_vacuumtime;
2579         }
2580         else
2581         {
2582                 /* last_anl_tuples must never exceed n_live_tuples */
2583                 tabentry->last_anl_tuples = Min(tabentry->last_anl_tuples,
2584                                                                                 msg->m_tuples);
2585         }
2586 }
2587
2588 /* ----------
2589  * pgstat_recv_analyze() -
2590  *
2591  *      Process an ANALYZE message.
2592  * ----------
2593  */
2594 static void
2595 pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len)
2596 {
2597         PgStat_StatDBEntry *dbentry;
2598         PgStat_StatTabEntry *tabentry;
2599
2600         /*
2601          * Don't create either the database or table entry if it doesn't already
2602          * exist.  This avoids bloating the stats with entries for stuff that is
2603          * only touched by analyze and not by live operations.
2604          */
2605         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
2606         if (dbentry == NULL)
2607                 return;
2608
2609         tabentry = hash_search(dbentry->tables, &(msg->m_tableoid),
2610                                                    HASH_FIND, NULL);
2611         if (tabentry == NULL)
2612                 return;
2613
2614         if (msg->m_autovacuum)
2615                 tabentry->autovac_analyze_timestamp = msg->m_analyzetime;
2616         else
2617                 tabentry->analyze_timestamp = msg->m_analyzetime;
2618         tabentry->n_live_tuples = msg->m_live_tuples;
2619         tabentry->n_dead_tuples = msg->m_dead_tuples;
2620         tabentry->last_anl_tuples = msg->m_live_tuples + msg->m_dead_tuples;
2621 }