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-2010, PostgreSQL Global Development Group
  15  *
  16  *      $PostgreSQL: pgsql/src/backend/postmaster/pgstat.c,v 1.205 2010/08/08 16:27:03 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/twophase_rmgr.h"
  43 #include "access/xact.h"
  44 #include "catalog/pg_database.h"
  45 #include "catalog/pg_proc.h"
  46 #include "libpq/ip.h"
  47 #include "libpq/libpq.h"
  48 #include "libpq/pqsignal.h"
  49 #include "mb/pg_wchar.h"
  50 #include "miscadmin.h"
  51 #include "pg_trace.h"
  52 #include "postmaster/autovacuum.h"
  53 #include "postmaster/fork_process.h"
  54 #include "postmaster/postmaster.h"
  55 #include "storage/backendid.h"
  56 #include "storage/fd.h"
  57 #include "storage/ipc.h"
  58 #include "storage/pg_shmem.h"
  59 #include "storage/pmsignal.h"
  60 #include "utils/guc.h"
  61 #include "utils/memutils.h"
  62 #include "utils/ps_status.h"
  63 #include "utils/rel.h"
  64 #include "utils/tqual.h"
  65
  66
  67 /* ----------
  68  * Paths for the statistics files (relative to installation's $PGDATA).
  69  * ----------
  70  */
  71 #define PGSTAT_STAT_PERMANENT_FILENAME          "global/pgstat.stat"
  72 #define PGSTAT_STAT_PERMANENT_TMPFILE           "global/pgstat.tmp"
  73
  74 /* ----------
  75  * Timer definitions.
  76  * ----------
  77  */
  78 #define PGSTAT_STAT_INTERVAL    500             /* Minimum time between stats file
  79                                                                                  * updates; in milliseconds. */
  80
  81 #define PGSTAT_RETRY_DELAY              10              /* How long to wait between statistics
  82                                                                                  * update requests; in milliseconds. */
  83
  84 #define PGSTAT_MAX_WAIT_TIME    5000    /* Maximum time to wait for a stats
  85                                                                                  * file update; in milliseconds. */
  86
  87 #define PGSTAT_RESTART_INTERVAL 60              /* How often to attempt to restart a
  88                                                                                  * failed statistics collector; in
  89                                                                                  * seconds. */
  90
  91 #define PGSTAT_SELECT_TIMEOUT   2               /* How often to check for postmaster
  92                                                                                  * death; in seconds. */
  93
  94 #define PGSTAT_POLL_LOOP_COUNT  (PGSTAT_MAX_WAIT_TIME / PGSTAT_RETRY_DELAY)
  95
  96
  97 /* ----------
  98  * The initial size hints for the hash tables used in the collector.
  99  * ----------
 100  */
 101 #define PGSTAT_DB_HASH_SIZE             16
 102 #define PGSTAT_TAB_HASH_SIZE    512
 103 #define PGSTAT_FUNCTION_HASH_SIZE       512
 104
 105
 106 /* ----------
 107  * GUC parameters
 108  * ----------
 109  */
 110 bool            pgstat_track_activities = false;
 111 bool            pgstat_track_counts = false;
 112 int                     pgstat_track_functions = TRACK_FUNC_OFF;
 113 int                     pgstat_track_activity_query_size = 1024;
 114
 115 /* ----------
 116  * Built from GUC parameter
 117  * ----------
 118  */
 119 char       *pgstat_stat_filename = NULL;
 120 char       *pgstat_stat_tmpname = NULL;
 121
 122 /*
 123  * BgWriter global statistics counters (unused in other processes).
 124  * Stored directly in a stats message structure so it can be sent
 125  * without needing to copy things around.  We assume this inits to zeroes.
 126  */
 127 PgStat_MsgBgWriter BgWriterStats;
 128
 129 /* ----------
 130  * Local data
 131  * ----------
 132  */
 133 NON_EXEC_STATIC pgsocket pgStatSock = PGINVALID_SOCKET;
 134
 135 static struct sockaddr_storage pgStatAddr;
 136
 137 static time_t last_pgstat_start_time;
 138
 139 static bool pgStatRunningInCollector = false;
 140
 141 /*
 142  * Structures in which backends store per-table info that's waiting to be
 143  * sent to the collector.
 144  *
 145  * NOTE: once allocated, TabStatusArray structures are never moved or deleted
 146  * for the life of the backend.  Also, we zero out the t_id fields of the
 147  * contained PgStat_TableStatus structs whenever they are not actively in use.
 148  * This allows relcache pgstat_info pointers to be treated as long-lived data,
 149  * avoiding repeated searches in pgstat_initstats() when a relation is
 150  * repeatedly opened during a transaction.
 151  */
 152 #define TABSTAT_QUANTUM         100 /* we alloc this many at a time */
 153
 154 typedef struct TabStatusArray
 155 {
 156         struct TabStatusArray *tsa_next;        /* link to next array, if any */
 157         int                     tsa_used;               /* # entries currently used */
 158         PgStat_TableStatus tsa_entries[TABSTAT_QUANTUM];        /* per-table data */
 159 } TabStatusArray;
 160
 161 static TabStatusArray *pgStatTabList = NULL;
 162
 163 /*
 164  * Backends store per-function info that's waiting to be sent to the collector
 165  * in this hash table (indexed by function OID).
 166  */
 167 static HTAB *pgStatFunctions = NULL;
 168
 169 /*
 170  * Indicates if backend has some function stats that it hasn't yet
 171  * sent to the collector.
 172  */
 173 static bool have_function_stats = false;
 174
 175 /*
 176  * Tuple insertion/deletion counts for an open transaction can't be propagated
 177  * into PgStat_TableStatus counters until we know if it is going to commit
 178  * or abort.  Hence, we keep these counts in per-subxact structs that live
 179  * in TopTransactionContext.  This data structure is designed on the assumption
 180  * that subxacts won't usually modify very many tables.
 181  */
 182 typedef struct PgStat_SubXactStatus
 183 {
 184         int                     nest_level;             /* subtransaction nest level */
 185         struct PgStat_SubXactStatus *prev;      /* higher-level subxact if any */
 186         PgStat_TableXactStatus *first;          /* head of list for this subxact */
 187 } PgStat_SubXactStatus;
 188
 189 static PgStat_SubXactStatus *pgStatXactStack = NULL;
 190
 191 static int      pgStatXactCommit = 0;
 192 static int      pgStatXactRollback = 0;
 193
 194 /* Record that's written to 2PC state file when pgstat state is persisted */
 195 typedef struct TwoPhasePgStatRecord
 196 {
 197         PgStat_Counter tuples_inserted;         /* tuples inserted in xact */
 198         PgStat_Counter tuples_updated;          /* tuples updated in xact */
 199         PgStat_Counter tuples_deleted;          /* tuples deleted in xact */
 200         Oid                     t_id;                   /* table's OID */
 201         bool            t_shared;               /* is it a shared catalog? */
 202 } TwoPhasePgStatRecord;
 203
 204 /*
 205  * Info about current "snapshot" of stats file
 206  */
 207 static MemoryContext pgStatLocalContext = NULL;
 208 static HTAB *pgStatDBHash = NULL;
 209 static PgBackendStatus *localBackendStatusTable = NULL;
 210 static int      localNumBackends = 0;
 211
 212 /*
 213  * Cluster wide statistics, kept in the stats collector.
 214  * Contains statistics that are not collected per database
 215  * or per table.
 216  */
 217 static PgStat_GlobalStats globalStats;
 218
 219 /* Last time the collector successfully wrote the stats file */
 220 static TimestampTz last_statwrite;
 221
 222 /* Latest statistics request time from backends */
 223 static TimestampTz last_statrequest;
 224
 225 static volatile bool need_exit = false;
 226 static volatile bool got_SIGHUP = false;
 227
 228 /*
 229  * Total time charged to functions so far in the current backend.
 230  * We use this to help separate "self" and "other" time charges.
 231  * (We assume this initializes to zero.)
 232  */
 233 static instr_time total_func_time;
 234
 235
 236 /* ----------
 237  * Local function forward declarations
 238  * ----------
 239  */
 240 #ifdef EXEC_BACKEND
 241 static pid_t pgstat_forkexec(void);
 242 #endif
 243
 244 NON_EXEC_STATIC void PgstatCollectorMain(int argc, char *argv[]);
 245 static void pgstat_exit(SIGNAL_ARGS);
 246 static void pgstat_beshutdown_hook(int code, Datum arg);
 247 static void pgstat_sighup_handler(SIGNAL_ARGS);
 248
 249 static PgStat_StatDBEntry *pgstat_get_db_entry(Oid databaseid, bool create);
 250 static PgStat_StatTabEntry *pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry,
 251                                          Oid tableoid, bool create);
 252 static void pgstat_write_statsfile(bool permanent);
 253 static HTAB *pgstat_read_statsfile(Oid onlydb, bool permanent);
 254 static void backend_read_statsfile(void);
 255 static void pgstat_read_current_status(void);
 256
 257 static void pgstat_send_tabstat(PgStat_MsgTabstat *tsmsg);
 258 static void pgstat_send_funcstats(void);
 259 static HTAB *pgstat_collect_oids(Oid catalogid);
 260
 261 static PgStat_TableStatus *get_tabstat_entry(Oid rel_id, bool isshared);
 262
 263 static void pgstat_setup_memcxt(void);
 264
 265 static void pgstat_setheader(PgStat_MsgHdr *hdr, StatMsgType mtype);
 266 static void pgstat_send(void *msg, int len);
 267
 268 static void pgstat_recv_inquiry(PgStat_MsgInquiry *msg, int len);
 269 static void pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len);
 270 static void pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len);
 271 static void pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len);
 272 static void pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len);
 273 static void pgstat_recv_resetsharedcounter(PgStat_MsgResetsharedcounter *msg, int len);
 274 static void pgstat_recv_resetsinglecounter(PgStat_MsgResetsinglecounter *msg, int len);
 275 static void pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len);
 276 static void pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len);
 277 static void pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len);
 278 static void pgstat_recv_bgwriter(PgStat_MsgBgWriter *msg, int len);
 279 static void pgstat_recv_funcstat(PgStat_MsgFuncstat *msg, int len);
 280 static void pgstat_recv_funcpurge(PgStat_MsgFuncpurge *msg, int len);
 281
 282
 283 /* ------------------------------------------------------------
 284  * Public functions called from postmaster follow
 285  * ------------------------------------------------------------
 286  */
 287
 288 /* ----------
 289  * pgstat_init() -
 290  *
 291  *      Called from postmaster at startup. Create the resources required
 292  *      by the statistics collector process.  If unable to do so, do not
 293  *      fail --- better to let the postmaster start with stats collection
 294  *      disabled.
 295  * ----------
 296  */
 297 void
 298 pgstat_init(void)
 299 {
 300         ACCEPT_TYPE_ARG3 alen;
 301         struct addrinfo *addrs = NULL,
 302                            *addr,
 303                                 hints;
 304         int                     ret;
 305         fd_set          rset;
 306         struct timeval tv;
 307         char            test_byte;
 308         int                     sel_res;
 309         int                     tries = 0;
 310
 311 #define TESTBYTEVAL ((char) 199)
 312
 313         /*
 314          * Create the UDP socket for sending and receiving statistic messages
 315          */
 316         hints.ai_flags = AI_PASSIVE;
 317         hints.ai_family = PF_UNSPEC;
 318         hints.ai_socktype = SOCK_DGRAM;
 319         hints.ai_protocol = 0;
 320         hints.ai_addrlen = 0;
 321         hints.ai_addr = NULL;
 322         hints.ai_canonname = NULL;
 323         hints.ai_next = NULL;
 324         ret = pg_getaddrinfo_all("localhost", NULL, &hints, &addrs);
 325         if (ret || !addrs)
 326         {
 327                 ereport(LOG,
 328                                 (errmsg("could not resolve \"localhost\": %s",
 329                                                 gai_strerror(ret))));
 330                 goto startup_failed;
 331         }
 332
 333         /*
 334          * On some platforms, pg_getaddrinfo_all() may return multiple addresses
 335          * only one of which will actually work (eg, both IPv6 and IPv4 addresses
 336          * when kernel will reject IPv6).  Worse, the failure may occur at the
 337          * bind() or perhaps even connect() stage.      So we must loop through the
 338          * results till we find a working combination. We will generate LOG
 339          * messages, but no error, for bogus combinations.
 340          */
 341         for (addr = addrs; addr; addr = addr->ai_next)
 342         {
 343 #ifdef HAVE_UNIX_SOCKETS
 344                 /* Ignore AF_UNIX sockets, if any are returned. */
 345                 if (addr->ai_family == AF_UNIX)
 346                         continue;
 347 #endif
 348
 349                 if (++tries > 1)
 350                         ereport(LOG,
 351                         (errmsg("trying another address for the statistics collector")));
 352
 353                 /*
 354                  * Create the socket.
 355                  */
 356                 if ((pgStatSock = socket(addr->ai_family, SOCK_DGRAM, 0)) == PGINVALID_SOCKET)
 357                 {
 358                         ereport(LOG,
 359                                         (errcode_for_socket_access(),
 360                         errmsg("could not create socket for statistics collector: %m")));
 361                         continue;
 362                 }
 363
 364                 /*
 365                  * Bind it to a kernel assigned port on localhost and get the assigned
 366                  * port via getsockname().
 367                  */
 368                 if (bind(pgStatSock, addr->ai_addr, addr->ai_addrlen) < 0)
 369                 {
 370                         ereport(LOG,
 371                                         (errcode_for_socket_access(),
 372                           errmsg("could not bind socket for statistics collector: %m")));
 373                         closesocket(pgStatSock);
 374                         pgStatSock = PGINVALID_SOCKET;
 375                         continue;
 376                 }
 377
 378                 alen = sizeof(pgStatAddr);
 379                 if (getsockname(pgStatSock, (struct sockaddr *) & pgStatAddr, &alen) < 0)
 380                 {
 381                         ereport(LOG,
 382                                         (errcode_for_socket_access(),
 383                                          errmsg("could not get address of socket for statistics collector: %m")));
 384                         closesocket(pgStatSock);
 385                         pgStatSock = PGINVALID_SOCKET;
 386                         continue;
 387                 }
 388
 389                 /*
 390                  * Connect the socket to its own address.  This saves a few cycles by
 391                  * not having to respecify the target address on every send. This also
 392                  * provides a kernel-level check that only packets from this same
 393                  * address will be received.
 394                  */
 395                 if (connect(pgStatSock, (struct sockaddr *) & pgStatAddr, alen) < 0)
 396                 {
 397                         ereport(LOG,
 398                                         (errcode_for_socket_access(),
 399                         errmsg("could not connect socket for statistics collector: %m")));
 400                         closesocket(pgStatSock);
 401                         pgStatSock = PGINVALID_SOCKET;
 402                         continue;
 403                 }
 404
 405                 /*
 406                  * Try to send and receive a one-byte test message on the socket. This
 407                  * is to catch situations where the socket can be created but will not
 408                  * actually pass data (for instance, because kernel packet filtering
 409                  * rules prevent it).
 410                  */
 411                 test_byte = TESTBYTEVAL;
 412
 413 retry1:
 414                 if (send(pgStatSock, &test_byte, 1, 0) != 1)
 415                 {
 416                         if (errno == EINTR)
 417                                 goto retry1;    /* if interrupted, just retry */
 418                         ereport(LOG,
 419                                         (errcode_for_socket_access(),
 420                                          errmsg("could not send test message on socket for statistics collector: %m")));
 421                         closesocket(pgStatSock);
 422                         pgStatSock = PGINVALID_SOCKET;
 423                         continue;
 424                 }
 425
 426                 /*
 427                  * There could possibly be a little delay before the message can be
 428                  * received.  We arbitrarily allow up to half a second before deciding
 429                  * it's broken.
 430                  */
 431                 for (;;)                                /* need a loop to handle EINTR */
 432                 {
 433                         FD_ZERO(&rset);
 434                         FD_SET(pgStatSock, &rset);
 435
 436                         tv.tv_sec = 0;
 437                         tv.tv_usec = 500000;
 438                         sel_res = select(pgStatSock + 1, &rset, NULL, NULL, &tv);
 439                         if (sel_res >= 0 || errno != EINTR)
 440                                 break;
 441                 }
 442                 if (sel_res < 0)
 443                 {
 444                         ereport(LOG,
 445                                         (errcode_for_socket_access(),
 446                                          errmsg("select() failed in statistics collector: %m")));
 447                         closesocket(pgStatSock);
 448                         pgStatSock = PGINVALID_SOCKET;
 449                         continue;
 450                 }
 451                 if (sel_res == 0 || !FD_ISSET(pgStatSock, &rset))
 452                 {
 453                         /*
 454                          * This is the case we actually think is likely, so take pains to
 455                          * give a specific message for it.
 456                          *
 457                          * errno will not be set meaningfully here, so don't use it.
 458                          */
 459                         ereport(LOG,
 460                                         (errcode(ERRCODE_CONNECTION_FAILURE),
 461                                          errmsg("test message did not get through on socket for statistics collector")));
 462                         closesocket(pgStatSock);
 463                         pgStatSock = PGINVALID_SOCKET;
 464                         continue;
 465                 }
 466
 467                 test_byte++;                    /* just make sure variable is changed */
 468
 469 retry2:
 470                 if (recv(pgStatSock, &test_byte, 1, 0) != 1)
 471                 {
 472                         if (errno == EINTR)
 473                                 goto retry2;    /* if interrupted, just retry */
 474                         ereport(LOG,
 475                                         (errcode_for_socket_access(),
 476                                          errmsg("could not receive test message on socket for statistics collector: %m")));
 477                         closesocket(pgStatSock);
 478                         pgStatSock = PGINVALID_SOCKET;
 479                         continue;
 480                 }
 481
 482                 if (test_byte != TESTBYTEVAL)   /* strictly paranoia ... */
 483                 {
 484                         ereport(LOG,
 485                                         (errcode(ERRCODE_INTERNAL_ERROR),
 486                                          errmsg("incorrect test message transmission on socket for statistics collector")));
 487                         closesocket(pgStatSock);
 488                         pgStatSock = PGINVALID_SOCKET;
 489                         continue;
 490                 }
 491
 492                 /* If we get here, we have a working socket */
 493                 break;
 494         }
 495
 496         /* Did we find a working address? */
 497         if (!addr || pgStatSock == PGINVALID_SOCKET)
 498                 goto startup_failed;
 499
 500         /*
 501          * Set the socket to non-blocking IO.  This ensures that if the collector
 502          * falls behind, statistics messages will be discarded; backends won't
 503          * block waiting to send messages to the collector.
 504          */
 505         if (!pg_set_noblock(pgStatSock))
 506         {
 507                 ereport(LOG,
 508                                 (errcode_for_socket_access(),
 509                                  errmsg("could not set statistics collector socket to nonblocking mode: %m")));
 510                 goto startup_failed;
 511         }
 512
 513         pg_freeaddrinfo_all(hints.ai_family, addrs);
 514
 515         return;
 516
 517 startup_failed:
 518         ereport(LOG,
 519           (errmsg("disabling statistics collector for lack of working socket")));
 520
 521         if (addrs)
 522                 pg_freeaddrinfo_all(hints.ai_family, addrs);
 523
 524         if (pgStatSock != PGINVALID_SOCKET)
 525                 closesocket(pgStatSock);
 526         pgStatSock = PGINVALID_SOCKET;
 527
 528         /*
 529          * Adjust GUC variables to suppress useless activity, and for debugging
 530          * purposes (seeing track_counts off is a clue that we failed here). We
 531          * use PGC_S_OVERRIDE because there is no point in trying to turn it back
 532          * on from postgresql.conf without a restart.
 533          */
 534         SetConfigOption("track_counts", "off", PGC_INTERNAL, PGC_S_OVERRIDE);
 535 }
 536
 537 /*
 538  * pgstat_reset_all() -
 539  *
 540  * Remove the stats file.  This is currently used only if WAL
 541  * recovery is needed after a crash.
 542  */
 543 void
 544 pgstat_reset_all(void)
 545 {
 546         unlink(pgstat_stat_filename);
 547         unlink(PGSTAT_STAT_PERMANENT_FILENAME);
 548 }
 549
 550 #ifdef EXEC_BACKEND
 551
 552 /*
 553  * pgstat_forkexec() -
 554  *
 555  * Format up the arglist for, then fork and exec, statistics collector process
 556  */
 557 static pid_t
 558 pgstat_forkexec(void)
 559 {
 560         char       *av[10];
 561         int                     ac = 0;
 562
 563         av[ac++] = "postgres";
 564         av[ac++] = "--forkcol";
 565         av[ac++] = NULL;                        /* filled in by postmaster_forkexec */
 566
 567         av[ac] = NULL;
 568         Assert(ac < lengthof(av));
 569
 570         return postmaster_forkexec(ac, av);
 571 }
 572 #endif   /* EXEC_BACKEND */
 573
 574
 575 /*
 576  * pgstat_start() -
 577  *
 578  *      Called from postmaster at startup or after an existing collector
 579  *      died.  Attempt to fire up a fresh statistics collector.
 580  *
 581  *      Returns PID of child process, or 0 if fail.
 582  *
 583  *      Note: if fail, we will be called again from the postmaster main loop.
 584  */
 585 int
 586 pgstat_start(void)
 587 {
 588         time_t          curtime;
 589         pid_t           pgStatPid;
 590
 591         /*
 592          * Check that the socket is there, else pgstat_init failed and we can do
 593          * nothing useful.
 594          */
 595         if (pgStatSock == PGINVALID_SOCKET)
 596                 return 0;
 597
 598         /*
 599          * Do nothing if too soon since last collector start.  This is a safety
 600          * valve to protect against continuous respawn attempts if the collector
 601          * is dying immediately at launch.      Note that since we will be re-called
 602          * from the postmaster main loop, we will get another chance later.
 603          */
 604         curtime = time(NULL);
 605         if ((unsigned int) (curtime - last_pgstat_start_time) <
 606                 (unsigned int) PGSTAT_RESTART_INTERVAL)
 607                 return 0;
 608         last_pgstat_start_time = curtime;
 609
 610         /*
 611          * Okay, fork off the collector.
 612          */
 613 #ifdef EXEC_BACKEND
 614         switch ((pgStatPid = pgstat_forkexec()))
 615 #else
 616         switch ((pgStatPid = fork_process()))
 617 #endif
 618         {
 619                 case -1:
 620                         ereport(LOG,
 621                                         (errmsg("could not fork statistics collector: %m")));
 622                         return 0;
 623
 624 #ifndef EXEC_BACKEND
 625                 case 0:
 626                         /* in postmaster child ... */
 627                         /* Close the postmaster's sockets */
 628                         ClosePostmasterPorts(false);
 629
 630                         /* Lose the postmaster's on-exit routines */
 631                         on_exit_reset();
 632
 633                         /* Drop our connection to postmaster's shared memory, as well */
 634                         PGSharedMemoryDetach();
 635
 636                         PgstatCollectorMain(0, NULL);
 637                         break;
 638 #endif
 639
 640                 default:
 641                         return (int) pgStatPid;
 642         }
 643
 644         /* shouldn't get here */
 645         return 0;
 646 }
 647
 648 void
 649 allow_immediate_pgstat_restart(void)
 650 {
 651         last_pgstat_start_time = 0;
 652 }
 653
 654 /* ------------------------------------------------------------
 655  * Public functions used by backends follow
 656  *------------------------------------------------------------
 657  */
 658
 659
 660 /* ----------
 661  * pgstat_report_stat() -
 662  *
 663  *      Called from tcop/postgres.c to send the so far collected per-table
 664  *      and function usage statistics to the collector.  Note that this is
 665  *      called only when not within a transaction, so it is fair to use
 666  *      transaction stop time as an approximation of current time.
 667  * ----------
 668  */
 669 void
 670 pgstat_report_stat(bool force)
 671 {
 672         /* we assume this inits to all zeroes: */
 673         static const PgStat_TableCounts all_zeroes;
 674         static TimestampTz last_report = 0;
 675
 676         TimestampTz now;
 677         PgStat_MsgTabstat regular_msg;
 678         PgStat_MsgTabstat shared_msg;
 679         TabStatusArray *tsa;
 680         int                     i;
 681
 682         /* Don't expend a clock check if nothing to do */
 683         if ((pgStatTabList == NULL || pgStatTabList->tsa_used == 0)
 684                 && !have_function_stats)
 685                 return;
 686
 687         /*
 688          * Don't send a message unless it's been at least PGSTAT_STAT_INTERVAL
 689          * msec since we last sent one, or the caller wants to force stats out.
 690          */
 691         now = GetCurrentTransactionStopTimestamp();
 692         if (!force &&
 693                 !TimestampDifferenceExceeds(last_report, now, PGSTAT_STAT_INTERVAL))
 694                 return;
 695         last_report = now;
 696
 697         /*
 698          * Scan through the TabStatusArray struct(s) to find tables that actually
 699          * have counts, and build messages to send.  We have to separate shared
 700          * relations from regular ones because the databaseid field in the message
 701          * header has to depend on that.
 702          */
 703         regular_msg.m_databaseid = MyDatabaseId;
 704         shared_msg.m_databaseid = InvalidOid;
 705         regular_msg.m_nentries = 0;
 706         shared_msg.m_nentries = 0;
 707
 708         for (tsa = pgStatTabList; tsa != NULL; tsa = tsa->tsa_next)
 709         {
 710                 for (i = 0; i < tsa->tsa_used; i++)
 711                 {
 712                         PgStat_TableStatus *entry = &tsa->tsa_entries[i];
 713                         PgStat_MsgTabstat *this_msg;
 714                         PgStat_TableEntry *this_ent;
 715
 716                         /* Shouldn't have any pending transaction-dependent counts */
 717                         Assert(entry->trans == NULL);
 718
 719                         /*
 720                          * Ignore entries that didn't accumulate any actual counts, such
 721                          * as indexes that were opened by the planner but not used.
 722                          */
 723                         if (memcmp(&entry->t_counts, &all_zeroes,
 724                                            sizeof(PgStat_TableCounts)) == 0)
 725                                 continue;
 726
 727                         /*
 728                          * OK, insert data into the appropriate message, and send if full.
 729                          */
 730                         this_msg = entry->t_shared ? &shared_msg : &regular_msg;
 731                         this_ent = &this_msg->m_entry[this_msg->m_nentries];
 732                         this_ent->t_id = entry->t_id;
 733                         memcpy(&this_ent->t_counts, &entry->t_counts,
 734                                    sizeof(PgStat_TableCounts));
 735                         if (++this_msg->m_nentries >= PGSTAT_NUM_TABENTRIES)
 736                         {
 737                                 pgstat_send_tabstat(this_msg);
 738                                 this_msg->m_nentries = 0;
 739                         }
 740                 }
 741                 /* zero out TableStatus structs after use */
 742                 MemSet(tsa->tsa_entries, 0,
 743                            tsa->tsa_used * sizeof(PgStat_TableStatus));
 744                 tsa->tsa_used = 0;
 745         }
 746
 747         /*
 748          * Send partial messages.  If force is true, make sure that any pending
 749          * xact commit/abort gets counted, even if no table stats to send.
 750          */
 751         if (regular_msg.m_nentries > 0 ||
 752                 (force && (pgStatXactCommit > 0 || pgStatXactRollback > 0)))
 753                 pgstat_send_tabstat(&regular_msg);
 754         if (shared_msg.m_nentries > 0)
 755                 pgstat_send_tabstat(&shared_msg);
 756
 757         /* Now, send function statistics */
 758         pgstat_send_funcstats();
 759 }
 760
 761 /*
 762  * Subroutine for pgstat_report_stat: finish and send a tabstat message
 763  */
 764 static void
 765 pgstat_send_tabstat(PgStat_MsgTabstat *tsmsg)
 766 {
 767         int                     n;
 768         int                     len;
 769
 770         /* It's unlikely we'd get here with no socket, but maybe not impossible */
 771         if (pgStatSock == PGINVALID_SOCKET)
 772                 return;
 773
 774         /*
 775          * Report accumulated xact commit/rollback whenever we send a normal
 776          * tabstat message
 777          */
 778         if (OidIsValid(tsmsg->m_databaseid))
 779         {
 780                 tsmsg->m_xact_commit = pgStatXactCommit;
 781                 tsmsg->m_xact_rollback = pgStatXactRollback;
 782                 pgStatXactCommit = 0;
 783                 pgStatXactRollback = 0;
 784         }
 785         else
 786         {
 787                 tsmsg->m_xact_commit = 0;
 788                 tsmsg->m_xact_rollback = 0;
 789         }
 790
 791         n = tsmsg->m_nentries;
 792         len = offsetof(PgStat_MsgTabstat, m_entry[0]) +
 793                 n * sizeof(PgStat_TableEntry);
 794
 795         pgstat_setheader(&tsmsg->m_hdr, PGSTAT_MTYPE_TABSTAT);
 796         pgstat_send(tsmsg, len);
 797 }
 798
 799 /*
 800  * Subroutine for pgstat_report_stat: populate and send a function stat message
 801  */
 802 static void
 803 pgstat_send_funcstats(void)
 804 {
 805         /* we assume this inits to all zeroes: */
 806         static const PgStat_FunctionCounts all_zeroes;
 807
 808         PgStat_MsgFuncstat msg;
 809         PgStat_BackendFunctionEntry *entry;
 810         HASH_SEQ_STATUS fstat;
 811
 812         if (pgStatFunctions == NULL)
 813                 return;
 814
 815         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_FUNCSTAT);
 816         msg.m_databaseid = MyDatabaseId;
 817         msg.m_nentries = 0;
 818
 819         hash_seq_init(&fstat, pgStatFunctions);
 820         while ((entry = (PgStat_BackendFunctionEntry *) hash_seq_search(&fstat)) != NULL)
 821         {
 822                 PgStat_FunctionEntry *m_ent;
 823
 824                 /* Skip it if no counts accumulated since last time */
 825                 if (memcmp(&entry->f_counts, &all_zeroes,
 826                                    sizeof(PgStat_FunctionCounts)) == 0)
 827                         continue;
 828
 829                 /* need to convert format of time accumulators */
 830                 m_ent = &msg.m_entry[msg.m_nentries];
 831                 m_ent->f_id = entry->f_id;
 832                 m_ent->f_numcalls = entry->f_counts.f_numcalls;
 833                 m_ent->f_time = INSTR_TIME_GET_MICROSEC(entry->f_counts.f_time);
 834                 m_ent->f_time_self = INSTR_TIME_GET_MICROSEC(entry->f_counts.f_time_self);
 835
 836                 if (++msg.m_nentries >= PGSTAT_NUM_FUNCENTRIES)
 837                 {
 838                         pgstat_send(&msg, offsetof(PgStat_MsgFuncstat, m_entry[0]) +
 839                                                 msg.m_nentries * sizeof(PgStat_FunctionEntry));
 840                         msg.m_nentries = 0;
 841                 }
 842
 843                 /* reset the entry's counts */
 844                 MemSet(&entry->f_counts, 0, sizeof(PgStat_FunctionCounts));
 845         }
 846
 847         if (msg.m_nentries > 0)
 848                 pgstat_send(&msg, offsetof(PgStat_MsgFuncstat, m_entry[0]) +
 849                                         msg.m_nentries * sizeof(PgStat_FunctionEntry));
 850
 851         have_function_stats = false;
 852 }
 853
 854
 855 /* ----------
 856  * pgstat_vacuum_stat() -
 857  *
 858  *      Will tell the collector about objects he can get rid of.
 859  * ----------
 860  */
 861 void
 862 pgstat_vacuum_stat(void)
 863 {
 864         HTAB       *htab;
 865         PgStat_MsgTabpurge msg;
 866         PgStat_MsgFuncpurge f_msg;
 867         HASH_SEQ_STATUS hstat;
 868         PgStat_StatDBEntry *dbentry;
 869         PgStat_StatTabEntry *tabentry;
 870         PgStat_StatFuncEntry *funcentry;
 871         int                     len;
 872
 873         if (pgStatSock == PGINVALID_SOCKET)
 874                 return;
 875
 876         /*
 877          * If not done for this transaction, read the statistics collector stats
 878          * file into some hash tables.
 879          */
 880         backend_read_statsfile();
 881
 882         /*
 883          * Read pg_database and make a list of OIDs of all existing databases
 884          */
 885         htab = pgstat_collect_oids(DatabaseRelationId);
 886
 887         /*
 888          * Search the database hash table for dead databases and tell the
 889          * collector to drop them.
 890          */
 891         hash_seq_init(&hstat, pgStatDBHash);
 892         while ((dbentry = (PgStat_StatDBEntry *) hash_seq_search(&hstat)) != NULL)
 893         {
 894                 Oid                     dbid = dbentry->databaseid;
 895
 896                 CHECK_FOR_INTERRUPTS();
 897
 898                 /* the DB entry for shared tables (with InvalidOid) is never dropped */
 899                 if (OidIsValid(dbid) &&
 900                         hash_search(htab, (void *) &dbid, HASH_FIND, NULL) == NULL)
 901                         pgstat_drop_database(dbid);
 902         }
 903
 904         /* Clean up */
 905         hash_destroy(htab);
 906
 907         /*
 908          * Lookup our own database entry; if not found, nothing more to do.
 909          */
 910         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
 911                                                                                                  (void *) &MyDatabaseId,
 912                                                                                                  HASH_FIND, NULL);
 913         if (dbentry == NULL || dbentry->tables == NULL)
 914                 return;
 915
 916         /*
 917          * Similarly to above, make a list of all known relations in this DB.
 918          */
 919         htab = pgstat_collect_oids(RelationRelationId);
 920
 921         /*
 922          * Initialize our messages table counter to zero
 923          */
 924         msg.m_nentries = 0;
 925
 926         /*
 927          * Check for all tables listed in stats hashtable if they still exist.
 928          */
 929         hash_seq_init(&hstat, dbentry->tables);
 930         while ((tabentry = (PgStat_StatTabEntry *) hash_seq_search(&hstat)) != NULL)
 931         {
 932                 Oid                     tabid = tabentry->tableid;
 933
 934                 CHECK_FOR_INTERRUPTS();
 935
 936                 if (hash_search(htab, (void *) &tabid, HASH_FIND, NULL) != NULL)
 937                         continue;
 938
 939                 /*
 940                  * Not there, so add this table's Oid to the message
 941                  */
 942                 msg.m_tableid[msg.m_nentries++] = tabid;
 943
 944                 /*
 945                  * If the message is full, send it out and reinitialize to empty
 946                  */
 947                 if (msg.m_nentries >= PGSTAT_NUM_TABPURGE)
 948                 {
 949                         len = offsetof(PgStat_MsgTabpurge, m_tableid[0])
 950                                 +msg.m_nentries * sizeof(Oid);
 951
 952                         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
 953                         msg.m_databaseid = MyDatabaseId;
 954                         pgstat_send(&msg, len);
 955
 956                         msg.m_nentries = 0;
 957                 }
 958         }
 959
 960         /*
 961          * Send the rest
 962          */
 963         if (msg.m_nentries > 0)
 964         {
 965                 len = offsetof(PgStat_MsgTabpurge, m_tableid[0])
 966                         +msg.m_nentries * sizeof(Oid);
 967
 968                 pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
 969                 msg.m_databaseid = MyDatabaseId;
 970                 pgstat_send(&msg, len);
 971         }
 972
 973         /* Clean up */
 974         hash_destroy(htab);
 975
 976         /*
 977          * Now repeat the above steps for functions.  However, we needn't bother
 978          * in the common case where no function stats are being collected.
 979          */
 980         if (dbentry->functions != NULL &&
 981                 hash_get_num_entries(dbentry->functions) > 0)
 982         {
 983                 htab = pgstat_collect_oids(ProcedureRelationId);
 984
 985                 pgstat_setheader(&f_msg.m_hdr, PGSTAT_MTYPE_FUNCPURGE);
 986                 f_msg.m_databaseid = MyDatabaseId;
 987                 f_msg.m_nentries = 0;
 988
 989                 hash_seq_init(&hstat, dbentry->functions);
 990                 while ((funcentry = (PgStat_StatFuncEntry *) hash_seq_search(&hstat)) != NULL)
 991                 {
 992                         Oid                     funcid = funcentry->functionid;
 993
 994                         CHECK_FOR_INTERRUPTS();
 995
 996                         if (hash_search(htab, (void *) &funcid, HASH_FIND, NULL) != NULL)
 997                                 continue;
 998
 999                         /*
1000                          * Not there, so add this function's Oid to the message
1001                          */
1002                         f_msg.m_functionid[f_msg.m_nentries++] = funcid;
1003
1004                         /*
1005                          * If the message is full, send it out and reinitialize to empty
1006                          */
1007                         if (f_msg.m_nentries >= PGSTAT_NUM_FUNCPURGE)
1008                         {
1009                                 len = offsetof(PgStat_MsgFuncpurge, m_functionid[0])
1010                                         +f_msg.m_nentries * sizeof(Oid);
1011
1012                                 pgstat_send(&f_msg, len);
1013
1014                                 f_msg.m_nentries = 0;
1015                         }
1016                 }
1017
1018                 /*
1019                  * Send the rest
1020                  */
1021                 if (f_msg.m_nentries > 0)
1022                 {
1023                         len = offsetof(PgStat_MsgFuncpurge, m_functionid[0])
1024                                 +f_msg.m_nentries * sizeof(Oid);
1025
1026                         pgstat_send(&f_msg, len);
1027                 }
1028
1029                 hash_destroy(htab);
1030         }
1031 }
1032
1033
1034 /* ----------
1035  * pgstat_collect_oids() -
1036  *
1037  *      Collect the OIDs of all objects listed in the specified system catalog
1038  *      into a temporary hash table.  Caller should hash_destroy the result
1039  *      when done with it.      (However, we make the table in CurrentMemoryContext
1040  *      so that it will be freed properly in event of an error.)
1041  * ----------
1042  */
1043 static HTAB *
1044 pgstat_collect_oids(Oid catalogid)
1045 {
1046         HTAB       *htab;
1047         HASHCTL         hash_ctl;
1048         Relation        rel;
1049         HeapScanDesc scan;
1050         HeapTuple       tup;
1051
1052         memset(&hash_ctl, 0, sizeof(hash_ctl));
1053         hash_ctl.keysize = sizeof(Oid);
1054         hash_ctl.entrysize = sizeof(Oid);
1055         hash_ctl.hash = oid_hash;
1056         hash_ctl.hcxt = CurrentMemoryContext;
1057         htab = hash_create("Temporary table of OIDs",
1058                                            PGSTAT_TAB_HASH_SIZE,
1059                                            &hash_ctl,
1060                                            HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
1061
1062         rel = heap_open(catalogid, AccessShareLock);
1063         scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
1064         while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)
1065         {
1066                 Oid                     thisoid = HeapTupleGetOid(tup);
1067
1068                 CHECK_FOR_INTERRUPTS();
1069
1070                 (void) hash_search(htab, (void *) &thisoid, HASH_ENTER, NULL);
1071         }
1072         heap_endscan(scan);
1073         heap_close(rel, AccessShareLock);
1074
1075         return htab;
1076 }
1077
1078
1079 /* ----------
1080  * pgstat_drop_database() -
1081  *
1082  *      Tell the collector that we just dropped a database.
1083  *      (If the message gets lost, we will still clean the dead DB eventually
1084  *      via future invocations of pgstat_vacuum_stat().)
1085  * ----------
1086  */
1087 void
1088 pgstat_drop_database(Oid databaseid)
1089 {
1090         PgStat_MsgDropdb msg;
1091
1092         if (pgStatSock == PGINVALID_SOCKET)
1093                 return;
1094
1095         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_DROPDB);
1096         msg.m_databaseid = databaseid;
1097         pgstat_send(&msg, sizeof(msg));
1098 }
1099
1100
1101 /* ----------
1102  * pgstat_drop_relation() -
1103  *
1104  *      Tell the collector that we just dropped a relation.
1105  *      (If the message gets lost, we will still clean the dead entry eventually
1106  *      via future invocations of pgstat_vacuum_stat().)
1107  *
1108  *      Currently not used for lack of any good place to call it; we rely
1109  *      entirely on pgstat_vacuum_stat() to clean out stats for dead rels.
1110  * ----------
1111  */
1112 #ifdef NOT_USED
1113 void
1114 pgstat_drop_relation(Oid relid)
1115 {
1116         PgStat_MsgTabpurge msg;
1117         int                     len;
1118
1119         if (pgStatSock == PGINVALID_SOCKET)
1120                 return;
1121
1122         msg.m_tableid[0] = relid;
1123         msg.m_nentries = 1;
1124
1125         len = offsetof(PgStat_MsgTabpurge, m_tableid[0]) +sizeof(Oid);
1126
1127         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
1128         msg.m_databaseid = MyDatabaseId;
1129         pgstat_send(&msg, len);
1130 }
1131 #endif   /* NOT_USED */
1132
1133
1134 /* ----------
1135  * pgstat_reset_counters() -
1136  *
1137  *      Tell the statistics collector to reset counters for our database.
1138  * ----------
1139  */
1140 void
1141 pgstat_reset_counters(void)
1142 {
1143         PgStat_MsgResetcounter msg;
1144
1145         if (pgStatSock == PGINVALID_SOCKET)
1146                 return;
1147
1148         if (!superuser())
1149                 ereport(ERROR,
1150                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1151                                  errmsg("must be superuser to reset statistics counters")));
1152
1153         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RESETCOUNTER);
1154         msg.m_databaseid = MyDatabaseId;
1155         pgstat_send(&msg, sizeof(msg));
1156 }
1157
1158 /* ----------
1159  * pgstat_reset_shared_counters() -
1160  *
1161  *      Tell the statistics collector to reset cluster-wide shared counters.
1162  * ----------
1163  */
1164 void
1165 pgstat_reset_shared_counters(const char *target)
1166 {
1167         PgStat_MsgResetsharedcounter msg;
1168
1169         if (pgStatSock == PGINVALID_SOCKET)
1170                 return;
1171
1172         if (!superuser())
1173                 ereport(ERROR,
1174                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1175                                  errmsg("must be superuser to reset statistics counters")));
1176
1177         if (strcmp(target, "bgwriter") == 0)
1178                 msg.m_resettarget = RESET_BGWRITER;
1179         else
1180                 ereport(ERROR,
1181                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1182                                  errmsg("unrecognized reset target: \"%s\"", target),
1183                                  errhint("Target must be \"bgwriter\".")));
1184
1185         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RESETSHAREDCOUNTER);
1186         pgstat_send(&msg, sizeof(msg));
1187 }
1188
1189 /* ----------
1190  * pgstat_reset_single_counter() -
1191  *
1192  *      Tell the statistics collector to reset a single counter.
1193  * ----------
1194  */
1195 void
1196 pgstat_reset_single_counter(Oid objoid, PgStat_Single_Reset_Type type)
1197 {
1198         PgStat_MsgResetsinglecounter msg;
1199
1200         if (pgStatSock == PGINVALID_SOCKET)
1201                 return;
1202
1203         if (!superuser())
1204                 ereport(ERROR,
1205                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1206                                  errmsg("must be superuser to reset statistics counters")));
1207
1208         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RESETSINGLECOUNTER);
1209         msg.m_databaseid = MyDatabaseId;
1210         msg.m_resettype = type;
1211         msg.m_objectid = objoid;
1212
1213         pgstat_send(&msg, sizeof(msg));
1214 }
1215
1216 /* ----------
1217  * pgstat_report_autovac() -
1218  *
1219  *      Called from autovacuum.c to report startup of an autovacuum process.
1220  *      We are called before InitPostgres is done, so can't rely on MyDatabaseId;
1221  *      the db OID must be passed in, instead.
1222  * ----------
1223  */
1224 void
1225 pgstat_report_autovac(Oid dboid)
1226 {
1227         PgStat_MsgAutovacStart msg;
1228
1229         if (pgStatSock == PGINVALID_SOCKET)
1230                 return;
1231
1232         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_AUTOVAC_START);
1233         msg.m_databaseid = dboid;
1234         msg.m_start_time = GetCurrentTimestamp();
1235
1236         pgstat_send(&msg, sizeof(msg));
1237 }
1238
1239
1240 /* ---------
1241  * pgstat_report_vacuum() -
1242  *
1243  *      Tell the collector about the table we just vacuumed.
1244  * ---------
1245  */
1246 void
1247 pgstat_report_vacuum(Oid tableoid, bool shared, bool adopt_counts,
1248                                          PgStat_Counter tuples)
1249 {
1250         PgStat_MsgVacuum msg;
1251
1252         if (pgStatSock == PGINVALID_SOCKET || !pgstat_track_counts)
1253                 return;
1254
1255         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_VACUUM);
1256         msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
1257         msg.m_tableoid = tableoid;
1258         msg.m_adopt_counts = adopt_counts;
1259         msg.m_autovacuum = IsAutoVacuumWorkerProcess();
1260         msg.m_vacuumtime = GetCurrentTimestamp();
1261         msg.m_tuples = tuples;
1262         pgstat_send(&msg, sizeof(msg));
1263 }
1264
1265 /* --------
1266  * pgstat_report_analyze() -
1267  *
1268  *      Tell the collector about the table we just analyzed.
1269  * --------
1270  */
1271 void
1272 pgstat_report_analyze(Relation rel, bool adopt_counts,
1273                                           PgStat_Counter livetuples, PgStat_Counter deadtuples)
1274 {
1275         PgStat_MsgAnalyze msg;
1276
1277         if (pgStatSock == PGINVALID_SOCKET || !pgstat_track_counts)
1278                 return;
1279
1280         /*
1281          * Unlike VACUUM, ANALYZE might be running inside a transaction that has
1282          * already inserted and/or deleted rows in the target table. ANALYZE will
1283          * have counted such rows as live or dead respectively. Because we will
1284          * report our counts of such rows at transaction end, we should subtract
1285          * off these counts from what we send to the collector now, else they'll
1286          * be double-counted after commit.      (This approach also ensures that the
1287          * collector ends up with the right numbers if we abort instead of
1288          * committing.)
1289          */
1290         if (rel->pgstat_info != NULL)
1291         {
1292                 PgStat_TableXactStatus *trans;
1293
1294                 for (trans = rel->pgstat_info->trans; trans; trans = trans->upper)
1295                 {
1296                         livetuples -= trans->tuples_inserted - trans->tuples_deleted;
1297                         deadtuples -= trans->tuples_updated + trans->tuples_deleted;
1298                 }
1299                 /* count stuff inserted by already-aborted subxacts, too */
1300                 deadtuples -= rel->pgstat_info->t_counts.t_delta_dead_tuples;
1301                 /* Since ANALYZE's counts are estimates, we could have underflowed */
1302                 livetuples = Max(livetuples, 0);
1303                 deadtuples = Max(deadtuples, 0);
1304         }
1305
1306         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_ANALYZE);
1307         msg.m_databaseid = rel->rd_rel->relisshared ? InvalidOid : MyDatabaseId;
1308         msg.m_tableoid = RelationGetRelid(rel);
1309         msg.m_adopt_counts = adopt_counts;
1310         msg.m_autovacuum = IsAutoVacuumWorkerProcess();
1311         msg.m_analyzetime = GetCurrentTimestamp();
1312         msg.m_live_tuples = livetuples;
1313         msg.m_dead_tuples = deadtuples;
1314         pgstat_send(&msg, sizeof(msg));
1315 }
1316
1317
1318 /* ----------
1319  * pgstat_ping() -
1320  *
1321  *      Send some junk data to the collector to increase traffic.
1322  * ----------
1323  */
1324 void
1325 pgstat_ping(void)
1326 {
1327         PgStat_MsgDummy msg;
1328
1329         if (pgStatSock == PGINVALID_SOCKET)
1330                 return;
1331
1332         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_DUMMY);
1333         pgstat_send(&msg, sizeof(msg));
1334 }
1335
1336 /* ----------
1337  * pgstat_send_inquiry() -
1338  *
1339  *      Notify collector that we need fresh data.
1340  *      ts specifies the minimum acceptable timestamp for the stats file.
1341  * ----------
1342  */
1343 static void
1344 pgstat_send_inquiry(TimestampTz ts)
1345 {
1346         PgStat_MsgInquiry msg;
1347
1348         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_INQUIRY);
1349         msg.inquiry_time = ts;
1350         pgstat_send(&msg, sizeof(msg));
1351 }
1352
1353
1354 /*
1355  * Initialize function call usage data.
1356  * Called by the executor before invoking a function.
1357  */
1358 void
1359 pgstat_init_function_usage(FunctionCallInfoData *fcinfo,
1360                                                    PgStat_FunctionCallUsage *fcu)
1361 {
1362         PgStat_BackendFunctionEntry *htabent;
1363         bool            found;
1364
1365         if (pgstat_track_functions <= fcinfo->flinfo->fn_stats)
1366         {
1367                 /* stats not wanted */
1368                 fcu->fs = NULL;
1369                 return;
1370         }
1371
1372         if (!pgStatFunctions)
1373         {
1374                 /* First time through - initialize function stat table */
1375                 HASHCTL         hash_ctl;
1376
1377                 memset(&hash_ctl, 0, sizeof(hash_ctl));
1378                 hash_ctl.keysize = sizeof(Oid);
1379                 hash_ctl.entrysize = sizeof(PgStat_BackendFunctionEntry);
1380                 hash_ctl.hash = oid_hash;
1381                 pgStatFunctions = hash_create("Function stat entries",
1382                                                                           PGSTAT_FUNCTION_HASH_SIZE,
1383                                                                           &hash_ctl,
1384                                                                           HASH_ELEM | HASH_FUNCTION);
1385         }
1386
1387         /* Get the stats entry for this function, create if necessary */
1388         htabent = hash_search(pgStatFunctions, &fcinfo->flinfo->fn_oid,
1389                                                   HASH_ENTER, &found);
1390         if (!found)
1391                 MemSet(&htabent->f_counts, 0, sizeof(PgStat_FunctionCounts));
1392
1393         fcu->fs = &htabent->f_counts;
1394
1395         /* save stats for this function, later used to compensate for recursion */
1396         fcu->save_f_time = htabent->f_counts.f_time;
1397
1398         /* save current backend-wide total time */
1399         fcu->save_total = total_func_time;
1400
1401         /* get clock time as of function start */
1402         INSTR_TIME_SET_CURRENT(fcu->f_start);
1403 }
1404
1405 /*
1406  * find_funcstat_entry - find any existing PgStat_BackendFunctionEntry entry
1407  *              for specified function
1408  *
1409  * If no entry, return NULL, don't create a new one
1410  */
1411 PgStat_BackendFunctionEntry *
1412 find_funcstat_entry(Oid func_id)
1413 {
1414         if (pgStatFunctions == NULL)
1415                 return NULL;
1416
1417         return (PgStat_BackendFunctionEntry *) hash_search(pgStatFunctions,
1418                                                                                                            (void *) &func_id,
1419                                                                                                            HASH_FIND, NULL);
1420 }
1421
1422 /*
1423  * Calculate function call usage and update stat counters.
1424  * Called by the executor after invoking a function.
1425  *
1426  * In the case of a set-returning function that runs in value-per-call mode,
1427  * we will see multiple pgstat_init_function_usage/pgstat_end_function_usage
1428  * calls for what the user considers a single call of the function.  The
1429  * finalize flag should be TRUE on the last call.
1430  */
1431 void
1432 pgstat_end_function_usage(PgStat_FunctionCallUsage *fcu, bool finalize)
1433 {
1434         PgStat_FunctionCounts *fs = fcu->fs;
1435         instr_time      f_total;
1436         instr_time      f_others;
1437         instr_time      f_self;
1438
1439         /* stats not wanted? */
1440         if (fs == NULL)
1441                 return;
1442
1443         /* total elapsed time in this function call */
1444         INSTR_TIME_SET_CURRENT(f_total);
1445         INSTR_TIME_SUBTRACT(f_total, fcu->f_start);
1446
1447         /* self usage: elapsed minus anything already charged to other calls */
1448         f_others = total_func_time;
1449         INSTR_TIME_SUBTRACT(f_others, fcu->save_total);
1450         f_self = f_total;
1451         INSTR_TIME_SUBTRACT(f_self, f_others);
1452
1453         /* update backend-wide total time */
1454         INSTR_TIME_ADD(total_func_time, f_self);
1455
1456         /*
1457          * Compute the new total f_time as the total elapsed time added to the
1458          * pre-call value of f_time.  This is necessary to avoid double-counting
1459          * any time taken by recursive calls of myself.  (We do not need any
1460          * similar kluge for self time, since that already excludes any recursive
1461          * calls.)
1462          */
1463         INSTR_TIME_ADD(f_total, fcu->save_f_time);
1464
1465         /* update counters in function stats table */
1466         if (finalize)
1467                 fs->f_numcalls++;
1468         fs->f_time = f_total;
1469         INSTR_TIME_ADD(fs->f_time_self, f_self);
1470
1471         /* indicate that we have something to send */
1472         have_function_stats = true;
1473 }
1474
1475
1476 /* ----------
1477  * pgstat_initstats() -
1478  *
1479  *      Initialize a relcache entry to count access statistics.
1480  *      Called whenever a relation is opened.
1481  *
1482  *      We assume that a relcache entry's pgstat_info field is zeroed by
1483  *      relcache.c when the relcache entry is made; thereafter it is long-lived
1484  *      data.  We can avoid repeated searches of the TabStatus arrays when the
1485  *      same relation is touched repeatedly within a transaction.
1486  * ----------
1487  */
1488 void
1489 pgstat_initstats(Relation rel)
1490 {
1491         Oid                     rel_id = rel->rd_id;
1492         char            relkind = rel->rd_rel->relkind;
1493
1494         /* We only count stats for things that have storage */
1495         if (!(relkind == RELKIND_RELATION ||
1496                   relkind == RELKIND_INDEX ||
1497                   relkind == RELKIND_TOASTVALUE ||
1498                   relkind == RELKIND_SEQUENCE))
1499         {
1500                 rel->pgstat_info = NULL;
1501                 return;
1502         }
1503
1504         if (pgStatSock == PGINVALID_SOCKET || !pgstat_track_counts)
1505         {
1506                 /* We're not counting at all */
1507                 rel->pgstat_info = NULL;
1508                 return;
1509         }
1510
1511         /*
1512          * If we already set up this relation in the current transaction, nothing
1513          * to do.
1514          */
1515         if (rel->pgstat_info != NULL &&
1516                 rel->pgstat_info->t_id == rel_id)
1517                 return;
1518
1519         /* Else find or make the PgStat_TableStatus entry, and update link */
1520         rel->pgstat_info = get_tabstat_entry(rel_id, rel->rd_rel->relisshared);
1521 }
1522
1523 /*
1524  * get_tabstat_entry - find or create a PgStat_TableStatus entry for rel
1525  */
1526 static PgStat_TableStatus *
1527 get_tabstat_entry(Oid rel_id, bool isshared)
1528 {
1529         PgStat_TableStatus *entry;
1530         TabStatusArray *tsa;
1531         TabStatusArray *prev_tsa;
1532         int                     i;
1533
1534         /*
1535          * Search the already-used tabstat slots for this relation.
1536          */
1537         prev_tsa = NULL;
1538         for (tsa = pgStatTabList; tsa != NULL; prev_tsa = tsa, tsa = tsa->tsa_next)
1539         {
1540                 for (i = 0; i < tsa->tsa_used; i++)
1541                 {
1542                         entry = &tsa->tsa_entries[i];
1543                         if (entry->t_id == rel_id)
1544                                 return entry;
1545                 }
1546
1547                 if (tsa->tsa_used < TABSTAT_QUANTUM)
1548                 {
1549                         /*
1550                          * It must not be present, but we found a free slot instead. Fine,
1551                          * let's use this one.  We assume the entry was already zeroed,
1552                          * either at creation or after last use.
1553                          */
1554                         entry = &tsa->tsa_entries[tsa->tsa_used++];
1555                         entry->t_id = rel_id;
1556                         entry->t_shared = isshared;
1557                         return entry;
1558                 }
1559         }
1560
1561         /*
1562          * We ran out of tabstat slots, so allocate more.  Be sure they're zeroed.
1563          */
1564         tsa = (TabStatusArray *) MemoryContextAllocZero(TopMemoryContext,
1565                                                                                                         sizeof(TabStatusArray));
1566         if (prev_tsa)
1567                 prev_tsa->tsa_next = tsa;
1568         else
1569                 pgStatTabList = tsa;
1570
1571         /*
1572          * Use the first entry of the new TabStatusArray.
1573          */
1574         entry = &tsa->tsa_entries[tsa->tsa_used++];
1575         entry->t_id = rel_id;
1576         entry->t_shared = isshared;
1577         return entry;
1578 }
1579
1580 /*
1581  * find_tabstat_entry - find any existing PgStat_TableStatus entry for rel
1582  *
1583  * If no entry, return NULL, don't create a new one
1584  */
1585 PgStat_TableStatus *
1586 find_tabstat_entry(Oid rel_id)
1587 {
1588         PgStat_TableStatus *entry;
1589         TabStatusArray *tsa;
1590         int                     i;
1591
1592         for (tsa = pgStatTabList; tsa != NULL; tsa = tsa->tsa_next)
1593         {
1594                 for (i = 0; i < tsa->tsa_used; i++)
1595                 {
1596                         entry = &tsa->tsa_entries[i];
1597                         if (entry->t_id == rel_id)
1598                                 return entry;
1599                 }
1600         }
1601
1602         /* Not present */
1603         return NULL;
1604 }
1605
1606 /*
1607  * get_tabstat_stack_level - add a new (sub)transaction stack entry if needed
1608  */
1609 static PgStat_SubXactStatus *
1610 get_tabstat_stack_level(int nest_level)
1611 {
1612         PgStat_SubXactStatus *xact_state;
1613
1614         xact_state = pgStatXactStack;
1615         if (xact_state == NULL || xact_state->nest_level != nest_level)
1616         {
1617                 xact_state = (PgStat_SubXactStatus *)
1618                         MemoryContextAlloc(TopTransactionContext,
1619                                                            sizeof(PgStat_SubXactStatus));
1620                 xact_state->nest_level = nest_level;
1621                 xact_state->prev = pgStatXactStack;
1622                 xact_state->first = NULL;
1623                 pgStatXactStack = xact_state;
1624         }
1625         return xact_state;
1626 }
1627
1628 /*
1629  * add_tabstat_xact_level - add a new (sub)transaction state record
1630  */
1631 static void
1632 add_tabstat_xact_level(PgStat_TableStatus *pgstat_info, int nest_level)
1633 {
1634         PgStat_SubXactStatus *xact_state;
1635         PgStat_TableXactStatus *trans;
1636
1637         /*
1638          * If this is the first rel to be modified at the current nest level, we
1639          * first have to push a transaction stack entry.
1640          */
1641         xact_state = get_tabstat_stack_level(nest_level);
1642
1643         /* Now make a per-table stack entry */
1644         trans = (PgStat_TableXactStatus *)
1645                 MemoryContextAllocZero(TopTransactionContext,
1646                                                            sizeof(PgStat_TableXactStatus));
1647         trans->nest_level = nest_level;
1648         trans->upper = pgstat_info->trans;
1649         trans->parent = pgstat_info;
1650         trans->next = xact_state->first;
1651         xact_state->first = trans;
1652         pgstat_info->trans = trans;
1653 }
1654
1655 /*
1656  * pgstat_count_heap_insert - count a tuple insertion
1657  */
1658 void
1659 pgstat_count_heap_insert(Relation rel)
1660 {
1661         PgStat_TableStatus *pgstat_info = rel->pgstat_info;
1662
1663         if (pgstat_track_counts && pgstat_info != NULL)
1664         {
1665                 /* We have to log the effect at the proper transactional level */
1666                 int                     nest_level = GetCurrentTransactionNestLevel();
1667
1668                 if (pgstat_info->trans == NULL ||
1669                         pgstat_info->trans->nest_level != nest_level)
1670                         add_tabstat_xact_level(pgstat_info, nest_level);
1671
1672                 pgstat_info->trans->tuples_inserted++;
1673         }
1674 }
1675
1676 /*
1677  * pgstat_count_heap_update - count a tuple update
1678  */
1679 void
1680 pgstat_count_heap_update(Relation rel, bool hot)
1681 {
1682         PgStat_TableStatus *pgstat_info = rel->pgstat_info;
1683
1684         if (pgstat_track_counts && pgstat_info != NULL)
1685         {
1686                 /* We have to log the effect at the proper transactional level */
1687                 int                     nest_level = GetCurrentTransactionNestLevel();
1688
1689                 if (pgstat_info->trans == NULL ||
1690                         pgstat_info->trans->nest_level != nest_level)
1691                         add_tabstat_xact_level(pgstat_info, nest_level);
1692
1693                 pgstat_info->trans->tuples_updated++;
1694
1695                 /* t_tuples_hot_updated is nontransactional, so just advance it */
1696                 if (hot)
1697                         pgstat_info->t_counts.t_tuples_hot_updated++;
1698         }
1699 }
1700
1701 /*
1702  * pgstat_count_heap_delete - count a tuple deletion
1703  */
1704 void
1705 pgstat_count_heap_delete(Relation rel)
1706 {
1707         PgStat_TableStatus *pgstat_info = rel->pgstat_info;
1708
1709         if (pgstat_track_counts && pgstat_info != NULL)
1710         {
1711                 /* We have to log the effect at the proper transactional level */
1712                 int                     nest_level = GetCurrentTransactionNestLevel();
1713
1714                 if (pgstat_info->trans == NULL ||
1715                         pgstat_info->trans->nest_level != nest_level)
1716                         add_tabstat_xact_level(pgstat_info, nest_level);
1717
1718                 pgstat_info->trans->tuples_deleted++;
1719         }
1720 }
1721
1722 /*
1723  * pgstat_update_heap_dead_tuples - update dead-tuples count
1724  *
1725  * The semantics of this are that we are reporting the nontransactional
1726  * recovery of "delta" dead tuples; so t_delta_dead_tuples decreases
1727  * rather than increasing, and the change goes straight into the per-table
1728  * counter, not into transactional state.
1729  */
1730 void
1731 pgstat_update_heap_dead_tuples(Relation rel, int delta)
1732 {
1733         PgStat_TableStatus *pgstat_info = rel->pgstat_info;
1734
1735         if (pgstat_track_counts && pgstat_info != NULL)
1736                 pgstat_info->t_counts.t_delta_dead_tuples -= delta;
1737 }
1738
1739
1740 /* ----------
1741  * AtEOXact_PgStat
1742  *
1743  *      Called from access/transam/xact.c at top-level transaction commit/abort.
1744  * ----------
1745  */
1746 void
1747 AtEOXact_PgStat(bool isCommit)
1748 {
1749         PgStat_SubXactStatus *xact_state;
1750
1751         /*
1752          * Count transaction commit or abort.  (We use counters, not just bools,
1753          * in case the reporting message isn't sent right away.)
1754          */
1755         if (isCommit)
1756                 pgStatXactCommit++;
1757         else
1758                 pgStatXactRollback++;
1759
1760         /*
1761          * Transfer transactional insert/update counts into the base tabstat
1762          * entries.  We don't bother to free any of the transactional state, since
1763          * it's all in TopTransactionContext and will go away anyway.
1764          */
1765         xact_state = pgStatXactStack;
1766         if (xact_state != NULL)
1767         {
1768                 PgStat_TableXactStatus *trans;
1769
1770                 Assert(xact_state->nest_level == 1);
1771                 Assert(xact_state->prev == NULL);
1772                 for (trans = xact_state->first; trans != NULL; trans = trans->next)
1773                 {
1774                         PgStat_TableStatus *tabstat;
1775
1776                         Assert(trans->nest_level == 1);
1777                         Assert(trans->upper == NULL);
1778                         tabstat = trans->parent;
1779                         Assert(tabstat->trans == trans);
1780                         /* count attempted actions regardless of commit/abort */
1781                         tabstat->t_counts.t_tuples_inserted += trans->tuples_inserted;
1782                         tabstat->t_counts.t_tuples_updated += trans->tuples_updated;
1783                         tabstat->t_counts.t_tuples_deleted += trans->tuples_deleted;
1784                         if (isCommit)
1785                         {
1786                                 /* insert adds a live tuple, delete removes one */
1787                                 tabstat->t_counts.t_delta_live_tuples +=
1788                                         trans->tuples_inserted - trans->tuples_deleted;
1789                                 /* update and delete each create a dead tuple */
1790                                 tabstat->t_counts.t_delta_dead_tuples +=
1791                                         trans->tuples_updated + trans->tuples_deleted;
1792                                 /* insert, update, delete each count as one change event */
1793                                 tabstat->t_counts.t_changed_tuples +=
1794                                         trans->tuples_inserted + trans->tuples_updated +
1795                                         trans->tuples_deleted;
1796                         }
1797                         else
1798                         {
1799                                 /* inserted tuples are dead, deleted tuples are unaffected */
1800                                 tabstat->t_counts.t_delta_dead_tuples +=
1801                                         trans->tuples_inserted + trans->tuples_updated;
1802                                 /* an aborted xact generates no changed_tuple events */
1803                         }
1804                         tabstat->trans = NULL;
1805                 }
1806         }
1807         pgStatXactStack = NULL;
1808
1809         /* Make sure any stats snapshot is thrown away */
1810         pgstat_clear_snapshot();
1811 }
1812
1813 /* ----------
1814  * AtEOSubXact_PgStat
1815  *
1816  *      Called from access/transam/xact.c at subtransaction commit/abort.
1817  * ----------
1818  */
1819 void
1820 AtEOSubXact_PgStat(bool isCommit, int nestDepth)
1821 {
1822         PgStat_SubXactStatus *xact_state;
1823
1824         /*
1825          * Transfer transactional insert/update counts into the next higher
1826          * subtransaction state.
1827          */
1828         xact_state = pgStatXactStack;
1829         if (xact_state != NULL &&
1830                 xact_state->nest_level >= nestDepth)
1831         {
1832                 PgStat_TableXactStatus *trans;
1833                 PgStat_TableXactStatus *next_trans;
1834
1835                 /* delink xact_state from stack immediately to simplify reuse case */
1836                 pgStatXactStack = xact_state->prev;
1837
1838                 for (trans = xact_state->first; trans != NULL; trans = next_trans)
1839                 {
1840                         PgStat_TableStatus *tabstat;
1841
1842                         next_trans = trans->next;
1843                         Assert(trans->nest_level == nestDepth);
1844                         tabstat = trans->parent;
1845                         Assert(tabstat->trans == trans);
1846                         if (isCommit)
1847                         {
1848                                 if (trans->upper && trans->upper->nest_level == nestDepth - 1)
1849                                 {
1850                                         trans->upper->tuples_inserted += trans->tuples_inserted;
1851                                         trans->upper->tuples_updated += trans->tuples_updated;
1852                                         trans->upper->tuples_deleted += trans->tuples_deleted;
1853                                         tabstat->trans = trans->upper;
1854                                         pfree(trans);
1855                                 }
1856                                 else
1857                                 {
1858                                         /*
1859                                          * When there isn't an immediate parent state, we can just
1860                                          * reuse the record instead of going through a
1861                                          * palloc/pfree pushup (this works since it's all in
1862                                          * TopTransactionContext anyway).  We have to re-link it
1863                                          * into the parent level, though, and that might mean
1864                                          * pushing a new entry into the pgStatXactStack.
1865                                          */
1866                                         PgStat_SubXactStatus *upper_xact_state;
1867
1868                                         upper_xact_state = get_tabstat_stack_level(nestDepth - 1);
1869                                         trans->next = upper_xact_state->first;
1870                                         upper_xact_state->first = trans;
1871                                         trans->nest_level = nestDepth - 1;
1872                                 }
1873                         }
1874                         else
1875                         {
1876                                 /*
1877                                  * On abort, update top-level tabstat counts, then forget the
1878                                  * subtransaction
1879                                  */
1880
1881                                 /* count attempted actions regardless of commit/abort */
1882                                 tabstat->t_counts.t_tuples_inserted += trans->tuples_inserted;
1883                                 tabstat->t_counts.t_tuples_updated += trans->tuples_updated;
1884                                 tabstat->t_counts.t_tuples_deleted += trans->tuples_deleted;
1885                                 /* inserted tuples are dead, deleted tuples are unaffected */
1886                                 tabstat->t_counts.t_delta_dead_tuples +=
1887                                         trans->tuples_inserted + trans->tuples_updated;
1888                                 tabstat->trans = trans->upper;
1889                                 pfree(trans);
1890                         }
1891                 }
1892                 pfree(xact_state);
1893         }
1894 }
1895
1896
1897 /*
1898  * AtPrepare_PgStat
1899  *              Save the transactional stats state at 2PC transaction prepare.
1900  *
1901  * In this phase we just generate 2PC records for all the pending
1902  * transaction-dependent stats work.
1903  */
1904 void
1905 AtPrepare_PgStat(void)
1906 {
1907         PgStat_SubXactStatus *xact_state;
1908
1909         xact_state = pgStatXactStack;
1910         if (xact_state != NULL)
1911         {
1912                 PgStat_TableXactStatus *trans;
1913
1914                 Assert(xact_state->nest_level == 1);
1915                 Assert(xact_state->prev == NULL);
1916                 for (trans = xact_state->first; trans != NULL; trans = trans->next)
1917                 {
1918                         PgStat_TableStatus *tabstat;
1919                         TwoPhasePgStatRecord record;
1920
1921                         Assert(trans->nest_level == 1);
1922                         Assert(trans->upper == NULL);
1923                         tabstat = trans->parent;
1924                         Assert(tabstat->trans == trans);
1925
1926                         record.tuples_inserted = trans->tuples_inserted;
1927                         record.tuples_updated = trans->tuples_updated;
1928                         record.tuples_deleted = trans->tuples_deleted;
1929                         record.t_id = tabstat->t_id;
1930                         record.t_shared = tabstat->t_shared;
1931
1932                         RegisterTwoPhaseRecord(TWOPHASE_RM_PGSTAT_ID, 0,
1933                                                                    &record, sizeof(TwoPhasePgStatRecord));
1934                 }
1935         }
1936 }
1937
1938 /*
1939  * PostPrepare_PgStat
1940  *              Clean up after successful PREPARE.
1941  *
1942  * All we need do here is unlink the transaction stats state from the
1943  * nontransactional state.      The nontransactional action counts will be
1944  * reported to the stats collector immediately, while the effects on live
1945  * and dead tuple counts are preserved in the 2PC state file.
1946  *
1947  * Note: AtEOXact_PgStat is not called during PREPARE.
1948  */
1949 void
1950 PostPrepare_PgStat(void)
1951 {
1952         PgStat_SubXactStatus *xact_state;
1953
1954         /*
1955          * We don't bother to free any of the transactional state, since it's all
1956          * in TopTransactionContext and will go away anyway.
1957          */
1958         xact_state = pgStatXactStack;
1959         if (xact_state != NULL)
1960         {
1961                 PgStat_TableXactStatus *trans;
1962
1963                 for (trans = xact_state->first; trans != NULL; trans = trans->next)
1964                 {
1965                         PgStat_TableStatus *tabstat;
1966
1967                         tabstat = trans->parent;
1968                         tabstat->trans = NULL;
1969                 }
1970         }
1971         pgStatXactStack = NULL;
1972
1973         /* Make sure any stats snapshot is thrown away */
1974         pgstat_clear_snapshot();
1975 }
1976
1977 /*
1978  * 2PC processing routine for COMMIT PREPARED case.
1979  *
1980  * Load the saved counts into our local pgstats state.
1981  */
1982 void
1983 pgstat_twophase_postcommit(TransactionId xid, uint16 info,
1984                                                    void *recdata, uint32 len)
1985 {
1986         TwoPhasePgStatRecord *rec = (TwoPhasePgStatRecord *) recdata;
1987         PgStat_TableStatus *pgstat_info;
1988
1989         /* Find or create a tabstat entry for the rel */
1990         pgstat_info = get_tabstat_entry(rec->t_id, rec->t_shared);
1991
1992         /* Same math as in AtEOXact_PgStat, commit case */
1993         pgstat_info->t_counts.t_tuples_inserted += rec->tuples_inserted;
1994         pgstat_info->t_counts.t_tuples_updated += rec->tuples_updated;
1995         pgstat_info->t_counts.t_tuples_deleted += rec->tuples_deleted;
1996         pgstat_info->t_counts.t_delta_live_tuples +=
1997                 rec->tuples_inserted - rec->tuples_deleted;
1998         pgstat_info->t_counts.t_delta_dead_tuples +=
1999                 rec->tuples_updated + rec->tuples_deleted;
2000         pgstat_info->t_counts.t_changed_tuples +=
2001                 rec->tuples_inserted + rec->tuples_updated +
2002                 rec->tuples_deleted;
2003 }
2004
2005 /*
2006  * 2PC processing routine for ROLLBACK PREPARED case.
2007  *
2008  * Load the saved counts into our local pgstats state, but treat them
2009  * as aborted.
2010  */
2011 void
2012 pgstat_twophase_postabort(TransactionId xid, uint16 info,
2013                                                   void *recdata, uint32 len)
2014 {
2015         TwoPhasePgStatRecord *rec = (TwoPhasePgStatRecord *) recdata;
2016         PgStat_TableStatus *pgstat_info;
2017
2018         /* Find or create a tabstat entry for the rel */
2019         pgstat_info = get_tabstat_entry(rec->t_id, rec->t_shared);
2020
2021         /* Same math as in AtEOXact_PgStat, abort case */
2022         pgstat_info->t_counts.t_tuples_inserted += rec->tuples_inserted;
2023         pgstat_info->t_counts.t_tuples_updated += rec->tuples_updated;
2024         pgstat_info->t_counts.t_tuples_deleted += rec->tuples_deleted;
2025         pgstat_info->t_counts.t_delta_dead_tuples +=
2026                 rec->tuples_inserted + rec->tuples_updated;
2027 }
2028
2029
2030 /* ----------
2031  * pgstat_fetch_stat_dbentry() -
2032  *
2033  *      Support function for the SQL-callable pgstat* functions. Returns
2034  *      the collected statistics for one database or NULL. NULL doesn't mean
2035  *      that the database doesn't exist, it is just not yet known by the
2036  *      collector, so the caller is better off to report ZERO instead.
2037  * ----------
2038  */
2039 PgStat_StatDBEntry *
2040 pgstat_fetch_stat_dbentry(Oid dbid)
2041 {
2042         /*
2043          * If not done for this transaction, read the statistics collector stats
2044          * file into some hash tables.
2045          */
2046         backend_read_statsfile();
2047
2048         /*
2049          * Lookup the requested database; return NULL if not found
2050          */
2051         return (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
2052                                                                                           (void *) &dbid,
2053                                                                                           HASH_FIND, NULL);
2054 }
2055
2056
2057 /* ----------
2058  * pgstat_fetch_stat_tabentry() -
2059  *
2060  *      Support function for the SQL-callable pgstat* functions. Returns
2061  *      the collected statistics for one table or NULL. NULL doesn't mean
2062  *      that the table doesn't exist, it is just not yet known by the
2063  *      collector, so the caller is better off to report ZERO instead.
2064  * ----------
2065  */
2066 PgStat_StatTabEntry *
2067 pgstat_fetch_stat_tabentry(Oid relid)
2068 {
2069         Oid                     dbid;
2070         PgStat_StatDBEntry *dbentry;
2071         PgStat_StatTabEntry *tabentry;
2072
2073         /*
2074          * If not done for this transaction, read the statistics collector stats
2075          * file into some hash tables.
2076          */
2077         backend_read_statsfile();
2078
2079         /*
2080          * Lookup our database, then look in its table hash table.
2081          */
2082         dbid = MyDatabaseId;
2083         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
2084                                                                                                  (void *) &dbid,
2085                                                                                                  HASH_FIND, NULL);
2086         if (dbentry != NULL && dbentry->tables != NULL)
2087         {
2088                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
2089                                                                                                            (void *) &relid,
2090                                                                                                            HASH_FIND, NULL);
2091                 if (tabentry)
2092                         return tabentry;
2093         }
2094
2095         /*
2096          * If we didn't find it, maybe it's a shared table.
2097          */
2098         dbid = InvalidOid;
2099         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
2100                                                                                                  (void *) &dbid,
2101                                                                                                  HASH_FIND, NULL);
2102         if (dbentry != NULL && dbentry->tables != NULL)
2103         {
2104                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
2105                                                                                                            (void *) &relid,
2106                                                                                                            HASH_FIND, NULL);
2107                 if (tabentry)
2108                         return tabentry;
2109         }
2110
2111         return NULL;
2112 }
2113
2114
2115 /* ----------
2116  * pgstat_fetch_stat_funcentry() -
2117  *
2118  *      Support function for the SQL-callable pgstat* functions. Returns
2119  *      the collected statistics for one function or NULL.
2120  * ----------
2121  */
2122 PgStat_StatFuncEntry *
2123 pgstat_fetch_stat_funcentry(Oid func_id)
2124 {
2125         PgStat_StatDBEntry *dbentry;
2126         PgStat_StatFuncEntry *funcentry = NULL;
2127
2128         /* load the stats file if needed */
2129         backend_read_statsfile();
2130
2131         /* Lookup our database, then find the requested function.  */
2132         dbentry = pgstat_fetch_stat_dbentry(MyDatabaseId);
2133         if (dbentry != NULL && dbentry->functions != NULL)
2134         {
2135                 funcentry = (PgStat_StatFuncEntry *) hash_search(dbentry->functions,
2136                                                                                                                  (void *) &func_id,
2137                                                                                                                  HASH_FIND, NULL);
2138         }
2139
2140         return funcentry;
2141 }
2142
2143
2144 /* ----------
2145  * pgstat_fetch_stat_beentry() -
2146  *
2147  *      Support function for the SQL-callable pgstat* functions. Returns
2148  *      our local copy of the current-activity entry for one backend.
2149  *
2150  *      NB: caller is responsible for a check if the user is permitted to see
2151  *      this info (especially the querystring).
2152  * ----------
2153  */
2154 PgBackendStatus *
2155 pgstat_fetch_stat_beentry(int beid)
2156 {
2157         pgstat_read_current_status();
2158
2159         if (beid < 1 || beid > localNumBackends)
2160                 return NULL;
2161
2162         return &localBackendStatusTable[beid - 1];
2163 }
2164
2165
2166 /* ----------
2167  * pgstat_fetch_stat_numbackends() -
2168  *
2169  *      Support function for the SQL-callable pgstat* functions. Returns
2170  *      the maximum current backend id.
2171  * ----------
2172  */
2173 int
2174 pgstat_fetch_stat_numbackends(void)
2175 {
2176         pgstat_read_current_status();
2177
2178         return localNumBackends;
2179 }
2180
2181 /*
2182  * ---------
2183  * pgstat_fetch_global() -
2184  *
2185  *      Support function for the SQL-callable pgstat* functions. Returns
2186  *      a pointer to the global statistics struct.
2187  * ---------
2188  */
2189 PgStat_GlobalStats *
2190 pgstat_fetch_global(void)
2191 {
2192         backend_read_statsfile();
2193
2194         return &globalStats;
2195 }
2196
2197
2198 /* ------------------------------------------------------------
2199  * Functions for management of the shared-memory PgBackendStatus array
2200  * ------------------------------------------------------------
2201  */
2202
2203 static PgBackendStatus *BackendStatusArray = NULL;
2204 static PgBackendStatus *MyBEEntry = NULL;
2205 static char *BackendAppnameBuffer = NULL;
2206 static char *BackendActivityBuffer = NULL;
2207
2208
2209 /*
2210  * Report shared-memory space needed by CreateSharedBackendStatus.
2211  */
2212 Size
2213 BackendStatusShmemSize(void)
2214 {
2215         Size            size;
2216
2217         size = mul_size(sizeof(PgBackendStatus), MaxBackends);
2218         size = add_size(size,
2219                                         mul_size(NAMEDATALEN, MaxBackends));
2220         size = add_size(size,
2221                                         mul_size(pgstat_track_activity_query_size, MaxBackends));
2222         return size;
2223 }
2224
2225 /*
2226  * Initialize the shared status array and activity/appname string buffers
2227  * during postmaster startup.
2228  */
2229 void
2230 CreateSharedBackendStatus(void)
2231 {
2232         Size            size;
2233         bool            found;
2234         int                     i;
2235         char       *buffer;
2236
2237         /* Create or attach to the shared array */
2238         size = mul_size(sizeof(PgBackendStatus), MaxBackends);
2239         BackendStatusArray = (PgBackendStatus *)
2240                 ShmemInitStruct("Backend Status Array", size, &found);
2241
2242         if (!found)
2243         {
2244                 /*
2245                  * We're the first - initialize.
2246                  */
2247                 MemSet(BackendStatusArray, 0, size);
2248         }
2249
2250         /* Create or attach to the shared appname buffer */
2251         size = mul_size(NAMEDATALEN, MaxBackends);
2252         BackendAppnameBuffer = (char *)
2253                 ShmemInitStruct("Backend Application Name Buffer", size, &found);
2254
2255         if (!found)
2256         {
2257                 MemSet(BackendAppnameBuffer, 0, size);
2258
2259                 /* Initialize st_appname pointers. */
2260                 buffer = BackendAppnameBuffer;
2261                 for (i = 0; i < MaxBackends; i++)
2262                 {
2263                         BackendStatusArray[i].st_appname = buffer;
2264                         buffer += NAMEDATALEN;
2265                 }
2266         }
2267
2268         /* Create or attach to the shared activity buffer */
2269         size = mul_size(pgstat_track_activity_query_size, MaxBackends);
2270         BackendActivityBuffer = (char *)
2271                 ShmemInitStruct("Backend Activity Buffer", size, &found);
2272
2273         if (!found)
2274         {
2275                 MemSet(BackendActivityBuffer, 0, size);
2276
2277                 /* Initialize st_activity pointers. */
2278                 buffer = BackendActivityBuffer;
2279                 for (i = 0; i < MaxBackends; i++)
2280                 {
2281                         BackendStatusArray[i].st_activity = buffer;
2282                         buffer += pgstat_track_activity_query_size;
2283                 }
2284         }
2285 }
2286
2287
2288 /* ----------
2289  * pgstat_initialize() -
2290  *
2291  *      Initialize pgstats state, and set up our on-proc-exit hook.
2292  *      Called from InitPostgres.  MyBackendId must be set,
2293  *      but we must not have started any transaction yet (since the
2294  *      exit hook must run after the last transaction exit).
2295  *      NOTE: MyDatabaseId isn't set yet; so the shutdown hook has to be careful.
2296  * ----------
2297  */
2298 void
2299 pgstat_initialize(void)
2300 {
2301         /* Initialize MyBEEntry */
2302         Assert(MyBackendId >= 1 && MyBackendId <= MaxBackends);
2303         MyBEEntry = &BackendStatusArray[MyBackendId - 1];
2304
2305         /* Set up a process-exit hook to clean up */
2306         on_shmem_exit(pgstat_beshutdown_hook, 0);
2307 }
2308
2309 /* ----------
2310  * pgstat_bestart() -
2311  *
2312  *      Initialize this backend's entry in the PgBackendStatus array.
2313  *      Called from InitPostgres.
2314  *      MyDatabaseId, session userid, and application_name must be set
2315  *      (hence, this cannot be combined with pgstat_initialize).
2316  * ----------
2317  */
2318 void
2319 pgstat_bestart(void)
2320 {
2321         TimestampTz proc_start_timestamp;
2322         Oid                     userid;
2323         SockAddr        clientaddr;
2324         volatile PgBackendStatus *beentry;
2325
2326         /*
2327          * To minimize the time spent modifying the PgBackendStatus entry, fetch
2328          * all the needed data first.
2329          *
2330          * If we have a MyProcPort, use its session start time (for consistency,
2331          * and to save a kernel call).
2332          */
2333         if (MyProcPort)
2334                 proc_start_timestamp = MyProcPort->SessionStartTime;
2335         else
2336                 proc_start_timestamp = GetCurrentTimestamp();
2337         userid = GetSessionUserId();
2338
2339         /*
2340          * We may not have a MyProcPort (eg, if this is the autovacuum process).
2341          * If so, use all-zeroes client address, which is dealt with specially in
2342          * pg_stat_get_backend_client_addr and pg_stat_get_backend_client_port.
2343          */
2344         if (MyProcPort)
2345                 memcpy(&clientaddr, &MyProcPort->raddr, sizeof(clientaddr));
2346         else
2347                 MemSet(&clientaddr, 0, sizeof(clientaddr));
2348
2349         /*
2350          * Initialize my status entry, following the protocol of bumping
2351          * st_changecount before and after; and make sure it's even afterwards. We
2352          * use a volatile pointer here to ensure the compiler doesn't try to get
2353          * cute.
2354          */
2355         beentry = MyBEEntry;
2356         do
2357         {
2358                 beentry->st_changecount++;
2359         } while ((beentry->st_changecount & 1) == 0);
2360
2361         beentry->st_procpid = MyProcPid;
2362         beentry->st_proc_start_timestamp = proc_start_timestamp;
2363         beentry->st_activity_start_timestamp = 0;
2364         beentry->st_xact_start_timestamp = 0;
2365         beentry->st_databaseid = MyDatabaseId;
2366         beentry->st_userid = userid;
2367         beentry->st_clientaddr = clientaddr;
2368         beentry->st_waiting = false;
2369         beentry->st_appname[0] = '\0';
2370         beentry->st_activity[0] = '\0';
2371         /* Also make sure the last byte in each string area is always 0 */
2372         beentry->st_appname[NAMEDATALEN - 1] = '\0';
2373         beentry->st_activity[pgstat_track_activity_query_size - 1] = '\0';
2374
2375         beentry->st_changecount++;
2376         Assert((beentry->st_changecount & 1) == 0);
2377
2378         /* Update app name to current GUC setting */
2379         if (application_name)
2380                 pgstat_report_appname(application_name);
2381 }
2382
2383 /*
2384  * Shut down a single backend's statistics reporting at process exit.
2385  *
2386  * Flush any remaining statistics counts out to the collector.
2387  * Without this, operations triggered during backend exit (such as
2388  * temp table deletions) won't be counted.
2389  *
2390  * Lastly, clear out our entry in the PgBackendStatus array.
2391  */
2392 static void
2393 pgstat_beshutdown_hook(int code, Datum arg)
2394 {
2395         volatile PgBackendStatus *beentry = MyBEEntry;
2396
2397         /*
2398          * If we got as far as discovering our own database ID, we can report what
2399          * we did to the collector.  Otherwise, we'd be sending an invalid
2400          * database ID, so forget it.  (This means that accesses to pg_database
2401          * during failed backend starts might never get counted.)
2402          */
2403         if (OidIsValid(MyDatabaseId))
2404                 pgstat_report_stat(true);
2405
2406         /*
2407          * Clear my status entry, following the protocol of bumping st_changecount
2408          * before and after.  We use a volatile pointer here to ensure the
2409          * compiler doesn't try to get cute.
2410          */
2411         beentry->st_changecount++;
2412
2413         beentry->st_procpid = 0;        /* mark invalid */
2414
2415         beentry->st_changecount++;
2416         Assert((beentry->st_changecount & 1) == 0);
2417 }
2418
2419
2420 /* ----------
2421  * pgstat_report_activity() -
2422  *
2423  *      Called from tcop/postgres.c to report what the backend is actually doing
2424  *      (usually "<IDLE>" or the start of the query to be executed).
2425  * ----------
2426  */
2427 void
2428 pgstat_report_activity(const char *cmd_str)
2429 {
2430         volatile PgBackendStatus *beentry = MyBEEntry;
2431         TimestampTz start_timestamp;
2432         int                     len;
2433
2434         TRACE_POSTGRESQL_STATEMENT_STATUS(cmd_str);
2435
2436         if (!pgstat_track_activities || !beentry)
2437                 return;
2438
2439         /*
2440          * To minimize the time spent modifying the entry, fetch all the needed
2441          * data first.
2442          */
2443         start_timestamp = GetCurrentStatementStartTimestamp();
2444
2445         len = strlen(cmd_str);
2446         len = pg_mbcliplen(cmd_str, len, pgstat_track_activity_query_size - 1);
2447
2448         /*
2449          * Update my status entry, following the protocol of bumping
2450          * st_changecount before and after.  We use a volatile pointer here to
2451          * ensure the compiler doesn't try to get cute.
2452          */
2453         beentry->st_changecount++;
2454
2455         beentry->st_activity_start_timestamp = start_timestamp;
2456         memcpy((char *) beentry->st_activity, cmd_str, len);
2457         beentry->st_activity[len] = '\0';
2458
2459         beentry->st_changecount++;
2460         Assert((beentry->st_changecount & 1) == 0);
2461 }
2462
2463 /* ----------
2464  * pgstat_report_appname() -
2465  *
2466  *      Called to update our application name.
2467  * ----------
2468  */
2469 void
2470 pgstat_report_appname(const char *appname)
2471 {
2472         volatile PgBackendStatus *beentry = MyBEEntry;
2473         int                     len;
2474
2475         if (!beentry)
2476                 return;
2477
2478         /* This should be unnecessary if GUC did its job, but be safe */
2479         len = pg_mbcliplen(appname, strlen(appname), NAMEDATALEN - 1);
2480
2481         /*
2482          * Update my status entry, following the protocol of bumping
2483          * st_changecount before and after.  We use a volatile pointer here to
2484          * ensure the compiler doesn't try to get cute.
2485          */
2486         beentry->st_changecount++;
2487
2488         memcpy((char *) beentry->st_appname, appname, len);
2489         beentry->st_appname[len] = '\0';
2490
2491         beentry->st_changecount++;
2492         Assert((beentry->st_changecount & 1) == 0);
2493 }
2494
2495 /*
2496  * Report current transaction start timestamp as the specified value.
2497  * Zero means there is no active transaction.
2498  */
2499 void
2500 pgstat_report_xact_timestamp(TimestampTz tstamp)
2501 {
2502         volatile PgBackendStatus *beentry = MyBEEntry;
2503
2504         if (!pgstat_track_activities || !beentry)
2505                 return;
2506
2507         /*
2508          * Update my status entry, following the protocol of bumping
2509          * st_changecount before and after.  We use a volatile pointer here to
2510          * ensure the compiler doesn't try to get cute.
2511          */
2512         beentry->st_changecount++;
2513         beentry->st_xact_start_timestamp = tstamp;
2514         beentry->st_changecount++;
2515         Assert((beentry->st_changecount & 1) == 0);
2516 }
2517
2518 /* ----------
2519  * pgstat_report_waiting() -
2520  *
2521  *      Called from lock manager to report beginning or end of a lock wait.
2522  *
2523  * NB: this *must* be able to survive being called before MyBEEntry has been
2524  * initialized.
2525  * ----------
2526  */
2527 void
2528 pgstat_report_waiting(bool waiting)
2529 {
2530         volatile PgBackendStatus *beentry = MyBEEntry;
2531
2532         if (!pgstat_track_activities || !beentry)
2533                 return;
2534
2535         /*
2536          * Since this is a single-byte field in a struct that only this process
2537          * may modify, there seems no need to bother with the st_changecount
2538          * protocol.  The update must appear atomic in any case.
2539          */
2540         beentry->st_waiting = waiting;
2541 }
2542
2543
2544 /* ----------
2545  * pgstat_read_current_status() -
2546  *
2547  *      Copy the current contents of the PgBackendStatus array to local memory,
2548  *      if not already done in this transaction.
2549  * ----------
2550  */
2551 static void
2552 pgstat_read_current_status(void)
2553 {
2554         volatile PgBackendStatus *beentry;
2555         PgBackendStatus *localtable;
2556         PgBackendStatus *localentry;
2557         char       *localappname,
2558                            *localactivity;
2559         int                     i;
2560
2561         Assert(!pgStatRunningInCollector);
2562         if (localBackendStatusTable)
2563                 return;                                 /* already done */
2564
2565         pgstat_setup_memcxt();
2566
2567         localtable = (PgBackendStatus *)
2568                 MemoryContextAlloc(pgStatLocalContext,
2569                                                    sizeof(PgBackendStatus) * MaxBackends);
2570         localappname = (char *)
2571                 MemoryContextAlloc(pgStatLocalContext,
2572                                                    NAMEDATALEN * MaxBackends);
2573         localactivity = (char *)
2574                 MemoryContextAlloc(pgStatLocalContext,
2575                                                    pgstat_track_activity_query_size * MaxBackends);
2576         localNumBackends = 0;
2577
2578         beentry = BackendStatusArray;
2579         localentry = localtable;
2580         for (i = 1; i <= MaxBackends; i++)
2581         {
2582                 /*
2583                  * Follow the protocol of retrying if st_changecount changes while we
2584                  * copy the entry, or if it's odd.  (The check for odd is needed to
2585                  * cover the case where we are able to completely copy the entry while
2586                  * the source backend is between increment steps.)      We use a volatile
2587                  * pointer here to ensure the compiler doesn't try to get cute.
2588                  */
2589                 for (;;)
2590                 {
2591                         int                     save_changecount = beentry->st_changecount;
2592
2593                         localentry->st_procpid = beentry->st_procpid;
2594                         if (localentry->st_procpid > 0)
2595                         {
2596                                 memcpy(localentry, (char *) beentry, sizeof(PgBackendStatus));
2597
2598                                 /*
2599                                  * strcpy is safe even if the string is modified concurrently,
2600                                  * because there's always a \0 at the end of the buffer.
2601                                  */
2602                                 strcpy(localappname, (char *) beentry->st_appname);
2603                                 localentry->st_appname = localappname;
2604                                 strcpy(localactivity, (char *) beentry->st_activity);
2605                                 localentry->st_activity = localactivity;
2606                         }
2607
2608                         if (save_changecount == beentry->st_changecount &&
2609                                 (save_changecount & 1) == 0)
2610                                 break;
2611
2612                         /* Make sure we can break out of loop if stuck... */
2613                         CHECK_FOR_INTERRUPTS();
2614                 }
2615
2616                 beentry++;
2617                 /* Only valid entries get included into the local array */
2618                 if (localentry->st_procpid > 0)
2619                 {
2620                         localentry++;
2621                         localappname += NAMEDATALEN;
2622                         localactivity += pgstat_track_activity_query_size;
2623                         localNumBackends++;
2624                 }
2625         }
2626
2627         /* Set the pointer only after completion of a valid table */
2628         localBackendStatusTable = localtable;
2629 }
2630
2631
2632 /* ----------
2633  * pgstat_get_backend_current_activity() -
2634  *
2635  *      Return a string representing the current activity of the backend with
2636  *      the specified PID.      This looks directly at the BackendStatusArray,
2637  *      and so will provide current information regardless of the age of our
2638  *      transaction's snapshot of the status array.
2639  *
2640  *      It is the caller's responsibility to invoke this only for backends whose
2641  *      state is expected to remain stable while the result is in use.  The
2642  *      only current use is in deadlock reporting, where we can expect that
2643  *      the target backend is blocked on a lock.  (There are corner cases
2644  *      where the target's wait could get aborted while we are looking at it,
2645  *      but the very worst consequence is to return a pointer to a string
2646  *      that's been changed, so we won't worry too much.)
2647  *
2648  *      Note: return strings for special cases match pg_stat_get_backend_activity.
2649  * ----------
2650  */
2651 const char *
2652 pgstat_get_backend_current_activity(int pid, bool checkUser)
2653 {
2654         PgBackendStatus *beentry;
2655         int                     i;
2656
2657         beentry = BackendStatusArray;
2658         for (i = 1; i <= MaxBackends; i++)
2659         {
2660                 /*
2661                  * Although we expect the target backend's entry to be stable, that
2662                  * doesn't imply that anyone else's is.  To avoid identifying the
2663                  * wrong backend, while we check for a match to the desired PID we
2664                  * must follow the protocol of retrying if st_changecount changes
2665                  * while we examine the entry, or if it's odd.  (This might be
2666                  * unnecessary, since fetching or storing an int is almost certainly
2667                  * atomic, but let's play it safe.)  We use a volatile pointer here to
2668                  * ensure the compiler doesn't try to get cute.
2669                  */
2670                 volatile PgBackendStatus *vbeentry = beentry;
2671                 bool            found;
2672
2673                 for (;;)
2674                 {
2675                         int                     save_changecount = vbeentry->st_changecount;
2676
2677                         found = (vbeentry->st_procpid == pid);
2678
2679                         if (save_changecount == vbeentry->st_changecount &&
2680                                 (save_changecount & 1) == 0)
2681                                 break;
2682
2683                         /* Make sure we can break out of loop if stuck... */
2684                         CHECK_FOR_INTERRUPTS();
2685                 }
2686
2687                 if (found)
2688                 {
2689                         /* Now it is safe to use the non-volatile pointer */
2690                         if (checkUser && !superuser() && beentry->st_userid != GetUserId())
2691                                 return "<insufficient privilege>";
2692                         else if (*(beentry->st_activity) == '\0')
2693                                 return "<command string not enabled>";
2694                         else
2695                                 return beentry->st_activity;
2696                 }
2697
2698                 beentry++;
2699         }
2700
2701         /* If we get here, caller is in error ... */
2702         return "<backend information not available>";
2703 }
2704
2705
2706 /* ------------------------------------------------------------
2707  * Local support functions follow
2708  * ------------------------------------------------------------
2709  */
2710
2711
2712 /* ----------
2713  * pgstat_setheader() -
2714  *
2715  *              Set common header fields in a statistics message
2716  * ----------
2717  */
2718 static void
2719 pgstat_setheader(PgStat_MsgHdr *hdr, StatMsgType mtype)
2720 {
2721         hdr->m_type = mtype;
2722 }
2723
2724
2725 /* ----------
2726  * pgstat_send() -
2727  *
2728  *              Send out one statistics message to the collector
2729  * ----------
2730  */
2731 static void
2732 pgstat_send(void *msg, int len)
2733 {
2734         int                     rc;
2735
2736         if (pgStatSock == PGINVALID_SOCKET)
2737                 return;
2738
2739         ((PgStat_MsgHdr *) msg)->m_size = len;
2740
2741         /* We'll retry after EINTR, but ignore all other failures */
2742         do
2743         {
2744                 rc = send(pgStatSock, msg, len, 0);
2745         } while (rc < 0 && errno == EINTR);
2746
2747 #ifdef USE_ASSERT_CHECKING
2748         /* In debug builds, log send failures ... */
2749         if (rc < 0)
2750                 elog(LOG, "could not send to statistics collector: %m");
2751 #endif
2752 }
2753
2754 /* ----------
2755  * pgstat_send_bgwriter() -
2756  *
2757  *              Send bgwriter statistics to the collector
2758  * ----------
2759  */
2760 void
2761 pgstat_send_bgwriter(void)
2762 {
2763         /* We assume this initializes to zeroes */
2764         static const PgStat_MsgBgWriter all_zeroes;
2765
2766         /*
2767          * This function can be called even if nothing at all has happened. In
2768          * this case, avoid sending a completely empty message to the stats
2769          * collector.
2770          */
2771         if (memcmp(&BgWriterStats, &all_zeroes, sizeof(PgStat_MsgBgWriter)) == 0)
2772                 return;
2773
2774         /*
2775          * Prepare and send the message
2776          */
2777         pgstat_setheader(&BgWriterStats.m_hdr, PGSTAT_MTYPE_BGWRITER);
2778         pgstat_send(&BgWriterStats, sizeof(BgWriterStats));
2779
2780         /*
2781          * Clear out the statistics buffer, so it can be re-used.
2782          */
2783         MemSet(&BgWriterStats, 0, sizeof(BgWriterStats));
2784 }
2785
2786
2787 /* ----------
2788  * PgstatCollectorMain() -
2789  *
2790  *      Start up the statistics collector process.      This is the body of the
2791  *      postmaster child process.
2792  *
2793  *      The argc/argv parameters are valid only in EXEC_BACKEND case.
2794  * ----------
2795  */
2796 NON_EXEC_STATIC void
2797 PgstatCollectorMain(int argc, char *argv[])
2798 {
2799         int                     len;
2800         PgStat_Msg      msg;
2801
2802 #ifndef WIN32
2803 #ifdef HAVE_POLL
2804         struct pollfd input_fd;
2805 #else
2806         struct timeval sel_timeout;
2807         fd_set          rfds;
2808 #endif
2809 #endif
2810
2811         IsUnderPostmaster = true;       /* we are a postmaster subprocess now */
2812
2813         MyProcPid = getpid();           /* reset MyProcPid */
2814
2815         MyStartTime = time(NULL);       /* record Start Time for logging */
2816
2817         /*
2818          * If possible, make this process a group leader, so that the postmaster
2819          * can signal any child processes too.  (pgstat probably never has any
2820          * child processes, but for consistency we make all postmaster child
2821          * processes do this.)
2822          */
2823 #ifdef HAVE_SETSID
2824         if (setsid() < 0)
2825                 elog(FATAL, "setsid() failed: %m");
2826 #endif
2827
2828         /*
2829          * Ignore all signals usually bound to some action in the postmaster,
2830          * except SIGQUIT.
2831          */
2832         pqsignal(SIGHUP, pgstat_sighup_handler);
2833         pqsignal(SIGINT, SIG_IGN);
2834         pqsignal(SIGTERM, SIG_IGN);
2835         pqsignal(SIGQUIT, pgstat_exit);
2836         pqsignal(SIGALRM, SIG_IGN);
2837         pqsignal(SIGPIPE, SIG_IGN);
2838         pqsignal(SIGUSR1, SIG_IGN);
2839         pqsignal(SIGUSR2, SIG_IGN);
2840         pqsignal(SIGCHLD, SIG_DFL);
2841         pqsignal(SIGTTIN, SIG_DFL);
2842         pqsignal(SIGTTOU, SIG_DFL);
2843         pqsignal(SIGCONT, SIG_DFL);
2844         pqsignal(SIGWINCH, SIG_DFL);
2845         PG_SETMASK(&UnBlockSig);
2846
2847         /*
2848          * Identify myself via ps
2849          */
2850         init_ps_display("stats collector process", "", "", "");
2851
2852         /*
2853          * Arrange to write the initial status file right away
2854          */
2855         last_statrequest = GetCurrentTimestamp();
2856         last_statwrite = last_statrequest - 1;
2857
2858         /*
2859          * Read in an existing statistics stats file or initialize the stats to
2860          * zero.
2861          */
2862         pgStatRunningInCollector = true;
2863         pgStatDBHash = pgstat_read_statsfile(InvalidOid, true);
2864
2865         /*
2866          * Setup the descriptor set for select(2).      Since only one bit in the set
2867          * ever changes, we need not repeat FD_ZERO each time.
2868          */
2869 #if !defined(HAVE_POLL) && !defined(WIN32)
2870         FD_ZERO(&rfds);
2871 #endif
2872
2873         /*
2874          * Loop to process messages until we get SIGQUIT or detect ungraceful
2875          * death of our parent postmaster.
2876          *
2877          * For performance reasons, we don't want to do a PostmasterIsAlive() test
2878          * after every message; instead, do it only when select()/poll() is
2879          * interrupted by timeout.      In essence, we'll stay alive as long as
2880          * backends keep sending us stuff often, even if the postmaster is gone.
2881          */
2882         for (;;)
2883         {
2884                 int                     got_data;
2885
2886                 /*
2887                  * Quit if we get SIGQUIT from the postmaster.
2888                  */
2889                 if (need_exit)
2890                         break;
2891
2892                 /*
2893                  * Reload configuration if we got SIGHUP from the postmaster.
2894                  */
2895                 if (got_SIGHUP)
2896                 {
2897                         ProcessConfigFile(PGC_SIGHUP);
2898                         got_SIGHUP = false;
2899                 }
2900
2901                 /*
2902                  * Write the stats file if a new request has arrived that is not
2903                  * satisfied by existing file.
2904                  */
2905                 if (last_statwrite < last_statrequest)
2906                         pgstat_write_statsfile(false);
2907
2908                 /*
2909                  * Wait for a message to arrive; but not for more than
2910                  * PGSTAT_SELECT_TIMEOUT seconds. (This determines how quickly we will
2911                  * shut down after an ungraceful postmaster termination; so it needn't
2912                  * be very fast.  However, on some systems SIGQUIT won't interrupt the
2913                  * poll/select call, so this also limits speed of response to SIGQUIT,
2914                  * which is more important.)
2915                  *
2916                  * We use poll(2) if available, otherwise select(2). Win32 has its own
2917                  * implementation.
2918                  */
2919 #ifndef WIN32
2920 #ifdef HAVE_POLL
2921                 input_fd.fd = pgStatSock;
2922                 input_fd.events = POLLIN | POLLERR;
2923                 input_fd.revents = 0;
2924
2925                 if (poll(&input_fd, 1, PGSTAT_SELECT_TIMEOUT * 1000) < 0)
2926                 {
2927                         if (errno == EINTR)
2928                                 continue;
2929                         ereport(ERROR,
2930                                         (errcode_for_socket_access(),
2931                                          errmsg("poll() failed in statistics collector: %m")));
2932                 }
2933
2934                 got_data = (input_fd.revents != 0);
2935 #else                                                   /* !HAVE_POLL */
2936
2937                 FD_SET(pgStatSock, &rfds);
2938
2939                 /*
2940                  * timeout struct is modified by select() on some operating systems,
2941                  * so re-fill it each time.
2942                  */
2943                 sel_timeout.tv_sec = PGSTAT_SELECT_TIMEOUT;
2944                 sel_timeout.tv_usec = 0;
2945
2946                 if (select(pgStatSock + 1, &rfds, NULL, NULL, &sel_timeout) < 0)
2947                 {
2948                         if (errno == EINTR)
2949                                 continue;
2950                         ereport(ERROR,
2951                                         (errcode_for_socket_access(),
2952                                          errmsg("select() failed in statistics collector: %m")));
2953                 }
2954
2955                 got_data = FD_ISSET(pgStatSock, &rfds);
2956 #endif   /* HAVE_POLL */
2957 #else                                                   /* WIN32 */
2958                 got_data = pgwin32_waitforsinglesocket(pgStatSock, FD_READ,
2959                                                                                            PGSTAT_SELECT_TIMEOUT * 1000);
2960 #endif
2961
2962                 /*
2963                  * If there is a message on the socket, read it and check for
2964                  * validity.
2965                  */
2966                 if (got_data)
2967                 {
2968                         len = recv(pgStatSock, (char *) &msg,
2969                                            sizeof(PgStat_Msg), 0);
2970                         if (len < 0)
2971                         {
2972                                 if (errno == EINTR)
2973                                         continue;
2974                                 ereport(ERROR,
2975                                                 (errcode_for_socket_access(),
2976                                                  errmsg("could not read statistics message: %m")));
2977                         }
2978
2979                         /*
2980                          * We ignore messages that are smaller than our common header
2981                          */
2982                         if (len < sizeof(PgStat_MsgHdr))
2983                                 continue;
2984
2985                         /*
2986                          * The received length must match the length in the header
2987                          */
2988                         if (msg.msg_hdr.m_size != len)
2989                                 continue;
2990
2991                         /*
2992                          * O.K. - we accept this message.  Process it.
2993                          */
2994                         switch (msg.msg_hdr.m_type)
2995                         {
2996                                 case PGSTAT_MTYPE_DUMMY:
2997                                         break;
2998
2999                                 case PGSTAT_MTYPE_INQUIRY:
3000                                         pgstat_recv_inquiry((PgStat_MsgInquiry *) &msg, len);
3001                                         break;
3002
3003                                 case PGSTAT_MTYPE_TABSTAT:
3004                                         pgstat_recv_tabstat((PgStat_MsgTabstat *) &msg, len);
3005                                         break;
3006
3007                                 case PGSTAT_MTYPE_TABPURGE:
3008                                         pgstat_recv_tabpurge((PgStat_MsgTabpurge *) &msg, len);
3009                                         break;
3010
3011                                 case PGSTAT_MTYPE_DROPDB:
3012                                         pgstat_recv_dropdb((PgStat_MsgDropdb *) &msg, len);
3013                                         break;
3014
3015                                 case PGSTAT_MTYPE_RESETCOUNTER:
3016                                         pgstat_recv_resetcounter((PgStat_MsgResetcounter *) &msg,
3017                                                                                          len);
3018                                         break;
3019
3020                                 case PGSTAT_MTYPE_RESETSHAREDCOUNTER:
3021                                         pgstat_recv_resetsharedcounter(
3022                                                                            (PgStat_MsgResetsharedcounter *) &msg,
3023                                                                                                    len);
3024                                         break;
3025
3026                                 case PGSTAT_MTYPE_RESETSINGLECOUNTER:
3027                                         pgstat_recv_resetsinglecounter(
3028                                                                            (PgStat_MsgResetsinglecounter *) &msg,
3029                                                                                                    len);
3030                                         break;
3031
3032                                 case PGSTAT_MTYPE_AUTOVAC_START:
3033                                         pgstat_recv_autovac((PgStat_MsgAutovacStart *) &msg, len);
3034                                         break;
3035
3036                                 case PGSTAT_MTYPE_VACUUM:
3037                                         pgstat_recv_vacuum((PgStat_MsgVacuum *) &msg, len);
3038                                         break;
3039
3040                                 case PGSTAT_MTYPE_ANALYZE:
3041                                         pgstat_recv_analyze((PgStat_MsgAnalyze *) &msg, len);
3042                                         break;
3043
3044                                 case PGSTAT_MTYPE_BGWRITER:
3045                                         pgstat_recv_bgwriter((PgStat_MsgBgWriter *) &msg, len);
3046                                         break;
3047
3048                                 case PGSTAT_MTYPE_FUNCSTAT:
3049                                         pgstat_recv_funcstat((PgStat_MsgFuncstat *) &msg, len);
3050                                         break;
3051
3052                                 case PGSTAT_MTYPE_FUNCPURGE:
3053                                         pgstat_recv_funcpurge((PgStat_MsgFuncpurge *) &msg, len);
3054                                         break;
3055
3056                                 default:
3057                                         break;
3058                         }
3059                 }
3060                 else
3061                 {
3062                         /*
3063                          * We can only get here if the select/poll timeout elapsed. Check
3064                          * for postmaster death.
3065                          */
3066                         if (!PostmasterIsAlive(true))
3067                                 break;
3068                 }
3069         }                                                       /* end of message-processing loop */
3070
3071         /*
3072          * Save the final stats to reuse at next startup.
3073          */
3074         pgstat_write_statsfile(true);
3075
3076         exit(0);
3077 }
3078
3079
3080 /* SIGQUIT signal handler for collector process */
3081 static void
3082 pgstat_exit(SIGNAL_ARGS)
3083 {
3084         need_exit = true;
3085 }
3086
3087 /* SIGHUP handler for collector process */
3088 static void
3089 pgstat_sighup_handler(SIGNAL_ARGS)
3090 {
3091         got_SIGHUP = true;
3092 }
3093
3094
3095 /*
3096  * Lookup the hash table entry for the specified database. If no hash
3097  * table entry exists, initialize it, if the create parameter is true.
3098  * Else, return NULL.
3099  */
3100 static PgStat_StatDBEntry *
3101 pgstat_get_db_entry(Oid databaseid, bool create)
3102 {
3103         PgStat_StatDBEntry *result;
3104         bool            found;
3105         HASHACTION      action = (create ? HASH_ENTER : HASH_FIND);
3106
3107         /* Lookup or create the hash table entry for this database */
3108         result = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
3109                                                                                                 &databaseid,
3110                                                                                                 action, &found);
3111
3112         if (!create && !found)
3113                 return NULL;
3114
3115         /* If not found, initialize the new one. */
3116         if (!found)
3117         {
3118                 HASHCTL         hash_ctl;
3119
3120                 result->tables = NULL;
3121                 result->functions = NULL;
3122                 result->n_xact_commit = 0;
3123                 result->n_xact_rollback = 0;
3124                 result->n_blocks_fetched = 0;
3125                 result->n_blocks_hit = 0;
3126                 result->n_tuples_returned = 0;
3127                 result->n_tuples_fetched = 0;
3128                 result->n_tuples_inserted = 0;
3129                 result->n_tuples_updated = 0;
3130                 result->n_tuples_deleted = 0;
3131                 result->last_autovac_time = 0;
3132
3133                 memset(&hash_ctl, 0, sizeof(hash_ctl));
3134                 hash_ctl.keysize = sizeof(Oid);
3135                 hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
3136                 hash_ctl.hash = oid_hash;
3137                 result->tables = hash_create("Per-database table",
3138                                                                          PGSTAT_TAB_HASH_SIZE,
3139                                                                          &hash_ctl,
3140                                                                          HASH_ELEM | HASH_FUNCTION);
3141
3142                 hash_ctl.keysize = sizeof(Oid);
3143                 hash_ctl.entrysize = sizeof(PgStat_StatFuncEntry);
3144                 hash_ctl.hash = oid_hash;
3145                 result->functions = hash_create("Per-database function",
3146                                                                                 PGSTAT_FUNCTION_HASH_SIZE,
3147                                                                                 &hash_ctl,
3148                                                                                 HASH_ELEM | HASH_FUNCTION);
3149         }
3150
3151         return result;
3152 }
3153
3154
3155 /*
3156  * Lookup the hash table entry for the specified table. If no hash
3157  * table entry exists, initialize it, if the create parameter is true.
3158  * Else, return NULL.
3159  */
3160 static PgStat_StatTabEntry *
3161 pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry, Oid tableoid, bool create)
3162 {
3163         PgStat_StatTabEntry *result;
3164         bool            found;
3165         HASHACTION      action = (create ? HASH_ENTER : HASH_FIND);
3166
3167         /* Lookup or create the hash table entry for this table */
3168         result = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
3169                                                                                                  &tableoid,
3170                                                                                                  action, &found);
3171
3172         if (!create && !found)
3173                 return NULL;
3174
3175         /* If not found, initialize the new one. */
3176         if (!found)
3177         {
3178                 result->numscans = 0;
3179                 result->tuples_returned = 0;
3180                 result->tuples_fetched = 0;
3181                 result->tuples_inserted = 0;
3182                 result->tuples_updated = 0;
3183                 result->tuples_deleted = 0;
3184                 result->tuples_hot_updated = 0;
3185                 result->n_live_tuples = 0;
3186                 result->n_dead_tuples = 0;
3187                 result->changes_since_analyze = 0;
3188                 result->blocks_fetched = 0;
3189                 result->blocks_hit = 0;
3190
3191                 result->vacuum_timestamp = 0;
3192                 result->autovac_vacuum_timestamp = 0;
3193                 result->analyze_timestamp = 0;
3194                 result->autovac_analyze_timestamp = 0;
3195         }
3196
3197         return result;
3198 }
3199
3200
3201 /* ----------
3202  * pgstat_write_statsfile() -
3203  *
3204  *      Tell the news.
3205  *      If writing to the permanent file (happens when the collector is
3206  *      shutting down only), remove the temporary file so that backends
3207  *      starting up under a new postmaster can't read the old data before
3208  *      the new collector is ready.
3209  * ----------
3210  */
3211 static void
3212 pgstat_write_statsfile(bool permanent)
3213 {
3214         HASH_SEQ_STATUS hstat;
3215         HASH_SEQ_STATUS tstat;
3216         HASH_SEQ_STATUS fstat;
3217         PgStat_StatDBEntry *dbentry;
3218         PgStat_StatTabEntry *tabentry;
3219         PgStat_StatFuncEntry *funcentry;
3220         FILE       *fpout;
3221         int32           format_id;
3222         const char *tmpfile = permanent ? PGSTAT_STAT_PERMANENT_TMPFILE : pgstat_stat_tmpname;
3223         const char *statfile = permanent ? PGSTAT_STAT_PERMANENT_FILENAME : pgstat_stat_filename;
3224
3225         /*
3226          * Open the statistics temp file to write out the current values.
3227          */
3228         fpout = AllocateFile(tmpfile, PG_BINARY_W);
3229         if (fpout == NULL)
3230         {
3231                 ereport(LOG,
3232                                 (errcode_for_file_access(),
3233                                  errmsg("could not open temporary statistics file \"%s\": %m",
3234                                                 tmpfile)));
3235                 return;
3236         }
3237
3238         /*
3239          * Set the timestamp of the stats file.
3240          */
3241         globalStats.stats_timestamp = GetCurrentTimestamp();
3242
3243         /*
3244          * Write the file header --- currently just a format ID.
3245          */
3246         format_id = PGSTAT_FILE_FORMAT_ID;
3247         fwrite(&format_id, sizeof(format_id), 1, fpout);
3248
3249         /*
3250          * Write global stats struct
3251          */
3252         fwrite(&globalStats, sizeof(globalStats), 1, fpout);
3253
3254         /*
3255          * Walk through the database table.
3256          */
3257         hash_seq_init(&hstat, pgStatDBHash);
3258         while ((dbentry = (PgStat_StatDBEntry *) hash_seq_search(&hstat)) != NULL)
3259         {
3260                 /*
3261                  * Write out the DB entry including the number of live backends. We
3262                  * don't write the tables or functions pointers, since they're of no
3263                  * use to any other process.
3264                  */
3265                 fputc('D', fpout);
3266                 fwrite(dbentry, offsetof(PgStat_StatDBEntry, tables), 1, fpout);
3267
3268                 /*
3269                  * Walk through the database's access stats per table.
3270                  */
3271                 hash_seq_init(&tstat, dbentry->tables);
3272                 while ((tabentry = (PgStat_StatTabEntry *) hash_seq_search(&tstat)) != NULL)
3273                 {
3274                         fputc('T', fpout);
3275                         fwrite(tabentry, sizeof(PgStat_StatTabEntry), 1, fpout);
3276                 }
3277
3278                 /*
3279                  * Walk through the database's function stats table.
3280                  */
3281                 hash_seq_init(&fstat, dbentry->functions);
3282                 while ((funcentry = (PgStat_StatFuncEntry *) hash_seq_search(&fstat)) != NULL)
3283                 {
3284                         fputc('F', fpout);
3285                         fwrite(funcentry, sizeof(PgStat_StatFuncEntry), 1, fpout);
3286                 }
3287
3288                 /*
3289                  * Mark the end of this DB
3290                  */
3291                 fputc('d', fpout);
3292         }
3293
3294         /*
3295          * No more output to be done. Close the temp file and replace the old
3296          * pgstat.stat with it.  The ferror() check replaces testing for error
3297          * after each individual fputc or fwrite above.
3298          */
3299         fputc('E', fpout);
3300
3301         if (ferror(fpout))
3302         {
3303                 ereport(LOG,
3304                                 (errcode_for_file_access(),
3305                            errmsg("could not write temporary statistics file \"%s\": %m",
3306                                           tmpfile)));
3307                 FreeFile(fpout);
3308                 unlink(tmpfile);
3309         }
3310         else if (FreeFile(fpout) < 0)
3311         {
3312                 ereport(LOG,
3313                                 (errcode_for_file_access(),
3314                            errmsg("could not close temporary statistics file \"%s\": %m",
3315                                           tmpfile)));
3316                 unlink(tmpfile);
3317         }
3318         else if (rename(tmpfile, statfile) < 0)
3319         {
3320                 ereport(LOG,
3321                                 (errcode_for_file_access(),
3322                                  errmsg("could not rename temporary statistics file \"%s\" to \"%s\": %m",
3323                                                 tmpfile, statfile)));
3324                 unlink(tmpfile);
3325         }
3326         else
3327         {
3328                 /*
3329                  * Successful write, so update last_statwrite.
3330                  */
3331                 last_statwrite = globalStats.stats_timestamp;
3332
3333                 /*
3334                  * If there is clock skew between backends and the collector, we could
3335                  * receive a stats request time that's in the future.  If so, complain
3336                  * and reset last_statrequest.  Resetting ensures that no inquiry
3337                  * message can cause more than one stats file write to occur.
3338                  */
3339                 if (last_statrequest > last_statwrite)
3340                 {
3341                         char       *reqtime;
3342                         char       *mytime;
3343
3344                         /* Copy because timestamptz_to_str returns a static buffer */
3345                         reqtime = pstrdup(timestamptz_to_str(last_statrequest));
3346                         mytime = pstrdup(timestamptz_to_str(last_statwrite));
3347                         elog(LOG, "last_statrequest %s is later than collector's time %s",
3348                                  reqtime, mytime);
3349                         pfree(reqtime);
3350                         pfree(mytime);
3351
3352                         last_statrequest = last_statwrite;
3353                 }
3354         }
3355
3356         if (permanent)
3357                 unlink(pgstat_stat_filename);
3358 }
3359
3360
3361 /* ----------
3362  * pgstat_read_statsfile() -
3363  *
3364  *      Reads in an existing statistics collector file and initializes the
3365  *      databases' hash table (whose entries point to the tables' hash tables).
3366  * ----------
3367  */
3368 static HTAB *
3369 pgstat_read_statsfile(Oid onlydb, bool permanent)
3370 {
3371         PgStat_StatDBEntry *dbentry;
3372         PgStat_StatDBEntry dbbuf;
3373         PgStat_StatTabEntry *tabentry;
3374         PgStat_StatTabEntry tabbuf;
3375         PgStat_StatFuncEntry funcbuf;
3376         PgStat_StatFuncEntry *funcentry;
3377         HASHCTL         hash_ctl;
3378         HTAB       *dbhash;
3379         HTAB       *tabhash = NULL;
3380         HTAB       *funchash = NULL;
3381         FILE       *fpin;
3382         int32           format_id;
3383         bool            found;
3384         const char *statfile = permanent ? PGSTAT_STAT_PERMANENT_FILENAME : pgstat_stat_filename;
3385
3386         /*
3387          * The tables will live in pgStatLocalContext.
3388          */
3389         pgstat_setup_memcxt();
3390
3391         /*
3392          * Create the DB hashtable
3393          */
3394         memset(&hash_ctl, 0, sizeof(hash_ctl));
3395         hash_ctl.keysize = sizeof(Oid);
3396         hash_ctl.entrysize = sizeof(PgStat_StatDBEntry);
3397         hash_ctl.hash = oid_hash;
3398         hash_ctl.hcxt = pgStatLocalContext;
3399         dbhash = hash_create("Databases hash", PGSTAT_DB_HASH_SIZE, &hash_ctl,
3400                                                  HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
3401
3402         /*
3403          * Clear out global statistics so they start from zero in case we can't
3404          * load an existing statsfile.
3405          */
3406         memset(&globalStats, 0, sizeof(globalStats));
3407
3408         /*
3409          * Try to open the status file. If it doesn't exist, the backends simply
3410          * return zero for anything and the collector simply starts from scratch
3411          * with empty counters.
3412          *
3413          * ENOENT is a possibility if the stats collector is not running or has
3414          * not yet written the stats file the first time.  Any other failure
3415          * condition is suspicious.
3416          */
3417         if ((fpin = AllocateFile(statfile, PG_BINARY_R)) == NULL)
3418         {
3419                 if (errno != ENOENT)
3420                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3421                                         (errcode_for_file_access(),
3422                                          errmsg("could not open statistics file \"%s\": %m",
3423                                                         statfile)));
3424                 return dbhash;
3425         }
3426
3427         /*
3428          * Verify it's of the expected format.
3429          */
3430         if (fread(&format_id, 1, sizeof(format_id), fpin) != sizeof(format_id)
3431                 || format_id != PGSTAT_FILE_FORMAT_ID)
3432         {
3433                 ereport(pgStatRunningInCollector ? LOG : WARNING,
3434                                 (errmsg("corrupted statistics file \"%s\"", statfile)));
3435                 goto done;
3436         }
3437
3438         /*
3439          * Read global stats struct
3440          */
3441         if (fread(&globalStats, 1, sizeof(globalStats), fpin) != sizeof(globalStats))
3442         {
3443                 ereport(pgStatRunningInCollector ? LOG : WARNING,
3444                                 (errmsg("corrupted statistics file \"%s\"", statfile)));
3445                 goto done;
3446         }
3447
3448         /*
3449          * We found an existing collector stats file. Read it and put all the
3450          * hashtable entries into place.
3451          */
3452         for (;;)
3453         {
3454                 switch (fgetc(fpin))
3455                 {
3456                                 /*
3457                                  * 'D'  A PgStat_StatDBEntry struct describing a database
3458                                  * follows. Subsequently, zero to many 'T' and 'F' entries
3459                                  * will follow until a 'd' is encountered.
3460                                  */
3461                         case 'D':
3462                                 if (fread(&dbbuf, 1, offsetof(PgStat_StatDBEntry, tables),
3463                                                   fpin) != offsetof(PgStat_StatDBEntry, tables))
3464                                 {
3465                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3466                                                         (errmsg("corrupted statistics file \"%s\"",
3467                                                                         statfile)));
3468                                         goto done;
3469                                 }
3470
3471                                 /*
3472                                  * Add to the DB hash
3473                                  */
3474                                 dbentry = (PgStat_StatDBEntry *) hash_search(dbhash,
3475                                                                                                   (void *) &dbbuf.databaseid,
3476                                                                                                                          HASH_ENTER,
3477                                                                                                                          &found);
3478                                 if (found)
3479                                 {
3480                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3481                                                         (errmsg("corrupted statistics file \"%s\"",
3482                                                                         statfile)));
3483                                         goto done;
3484                                 }
3485
3486                                 memcpy(dbentry, &dbbuf, sizeof(PgStat_StatDBEntry));
3487                                 dbentry->tables = NULL;
3488                                 dbentry->functions = NULL;
3489
3490                                 /*
3491                                  * Don't collect tables if not the requested DB (or the
3492                                  * shared-table info)
3493                                  */
3494                                 if (onlydb != InvalidOid)
3495                                 {
3496                                         if (dbbuf.databaseid != onlydb &&
3497                                                 dbbuf.databaseid != InvalidOid)
3498                                                 break;
3499                                 }
3500
3501                                 memset(&hash_ctl, 0, sizeof(hash_ctl));
3502                                 hash_ctl.keysize = sizeof(Oid);
3503                                 hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
3504                                 hash_ctl.hash = oid_hash;
3505                                 hash_ctl.hcxt = pgStatLocalContext;
3506                                 dbentry->tables = hash_create("Per-database table",
3507                                                                                           PGSTAT_TAB_HASH_SIZE,
3508                                                                                           &hash_ctl,
3509                                                                    HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
3510
3511                                 hash_ctl.keysize = sizeof(Oid);
3512                                 hash_ctl.entrysize = sizeof(PgStat_StatFuncEntry);
3513                                 hash_ctl.hash = oid_hash;
3514                                 hash_ctl.hcxt = pgStatLocalContext;
3515                                 dbentry->functions = hash_create("Per-database function",
3516                                                                                                  PGSTAT_FUNCTION_HASH_SIZE,
3517                                                                                                  &hash_ctl,
3518                                                                    HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
3519
3520                                 /*
3521                                  * Arrange that following records add entries to this
3522                                  * database's hash tables.
3523                                  */
3524                                 tabhash = dbentry->tables;
3525                                 funchash = dbentry->functions;
3526                                 break;
3527
3528                                 /*
3529                                  * 'd'  End of this database.
3530                                  */
3531                         case 'd':
3532                                 tabhash = NULL;
3533                                 funchash = NULL;
3534                                 break;
3535
3536                                 /*
3537                                  * 'T'  A PgStat_StatTabEntry follows.
3538                                  */
3539                         case 'T':
3540                                 if (fread(&tabbuf, 1, sizeof(PgStat_StatTabEntry),
3541                                                   fpin) != sizeof(PgStat_StatTabEntry))
3542                                 {
3543                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3544                                                         (errmsg("corrupted statistics file \"%s\"",
3545                                                                         statfile)));
3546                                         goto done;
3547                                 }
3548
3549                                 /*
3550                                  * Skip if table belongs to a not requested database.
3551                                  */
3552                                 if (tabhash == NULL)
3553                                         break;
3554
3555                                 tabentry = (PgStat_StatTabEntry *) hash_search(tabhash,
3556                                                                                                         (void *) &tabbuf.tableid,
3557                                                                                                                  HASH_ENTER, &found);
3558
3559                                 if (found)
3560                                 {
3561                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3562                                                         (errmsg("corrupted statistics file \"%s\"",
3563                                                                         statfile)));
3564                                         goto done;
3565                                 }
3566
3567                                 memcpy(tabentry, &tabbuf, sizeof(tabbuf));
3568                                 break;
3569
3570                                 /*
3571                                  * 'F'  A PgStat_StatFuncEntry follows.
3572                                  */
3573                         case 'F':
3574                                 if (fread(&funcbuf, 1, sizeof(PgStat_StatFuncEntry),
3575                                                   fpin) != sizeof(PgStat_StatFuncEntry))
3576                                 {
3577                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3578                                                         (errmsg("corrupted statistics file \"%s\"",
3579                                                                         statfile)));
3580                                         goto done;
3581                                 }
3582
3583                                 /*
3584                                  * Skip if function belongs to a not requested database.
3585                                  */
3586                                 if (funchash == NULL)
3587                                         break;
3588
3589                                 funcentry = (PgStat_StatFuncEntry *) hash_search(funchash,
3590                                                                                                 (void *) &funcbuf.functionid,
3591                                                                                                                  HASH_ENTER, &found);
3592
3593                                 if (found)
3594                                 {
3595                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3596                                                         (errmsg("corrupted statistics file \"%s\"",
3597                                                                         statfile)));
3598                                         goto done;
3599                                 }
3600
3601                                 memcpy(funcentry, &funcbuf, sizeof(funcbuf));
3602                                 break;
3603
3604                                 /*
3605                                  * 'E'  The EOF marker of a complete stats file.
3606                                  */
3607                         case 'E':
3608                                 goto done;
3609
3610                         default:
3611                                 ereport(pgStatRunningInCollector ? LOG : WARNING,
3612                                                 (errmsg("corrupted statistics file \"%s\"",
3613                                                                 statfile)));
3614                                 goto done;
3615                 }
3616         }
3617
3618 done:
3619         FreeFile(fpin);
3620
3621         if (permanent)
3622                 unlink(PGSTAT_STAT_PERMANENT_FILENAME);
3623
3624         return dbhash;
3625 }
3626
3627 /* ----------
3628  * pgstat_read_statsfile_timestamp() -
3629  *
3630  *      Attempt to fetch the timestamp of an existing stats file.
3631  *      Returns TRUE if successful (timestamp is stored at *ts).
3632  * ----------
3633  */
3634 static bool
3635 pgstat_read_statsfile_timestamp(bool permanent, TimestampTz *ts)
3636 {
3637         PgStat_GlobalStats myGlobalStats;
3638         FILE       *fpin;
3639         int32           format_id;
3640         const char *statfile = permanent ? PGSTAT_STAT_PERMANENT_FILENAME : pgstat_stat_filename;
3641
3642         /*
3643          * Try to open the status file.  As above, anything but ENOENT is worthy
3644          * of complaining about.
3645          */
3646         if ((fpin = AllocateFile(statfile, PG_BINARY_R)) == NULL)
3647         {
3648                 if (errno != ENOENT)
3649                         ereport(pgStatRunningInCollector ? LOG : WARNING,
3650                                         (errcode_for_file_access(),
3651                                          errmsg("could not open statistics file \"%s\": %m",
3652                                                         statfile)));
3653                 return false;
3654         }
3655
3656         /*
3657          * Verify it's of the expected format.
3658          */
3659         if (fread(&format_id, 1, sizeof(format_id), fpin) != sizeof(format_id)
3660                 || format_id != PGSTAT_FILE_FORMAT_ID)
3661         {
3662                 ereport(pgStatRunningInCollector ? LOG : WARNING,
3663                                 (errmsg("corrupted statistics file \"%s\"", statfile)));
3664                 FreeFile(fpin);
3665                 return false;
3666         }
3667
3668         /*
3669          * Read global stats struct
3670          */
3671         if (fread(&myGlobalStats, 1, sizeof(myGlobalStats), fpin) != sizeof(myGlobalStats))
3672         {
3673                 ereport(pgStatRunningInCollector ? LOG : WARNING,
3674                                 (errmsg("corrupted statistics file \"%s\"", statfile)));
3675                 FreeFile(fpin);
3676                 return false;
3677         }
3678
3679         *ts = myGlobalStats.stats_timestamp;
3680
3681         FreeFile(fpin);
3682         return true;
3683 }
3684
3685 /*
3686  * If not already done, read the statistics collector stats file into
3687  * some hash tables.  The results will be kept until pgstat_clear_snapshot()
3688  * is called (typically, at end of transaction).
3689  */
3690 static void
3691 backend_read_statsfile(void)
3692 {
3693         TimestampTz min_ts;
3694         int                     count;
3695
3696         /* already read it? */
3697         if (pgStatDBHash)
3698                 return;
3699         Assert(!pgStatRunningInCollector);
3700
3701         /*
3702          * We set the minimum acceptable timestamp to PGSTAT_STAT_INTERVAL msec
3703          * before now.  This indirectly ensures that the collector needn't write
3704          * the file more often than PGSTAT_STAT_INTERVAL.  In an autovacuum
3705          * worker, however, we want a lower delay to avoid using stale data, so we
3706          * use PGSTAT_RETRY_DELAY (since the number of worker is low, this
3707          * shouldn't be a problem).
3708          *
3709          * Note that we don't recompute min_ts after sleeping; so we might end up
3710          * accepting a file a bit older than PGSTAT_STAT_INTERVAL.      In practice
3711          * that shouldn't happen, though, as long as the sleep time is less than
3712          * PGSTAT_STAT_INTERVAL; and we don't want to lie to the collector about
3713          * what our cutoff time really is.
3714          */
3715         if (IsAutoVacuumWorkerProcess())
3716                 min_ts = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
3717                                                                                          -PGSTAT_RETRY_DELAY);
3718         else
3719                 min_ts = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
3720                                                                                          -PGSTAT_STAT_INTERVAL);
3721
3722         /*
3723          * Loop until fresh enough stats file is available or we ran out of time.
3724          * The stats inquiry message is sent repeatedly in case collector drops
3725          * it.
3726          */
3727         for (count = 0; count < PGSTAT_POLL_LOOP_COUNT; count++)
3728         {
3729                 TimestampTz file_ts = 0;
3730
3731                 CHECK_FOR_INTERRUPTS();
3732
3733                 if (pgstat_read_statsfile_timestamp(false, &file_ts) &&
3734                         file_ts >= min_ts)
3735                         break;
3736
3737                 /* Not there or too old, so kick the collector and wait a bit */
3738                 pgstat_send_inquiry(min_ts);
3739                 pg_usleep(PGSTAT_RETRY_DELAY * 1000L);
3740         }
3741
3742         if (count >= PGSTAT_POLL_LOOP_COUNT)
3743                 elog(WARNING, "pgstat wait timeout");
3744
3745         /* Autovacuum launcher wants stats about all databases */
3746         if (IsAutoVacuumLauncherProcess())
3747                 pgStatDBHash = pgstat_read_statsfile(InvalidOid, false);
3748         else
3749                 pgStatDBHash = pgstat_read_statsfile(MyDatabaseId, false);
3750 }
3751
3752
3753 /* ----------
3754  * pgstat_setup_memcxt() -
3755  *
3756  *      Create pgStatLocalContext, if not already done.
3757  * ----------
3758  */
3759 static void
3760 pgstat_setup_memcxt(void)
3761 {
3762         if (!pgStatLocalContext)
3763                 pgStatLocalContext = AllocSetContextCreate(TopMemoryContext,
3764                                                                                                    "Statistics snapshot",
3765                                                                                                    ALLOCSET_SMALL_MINSIZE,
3766                                                                                                    ALLOCSET_SMALL_INITSIZE,
3767                                                                                                    ALLOCSET_SMALL_MAXSIZE);
3768 }
3769
3770
3771 /* ----------
3772  * pgstat_clear_snapshot() -
3773  *
3774  *      Discard any data collected in the current transaction.  Any subsequent
3775  *      request will cause new snapshots to be read.
3776  *
3777  *      This is also invoked during transaction commit or abort to discard
3778  *      the no-longer-wanted snapshot.
3779  * ----------
3780  */
3781 void
3782 pgstat_clear_snapshot(void)
3783 {
3784         /* Release memory, if any was allocated */
3785         if (pgStatLocalContext)
3786                 MemoryContextDelete(pgStatLocalContext);
3787
3788         /* Reset variables */
3789         pgStatLocalContext = NULL;
3790         pgStatDBHash = NULL;
3791         localBackendStatusTable = NULL;
3792         localNumBackends = 0;
3793 }
3794
3795
3796 /* ----------
3797  * pgstat_recv_inquiry() -
3798  *
3799  *      Process stat inquiry requests.
3800  * ----------
3801  */
3802 static void
3803 pgstat_recv_inquiry(PgStat_MsgInquiry *msg, int len)
3804 {
3805         if (msg->inquiry_time > last_statrequest)
3806                 last_statrequest = msg->inquiry_time;
3807 }
3808
3809
3810 /* ----------
3811  * pgstat_recv_tabstat() -
3812  *
3813  *      Count what the backend has done.
3814  * ----------
3815  */
3816 static void
3817 pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
3818 {
3819         PgStat_StatDBEntry *dbentry;
3820         PgStat_StatTabEntry *tabentry;
3821         int                     i;
3822         bool            found;
3823
3824         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
3825
3826         /*
3827          * Update database-wide stats.
3828          */
3829         dbentry->n_xact_commit += (PgStat_Counter) (msg->m_xact_commit);
3830         dbentry->n_xact_rollback += (PgStat_Counter) (msg->m_xact_rollback);
3831
3832         /*
3833          * Process all table entries in the message.
3834          */
3835         for (i = 0; i < msg->m_nentries; i++)
3836         {
3837                 PgStat_TableEntry *tabmsg = &(msg->m_entry[i]);
3838
3839                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
3840                                                                                                         (void *) &(tabmsg->t_id),
3841                                                                                                            HASH_ENTER, &found);
3842
3843                 if (!found)
3844                 {
3845                         /*
3846                          * If it's a new table entry, initialize counters to the values we
3847                          * just got.
3848                          */
3849                         tabentry->numscans = tabmsg->t_counts.t_numscans;
3850                         tabentry->tuples_returned = tabmsg->t_counts.t_tuples_returned;
3851                         tabentry->tuples_fetched = tabmsg->t_counts.t_tuples_fetched;
3852                         tabentry->tuples_inserted = tabmsg->t_counts.t_tuples_inserted;
3853                         tabentry->tuples_updated = tabmsg->t_counts.t_tuples_updated;
3854                         tabentry->tuples_deleted = tabmsg->t_counts.t_tuples_deleted;
3855                         tabentry->tuples_hot_updated = tabmsg->t_counts.t_tuples_hot_updated;
3856                         tabentry->n_live_tuples = tabmsg->t_counts.t_delta_live_tuples;
3857                         tabentry->n_dead_tuples = tabmsg->t_counts.t_delta_dead_tuples;
3858                         tabentry->changes_since_analyze = tabmsg->t_counts.t_changed_tuples;
3859                         tabentry->blocks_fetched = tabmsg->t_counts.t_blocks_fetched;
3860                         tabentry->blocks_hit = tabmsg->t_counts.t_blocks_hit;
3861
3862                         tabentry->vacuum_timestamp = 0;
3863                         tabentry->autovac_vacuum_timestamp = 0;
3864                         tabentry->analyze_timestamp = 0;
3865                         tabentry->autovac_analyze_timestamp = 0;
3866                 }
3867                 else
3868                 {
3869                         /*
3870                          * Otherwise add the values to the existing entry.
3871                          */
3872                         tabentry->numscans += tabmsg->t_counts.t_numscans;
3873                         tabentry->tuples_returned += tabmsg->t_counts.t_tuples_returned;
3874                         tabentry->tuples_fetched += tabmsg->t_counts.t_tuples_fetched;
3875                         tabentry->tuples_inserted += tabmsg->t_counts.t_tuples_inserted;
3876                         tabentry->tuples_updated += tabmsg->t_counts.t_tuples_updated;
3877                         tabentry->tuples_deleted += tabmsg->t_counts.t_tuples_deleted;
3878                         tabentry->tuples_hot_updated += tabmsg->t_counts.t_tuples_hot_updated;
3879                         tabentry->n_live_tuples += tabmsg->t_counts.t_delta_live_tuples;
3880                         tabentry->n_dead_tuples += tabmsg->t_counts.t_delta_dead_tuples;
3881                         tabentry->changes_since_analyze += tabmsg->t_counts.t_changed_tuples;
3882                         tabentry->blocks_fetched += tabmsg->t_counts.t_blocks_fetched;
3883                         tabentry->blocks_hit += tabmsg->t_counts.t_blocks_hit;
3884                 }
3885
3886                 /* Clamp n_live_tuples in case of negative delta_live_tuples */
3887                 tabentry->n_live_tuples = Max(tabentry->n_live_tuples, 0);
3888                 /* Likewise for n_dead_tuples */
3889                 tabentry->n_dead_tuples = Max(tabentry->n_dead_tuples, 0);
3890
3891                 /*
3892                  * Add per-table stats to the per-database entry, too.
3893                  */
3894                 dbentry->n_tuples_returned += tabmsg->t_counts.t_tuples_returned;
3895                 dbentry->n_tuples_fetched += tabmsg->t_counts.t_tuples_fetched;
3896                 dbentry->n_tuples_inserted += tabmsg->t_counts.t_tuples_inserted;
3897                 dbentry->n_tuples_updated += tabmsg->t_counts.t_tuples_updated;
3898                 dbentry->n_tuples_deleted += tabmsg->t_counts.t_tuples_deleted;
3899                 dbentry->n_blocks_fetched += tabmsg->t_counts.t_blocks_fetched;
3900                 dbentry->n_blocks_hit += tabmsg->t_counts.t_blocks_hit;
3901         }
3902 }
3903
3904
3905 /* ----------
3906  * pgstat_recv_tabpurge() -
3907  *
3908  *      Arrange for dead table removal.
3909  * ----------
3910  */
3911 static void
3912 pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len)
3913 {
3914         PgStat_StatDBEntry *dbentry;
3915         int                     i;
3916
3917         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
3918
3919         /*
3920          * No need to purge if we don't even know the database.
3921          */
3922         if (!dbentry || !dbentry->tables)
3923                 return;
3924
3925         /*
3926          * Process all table entries in the message.
3927          */
3928         for (i = 0; i < msg->m_nentries; i++)
3929         {
3930                 /* Remove from hashtable if present; we don't care if it's not. */
3931                 (void) hash_search(dbentry->tables,
3932                                                    (void *) &(msg->m_tableid[i]),
3933                                                    HASH_REMOVE, NULL);
3934         }
3935 }
3936
3937
3938 /* ----------
3939  * pgstat_recv_dropdb() -
3940  *
3941  *      Arrange for dead database removal
3942  * ----------
3943  */
3944 static void
3945 pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len)
3946 {
3947         PgStat_StatDBEntry *dbentry;
3948
3949         /*
3950          * Lookup the database in the hashtable.
3951          */
3952         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
3953
3954         /*
3955          * If found, remove it.
3956          */
3957         if (dbentry)
3958         {
3959                 if (dbentry->tables != NULL)
3960                         hash_destroy(dbentry->tables);
3961                 if (dbentry->functions != NULL)
3962                         hash_destroy(dbentry->functions);
3963
3964                 if (hash_search(pgStatDBHash,
3965                                                 (void *) &(dbentry->databaseid),
3966                                                 HASH_REMOVE, NULL) == NULL)
3967                         ereport(ERROR,
3968                                         (errmsg("database hash table corrupted "
3969                                                         "during cleanup --- abort")));
3970         }
3971 }
3972
3973
3974 /* ----------
3975  * pgstat_recv_resetcounter() -
3976  *
3977  *      Reset the statistics for the specified database.
3978  * ----------
3979  */
3980 static void
3981 pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len)
3982 {
3983         HASHCTL         hash_ctl;
3984         PgStat_StatDBEntry *dbentry;
3985
3986         /*
3987          * Lookup the database in the hashtable.  Nothing to do if not there.
3988          */
3989         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
3990
3991         if (!dbentry)
3992                 return;
3993
3994         /*
3995          * We simply throw away all the database's table entries by recreating a
3996          * new hash table for them.
3997          */
3998         if (dbentry->tables != NULL)
3999                 hash_destroy(dbentry->tables);
4000         if (dbentry->functions != NULL)
4001                 hash_destroy(dbentry->functions);
4002
4003         dbentry->tables = NULL;
4004         dbentry->functions = NULL;
4005         dbentry->n_xact_commit = 0;
4006         dbentry->n_xact_rollback = 0;
4007         dbentry->n_blocks_fetched = 0;
4008         dbentry->n_blocks_hit = 0;
4009
4010         memset(&hash_ctl, 0, sizeof(hash_ctl));
4011         hash_ctl.keysize = sizeof(Oid);
4012         hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
4013         hash_ctl.hash = oid_hash;
4014         dbentry->tables = hash_create("Per-database table",
4015                                                                   PGSTAT_TAB_HASH_SIZE,
4016                                                                   &hash_ctl,
4017                                                                   HASH_ELEM | HASH_FUNCTION);
4018
4019         hash_ctl.keysize = sizeof(Oid);
4020         hash_ctl.entrysize = sizeof(PgStat_StatFuncEntry);
4021         hash_ctl.hash = oid_hash;
4022         dbentry->functions = hash_create("Per-database function",
4023                                                                          PGSTAT_FUNCTION_HASH_SIZE,
4024                                                                          &hash_ctl,
4025                                                                          HASH_ELEM | HASH_FUNCTION);
4026 }
4027
4028 /* ----------
4029  * pgstat_recv_resetshared() -
4030  *
4031  *      Reset some shared statistics of the cluster.
4032  * ----------
4033  */
4034 static void
4035 pgstat_recv_resetsharedcounter(PgStat_MsgResetsharedcounter *msg, int len)
4036 {
4037         if (msg->m_resettarget == RESET_BGWRITER)
4038         {
4039                 /* Reset the global background writer statistics for the cluster. */
4040                 memset(&globalStats, 0, sizeof(globalStats));
4041         }
4042
4043         /*
4044          * Presumably the sender of this message validated the target, don't
4045          * complain here if it's not valid
4046          */
4047 }
4048
4049 /* ----------
4050  * pgstat_recv_resetsinglecounter() -
4051  *
4052  *      Reset a statistics for a single object
4053  * ----------
4054  */
4055 static void
4056 pgstat_recv_resetsinglecounter(PgStat_MsgResetsinglecounter *msg, int len)
4057 {
4058         PgStat_StatDBEntry *dbentry;
4059
4060         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
4061
4062         if (!dbentry)
4063                 return;
4064
4065
4066         /* Remove object if it exists, ignore it if not */
4067         if (msg->m_resettype == RESET_TABLE)
4068                 (void) hash_search(dbentry->tables, (void *) &(msg->m_objectid), HASH_REMOVE, NULL);
4069         else if (msg->m_resettype == RESET_FUNCTION)
4070                 (void) hash_search(dbentry->functions, (void *) &(msg->m_objectid), HASH_REMOVE, NULL);
4071 }
4072
4073 /* ----------
4074  * pgstat_recv_autovac() -
4075  *
4076  *      Process an autovacuum signalling message.
4077  * ----------
4078  */
4079 static void
4080 pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len)
4081 {
4082         PgStat_StatDBEntry *dbentry;
4083
4084         /*
4085          * Store the last autovacuum time in the database's hashtable entry.
4086          */
4087         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
4088
4089         dbentry->last_autovac_time = msg->m_start_time;
4090 }
4091
4092 /* ----------
4093  * pgstat_recv_vacuum() -
4094  *
4095  *      Process a VACUUM message.
4096  * ----------
4097  */
4098 static void
4099 pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len)
4100 {
4101         PgStat_StatDBEntry *dbentry;
4102         PgStat_StatTabEntry *tabentry;
4103
4104         /*
4105          * Store the data in the table's hashtable entry.
4106          */
4107         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
4108
4109         tabentry = pgstat_get_tab_entry(dbentry, msg->m_tableoid, true);
4110
4111         if (msg->m_adopt_counts)
4112                 tabentry->n_live_tuples = msg->m_tuples;
4113         /* Resetting dead_tuples to 0 is an approximation ... */
4114         tabentry->n_dead_tuples = 0;
4115
4116         if (msg->m_autovacuum)
4117                 tabentry->autovac_vacuum_timestamp = msg->m_vacuumtime;
4118         else
4119                 tabentry->vacuum_timestamp = msg->m_vacuumtime;
4120 }
4121
4122 /* ----------
4123  * pgstat_recv_analyze() -
4124  *
4125  *      Process an ANALYZE message.
4126  * ----------
4127  */
4128 static void
4129 pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len)
4130 {
4131         PgStat_StatDBEntry *dbentry;
4132         PgStat_StatTabEntry *tabentry;
4133
4134         /*
4135          * Store the data in the table's hashtable entry.
4136          */
4137         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
4138
4139         tabentry = pgstat_get_tab_entry(dbentry, msg->m_tableoid, true);
4140
4141         if (msg->m_adopt_counts)
4142         {
4143                 tabentry->n_live_tuples = msg->m_live_tuples;
4144                 tabentry->n_dead_tuples = msg->m_dead_tuples;
4145         }
4146
4147         /*
4148          * We reset changes_since_analyze to zero, forgetting any changes that
4149          * occurred while the ANALYZE was in progress.
4150          */
4151         tabentry->changes_since_analyze = 0;
4152
4153         if (msg->m_autovacuum)
4154                 tabentry->autovac_analyze_timestamp = msg->m_analyzetime;
4155         else
4156                 tabentry->analyze_timestamp = msg->m_analyzetime;
4157 }
4158
4159
4160 /* ----------
4161  * pgstat_recv_bgwriter() -
4162  *
4163  *      Process a BGWRITER message.
4164  * ----------
4165  */
4166 static void
4167 pgstat_recv_bgwriter(PgStat_MsgBgWriter *msg, int len)
4168 {
4169         globalStats.timed_checkpoints += msg->m_timed_checkpoints;
4170         globalStats.requested_checkpoints += msg->m_requested_checkpoints;
4171         globalStats.buf_written_checkpoints += msg->m_buf_written_checkpoints;
4172         globalStats.buf_written_clean += msg->m_buf_written_clean;
4173         globalStats.maxwritten_clean += msg->m_maxwritten_clean;
4174         globalStats.buf_written_backend += msg->m_buf_written_backend;
4175         globalStats.buf_alloc += msg->m_buf_alloc;
4176 }
4177
4178 /* ----------
4179  * pgstat_recv_funcstat() -
4180  *
4181  *      Count what the backend has done.
4182  * ----------
4183  */
4184 static void
4185 pgstat_recv_funcstat(PgStat_MsgFuncstat *msg, int len)
4186 {
4187         PgStat_FunctionEntry *funcmsg = &(msg->m_entry[0]);
4188         PgStat_StatDBEntry *dbentry;
4189         PgStat_StatFuncEntry *funcentry;
4190         int                     i;
4191         bool            found;
4192
4193         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
4194
4195         /*
4196          * Process all function entries in the message.
4197          */
4198         for (i = 0; i < msg->m_nentries; i++, funcmsg++)
4199         {
4200                 funcentry = (PgStat_StatFuncEntry *) hash_search(dbentry->functions,
4201                                                                                                    (void *) &(funcmsg->f_id),
4202                                                                                                                  HASH_ENTER, &found);
4203
4204                 if (!found)
4205                 {
4206                         /*
4207                          * If it's a new function entry, initialize counters to the values
4208                          * we just got.
4209                          */
4210                         funcentry->f_numcalls = funcmsg->f_numcalls;
4211                         funcentry->f_time = funcmsg->f_time;
4212                         funcentry->f_time_self = funcmsg->f_time_self;
4213                 }
4214                 else
4215                 {
4216                         /*
4217                          * Otherwise add the values to the existing entry.
4218                          */
4219                         funcentry->f_numcalls += funcmsg->f_numcalls;
4220                         funcentry->f_time += funcmsg->f_time;
4221                         funcentry->f_time_self += funcmsg->f_time_self;
4222                 }
4223         }
4224 }
4225
4226 /* ----------
4227  * pgstat_recv_funcpurge() -
4228  *
4229  *      Arrange for dead function removal.
4230  * ----------
4231  */
4232 static void
4233 pgstat_recv_funcpurge(PgStat_MsgFuncpurge *msg, int len)
4234 {
4235         PgStat_StatDBEntry *dbentry;
4236         int                     i;
4237
4238         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
4239
4240         /*
4241          * No need to purge if we don't even know the database.
4242          */
4243         if (!dbentry || !dbentry->functions)
4244                 return;
4245
4246         /*
4247          * Process all function entries in the message.
4248          */
4249         for (i = 0; i < msg->m_nentries; i++)
4250         {
4251                 /* Remove from hashtable if present; we don't care if it's not. */
4252                 (void) hash_search(dbentry->functions,
4253                                                    (void *) &(msg->m_functionid[i]),
4254                                                    HASH_REMOVE, NULL);
4255         }
4256 }