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