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