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-2005, PostgreSQL Global Development Group
  15  *
  16  *      $PostgreSQL: pgsql/src/backend/postmaster/pgstat.c,v 1.108 2005/09/24 17:53:14 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
  32 #include "pgstat.h"
  33
  34 #include "access/heapam.h"
  35 #include "access/xact.h"
  36 #include "catalog/pg_database.h"
  37 #include "libpq/libpq.h"
  38 #include "libpq/pqsignal.h"
  39 #include "mb/pg_wchar.h"
  40 #include "miscadmin.h"
  41 #include "postmaster/autovacuum.h"
  42 #include "postmaster/fork_process.h"
  43 #include "postmaster/postmaster.h"
  44 #include "storage/backendid.h"
  45 #include "storage/fd.h"
  46 #include "storage/ipc.h"
  47 #include "storage/pg_shmem.h"
  48 #include "storage/pmsignal.h"
  49 #include "storage/procarray.h"
  50 #include "tcop/tcopprot.h"
  51 #include "utils/hsearch.h"
  52 #include "utils/memutils.h"
  53 #include "utils/ps_status.h"
  54 #include "utils/rel.h"
  55 #include "utils/syscache.h"
  56
  57
  58 /* ----------
  59  * Paths for the statistics files (relative to installation's $PGDATA).
  60  * ----------
  61  */
  62 #define PGSTAT_STAT_FILENAME    "global/pgstat.stat"
  63 #define PGSTAT_STAT_TMPFILE             "global/pgstat.tmp"
  64
  65 /* ----------
  66  * Timer definitions.
  67  * ----------
  68  */
  69 #define PGSTAT_STAT_INTERVAL    500             /* How often to write the status
  70                                                                                  * file; in milliseconds. */
  71
  72 #define PGSTAT_DESTROY_DELAY    10000   /* How long to keep destroyed
  73                                                                                  * objects known, to give delayed
  74                                                                                  * UDP packets time to arrive; in
  75                                                                                  * milliseconds. */
  76
  77 #define PGSTAT_DESTROY_COUNT    (PGSTAT_DESTROY_DELAY / PGSTAT_STAT_INTERVAL)
  78
  79 #define PGSTAT_RESTART_INTERVAL 60              /* How often to attempt to restart
  80                                                                                  * a failed statistics collector;
  81                                                                                  * in seconds. */
  82
  83 /* ----------
  84  * Amount of space reserved in pgstat_recvbuffer().
  85  * ----------
  86  */
  87 #define PGSTAT_RECVBUFFERSZ             ((int) (1024 * sizeof(PgStat_Msg)))
  88
  89 /* ----------
  90  * The initial size hints for the hash tables used in the collector.
  91  * ----------
  92  */
  93 #define PGSTAT_DB_HASH_SIZE             16
  94 #define PGSTAT_BE_HASH_SIZE             512
  95 #define PGSTAT_TAB_HASH_SIZE    512
  96
  97
  98 /* ----------
  99  * GUC parameters
 100  * ----------
 101  */
 102 bool            pgstat_collect_startcollector = true;
 103 bool            pgstat_collect_resetonpmstart = false;
 104 bool            pgstat_collect_querystring = false;
 105 bool            pgstat_collect_tuplelevel = false;
 106 bool            pgstat_collect_blocklevel = false;
 107
 108 /* ----------
 109  * Local data
 110  * ----------
 111  */
 112 NON_EXEC_STATIC int pgStatSock = -1;
 113 NON_EXEC_STATIC int pgStatPipe[2] = {-1,-1};
 114 static struct sockaddr_storage pgStatAddr;
 115 static pid_t pgStatCollectorPid = 0;
 116
 117 static time_t last_pgstat_start_time;
 118
 119 static long pgStatNumMessages = 0;
 120
 121 static bool pgStatRunningInCollector = FALSE;
 122
 123 /*
 124  * Place where backends store per-table info to be sent to the collector.
 125  * We store shared relations separately from non-shared ones, to be able to
 126  * send them in separate messages.
 127  */
 128 typedef struct TabStatArray
 129 {
 130         int             tsa_alloc;                                      /* num allocated */
 131         int             tsa_used;                                       /* num actually used */
 132         PgStat_MsgTabstat **tsa_messages;       /* the array itself */
 133 } TabStatArray;
 134
 135 #define TABSTAT_QUANTUM         4       /* we alloc this many at a time */
 136
 137 static TabStatArray RegularTabStat = { 0, 0, NULL };
 138 static TabStatArray SharedTabStat = { 0, 0, NULL };
 139
 140 static int      pgStatXactCommit = 0;
 141 static int      pgStatXactRollback = 0;
 142
 143 static TransactionId pgStatDBHashXact = InvalidTransactionId;
 144 static HTAB *pgStatDBHash = NULL;
 145 static HTAB *pgStatBeDead = NULL;
 146 static PgStat_StatBeEntry *pgStatBeTable = NULL;
 147 static int      pgStatNumBackends = 0;
 148
 149
 150 /* ----------
 151  * Local function forward declarations
 152  * ----------
 153  */
 154 #ifdef EXEC_BACKEND
 155
 156 typedef enum STATS_PROCESS_TYPE
 157 {
 158         STAT_PROC_BUFFER,
 159         STAT_PROC_COLLECTOR
 160 }       STATS_PROCESS_TYPE;
 161
 162 static pid_t pgstat_forkexec(STATS_PROCESS_TYPE procType);
 163 static void pgstat_parseArgs(int argc, char *argv[]);
 164 #endif
 165
 166 NON_EXEC_STATIC void PgstatBufferMain(int argc, char *argv[]);
 167 NON_EXEC_STATIC void PgstatCollectorMain(int argc, char *argv[]);
 168 static void pgstat_recvbuffer(void);
 169 static void pgstat_exit(SIGNAL_ARGS);
 170 static void pgstat_die(SIGNAL_ARGS);
 171 static void pgstat_beshutdown_hook(int code, Datum arg);
 172
 173 static PgStat_StatDBEntry *pgstat_get_db_entry(Oid databaseid, bool create);
 174 static int      pgstat_add_backend(PgStat_MsgHdr *msg);
 175 static void pgstat_sub_backend(int procpid);
 176 static void pgstat_drop_database(Oid databaseid);
 177 static void pgstat_write_statsfile(void);
 178 static void pgstat_read_statsfile(HTAB **dbhash, Oid onlydb,
 179                                           PgStat_StatBeEntry **betab,
 180                                           int *numbackends);
 181 static void backend_read_statsfile(void);
 182
 183 static void pgstat_setheader(PgStat_MsgHdr *hdr, StatMsgType mtype);
 184 static void pgstat_send(void *msg, int len);
 185
 186 static void pgstat_recv_bestart(PgStat_MsgBestart *msg, int len);
 187 static void pgstat_recv_beterm(PgStat_MsgBeterm *msg, int len);
 188 static void pgstat_recv_activity(PgStat_MsgActivity *msg, int len);
 189 static void pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len);
 190 static void pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len);
 191 static void pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len);
 192 static void pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len);
 193 static void pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len);
 194 static void pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len);
 195 static void pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len);
 196
 197
 198 /* ------------------------------------------------------------
 199  * Public functions called from postmaster follow
 200  * ------------------------------------------------------------
 201  */
 202
 203 /* ----------
 204  * pgstat_init() -
 205  *
 206  *      Called from postmaster at startup. Create the resources required
 207  *      by the statistics collector process.  If unable to do so, do not
 208  *      fail --- better to let the postmaster start with stats collection
 209  *      disabled.
 210  * ----------
 211  */
 212 void
 213 pgstat_init(void)
 214 {
 215         ACCEPT_TYPE_ARG3 alen;
 216         struct addrinfo *addrs = NULL,
 217                            *addr,
 218                                 hints;
 219         int                     ret;
 220         fd_set          rset;
 221         struct timeval tv;
 222         char            test_byte;
 223         int                     sel_res;
 224
 225 #define TESTBYTEVAL ((char) 199)
 226
 227         /*
 228          * Force start of collector daemon if something to collect
 229          */
 230         if (pgstat_collect_querystring ||
 231                 pgstat_collect_tuplelevel ||
 232                 pgstat_collect_blocklevel)
 233                 pgstat_collect_startcollector = true;
 234
 235         /*
 236          * If we don't have to start a collector or should reset the collected
 237          * statistics on postmaster start, simply remove the stats file.
 238          */
 239         if (!pgstat_collect_startcollector || pgstat_collect_resetonpmstart)
 240                 pgstat_reset_all();
 241
 242         /*
 243          * Nothing else required if collector will not get started
 244          */
 245         if (!pgstat_collect_startcollector)
 246                 return;
 247
 248         /*
 249          * Create the UDP socket for sending and receiving statistic messages
 250          */
 251         hints.ai_flags = AI_PASSIVE;
 252         hints.ai_family = PF_UNSPEC;
 253         hints.ai_socktype = SOCK_DGRAM;
 254         hints.ai_protocol = 0;
 255         hints.ai_addrlen = 0;
 256         hints.ai_addr = NULL;
 257         hints.ai_canonname = NULL;
 258         hints.ai_next = NULL;
 259         ret = getaddrinfo_all("localhost", NULL, &hints, &addrs);
 260         if (ret || !addrs)
 261         {
 262                 ereport(LOG,
 263                                 (errmsg("could not resolve \"localhost\": %s",
 264                                                 gai_strerror(ret))));
 265                 goto startup_failed;
 266         }
 267
 268         /*
 269          * On some platforms, getaddrinfo_all() may return multiple addresses
 270          * only one of which will actually work (eg, both IPv6 and IPv4
 271          * addresses when kernel will reject IPv6).  Worse, the failure may
 272          * occur at the bind() or perhaps even connect() stage.  So we must
 273          * loop through the results till we find a working combination.  We
 274          * will generate LOG messages, but no error, for bogus combinations.
 275          */
 276         for (addr = addrs; addr; addr = addr->ai_next)
 277         {
 278 #ifdef HAVE_UNIX_SOCKETS
 279                 /* Ignore AF_UNIX sockets, if any are returned. */
 280                 if (addr->ai_family == AF_UNIX)
 281                         continue;
 282 #endif
 283
 284                 /*
 285                  * Create the socket.
 286                  */
 287                 if ((pgStatSock = socket(addr->ai_family, SOCK_DGRAM, 0)) < 0)
 288                 {
 289                         ereport(LOG,
 290                                         (errcode_for_socket_access(),
 291                                          errmsg("could not create socket for statistics collector: %m")));
 292                         continue;
 293                 }
 294
 295                 /*
 296                  * Bind it to a kernel assigned port on localhost and get the
 297                  * assigned port via getsockname().
 298                  */
 299                 if (bind(pgStatSock, addr->ai_addr, addr->ai_addrlen) < 0)
 300                 {
 301                         ereport(LOG,
 302                                         (errcode_for_socket_access(),
 303                                          errmsg("could not bind socket for statistics collector: %m")));
 304                         closesocket(pgStatSock);
 305                         pgStatSock = -1;
 306                         continue;
 307                 }
 308
 309                 alen = sizeof(pgStatAddr);
 310                 if (getsockname(pgStatSock, (struct sockaddr *) & pgStatAddr, &alen) < 0)
 311                 {
 312                         ereport(LOG,
 313                                         (errcode_for_socket_access(),
 314                                          errmsg("could not get address of socket for statistics collector: %m")));
 315                         closesocket(pgStatSock);
 316                         pgStatSock = -1;
 317                         continue;
 318                 }
 319
 320                 /*
 321                  * Connect the socket to its own address.  This saves a few cycles
 322                  * by not having to respecify the target address on every send.
 323                  * This also provides a kernel-level check that only packets from
 324                  * this same address will be received.
 325                  */
 326                 if (connect(pgStatSock, (struct sockaddr *) & pgStatAddr, alen) < 0)
 327                 {
 328                         ereport(LOG,
 329                                         (errcode_for_socket_access(),
 330                                          errmsg("could not connect socket for statistics collector: %m")));
 331                         closesocket(pgStatSock);
 332                         pgStatSock = -1;
 333                         continue;
 334                 }
 335
 336                 /*
 337                  * Try to send and receive a one-byte test message on the socket.
 338                  * This is to catch situations where the socket can be created but
 339                  * will not actually pass data (for instance, because kernel
 340                  * packet filtering rules prevent it).
 341                  */
 342                 test_byte = TESTBYTEVAL;
 343                 if (send(pgStatSock, &test_byte, 1, 0) != 1)
 344                 {
 345                         ereport(LOG,
 346                                         (errcode_for_socket_access(),
 347                                          errmsg("could not send test message on socket for statistics collector: %m")));
 348                         closesocket(pgStatSock);
 349                         pgStatSock = -1;
 350                         continue;
 351                 }
 352
 353                 /*
 354                  * There could possibly be a little delay before the message can
 355                  * be received.  We arbitrarily allow up to half a second before
 356                  * deciding it's broken.
 357                  */
 358                 for (;;)                                /* need a loop to handle EINTR */
 359                 {
 360                         FD_ZERO(&rset);
 361                         FD_SET(pgStatSock, &rset);
 362                         tv.tv_sec = 0;
 363                         tv.tv_usec = 500000;
 364                         sel_res = select(pgStatSock + 1, &rset, NULL, NULL, &tv);
 365                         if (sel_res >= 0 || errno != EINTR)
 366                                 break;
 367                 }
 368                 if (sel_res < 0)
 369                 {
 370                         ereport(LOG,
 371                                         (errcode_for_socket_access(),
 372                                  errmsg("select() failed in statistics collector: %m")));
 373                         closesocket(pgStatSock);
 374                         pgStatSock = -1;
 375                         continue;
 376                 }
 377                 if (sel_res == 0 || !FD_ISSET(pgStatSock, &rset))
 378                 {
 379                         /*
 380                          * This is the case we actually think is likely, so take pains
 381                          * to give a specific message for it.
 382                          *
 383                          * errno will not be set meaningfully here, so don't use it.
 384                          */
 385                         ereport(LOG,
 386                                         (errcode(ERRCODE_CONNECTION_FAILURE),
 387                                          errmsg("test message did not get through on socket for statistics collector")));
 388                         closesocket(pgStatSock);
 389                         pgStatSock = -1;
 390                         continue;
 391                 }
 392
 393                 test_byte++;                    /* just make sure variable is changed */
 394
 395                 if (recv(pgStatSock, &test_byte, 1, 0) != 1)
 396                 {
 397                         ereport(LOG,
 398                                         (errcode_for_socket_access(),
 399                                          errmsg("could not receive test message on socket for statistics collector: %m")));
 400                         closesocket(pgStatSock);
 401                         pgStatSock = -1;
 402                         continue;
 403                 }
 404
 405                 if (test_byte != TESTBYTEVAL)   /* strictly paranoia ... */
 406                 {
 407                         ereport(LOG,
 408                                         (errcode(ERRCODE_INTERNAL_ERROR),
 409                                          errmsg("incorrect test message transmission on socket for statistics collector")));
 410                         closesocket(pgStatSock);
 411                         pgStatSock = -1;
 412                         continue;
 413                 }
 414
 415                 /* If we get here, we have a working socket */
 416                 break;
 417         }
 418
 419         /* Did we find a working address? */
 420         if (!addr || pgStatSock < 0)
 421                 goto startup_failed;
 422
 423         /*
 424          * Set the socket to non-blocking IO.  This ensures that if the
 425          * collector falls behind (despite the buffering process), statistics
 426          * messages will be discarded; backends won't block waiting to send
 427          * messages to the collector.
 428          */
 429         if (!pg_set_noblock(pgStatSock))
 430         {
 431                 ereport(LOG,
 432                                 (errcode_for_socket_access(),
 433                                  errmsg("could not set statistics collector socket to nonblocking mode: %m")));
 434                 goto startup_failed;
 435         }
 436
 437         freeaddrinfo_all(hints.ai_family, addrs);
 438
 439         return;
 440
 441 startup_failed:
 442         ereport(LOG,
 443                         (errmsg("disabling statistics collector for lack of working socket")));
 444
 445         if (addrs)
 446                 freeaddrinfo_all(hints.ai_family, addrs);
 447
 448         if (pgStatSock >= 0)
 449                 closesocket(pgStatSock);
 450         pgStatSock = -1;
 451
 452         /* Adjust GUC variables to suppress useless activity */
 453         pgstat_collect_startcollector = false;
 454         pgstat_collect_querystring = false;
 455         pgstat_collect_tuplelevel = false;
 456         pgstat_collect_blocklevel = false;
 457 }
 458
 459 /*
 460  * pgstat_reset_all() -
 461  *
 462  * Remove the stats file.  This is used on server start if the
 463  * stats_reset_on_server_start feature is enabled, or if WAL
 464  * recovery is needed after a crash.
 465  */
 466 void
 467 pgstat_reset_all(void)
 468 {
 469         unlink(PGSTAT_STAT_FILENAME);
 470 }
 471
 472 #ifdef EXEC_BACKEND
 473
 474 /*
 475  * pgstat_forkexec() -
 476  *
 477  * Format up the arglist for, then fork and exec, statistics
 478  * (buffer and collector) processes
 479  */
 480 static pid_t
 481 pgstat_forkexec(STATS_PROCESS_TYPE procType)
 482 {
 483         char       *av[10];
 484         int                     ac = 0,
 485                                 bufc = 0,
 486                                 i;
 487         char            pgstatBuf[2][32];
 488
 489         av[ac++] = "postgres";
 490
 491         switch (procType)
 492         {
 493                 case STAT_PROC_BUFFER:
 494                         av[ac++] = "-forkbuf";
 495                         break;
 496
 497                 case STAT_PROC_COLLECTOR:
 498                         av[ac++] = "-forkcol";
 499                         break;
 500
 501                 default:
 502                         Assert(false);
 503         }
 504
 505         av[ac++] = NULL;                        /* filled in by postmaster_forkexec */
 506
 507         /* postgres_exec_path is not passed by write_backend_variables */
 508         av[ac++] = postgres_exec_path;
 509
 510         /* Add to the arg list */
 511         Assert(bufc <= lengthof(pgstatBuf));
 512         for (i = 0; i < bufc; i++)
 513                 av[ac++] = pgstatBuf[i];
 514
 515         av[ac] = NULL;
 516         Assert(ac < lengthof(av));
 517
 518         return postmaster_forkexec(ac, av);
 519 }
 520
 521
 522 /*
 523  * pgstat_parseArgs() -
 524  *
 525  * Extract data from the arglist for exec'ed statistics
 526  * (buffer and collector) processes
 527  */
 528 static void
 529 pgstat_parseArgs(int argc, char *argv[])
 530 {
 531         Assert(argc == 4);
 532
 533         argc = 3;
 534         StrNCpy(postgres_exec_path, argv[argc++], MAXPGPATH);
 535 }
 536 #endif   /* EXEC_BACKEND */
 537
 538
 539 /* ----------
 540  * pgstat_start() -
 541  *
 542  *      Called from postmaster at startup or after an existing collector
 543  *      died.  Attempt to fire up a fresh statistics collector.
 544  *
 545  *      Returns PID of child process, or 0 if fail.
 546  *
 547  *      Note: if fail, we will be called again from the postmaster main loop.
 548  * ----------
 549  */
 550 int
 551 pgstat_start(void)
 552 {
 553         time_t          curtime;
 554         pid_t           pgStatPid;
 555
 556         /*
 557          * Do nothing if no collector needed
 558          */
 559         if (!pgstat_collect_startcollector)
 560                 return 0;
 561
 562         /*
 563          * Do nothing if too soon since last collector start.  This is a
 564          * safety valve to protect against continuous respawn attempts if the
 565          * collector is dying immediately at launch.  Note that since we will
 566          * be re-called from the postmaster main loop, we will get another
 567          * chance later.
 568          */
 569         curtime = time(NULL);
 570         if ((unsigned int) (curtime - last_pgstat_start_time) <
 571                 (unsigned int) PGSTAT_RESTART_INTERVAL)
 572                 return 0;
 573         last_pgstat_start_time = curtime;
 574
 575         /*
 576          * Check that the socket is there, else pgstat_init failed.
 577          */
 578         if (pgStatSock < 0)
 579         {
 580                 ereport(LOG,
 581                                 (errmsg("statistics collector startup skipped")));
 582
 583                 /*
 584                  * We can only get here if someone tries to manually turn
 585                  * pgstat_collect_startcollector on after it had been off.
 586                  */
 587                 pgstat_collect_startcollector = false;
 588                 return 0;
 589         }
 590
 591         /*
 592          * Okay, fork off the collector.
 593          */
 594 #ifdef EXEC_BACKEND
 595         switch ((pgStatPid = pgstat_forkexec(STAT_PROC_BUFFER)))
 596 #else
 597         switch ((pgStatPid = fork_process()))
 598 #endif
 599         {
 600                 case -1:
 601                         ereport(LOG,
 602                                         (errmsg("could not fork statistics buffer: %m")));
 603                         return 0;
 604
 605 #ifndef EXEC_BACKEND
 606                 case 0:
 607                         /* in postmaster child ... */
 608                         /* Close the postmaster's sockets */
 609                         ClosePostmasterPorts(false);
 610
 611                         /* Drop our connection to postmaster's shared memory, as well */
 612                         PGSharedMemoryDetach();
 613
 614                         PgstatBufferMain(0, NULL);
 615                         break;
 616 #endif
 617
 618                 default:
 619                         return (int) pgStatPid;
 620         }
 621
 622         /* shouldn't get here */
 623         return 0;
 624 }
 625
 626
 627 /* ----------
 628  * pgstat_beterm() -
 629  *
 630  *      Called from postmaster to tell collector a backend terminated.
 631  * ----------
 632  */
 633 void
 634 pgstat_beterm(int pid)
 635 {
 636         PgStat_MsgBeterm msg;
 637
 638         if (pgStatSock < 0)
 639                 return;
 640
 641         /* can't use pgstat_setheader() because it's not called in a backend */
 642         MemSet(&(msg.m_hdr), 0, sizeof(msg.m_hdr));
 643         msg.m_hdr.m_type = PGSTAT_MTYPE_BETERM;
 644         msg.m_hdr.m_procpid = pid;
 645
 646         pgstat_send(&msg, sizeof(msg));
 647 }
 648
 649
 650 /* ----------
 651  * pgstat_report_autovac() -
 652  *
 653  *      Called from autovacuum.c to report startup of an autovacuum process.
 654  *      We are called before InitPostgres is done, so can't rely on MyDatabaseId;
 655  *      the db OID must be passed in, instead.
 656  * ----------
 657  */
 658 void
 659 pgstat_report_autovac(Oid dboid)
 660 {
 661         PgStat_MsgAutovacStart msg;
 662
 663         if (pgStatSock < 0)
 664                 return;
 665
 666         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_AUTOVAC_START);
 667         msg.m_databaseid = dboid;
 668         msg.m_start_time = GetCurrentTimestamp();
 669
 670         pgstat_send(&msg, sizeof(msg));
 671 }
 672
 673 /* ------------------------------------------------------------
 674  * Public functions used by backends follow
 675  *------------------------------------------------------------
 676  */
 677
 678
 679 /* ----------
 680  * pgstat_bestart() -
 681  *
 682  *      Tell the collector that this new backend is soon ready to process
 683  *      queries. Called from InitPostgres.
 684  * ----------
 685  */
 686 void
 687 pgstat_bestart(void)
 688 {
 689         PgStat_MsgBestart msg;
 690
 691         if (pgStatSock < 0)
 692                 return;
 693
 694         /*
 695          * We may not have a MyProcPort (eg, if this is the autovacuum process).
 696          * For the moment, punt and don't send BESTART --- would be better to
 697          * work out a clean way of handling "unknown clientaddr".
 698          */
 699         if (MyProcPort)
 700         {
 701                 pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_BESTART);
 702                 msg.m_databaseid = MyDatabaseId;
 703                 msg.m_userid = GetSessionUserId();
 704                 memcpy(&msg.m_clientaddr, &MyProcPort->raddr, sizeof(msg.m_clientaddr));
 705                 pgstat_send(&msg, sizeof(msg));
 706         }
 707
 708         /*
 709          * Set up a process-exit hook to ensure we flush the last batch of
 710          * statistics to the collector.
 711          */
 712         on_shmem_exit(pgstat_beshutdown_hook, 0);
 713 }
 714
 715 /* ---------
 716  * pgstat_report_vacuum() -
 717  *
 718  *      Tell the collector about the table we just vacuumed.
 719  * ---------
 720  */
 721 void
 722 pgstat_report_vacuum(Oid tableoid, bool shared,
 723                                          bool analyze, PgStat_Counter tuples)
 724 {
 725         PgStat_MsgVacuum msg;
 726
 727         if (pgStatSock < 0)
 728                 return;
 729
 730         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_VACUUM);
 731         msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
 732         msg.m_tableoid = tableoid;
 733         msg.m_analyze = analyze;
 734         msg.m_tuples = tuples;
 735         pgstat_send(&msg, sizeof(msg));
 736 }
 737
 738 /* --------
 739  * pgstat_report_analyze() -
 740  *
 741  *      Tell the collector about the table we just analyzed.
 742  * --------
 743  */
 744 void
 745 pgstat_report_analyze(Oid tableoid, bool shared, PgStat_Counter livetuples,
 746                                           PgStat_Counter deadtuples)
 747 {
 748         PgStat_MsgAnalyze msg;
 749
 750         if (pgStatSock < 0)
 751                 return;
 752
 753         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_ANALYZE);
 754         msg.m_databaseid = shared ? InvalidOid : MyDatabaseId;
 755         msg.m_tableoid = tableoid;
 756         msg.m_live_tuples = livetuples;
 757         msg.m_dead_tuples = deadtuples;
 758         pgstat_send(&msg, sizeof(msg));
 759 }
 760
 761 /*
 762  * Flush any remaining statistics counts out to the collector at process
 763  * exit.   Without this, operations triggered during backend exit (such as
 764  * temp table deletions) won't be counted.
 765  */
 766 static void
 767 pgstat_beshutdown_hook(int code, Datum arg)
 768 {
 769         pgstat_report_tabstat();
 770 }
 771
 772
 773 /* ----------
 774  * pgstat_report_activity() -
 775  *
 776  *      Called from tcop/postgres.c to tell the collector what the backend
 777  *      is actually doing (usually "<IDLE>" or the start of the query to
 778  *      be executed).
 779  * ----------
 780  */
 781 void
 782 pgstat_report_activity(const char *what)
 783 {
 784         PgStat_MsgActivity msg;
 785         int                     len;
 786
 787         if (!pgstat_collect_querystring || pgStatSock < 0)
 788                 return;
 789
 790         len = strlen(what);
 791         len = pg_mbcliplen(what, len, PGSTAT_ACTIVITY_SIZE - 1);
 792
 793         memcpy(msg.m_what, what, len);
 794         msg.m_what[len] = '\0';
 795         len += offsetof(PgStat_MsgActivity, m_what) +1;
 796
 797         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_ACTIVITY);
 798         pgstat_send(&msg, len);
 799 }
 800
 801
 802 /* ----------
 803  * pgstat_report_tabstat() -
 804  *
 805  *      Called from tcop/postgres.c to send the so far collected
 806  *      per table access statistics to the collector.
 807  * ----------
 808  */
 809 void
 810 pgstat_report_tabstat(void)
 811 {
 812         int                     i;
 813
 814         if (pgStatSock < 0 ||
 815                 !(pgstat_collect_querystring ||
 816                   pgstat_collect_tuplelevel ||
 817                   pgstat_collect_blocklevel))
 818         {
 819                 /* Not reporting stats, so just flush whatever we have */
 820                 RegularTabStat.tsa_used = 0;
 821                 SharedTabStat.tsa_used = 0;
 822                 return;
 823         }
 824
 825         /*
 826          * For each message buffer used during the last query set the header
 827          * fields and send it out.
 828          */
 829         for (i = 0; i < RegularTabStat.tsa_used; i++)
 830         {
 831                 PgStat_MsgTabstat *tsmsg = RegularTabStat.tsa_messages[i];
 832                 int                     n;
 833                 int                     len;
 834
 835                 n = tsmsg->m_nentries;
 836                 len = offsetof(PgStat_MsgTabstat, m_entry[0]) +
 837                         n * sizeof(PgStat_TableEntry);
 838
 839                 tsmsg->m_xact_commit = pgStatXactCommit;
 840                 tsmsg->m_xact_rollback = pgStatXactRollback;
 841                 pgStatXactCommit = 0;
 842                 pgStatXactRollback = 0;
 843
 844                 pgstat_setheader(&tsmsg->m_hdr, PGSTAT_MTYPE_TABSTAT);
 845                 tsmsg->m_databaseid = MyDatabaseId;
 846                 pgstat_send(tsmsg, len);
 847         }
 848         RegularTabStat.tsa_used = 0;
 849
 850         /* Ditto, for shared relations */
 851         for (i = 0; i < SharedTabStat.tsa_used; i++)
 852         {
 853                 PgStat_MsgTabstat *tsmsg = SharedTabStat.tsa_messages[i];
 854                 int                     n;
 855                 int                     len;
 856
 857                 n = tsmsg->m_nentries;
 858                 len = offsetof(PgStat_MsgTabstat, m_entry[0]) +
 859                         n * sizeof(PgStat_TableEntry);
 860
 861                 /* We don't report transaction commit/abort here */
 862                 tsmsg->m_xact_commit = 0;
 863                 tsmsg->m_xact_rollback = 0;
 864
 865                 pgstat_setheader(&tsmsg->m_hdr, PGSTAT_MTYPE_TABSTAT);
 866                 tsmsg->m_databaseid = InvalidOid;
 867                 pgstat_send(tsmsg, len);
 868         }
 869         SharedTabStat.tsa_used = 0;
 870 }
 871
 872
 873 /* ----------
 874  * pgstat_vacuum_tabstat() -
 875  *
 876  *      Will tell the collector about objects he can get rid of.
 877  * ----------
 878  */
 879 int
 880 pgstat_vacuum_tabstat(void)
 881 {
 882         Relation        dbrel;
 883         HeapScanDesc dbscan;
 884         HeapTuple       dbtup;
 885         Oid                *dbidlist;
 886         int                     dbidalloc;
 887         int                     dbidused;
 888         HASH_SEQ_STATUS hstat;
 889         PgStat_StatDBEntry *dbentry;
 890         PgStat_StatTabEntry *tabentry;
 891         HeapTuple       reltup;
 892         int                     nobjects = 0;
 893         PgStat_MsgTabpurge msg;
 894         int                     len;
 895         int                     i;
 896
 897         if (pgStatSock < 0)
 898                 return 0;
 899
 900         /*
 901          * If not done for this transaction, read the statistics collector
 902          * stats file into some hash tables.
 903          */
 904         backend_read_statsfile();
 905
 906         /*
 907          * Lookup our own database entry; if not found, nothing to do.
 908          */
 909         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
 910                                                                                                  (void *) &MyDatabaseId,
 911                                                                                                  HASH_FIND, NULL);
 912         if (dbentry == NULL)
 913                 return -1;
 914         if (dbentry->tables == NULL)
 915                 return 0;
 916
 917         /*
 918          * Initialize our messages table counter to zero
 919          */
 920         msg.m_nentries = 0;
 921
 922         /*
 923          * Check for all tables listed in stats hashtable if they still exist.
 924          */
 925         hash_seq_init(&hstat, dbentry->tables);
 926         while ((tabentry = (PgStat_StatTabEntry *) hash_seq_search(&hstat)) != NULL)
 927         {
 928                 /*
 929                  * Check if this relation is still alive by looking up it's
 930                  * pg_class tuple in the system catalog cache.
 931                  */
 932                 reltup = SearchSysCache(RELOID,
 933                                                                 ObjectIdGetDatum(tabentry->tableid),
 934                                                                 0, 0, 0);
 935                 if (HeapTupleIsValid(reltup))
 936                 {
 937                         ReleaseSysCache(reltup);
 938                         continue;
 939                 }
 940
 941                 /*
 942                  * Add this table's Oid to the message
 943                  */
 944                 msg.m_tableid[msg.m_nentries++] = tabentry->tableid;
 945                 nobjects++;
 946
 947                 /*
 948                  * If the message is full, send it out and reinitialize to zero
 949                  */
 950                 if (msg.m_nentries >= PGSTAT_NUM_TABPURGE)
 951                 {
 952                         len = offsetof(PgStat_MsgTabpurge, m_tableid[0])
 953                                 +msg.m_nentries * sizeof(Oid);
 954
 955                         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
 956                         msg.m_databaseid = MyDatabaseId;
 957                         pgstat_send(&msg, len);
 958
 959                         msg.m_nentries = 0;
 960                 }
 961         }
 962
 963         /*
 964          * Send the rest
 965          */
 966         if (msg.m_nentries > 0)
 967         {
 968                 len = offsetof(PgStat_MsgTabpurge, m_tableid[0])
 969                         +msg.m_nentries * sizeof(Oid);
 970
 971                 pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_TABPURGE);
 972                 msg.m_databaseid = MyDatabaseId;
 973                 pgstat_send(&msg, len);
 974         }
 975
 976         /*
 977          * Read pg_database and remember the Oid's of all existing databases
 978          */
 979         dbidalloc = 256;
 980         dbidused = 0;
 981         dbidlist = (Oid *) palloc(sizeof(Oid) * dbidalloc);
 982
 983         dbrel = heap_open(DatabaseRelationId, AccessShareLock);
 984         dbscan = heap_beginscan(dbrel, SnapshotNow, 0, NULL);
 985         while ((dbtup = heap_getnext(dbscan, ForwardScanDirection)) != NULL)
 986         {
 987                 if (dbidused >= dbidalloc)
 988                 {
 989                         dbidalloc *= 2;
 990                         dbidlist = (Oid *) repalloc((char *) dbidlist,
 991                                                                                 sizeof(Oid) * dbidalloc);
 992                 }
 993                 dbidlist[dbidused++] = HeapTupleGetOid(dbtup);
 994         }
 995         heap_endscan(dbscan);
 996         heap_close(dbrel, AccessShareLock);
 997
 998         /*
 999          * Search the database hash table for dead databases and tell the
1000          * collector to drop them as well.
1001          */
1002         hash_seq_init(&hstat, pgStatDBHash);
1003         while ((dbentry = (PgStat_StatDBEntry *) hash_seq_search(&hstat)) != NULL)
1004         {
1005                 Oid                     dbid = dbentry->databaseid;
1006
1007                 for (i = 0; i < dbidused; i++)
1008                 {
1009                         if (dbidlist[i] == dbid)
1010                         {
1011                                 dbid = InvalidOid;
1012                                 break;
1013                         }
1014                 }
1015
1016                 if (dbid != InvalidOid)
1017                 {
1018                         pgstat_drop_database(dbid);
1019                         nobjects++;
1020                 }
1021         }
1022
1023         /*
1024          * Free the dbid list.
1025          */
1026         pfree(dbidlist);
1027
1028         /*
1029          * Tell the caller how many removeable objects we found
1030          */
1031         return nobjects;
1032 }
1033
1034
1035 /* ----------
1036  * pgstat_drop_database() -
1037  *
1038  *      Tell the collector that we just dropped a database.
1039  *      This is the only message that shouldn't get lost in space. Otherwise
1040  *      the collector will keep the statistics for the dead DB until his
1041  *      stats file got removed while the postmaster is down.
1042  * ----------
1043  */
1044 static void
1045 pgstat_drop_database(Oid databaseid)
1046 {
1047         PgStat_MsgDropdb msg;
1048
1049         if (pgStatSock < 0)
1050                 return;
1051
1052         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_DROPDB);
1053         msg.m_databaseid = databaseid;
1054         pgstat_send(&msg, sizeof(msg));
1055 }
1056
1057
1058 /* ----------
1059  * pgstat_reset_counters() -
1060  *
1061  *      Tell the statistics collector to reset counters for our database.
1062  * ----------
1063  */
1064 void
1065 pgstat_reset_counters(void)
1066 {
1067         PgStat_MsgResetcounter msg;
1068
1069         if (pgStatSock < 0)
1070                 return;
1071
1072         if (!superuser())
1073                 ereport(ERROR,
1074                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1075                           errmsg("must be superuser to reset statistics counters")));
1076
1077         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RESETCOUNTER);
1078         msg.m_databaseid = MyDatabaseId;
1079         pgstat_send(&msg, sizeof(msg));
1080 }
1081
1082
1083 /* ----------
1084  * pgstat_ping() -
1085  *
1086  *      Send some junk data to the collector to increase traffic.
1087  * ----------
1088  */
1089 void
1090 pgstat_ping(void)
1091 {
1092         PgStat_MsgDummy msg;
1093
1094         if (pgStatSock < 0)
1095                 return;
1096
1097         pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_DUMMY);
1098         pgstat_send(&msg, sizeof(msg));
1099 }
1100
1101 /*
1102  * Enlarge a TabStatArray
1103  */
1104 static void
1105 more_tabstat_space(TabStatArray *tsarr)
1106 {
1107         PgStat_MsgTabstat *newMessages;
1108         PgStat_MsgTabstat **msgArray;
1109         int                     newAlloc;
1110         int                     i;
1111
1112         AssertArg(PointerIsValid(tsarr));
1113
1114         newAlloc = tsarr->tsa_alloc + TABSTAT_QUANTUM;
1115
1116         /* Create (another) quantum of message buffers */
1117         newMessages = (PgStat_MsgTabstat *)
1118                 MemoryContextAllocZero(TopMemoryContext,
1119                                                            sizeof(PgStat_MsgTabstat) * TABSTAT_QUANTUM);
1120
1121         /* Create or enlarge the pointer array */
1122         if (tsarr->tsa_messages == NULL)
1123                 msgArray = (PgStat_MsgTabstat **)
1124                         MemoryContextAlloc(TopMemoryContext,
1125                                                            sizeof(PgStat_MsgTabstat *) * newAlloc);
1126         else
1127                 msgArray = (PgStat_MsgTabstat **)
1128                         repalloc(tsarr->tsa_messages,
1129                                          sizeof(PgStat_MsgTabstat *) * newAlloc);
1130
1131         for (i = 0; i < TABSTAT_QUANTUM; i++)
1132                 msgArray[tsarr->tsa_alloc + i] = newMessages++;
1133         tsarr->tsa_messages = msgArray;
1134         tsarr->tsa_alloc = newAlloc;
1135
1136         Assert(tsarr->tsa_used < tsarr->tsa_alloc);
1137 }
1138
1139 /* ----------
1140  * pgstat_initstats() -
1141  *
1142  *      Called from various places usually dealing with initialization
1143  *      of Relation or Scan structures. The data placed into these
1144  *      structures from here tell where later to count for buffer reads,
1145  *      scans and tuples fetched.
1146  * ----------
1147  */
1148 void
1149 pgstat_initstats(PgStat_Info *stats, Relation rel)
1150 {
1151         Oid                     rel_id = rel->rd_id;
1152         PgStat_TableEntry *useent;
1153         TabStatArray    *tsarr;
1154         PgStat_MsgTabstat *tsmsg;
1155         int                     mb;
1156         int                     i;
1157
1158         /*
1159          * Initialize data not to count at all.
1160          */
1161         stats->tabentry = NULL;
1162         stats->no_stats = FALSE;
1163         stats->heap_scan_counted = FALSE;
1164         stats->index_scan_counted = FALSE;
1165
1166         if (pgStatSock < 0 ||
1167                 !(pgstat_collect_tuplelevel ||
1168                   pgstat_collect_blocklevel))
1169         {
1170                 stats->no_stats = TRUE;
1171                 return;
1172         }
1173
1174         tsarr = rel->rd_rel->relisshared ? &SharedTabStat : &RegularTabStat;
1175
1176         /*
1177          * Search the already-used message slots for this relation.
1178          */
1179         for (mb = 0; mb < tsarr->tsa_used; mb++)
1180         {
1181                 tsmsg = tsarr->tsa_messages[mb];
1182
1183                 for (i = tsmsg->m_nentries; --i >= 0;)
1184                 {
1185                         if (tsmsg->m_entry[i].t_id == rel_id)
1186                         {
1187                                 stats->tabentry = (void *) &(tsmsg->m_entry[i]);
1188                                 return;
1189                         }
1190                 }
1191
1192                 if (tsmsg->m_nentries >= PGSTAT_NUM_TABENTRIES)
1193                         continue;
1194
1195                 /*
1196                  * Not found, but found a message buffer with an empty slot
1197                  * instead. Fine, let's use this one.
1198                  */
1199                 i = tsmsg->m_nentries++;
1200                 useent = &tsmsg->m_entry[i];
1201                 MemSet(useent, 0, sizeof(PgStat_TableEntry));
1202                 useent->t_id = rel_id;
1203                 stats->tabentry = (void *) useent;
1204                 return;
1205         }
1206
1207         /*
1208          * If we ran out of message buffers, we just allocate more.
1209          */
1210         if (tsarr->tsa_used >= tsarr->tsa_alloc)
1211                 more_tabstat_space(tsarr);
1212
1213         /*
1214          * Use the first entry of the next message buffer.
1215          */
1216         mb = tsarr->tsa_used++;
1217         tsmsg = tsarr->tsa_messages[mb];
1218         tsmsg->m_nentries = 1;
1219         useent = &tsmsg->m_entry[0];
1220         MemSet(useent, 0, sizeof(PgStat_TableEntry));
1221         useent->t_id = rel_id;
1222         stats->tabentry = (void *) useent;
1223 }
1224
1225
1226 /* ----------
1227  * pgstat_count_xact_commit() -
1228  *
1229  *      Called from access/transam/xact.c to count transaction commits.
1230  * ----------
1231  */
1232 void
1233 pgstat_count_xact_commit(void)
1234 {
1235         if (!(pgstat_collect_querystring ||
1236                   pgstat_collect_tuplelevel ||
1237                   pgstat_collect_blocklevel))
1238                 return;
1239
1240         pgStatXactCommit++;
1241
1242         /*
1243          * If there was no relation activity yet, just make one existing
1244          * message buffer used without slots, causing the next report to tell
1245          * new xact-counters.
1246          */
1247         if (RegularTabStat.tsa_alloc == 0)
1248                 more_tabstat_space(&RegularTabStat);
1249
1250         if (RegularTabStat.tsa_used == 0)
1251         {
1252                 RegularTabStat.tsa_used++;
1253                 RegularTabStat.tsa_messages[0]->m_nentries = 0;
1254         }
1255 }
1256
1257
1258 /* ----------
1259  * pgstat_count_xact_rollback() -
1260  *
1261  *      Called from access/transam/xact.c to count transaction rollbacks.
1262  * ----------
1263  */
1264 void
1265 pgstat_count_xact_rollback(void)
1266 {
1267         if (!(pgstat_collect_querystring ||
1268                   pgstat_collect_tuplelevel ||
1269                   pgstat_collect_blocklevel))
1270                 return;
1271
1272         pgStatXactRollback++;
1273
1274         /*
1275          * If there was no relation activity yet, just make one existing
1276          * message buffer used without slots, causing the next report to tell
1277          * new xact-counters.
1278          */
1279         if (RegularTabStat.tsa_alloc == 0)
1280                 more_tabstat_space(&RegularTabStat);
1281
1282         if (RegularTabStat.tsa_used == 0)
1283         {
1284                 RegularTabStat.tsa_used++;
1285                 RegularTabStat.tsa_messages[0]->m_nentries = 0;
1286         }
1287 }
1288
1289
1290 /* ----------
1291  * pgstat_fetch_stat_dbentry() -
1292  *
1293  *      Support function for the SQL-callable pgstat* functions. Returns
1294  *      the collected statistics for one database or NULL. NULL doesn't mean
1295  *      that the database doesn't exist, it is just not yet known by the
1296  *      collector, so the caller is better off to report ZERO instead.
1297  * ----------
1298  */
1299 PgStat_StatDBEntry *
1300 pgstat_fetch_stat_dbentry(Oid dbid)
1301 {
1302         /*
1303          * If not done for this transaction, read the statistics collector
1304          * stats file into some hash tables.
1305          */
1306         backend_read_statsfile();
1307
1308         /*
1309          * Lookup the requested database; return NULL if not found
1310          */
1311         return (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
1312                                                                                           (void *) &dbid,
1313                                                                                           HASH_FIND, NULL);
1314 }
1315
1316
1317 /* ----------
1318  * pgstat_fetch_stat_tabentry() -
1319  *
1320  *      Support function for the SQL-callable pgstat* functions. Returns
1321  *      the collected statistics for one table or NULL. NULL doesn't mean
1322  *      that the table doesn't exist, it is just not yet known by the
1323  *      collector, so the caller is better off to report ZERO instead.
1324  * ----------
1325  */
1326 PgStat_StatTabEntry *
1327 pgstat_fetch_stat_tabentry(Oid relid)
1328 {
1329         Oid                     dbid;
1330         PgStat_StatDBEntry *dbentry;
1331         PgStat_StatTabEntry *tabentry;
1332
1333         /*
1334          * If not done for this transaction, read the statistics collector
1335          * stats file into some hash tables.
1336          */
1337         backend_read_statsfile();
1338
1339         /*
1340          * Lookup our database, then look in its table hash table.
1341          */
1342         dbid = MyDatabaseId;
1343         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
1344                                                                                                  (void *) &dbid,
1345                                                                                                  HASH_FIND, NULL);
1346         if (dbentry != NULL && dbentry->tables != NULL)
1347         {
1348                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
1349                                                                                                            (void *) &relid,
1350                                                                                                            HASH_FIND, NULL);
1351                 if (tabentry)
1352                         return tabentry;
1353         }
1354
1355         /*
1356          * If we didn't find it, maybe it's a shared table.
1357          */
1358         dbid = InvalidOid;
1359         dbentry = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
1360                                                                                                  (void *) &dbid,
1361                                                                                                  HASH_FIND, NULL);
1362         if (dbentry != NULL && dbentry->tables != NULL)
1363         {
1364                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
1365                                                                                                            (void *) &relid,
1366                                                                                                            HASH_FIND, NULL);
1367                 if (tabentry)
1368                         return tabentry;
1369         }
1370
1371         return NULL;
1372 }
1373
1374
1375 /* ----------
1376  * pgstat_fetch_stat_beentry() -
1377  *
1378  *      Support function for the SQL-callable pgstat* functions. Returns
1379  *      the actual activity slot of one active backend. The caller is
1380  *      responsible for a check if the actual user is permitted to see
1381  *      that info (especially the querystring).
1382  * ----------
1383  */
1384 PgStat_StatBeEntry *
1385 pgstat_fetch_stat_beentry(int beid)
1386 {
1387         backend_read_statsfile();
1388
1389         if (beid < 1 || beid > pgStatNumBackends)
1390                 return NULL;
1391
1392         return &pgStatBeTable[beid - 1];
1393 }
1394
1395
1396 /* ----------
1397  * pgstat_fetch_stat_numbackends() -
1398  *
1399  *      Support function for the SQL-callable pgstat* functions. Returns
1400  *      the maximum current backend id.
1401  * ----------
1402  */
1403 int
1404 pgstat_fetch_stat_numbackends(void)
1405 {
1406         backend_read_statsfile();
1407
1408         return pgStatNumBackends;
1409 }
1410
1411
1412
1413 /* ------------------------------------------------------------
1414  * Local support functions follow
1415  * ------------------------------------------------------------
1416  */
1417
1418
1419 /* ----------
1420  * pgstat_setheader() -
1421  *
1422  *              Set common header fields in a statistics message
1423  * ----------
1424  */
1425 static void
1426 pgstat_setheader(PgStat_MsgHdr *hdr, StatMsgType mtype)
1427 {
1428         hdr->m_type = mtype;
1429         hdr->m_backendid = MyBackendId;
1430         hdr->m_procpid = MyProcPid;
1431 }
1432
1433
1434 /* ----------
1435  * pgstat_send() -
1436  *
1437  *              Send out one statistics message to the collector
1438  * ----------
1439  */
1440 static void
1441 pgstat_send(void *msg, int len)
1442 {
1443         if (pgStatSock < 0)
1444                 return;
1445
1446         ((PgStat_MsgHdr *) msg)->m_size = len;
1447
1448 #ifdef USE_ASSERT_CHECKING
1449         if (send(pgStatSock, msg, len, 0) < 0)
1450                 elog(LOG, "could not send to statistics collector: %m");
1451 #else
1452         send(pgStatSock, msg, len, 0);
1453         /* We deliberately ignore any error from send() */
1454 #endif
1455 }
1456
1457
1458 /* ----------
1459  * PgstatBufferMain() -
1460  *
1461  *      Start up the statistics buffer process.  This is the body of the
1462  *      postmaster child process.
1463  *
1464  *      The argc/argv parameters are valid only in EXEC_BACKEND case.
1465  * ----------
1466  */
1467 NON_EXEC_STATIC void
1468 PgstatBufferMain(int argc, char *argv[])
1469 {
1470         IsUnderPostmaster = true;       /* we are a postmaster subprocess now */
1471
1472         MyProcPid = getpid();           /* reset MyProcPid */
1473
1474         /* Lose the postmaster's on-exit routines */
1475         on_exit_reset();
1476
1477         /*
1478          * Ignore all signals usually bound to some action in the postmaster,
1479          * except for SIGCHLD and SIGQUIT --- see pgstat_recvbuffer.
1480          */
1481         pqsignal(SIGHUP, SIG_IGN);
1482         pqsignal(SIGINT, SIG_IGN);
1483         pqsignal(SIGTERM, SIG_IGN);
1484         pqsignal(SIGQUIT, pgstat_exit);
1485         pqsignal(SIGALRM, SIG_IGN);
1486         pqsignal(SIGPIPE, SIG_IGN);
1487         pqsignal(SIGUSR1, SIG_IGN);
1488         pqsignal(SIGUSR2, SIG_IGN);
1489         pqsignal(SIGCHLD, pgstat_die);
1490         pqsignal(SIGTTIN, SIG_DFL);
1491         pqsignal(SIGTTOU, SIG_DFL);
1492         pqsignal(SIGCONT, SIG_DFL);
1493         pqsignal(SIGWINCH, SIG_DFL);
1494         /* unblock will happen in pgstat_recvbuffer */
1495
1496 #ifdef EXEC_BACKEND
1497         pgstat_parseArgs(argc, argv);
1498 #endif
1499
1500         /*
1501          * Start a buffering process to read from the socket, so we have a
1502          * little more time to process incoming messages.
1503          *
1504          * NOTE: the process structure is: postmaster is parent of buffer process
1505          * is parent of collector process.      This way, the buffer can detect
1506          * collector failure via SIGCHLD, whereas otherwise it wouldn't notice
1507          * collector failure until it tried to write on the pipe.  That would
1508          * mean that after the postmaster started a new collector, we'd have
1509          * two buffer processes competing to read from the UDP socket --- not
1510          * good.
1511          */
1512         if (pgpipe(pgStatPipe) < 0)
1513                 ereport(ERROR,
1514                                 (errcode_for_socket_access(),
1515                          errmsg("could not create pipe for statistics buffer: %m")));
1516
1517         /* child becomes collector process */
1518 #ifdef EXEC_BACKEND
1519         pgStatCollectorPid = pgstat_forkexec(STAT_PROC_COLLECTOR);
1520 #else
1521         pgStatCollectorPid = fork();
1522 #endif
1523         switch (pgStatCollectorPid)
1524         {
1525                 case -1:
1526                         ereport(ERROR,
1527                                         (errmsg("could not fork statistics collector: %m")));
1528
1529 #ifndef EXEC_BACKEND
1530                 case 0:
1531                         /* child becomes collector process */
1532                         PgstatCollectorMain(0, NULL);
1533                         break;
1534 #endif
1535
1536                 default:
1537                         /* parent becomes buffer process */
1538                         closesocket(pgStatPipe[0]);
1539                         pgstat_recvbuffer();
1540         }
1541         exit(0);
1542 }
1543
1544
1545 /* ----------
1546  * PgstatCollectorMain() -
1547  *
1548  *      Start up the statistics collector itself.  This is the body of the
1549  *      postmaster grandchild process.
1550  *
1551  *      The argc/argv parameters are valid only in EXEC_BACKEND case.
1552  * ----------
1553  */
1554 NON_EXEC_STATIC void
1555 PgstatCollectorMain(int argc, char *argv[])
1556 {
1557         PgStat_Msg      msg;
1558         fd_set          rfds;
1559         int                     readPipe;
1560         int                     nready;
1561         int                     len = 0;
1562         struct timeval timeout;
1563         struct timeval next_statwrite;
1564         bool            need_statwrite;
1565         HASHCTL         hash_ctl;
1566
1567         MyProcPid = getpid();           /* reset MyProcPid */
1568
1569         /*
1570          * Reset signal handling.  With the exception of restoring default
1571          * SIGCHLD and SIGQUIT handling, this is a no-op in the
1572          * non-EXEC_BACKEND case because we'll have inherited these settings
1573          * from the buffer process; but it's not a no-op for EXEC_BACKEND.
1574          */
1575         pqsignal(SIGHUP, SIG_IGN);
1576         pqsignal(SIGINT, SIG_IGN);
1577         pqsignal(SIGTERM, SIG_IGN);
1578 #ifndef WIN32
1579         pqsignal(SIGQUIT, SIG_IGN);
1580 #else
1581         /* kluge to allow buffer process to kill collector; FIXME */
1582         pqsignal(SIGQUIT, pgstat_exit);
1583 #endif
1584         pqsignal(SIGALRM, SIG_IGN);
1585         pqsignal(SIGPIPE, SIG_IGN);
1586         pqsignal(SIGUSR1, SIG_IGN);
1587         pqsignal(SIGUSR2, SIG_IGN);
1588         pqsignal(SIGCHLD, SIG_DFL);
1589         pqsignal(SIGTTIN, SIG_DFL);
1590         pqsignal(SIGTTOU, SIG_DFL);
1591         pqsignal(SIGCONT, SIG_DFL);
1592         pqsignal(SIGWINCH, SIG_DFL);
1593         PG_SETMASK(&UnBlockSig);
1594
1595 #ifdef EXEC_BACKEND
1596         pgstat_parseArgs(argc, argv);
1597 #endif
1598
1599         /* Close unwanted files */
1600         closesocket(pgStatPipe[1]);
1601         closesocket(pgStatSock);
1602
1603         /*
1604          * Identify myself via ps
1605          */
1606         init_ps_display("stats collector process", "", "");
1607         set_ps_display("");
1608
1609         /*
1610          * Arrange to write the initial status file right away
1611          */
1612         gettimeofday(&next_statwrite, NULL);
1613         need_statwrite = TRUE;
1614
1615         /*
1616          * Read in an existing statistics stats file or initialize the stats
1617          * to zero.
1618          */
1619         pgStatRunningInCollector = TRUE;
1620         pgstat_read_statsfile(&pgStatDBHash, InvalidOid, NULL, NULL);
1621
1622         /*
1623          * Create the dead backend hashtable
1624          */
1625         memset(&hash_ctl, 0, sizeof(hash_ctl));
1626         hash_ctl.keysize = sizeof(int);
1627         hash_ctl.entrysize = sizeof(PgStat_StatBeDead);
1628         hash_ctl.hash = tag_hash;
1629         pgStatBeDead = hash_create("Dead Backends", PGSTAT_BE_HASH_SIZE,
1630                                                            &hash_ctl, HASH_ELEM | HASH_FUNCTION);
1631
1632         /*
1633          * Create the known backends table
1634          */
1635         pgStatBeTable = (PgStat_StatBeEntry *)
1636                 palloc0(sizeof(PgStat_StatBeEntry) * MaxBackends);
1637
1638         readPipe = pgStatPipe[0];
1639
1640         /*
1641          * Process incoming messages and handle all the reporting stuff until
1642          * there are no more messages.
1643          */
1644         for (;;)
1645         {
1646                 /*
1647                  * If we need to write the status file again (there have been
1648                  * changes in the statistics since we wrote it last) calculate the
1649                  * timeout until we have to do so.
1650                  */
1651                 if (need_statwrite)
1652                 {
1653                         struct timeval now;
1654
1655                         gettimeofday(&now, NULL);
1656                         /* avoid assuming that tv_sec is signed */
1657                         if (now.tv_sec > next_statwrite.tv_sec ||
1658                                 (now.tv_sec == next_statwrite.tv_sec &&
1659                                  now.tv_usec >= next_statwrite.tv_usec))
1660                         {
1661                                 timeout.tv_sec = 0;
1662                                 timeout.tv_usec = 0;
1663                         }
1664                         else
1665                         {
1666                                 timeout.tv_sec = next_statwrite.tv_sec - now.tv_sec;
1667                                 timeout.tv_usec = next_statwrite.tv_usec - now.tv_usec;
1668                                 if (timeout.tv_usec < 0)
1669                                 {
1670                                         timeout.tv_sec--;
1671                                         timeout.tv_usec += 1000000;
1672                                 }
1673                         }
1674                 }
1675
1676                 /*
1677                  * Setup the descriptor set for select(2)
1678                  */
1679                 FD_ZERO(&rfds);
1680                 FD_SET(readPipe, &rfds);
1681
1682                 /*
1683                  * Now wait for something to do.
1684                  */
1685                 nready = select(readPipe + 1, &rfds, NULL, NULL,
1686                                                 (need_statwrite) ? &timeout : NULL);
1687                 if (nready < 0)
1688                 {
1689                         if (errno == EINTR)
1690                                 continue;
1691                         ereport(ERROR,
1692                                         (errcode_for_socket_access(),
1693                                  errmsg("select() failed in statistics collector: %m")));
1694                 }
1695
1696                 /*
1697                  * If there are no descriptors ready, our timeout for writing the
1698                  * stats file happened.
1699                  */
1700                 if (nready == 0)
1701                 {
1702                         pgstat_write_statsfile();
1703                         need_statwrite = FALSE;
1704
1705                         continue;
1706                 }
1707
1708                 /*
1709                  * Check if there is a new statistics message to collect.
1710                  */
1711                 if (FD_ISSET(readPipe, &rfds))
1712                 {
1713                         /*
1714                          * We may need to issue multiple read calls in case the buffer
1715                          * process didn't write the message in a single write, which
1716                          * is possible since it dumps its buffer bytewise. In any
1717                          * case, we'd need two reads since we don't know the message
1718                          * length initially.
1719                          */
1720                         int                     nread = 0;
1721                         int                     targetlen = sizeof(PgStat_MsgHdr);              /* initial */
1722                         bool            pipeEOF = false;
1723
1724                         while (nread < targetlen)
1725                         {
1726                                 len = piperead(readPipe, ((char *) &msg) + nread,
1727                                                            targetlen - nread);
1728                                 if (len < 0)
1729                                 {
1730                                         if (errno == EINTR)
1731                                                 continue;
1732                                         ereport(ERROR,
1733                                                         (errcode_for_socket_access(),
1734                                                          errmsg("could not read from statistics collector pipe: %m")));
1735                                 }
1736                                 if (len == 0)   /* EOF on the pipe! */
1737                                 {
1738                                         pipeEOF = true;
1739                                         break;
1740                                 }
1741                                 nread += len;
1742                                 if (nread == sizeof(PgStat_MsgHdr))
1743                                 {
1744                                         /* we have the header, compute actual msg length */
1745                                         targetlen = msg.msg_hdr.m_size;
1746                                         if (targetlen < (int) sizeof(PgStat_MsgHdr) ||
1747                                                 targetlen > (int) sizeof(msg))
1748                                         {
1749                                                 /*
1750                                                  * Bogus message length implies that we got out of
1751                                                  * sync with the buffer process somehow. Abort so
1752                                                  * that we can restart both processes.
1753                                                  */
1754                                                 ereport(ERROR,
1755                                                   (errmsg("invalid statistics message length")));
1756                                         }
1757                                 }
1758                         }
1759
1760                         /*
1761                          * EOF on the pipe implies that the buffer process exited.
1762                          * Fall out of outer loop.
1763                          */
1764                         if (pipeEOF)
1765                                 break;
1766
1767                         /*
1768                          * Distribute the message to the specific function handling
1769                          * it.
1770                          */
1771                         switch (msg.msg_hdr.m_type)
1772                         {
1773                                 case PGSTAT_MTYPE_DUMMY:
1774                                         break;
1775
1776                                 case PGSTAT_MTYPE_BESTART:
1777                                         pgstat_recv_bestart((PgStat_MsgBestart *) &msg, nread);
1778                                         break;
1779
1780                                 case PGSTAT_MTYPE_BETERM:
1781                                         pgstat_recv_beterm((PgStat_MsgBeterm *) &msg, nread);
1782                                         break;
1783
1784                                 case PGSTAT_MTYPE_TABSTAT:
1785                                         pgstat_recv_tabstat((PgStat_MsgTabstat *) &msg, nread);
1786                                         break;
1787
1788                                 case PGSTAT_MTYPE_TABPURGE:
1789                                         pgstat_recv_tabpurge((PgStat_MsgTabpurge *) &msg, nread);
1790                                         break;
1791
1792                                 case PGSTAT_MTYPE_ACTIVITY:
1793                                         pgstat_recv_activity((PgStat_MsgActivity *) &msg, nread);
1794                                         break;
1795
1796                                 case PGSTAT_MTYPE_DROPDB:
1797                                         pgstat_recv_dropdb((PgStat_MsgDropdb *) &msg, nread);
1798                                         break;
1799
1800                                 case PGSTAT_MTYPE_RESETCOUNTER:
1801                                         pgstat_recv_resetcounter((PgStat_MsgResetcounter *) &msg,
1802                                                                                          nread);
1803                                         break;
1804
1805                                 case PGSTAT_MTYPE_AUTOVAC_START:
1806                                         pgstat_recv_autovac((PgStat_MsgAutovacStart *) &msg, nread);
1807                                         break;
1808
1809                                 case PGSTAT_MTYPE_VACUUM:
1810                                         pgstat_recv_vacuum((PgStat_MsgVacuum *) &msg, nread);
1811                                         break;
1812
1813                                 case PGSTAT_MTYPE_ANALYZE:
1814                                         pgstat_recv_analyze((PgStat_MsgAnalyze *) &msg, nread);
1815                                         break;
1816
1817                                 default:
1818                                         break;
1819                         }
1820
1821                         /*
1822                          * Globally count messages.
1823                          */
1824                         pgStatNumMessages++;
1825
1826                         /*
1827                          * If this is the first message after we wrote the stats file
1828                          * the last time, setup the timeout that it'd be written.
1829                          */
1830                         if (!need_statwrite)
1831                         {
1832                                 gettimeofday(&next_statwrite, NULL);
1833                                 next_statwrite.tv_usec += ((PGSTAT_STAT_INTERVAL) * 1000);
1834                                 next_statwrite.tv_sec += (next_statwrite.tv_usec / 1000000);
1835                                 next_statwrite.tv_usec %= 1000000;
1836                                 need_statwrite = TRUE;
1837                         }
1838                 }
1839
1840                 /*
1841                  * Note that we do NOT check for postmaster exit inside the loop;
1842                  * only EOF on the buffer pipe causes us to fall out.  This
1843                  * ensures we don't exit prematurely if there are still a few
1844                  * messages in the buffer or pipe at postmaster shutdown.
1845                  */
1846         }
1847
1848         /*
1849          * Okay, we saw EOF on the buffer pipe, so there are no more messages
1850          * to process.  If the buffer process quit because of postmaster
1851          * shutdown, we want to save the final stats to reuse at next startup.
1852          * But if the buffer process failed, it seems best not to (there may
1853          * even now be a new collector firing up, and we don't want it to read
1854          * a partially-rewritten stats file).
1855          */
1856         if (!PostmasterIsAlive(false))
1857                 pgstat_write_statsfile();
1858 }
1859
1860
1861 /* ----------
1862  * pgstat_recvbuffer() -
1863  *
1864  *      This is the body of the separate buffering process. Its only
1865  *      purpose is to receive messages from the UDP socket as fast as
1866  *      possible and forward them over a pipe into the collector itself.
1867  *      If the collector is slow to absorb messages, they are buffered here.
1868  * ----------
1869  */
1870 static void
1871 pgstat_recvbuffer(void)
1872 {
1873         fd_set          rfds;
1874         fd_set          wfds;
1875         struct timeval timeout;
1876         int                     writePipe = pgStatPipe[1];
1877         int                     maxfd;
1878         int                     nready;
1879         int                     len;
1880         int                     xfr;
1881         int                     frm;
1882         PgStat_Msg      input_buffer;
1883         char       *msgbuffer;
1884         int                     msg_send = 0;   /* next send index in buffer */
1885         int                     msg_recv = 0;   /* next receive index */
1886         int                     msg_have = 0;   /* number of bytes stored */
1887         bool            overflow = false;
1888
1889         /*
1890          * Identify myself via ps
1891          */
1892         init_ps_display("stats buffer process", "", "");
1893         set_ps_display("");
1894
1895         /*
1896          * We want to die if our child collector process does.  There are two
1897          * ways we might notice that it has died: receive SIGCHLD, or get a
1898          * write failure on the pipe leading to the child.      We can set SIGPIPE
1899          * to kill us here.  Our SIGCHLD handler was already set up before we
1900          * forked (must do it that way, else it's a race condition).
1901          */
1902         pqsignal(SIGPIPE, SIG_DFL);
1903         PG_SETMASK(&UnBlockSig);
1904
1905         /*
1906          * Set the write pipe to nonblock mode, so that we cannot block when
1907          * the collector falls behind.
1908          */
1909         if (!pg_set_noblock(writePipe))
1910                 ereport(ERROR,
1911                                 (errcode_for_socket_access(),
1912                                  errmsg("could not set statistics collector pipe to nonblocking mode: %m")));
1913
1914         /*
1915          * Allocate the message buffer
1916          */
1917         msgbuffer = (char *) palloc(PGSTAT_RECVBUFFERSZ);
1918
1919         /*
1920          * Loop forever
1921          */
1922         for (;;)
1923         {
1924                 FD_ZERO(&rfds);
1925                 FD_ZERO(&wfds);
1926                 maxfd = -1;
1927
1928                 /*
1929                  * As long as we have buffer space we add the socket to the read
1930                  * descriptor set.
1931                  */
1932                 if (msg_have <= (int) (PGSTAT_RECVBUFFERSZ - sizeof(PgStat_Msg)))
1933                 {
1934                         FD_SET(pgStatSock, &rfds);
1935                         maxfd = pgStatSock;
1936                         overflow = false;
1937                 }
1938                 else
1939                 {
1940                         if (!overflow)
1941                         {
1942                                 ereport(LOG,
1943                                                 (errmsg("statistics buffer is full")));
1944                                 overflow = true;
1945                         }
1946                 }
1947
1948                 /*
1949                  * If we have messages to write out, we add the pipe to the write
1950                  * descriptor set.
1951                  */
1952                 if (msg_have > 0)
1953                 {
1954                         FD_SET(writePipe, &wfds);
1955                         if (writePipe > maxfd)
1956                                 maxfd = writePipe;
1957                 }
1958
1959                 /*
1960                  * Wait for some work to do; but not for more than 10 seconds.
1961                  * (This determines how quickly we will shut down after an
1962                  * ungraceful postmaster termination; so it needn't be very fast.)
1963                  */
1964                 timeout.tv_sec = 10;
1965                 timeout.tv_usec = 0;
1966
1967                 nready = select(maxfd + 1, &rfds, &wfds, NULL, &timeout);
1968                 if (nready < 0)
1969                 {
1970                         if (errno == EINTR)
1971                                 continue;
1972                         ereport(ERROR,
1973                                         (errcode_for_socket_access(),
1974                                          errmsg("select() failed in statistics buffer: %m")));
1975                 }
1976
1977                 /*
1978                  * If there is a message on the socket, read it and check for
1979                  * validity.
1980                  */
1981                 if (FD_ISSET(pgStatSock, &rfds))
1982                 {
1983                         len = recv(pgStatSock, (char *) &input_buffer,
1984                                            sizeof(PgStat_Msg), 0);
1985                         if (len < 0)
1986                                 ereport(ERROR,
1987                                                 (errcode_for_socket_access(),
1988                                            errmsg("could not read statistics message: %m")));
1989
1990                         /*
1991                          * We ignore messages that are smaller than our common header
1992                          */
1993                         if (len < sizeof(PgStat_MsgHdr))
1994                                 continue;
1995
1996                         /*
1997                          * The received length must match the length in the header
1998                          */
1999                         if (input_buffer.msg_hdr.m_size != len)
2000                                 continue;
2001
2002                         /*
2003                          * O.K. - we accept this message.  Copy it to the circular
2004                          * msgbuffer.
2005                          */
2006                         frm = 0;
2007                         while (len > 0)
2008                         {
2009                                 xfr = PGSTAT_RECVBUFFERSZ - msg_recv;
2010                                 if (xfr > len)
2011                                         xfr = len;
2012                                 Assert(xfr > 0);
2013                                 memcpy(msgbuffer + msg_recv,
2014                                            ((char *) &input_buffer) + frm,
2015                                            xfr);
2016                                 msg_recv += xfr;
2017                                 if (msg_recv == PGSTAT_RECVBUFFERSZ)
2018                                         msg_recv = 0;
2019                                 msg_have += xfr;
2020                                 frm += xfr;
2021                                 len -= xfr;
2022                         }
2023                 }
2024
2025                 /*
2026                  * If the collector is ready to receive, write some data into his
2027                  * pipe.  We may or may not be able to write all that we have.
2028                  *
2029                  * NOTE: if what we have is less than PIPE_BUF bytes but more than
2030                  * the space available in the pipe buffer, most kernels will
2031                  * refuse to write any of it, and will return EAGAIN.  This means
2032                  * we will busy-loop until the situation changes (either because
2033                  * the collector caught up, or because more data arrives so that
2034                  * we have more than PIPE_BUF bytes buffered).  This is not good,
2035                  * but is there any way around it?      We have no way to tell when
2036                  * the collector has caught up...
2037                  */
2038                 if (FD_ISSET(writePipe, &wfds))
2039                 {
2040                         xfr = PGSTAT_RECVBUFFERSZ - msg_send;
2041                         if (xfr > msg_have)
2042                                 xfr = msg_have;
2043                         Assert(xfr > 0);
2044                         len = pipewrite(writePipe, msgbuffer + msg_send, xfr);
2045                         if (len < 0)
2046                         {
2047                                 if (errno == EINTR || errno == EAGAIN)
2048                                         continue;       /* not enough space in pipe */
2049                                 ereport(ERROR,
2050                                                 (errcode_for_socket_access(),
2051                                                  errmsg("could not write to statistics collector pipe: %m")));
2052                         }
2053                         /* NB: len < xfr is okay */
2054                         msg_send += len;
2055                         if (msg_send == PGSTAT_RECVBUFFERSZ)
2056                                 msg_send = 0;
2057                         msg_have -= len;
2058                 }
2059
2060                 /*
2061                  * Make sure we forwarded all messages before we check for
2062                  * postmaster termination.
2063                  */
2064                 if (msg_have != 0 || FD_ISSET(pgStatSock, &rfds))
2065                         continue;
2066
2067                 /*
2068                  * If the postmaster has terminated, we die too.  (This is no
2069                  * longer the normal exit path, however.)
2070                  */
2071                 if (!PostmasterIsAlive(true))
2072                         exit(0);
2073         }
2074 }
2075
2076 /* SIGQUIT signal handler for buffer process */
2077 static void
2078 pgstat_exit(SIGNAL_ARGS)
2079 {
2080         /*
2081          * For now, we just nail the doors shut and get out of town.  It might
2082          * be cleaner to allow any pending messages to be sent, but that
2083          * creates a tradeoff against speed of exit.
2084          */
2085
2086         /*
2087          * If running in bufferer, kill our collector as well. On some broken
2088          * win32 systems, it does not shut down automatically because of issues
2089          * with socket inheritance.  XXX so why not fix the socket inheritance...
2090          */
2091 #ifdef WIN32
2092         if (pgStatCollectorPid > 0)
2093                 kill(pgStatCollectorPid, SIGQUIT);
2094 #endif
2095         exit(0);
2096 }
2097
2098 /* SIGCHLD signal handler for buffer process */
2099 static void
2100 pgstat_die(SIGNAL_ARGS)
2101 {
2102         exit(1);
2103 }
2104
2105
2106 /* ----------
2107  * pgstat_add_backend() -
2108  *
2109  *      Support function to keep our backend list up to date.
2110  * ----------
2111  */
2112 static int
2113 pgstat_add_backend(PgStat_MsgHdr *msg)
2114 {
2115         PgStat_StatBeEntry *beentry;
2116         PgStat_StatBeDead *deadbe;
2117
2118         /*
2119          * Check that the backend ID is valid
2120          */
2121         if (msg->m_backendid < 1 || msg->m_backendid > MaxBackends)
2122         {
2123                 ereport(LOG,
2124                          (errmsg("invalid server process ID %d", msg->m_backendid)));
2125                 return -1;
2126         }
2127
2128         /*
2129          * Get the slot for this backendid.
2130          */
2131         beentry = &pgStatBeTable[msg->m_backendid - 1];
2132
2133         /*
2134          * If the slot contains the PID of this backend, everything is
2135          * fine and we have nothing to do. Note that all the slots are
2136          * zero'd out when the collector is started. We assume that a slot
2137          * is "empty" iff procpid == 0.
2138          */
2139         if (beentry->procpid > 0 && beentry->procpid == msg->m_procpid)
2140                 return 0;
2141
2142         /*
2143          * Lookup if this backend is known to be dead. This can be caused due
2144          * to messages arriving in the wrong order - e.g. postmaster's BETERM
2145          * message might have arrived before we received all the backends
2146          * stats messages, or even a new backend with the same backendid was
2147          * faster in sending his BESTART.
2148          *
2149          * If the backend is known to be dead, we ignore this add.
2150          */
2151         deadbe = (PgStat_StatBeDead *) hash_search(pgStatBeDead,
2152                                                                                            (void *) &(msg->m_procpid),
2153                                                                                            HASH_FIND, NULL);
2154         if (deadbe)
2155                 return 1;
2156
2157         /*
2158          * Backend isn't known to be dead. If it's slot is currently used, we
2159          * have to kick out the old backend.
2160          */
2161         if (beentry->procpid > 0)
2162                 pgstat_sub_backend(beentry->procpid);
2163
2164         /* Must be able to distinguish between empty and non-empty slots */
2165         Assert(msg->m_procpid > 0);
2166
2167         /* Put this new backend into the slot */
2168         beentry->procpid = msg->m_procpid;
2169         beentry->start_timestamp = GetCurrentTimestamp();
2170         beentry->activity_start_timestamp = 0;
2171         beentry->activity[0] = '\0';
2172
2173         /*
2174          * We can't initialize the rest of the data in this slot until we
2175          * see the BESTART message. Therefore, we set the database and
2176          * user to sentinel values, to indicate "undefined". There is no
2177          * easy way to do this for the client address, so make sure to
2178          * check that the database or user are defined before accessing
2179          * the client address.
2180          */
2181         beentry->userid = InvalidOid;
2182         beentry->databaseid = InvalidOid;
2183
2184         return 0;
2185 }
2186
2187 /*
2188  * Lookup the hash table entry for the specified database. If no hash
2189  * table entry exists, initialize it, if the create parameter is true.
2190  * Else, return NULL.
2191  */
2192 static PgStat_StatDBEntry *
2193 pgstat_get_db_entry(Oid databaseid, bool create)
2194 {
2195         PgStat_StatDBEntry *result;
2196         bool found;
2197         HASHACTION action = (create ? HASH_ENTER : HASH_FIND);
2198
2199         /* Lookup or create the hash table entry for this database */
2200         result = (PgStat_StatDBEntry *) hash_search(pgStatDBHash,
2201                                                                                                 &databaseid,
2202                                                                                                 action, &found);
2203
2204         if (!create && !found)
2205                 return NULL;
2206
2207         /* If not found, initialize the new one. */
2208         if (!found)
2209         {
2210                 HASHCTL         hash_ctl;
2211
2212                 result->tables = NULL;
2213                 result->n_xact_commit = 0;
2214                 result->n_xact_rollback = 0;
2215                 result->n_blocks_fetched = 0;
2216                 result->n_blocks_hit = 0;
2217                 result->destroy = 0;
2218                 result->last_autovac_time = 0;
2219
2220                 memset(&hash_ctl, 0, sizeof(hash_ctl));
2221                 hash_ctl.keysize = sizeof(Oid);
2222                 hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
2223                 hash_ctl.hash = oid_hash;
2224                 result->tables = hash_create("Per-database table",
2225                                                                           PGSTAT_TAB_HASH_SIZE,
2226                                                                           &hash_ctl,
2227                                                                           HASH_ELEM | HASH_FUNCTION);
2228         }
2229
2230         return result;
2231 }
2232
2233 /* ----------
2234  * pgstat_sub_backend() -
2235  *
2236  *      Remove a backend from the actual backends list.
2237  * ----------
2238  */
2239 static void
2240 pgstat_sub_backend(int procpid)
2241 {
2242         int                     i;
2243         PgStat_StatBeDead *deadbe;
2244         bool            found;
2245
2246         /*
2247          * Search in the known-backends table for the slot containing this
2248          * PID.
2249          */
2250         for (i = 0; i < MaxBackends; i++)
2251         {
2252                 if (pgStatBeTable[i].procpid == procpid)
2253                 {
2254                         /*
2255                          * That's him. Add an entry to the known to be dead backends.
2256                          * Due to possible misorder in the arrival of UDP packets it's
2257                          * possible that even if we know the backend is dead, there
2258                          * could still be messages queued that arrive later. Those
2259                          * messages must not cause our number of backends statistics
2260                          * to get screwed up, so we remember for a couple of seconds
2261                          * that this PID is dead and ignore them (only the counting of
2262                          * backends, not the table access stats they sent).
2263                          */
2264                         deadbe = (PgStat_StatBeDead *) hash_search(pgStatBeDead,
2265                                                                                                            (void *) &procpid,
2266                                                                                                            HASH_ENTER,
2267                                                                                                            &found);
2268
2269                         if (!found)
2270                         {
2271                                 deadbe->backendid = i + 1;
2272                                 deadbe->destroy = PGSTAT_DESTROY_COUNT;
2273                         }
2274
2275                         /*
2276                          * Declare the backend slot empty.
2277                          */
2278                         pgStatBeTable[i].procpid = 0;
2279                         return;
2280                 }
2281         }
2282
2283         /*
2284          * No big problem if not found. This can happen if UDP messages arrive
2285          * out of order here.
2286          */
2287 }
2288
2289
2290 /* ----------
2291  * pgstat_write_statsfile() -
2292  *
2293  *      Tell the news.
2294  * ----------
2295  */
2296 static void
2297 pgstat_write_statsfile(void)
2298 {
2299         HASH_SEQ_STATUS hstat;
2300         HASH_SEQ_STATUS tstat;
2301         PgStat_StatDBEntry *dbentry;
2302         PgStat_StatTabEntry *tabentry;
2303         PgStat_StatBeDead *deadbe;
2304         FILE       *fpout;
2305         int                     i;
2306         int32           format_id;
2307
2308         /*
2309          * Open the statistics temp file to write out the current values.
2310          */
2311         fpout = fopen(PGSTAT_STAT_TMPFILE, PG_BINARY_W);
2312         if (fpout == NULL)
2313         {
2314                 ereport(LOG,
2315                                 (errcode_for_file_access(),
2316                         errmsg("could not open temporary statistics file \"%s\": %m",
2317                                    PGSTAT_STAT_TMPFILE)));
2318                 return;
2319         }
2320
2321         /*
2322          * Write the file header --- currently just a format ID.
2323          */
2324         format_id = PGSTAT_FILE_FORMAT_ID;
2325         fwrite(&format_id, sizeof(format_id), 1, fpout);
2326
2327         /*
2328          * Walk through the database table.
2329          */
2330         hash_seq_init(&hstat, pgStatDBHash);
2331         while ((dbentry = (PgStat_StatDBEntry *) hash_seq_search(&hstat)) != NULL)
2332         {
2333                 /*
2334                  * If this database is marked destroyed, count down and do so if
2335                  * it reaches 0.
2336                  */
2337                 if (dbentry->destroy > 0)
2338                 {
2339                         if (--(dbentry->destroy) == 0)
2340                         {
2341                                 if (dbentry->tables != NULL)
2342                                         hash_destroy(dbentry->tables);
2343
2344                                 if (hash_search(pgStatDBHash,
2345                                                                 (void *) &(dbentry->databaseid),
2346                                                                 HASH_REMOVE, NULL) == NULL)
2347                                         ereport(ERROR,
2348                                                         (errmsg("database hash table corrupted "
2349                                                                         "during cleanup --- abort")));
2350                         }
2351
2352                         /*
2353                          * Don't include statistics for it.
2354                          */
2355                         continue;
2356                 }
2357
2358                 /*
2359                  * Write out the DB line including the number of live backends.
2360                  */
2361                 fputc('D', fpout);
2362                 fwrite(dbentry, sizeof(PgStat_StatDBEntry), 1, fpout);
2363
2364                 /*
2365                  * Walk through the databases access stats per table.
2366                  */
2367                 hash_seq_init(&tstat, dbentry->tables);
2368                 while ((tabentry = (PgStat_StatTabEntry *) hash_seq_search(&tstat)) != NULL)
2369                 {
2370                         /*
2371                          * If table entry marked for destruction, same as above for
2372                          * the database entry.
2373                          */
2374                         if (tabentry->destroy > 0)
2375                         {
2376                                 if (--(tabentry->destroy) == 0)
2377                                 {
2378                                         if (hash_search(dbentry->tables,
2379                                                                         (void *) &(tabentry->tableid),
2380                                                                         HASH_REMOVE, NULL) == NULL)
2381                                         {
2382                                                 ereport(ERROR,
2383                                                                 (errmsg("tables hash table for "
2384                                                                                 "database %u corrupted during "
2385                                                                                 "cleanup --- abort",
2386                                                                                 dbentry->databaseid)));
2387                                         }
2388                                 }
2389                                 continue;
2390                         }
2391
2392                         /*
2393                          * At least we think this is still a live table. Print its
2394                          * access stats.
2395                          */
2396                         fputc('T', fpout);
2397                         fwrite(tabentry, sizeof(PgStat_StatTabEntry), 1, fpout);
2398                 }
2399
2400                 /*
2401                  * Mark the end of this DB
2402                  */
2403                 fputc('d', fpout);
2404         }
2405
2406         /*
2407          * Write out the known running backends to the stats file.
2408          */
2409         i = MaxBackends;
2410         fputc('M', fpout);
2411         fwrite(&i, sizeof(i), 1, fpout);
2412
2413         for (i = 0; i < MaxBackends; i++)
2414         {
2415                 if (pgStatBeTable[i].procpid > 0)
2416                 {
2417                         fputc('B', fpout);
2418                         fwrite(&pgStatBeTable[i], sizeof(PgStat_StatBeEntry), 1, fpout);
2419                 }
2420         }
2421
2422         /*
2423          * No more output to be done. Close the temp file and replace the old
2424          * pgstat.stat with it.
2425          */
2426         fputc('E', fpout);
2427         if (fclose(fpout) < 0)
2428         {
2429                 ereport(LOG,
2430                                 (errcode_for_file_access(),
2431                    errmsg("could not close temporary statistics file \"%s\": %m",
2432                                   PGSTAT_STAT_TMPFILE)));
2433         }
2434         else
2435         {
2436                 if (rename(PGSTAT_STAT_TMPFILE, PGSTAT_STAT_FILENAME) < 0)
2437                 {
2438                         ereport(LOG,
2439                                         (errcode_for_file_access(),
2440                                          errmsg("could not rename temporary statistics file \"%s\" to \"%s\": %m",
2441                                                         PGSTAT_STAT_TMPFILE, PGSTAT_STAT_FILENAME)));
2442                 }
2443         }
2444
2445         /*
2446          * Clear out the dead backends table
2447          */
2448         hash_seq_init(&hstat, pgStatBeDead);
2449         while ((deadbe = (PgStat_StatBeDead *) hash_seq_search(&hstat)) != NULL)
2450         {
2451                 /*
2452                  * Count down the destroy delay and remove entries where it
2453                  * reaches 0.
2454                  */
2455                 if (--(deadbe->destroy) <= 0)
2456                 {
2457                         if (hash_search(pgStatBeDead,
2458                                                         (void *) &(deadbe->procpid),
2459                                                         HASH_REMOVE, NULL) == NULL)
2460                         {
2461                                 ereport(ERROR,
2462                                           (errmsg("dead-server-process hash table corrupted "
2463                                                           "during cleanup --- abort")));
2464                         }
2465                 }
2466         }
2467 }
2468
2469
2470 /* ----------
2471  * pgstat_read_statsfile() -
2472  *
2473  *      Reads in an existing statistics collector and initializes the
2474  *      databases' hash table (whose entries point to the tables' hash tables)
2475  *      and the current backend table.
2476  * ----------
2477  */
2478 static void
2479 pgstat_read_statsfile(HTAB **dbhash, Oid onlydb,
2480                                           PgStat_StatBeEntry **betab, int *numbackends)
2481 {
2482         PgStat_StatDBEntry *dbentry;
2483         PgStat_StatDBEntry dbbuf;
2484         PgStat_StatTabEntry *tabentry;
2485         PgStat_StatTabEntry tabbuf;
2486         PgStat_StatBeEntry *beentry;
2487         HASHCTL         hash_ctl;
2488         HTAB       *tabhash = NULL;
2489         FILE       *fpin;
2490         int32           format_id;
2491         int                     maxbackends = 0;
2492         int                     havebackends = 0;
2493         bool            found;
2494         bool            check_pids;
2495         MemoryContext use_mcxt;
2496         int                     mcxt_flags;
2497
2498         /*
2499          * If running in the collector or the autovacuum process, we use the
2500          * DynaHashCxt memory context.  If running in a backend, we use the
2501          * TopTransactionContext instead, so the caller must only know the last
2502          * XactId when this call happened to know if his tables are still valid or
2503          * already gone!
2504          *
2505          * Also, if running in a regular backend, we check backend entries against
2506          * the PGPROC array so that we can detect stale entries.  This lets us
2507          * discard entries whose BETERM message got lost for some reason.
2508          */
2509         if (pgStatRunningInCollector || IsAutoVacuumProcess())
2510         {
2511                 use_mcxt = NULL;
2512                 mcxt_flags = 0;
2513                 check_pids = false;
2514         }
2515         else
2516         {
2517                 use_mcxt = TopTransactionContext;
2518                 mcxt_flags = HASH_CONTEXT;
2519                 check_pids = true;
2520         }
2521
2522         /*
2523          * Create the DB hashtable
2524          */
2525         memset(&hash_ctl, 0, sizeof(hash_ctl));
2526         hash_ctl.keysize = sizeof(Oid);
2527         hash_ctl.entrysize = sizeof(PgStat_StatDBEntry);
2528         hash_ctl.hash = oid_hash;
2529         hash_ctl.hcxt = use_mcxt;
2530         *dbhash = hash_create("Databases hash", PGSTAT_DB_HASH_SIZE, &hash_ctl,
2531                                                   HASH_ELEM | HASH_FUNCTION | mcxt_flags);
2532
2533         /*
2534          * Initialize the number of known backends to zero, just in case we do
2535          * a silent error return below.
2536          */
2537         if (numbackends != NULL)
2538                 *numbackends = 0;
2539         if (betab != NULL)
2540                 *betab = NULL;
2541
2542         /*
2543          * Try to open the status file. If it doesn't exist, the backends
2544          * simply return zero for anything and the collector simply starts
2545          * from scratch with empty counters.
2546          */
2547         if ((fpin = AllocateFile(PGSTAT_STAT_FILENAME, PG_BINARY_R)) == NULL)
2548                 return;
2549
2550         /*
2551          * Verify it's of the expected format.
2552          */
2553         if (fread(&format_id, 1, sizeof(format_id), fpin) != sizeof(format_id)
2554                 || format_id != PGSTAT_FILE_FORMAT_ID)
2555         {
2556                 ereport(pgStatRunningInCollector ? LOG : WARNING,
2557                                 (errmsg("corrupted pgstat.stat file")));
2558                 goto done;
2559         }
2560
2561         /*
2562          * We found an existing collector stats file. Read it and put all the
2563          * hashtable entries into place.
2564          */
2565         for (;;)
2566         {
2567                 switch (fgetc(fpin))
2568                 {
2569                                 /*
2570                                  * 'D'  A PgStat_StatDBEntry struct describing a database
2571                                  * follows. Subsequently, zero to many 'T' entries will
2572                                  * follow until a 'd' is encountered.
2573                                  */
2574                         case 'D':
2575                                 if (fread(&dbbuf, 1, sizeof(dbbuf), fpin) != sizeof(dbbuf))
2576                                 {
2577                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2578                                                         (errmsg("corrupted pgstat.stat file")));
2579                                         goto done;
2580                                 }
2581
2582                                 /*
2583                                  * Add to the DB hash
2584                                  */
2585                                 dbentry = (PgStat_StatDBEntry *) hash_search(*dbhash,
2586                                                                                           (void *) &dbbuf.databaseid,
2587                                                                                                                          HASH_ENTER,
2588                                                                                                                          &found);
2589                                 if (found)
2590                                 {
2591                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2592                                                         (errmsg("corrupted pgstat.stat file")));
2593                                         goto done;
2594                                 }
2595
2596                                 memcpy(dbentry, &dbbuf, sizeof(PgStat_StatDBEntry));
2597                                 dbentry->tables = NULL;
2598                                 dbentry->destroy = 0;
2599                                 dbentry->n_backends = 0;
2600
2601                                 /*
2602                                  * Don't collect tables if not the requested DB (or the
2603                                  * shared-table info)
2604                                  */
2605                                 if (onlydb != InvalidOid)
2606                                 {
2607                                         if (dbbuf.databaseid != onlydb &&
2608                                                 dbbuf.databaseid != InvalidOid)
2609                                         break;
2610                                 }
2611
2612                                 memset(&hash_ctl, 0, sizeof(hash_ctl));
2613                                 hash_ctl.keysize = sizeof(Oid);
2614                                 hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
2615                                 hash_ctl.hash = oid_hash;
2616                                 hash_ctl.hcxt = use_mcxt;
2617                                 dbentry->tables = hash_create("Per-database table",
2618                                                                                           PGSTAT_TAB_HASH_SIZE,
2619                                                                                           &hash_ctl,
2620                                                                                           HASH_ELEM | HASH_FUNCTION | mcxt_flags);
2621
2622                                 /*
2623                                  * Arrange that following 'T's add entries to this
2624                                  * databases tables hash table.
2625                                  */
2626                                 tabhash = dbentry->tables;
2627                                 break;
2628
2629                                 /*
2630                                  * 'd'  End of this database.
2631                                  */
2632                         case 'd':
2633                                 tabhash = NULL;
2634                                 break;
2635
2636                                 /*
2637                                  * 'T'  A PgStat_StatTabEntry follows.
2638                                  */
2639                         case 'T':
2640                                 if (fread(&tabbuf, 1, sizeof(tabbuf), fpin) != sizeof(tabbuf))
2641                                 {
2642                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2643                                                         (errmsg("corrupted pgstat.stat file")));
2644                                         goto done;
2645                                 }
2646
2647                                 /*
2648                                  * Skip if table belongs to a not requested database.
2649                                  */
2650                                 if (tabhash == NULL)
2651                                         break;
2652
2653                                 tabentry = (PgStat_StatTabEntry *) hash_search(tabhash,
2654                                                                                                 (void *) &tabbuf.tableid,
2655                                                                                                          HASH_ENTER, &found);
2656
2657                                 if (found)
2658                                 {
2659                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2660                                                         (errmsg("corrupted pgstat.stat file")));
2661                                         goto done;
2662                                 }
2663
2664                                 memcpy(tabentry, &tabbuf, sizeof(tabbuf));
2665                                 break;
2666
2667                                 /*
2668                                  * 'M'  The maximum number of backends to expect follows.
2669                                  */
2670                         case 'M':
2671                                 if (betab == NULL || numbackends == NULL)
2672                                         goto done;
2673                                 if (fread(&maxbackends, 1, sizeof(maxbackends), fpin) !=
2674                                         sizeof(maxbackends))
2675                                 {
2676                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2677                                                         (errmsg("corrupted pgstat.stat file")));
2678                                         goto done;
2679                                 }
2680                                 if (maxbackends == 0)
2681                                         goto done;
2682
2683                                 /*
2684                                  * Allocate space (in TopTransactionContext too) for the
2685                                  * backend table.
2686                                  */
2687                                 if (use_mcxt == NULL)
2688                                         *betab = (PgStat_StatBeEntry *)
2689                                                 palloc(sizeof(PgStat_StatBeEntry) * maxbackends);
2690                                 else
2691                                         *betab = (PgStat_StatBeEntry *)
2692                                                 MemoryContextAlloc(use_mcxt,
2693                                                                                    sizeof(PgStat_StatBeEntry) * maxbackends);
2694                                 break;
2695
2696                                 /*
2697                                  * 'B'  A PgStat_StatBeEntry follows.
2698                                  */
2699                         case 'B':
2700                                 if (betab == NULL || numbackends == NULL || *betab == NULL)
2701                                         goto done;
2702
2703                                 if (havebackends >= maxbackends)
2704                                         goto done;
2705
2706                                 /*
2707                                  * Read it directly into the table.
2708                                  */
2709                                 beentry = &(*betab)[havebackends];
2710
2711                                 if (fread(beentry, 1, sizeof(PgStat_StatBeEntry), fpin) !=
2712                                         sizeof(PgStat_StatBeEntry))
2713                                 {
2714                                         ereport(pgStatRunningInCollector ? LOG : WARNING,
2715                                                         (errmsg("corrupted pgstat.stat file")));
2716                                         goto done;
2717                                 }
2718
2719                                 /*
2720                                  * If possible, check PID to verify still running
2721                                  */
2722                                 if (check_pids && !IsBackendPid(beentry->procpid))
2723                                 {
2724                                         /*
2725                                          * Note: we could send a BETERM message to tell the
2726                                          * collector to drop the entry, but I'm a bit worried
2727                                          * about race conditions.  For now, just silently ignore
2728                                          * dead entries; they'll get recycled eventually anyway.
2729                                          */
2730
2731                                         /* Don't accept the entry */
2732                                         memset(beentry, 0, sizeof(PgStat_StatBeEntry));
2733                                         break;
2734                                 }
2735
2736                                 /*
2737                                  * Count backends per database here.
2738                                  */
2739                                 dbentry = (PgStat_StatDBEntry *)
2740                                         hash_search(*dbhash,
2741                                                                 &(beentry->databaseid),
2742                                                                 HASH_FIND,
2743                                                                 NULL);
2744                                 if (dbentry)
2745                                         dbentry->n_backends++;
2746
2747                                 havebackends++;
2748                                 *numbackends = havebackends;
2749
2750                                 break;
2751
2752                                 /*
2753                                  * 'E'  The EOF marker of a complete stats file.
2754                                  */
2755                         case 'E':
2756                                 goto done;
2757
2758                         default:
2759                                 ereport(pgStatRunningInCollector ? LOG : WARNING,
2760                                                 (errmsg("corrupted pgstat.stat file")));
2761                                 goto done;
2762                 }
2763         }
2764
2765 done:
2766         FreeFile(fpin);
2767 }
2768
2769 /*
2770  * If not done for this transaction, read the statistics collector
2771  * stats file into some hash tables.
2772  *
2773  * Because we store the hash tables in TopTransactionContext, the result
2774  * is good for the entire current main transaction.
2775  *
2776  * Inside the autovacuum process, the statfile is assumed to be valid
2777  * "forever", that is one iteration, within one database.  This means
2778  * we only consider the statistics as they were when the autovacuum
2779  * iteration started.
2780  */
2781 static void
2782 backend_read_statsfile(void)
2783 {
2784         if (IsAutoVacuumProcess())
2785         {
2786                 /* already read it? */
2787                 if (pgStatDBHash)
2788                         return;
2789                 Assert(!pgStatRunningInCollector);
2790                 pgstat_read_statsfile(&pgStatDBHash, InvalidOid,
2791                                                           &pgStatBeTable, &pgStatNumBackends);
2792         }
2793         else
2794         {
2795                 TransactionId topXid = GetTopTransactionId();
2796
2797                 if (!TransactionIdEquals(pgStatDBHashXact, topXid))
2798                 {
2799                         Assert(!pgStatRunningInCollector);
2800                         pgstat_read_statsfile(&pgStatDBHash, MyDatabaseId,
2801                                                                   &pgStatBeTable, &pgStatNumBackends);
2802                         pgStatDBHashXact = topXid;
2803                 }
2804         }
2805 }
2806
2807
2808 /* ----------
2809  * pgstat_recv_bestart() -
2810  *
2811  *      Process a backend startup message.
2812  * ----------
2813  */
2814 static void
2815 pgstat_recv_bestart(PgStat_MsgBestart *msg, int len)
2816 {
2817         PgStat_StatBeEntry *entry;
2818
2819         /*
2820          * If the backend is known dead, we ignore the message -- we don't
2821          * want to update the backend entry's state since this BESTART
2822          * message refers to an old, dead backend
2823          */
2824         if (pgstat_add_backend(&msg->m_hdr) != 0)
2825                 return;
2826
2827         entry = &(pgStatBeTable[msg->m_hdr.m_backendid - 1]);
2828         entry->userid = msg->m_userid;
2829         memcpy(&entry->clientaddr, &msg->m_clientaddr, sizeof(entry->clientaddr));
2830         entry->databaseid = msg->m_databaseid;
2831 }
2832
2833
2834 /* ----------
2835  * pgstat_recv_beterm() -
2836  *
2837  *      Process a backend termination message.
2838  * ----------
2839  */
2840 static void
2841 pgstat_recv_beterm(PgStat_MsgBeterm *msg, int len)
2842 {
2843         pgstat_sub_backend(msg->m_hdr.m_procpid);
2844 }
2845
2846 /* ----------
2847  * pgstat_recv_autovac() -
2848  *
2849  *      Process an autovacuum signalling message.
2850  * ----------
2851  */
2852 static void
2853 pgstat_recv_autovac(PgStat_MsgAutovacStart *msg, int len)
2854 {
2855         PgStat_StatDBEntry *dbentry;
2856
2857         /*
2858          * Lookup the database in the hashtable.  Don't create the entry if it
2859          * doesn't exist, because autovacuum may be processing a template
2860          * database.  If this isn't the case, the database is most likely to
2861          * have an entry already.  (If it doesn't, not much harm is done
2862          * anyway -- it'll get created as soon as somebody actually uses
2863          * the database.)
2864          */
2865         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
2866         if (dbentry == NULL)
2867                 return;
2868
2869         /*
2870          * Store the last autovacuum time in the database entry.
2871          */
2872         dbentry->last_autovac_time = msg->m_start_time;
2873 }
2874
2875 /* ----------
2876  * pgstat_recv_vacuum() -
2877  *
2878  *      Process a VACUUM message.
2879  * ----------
2880  */
2881 static void
2882 pgstat_recv_vacuum(PgStat_MsgVacuum *msg, int len)
2883 {
2884         PgStat_StatDBEntry *dbentry;
2885         PgStat_StatTabEntry *tabentry;
2886         bool            found;
2887         bool            create;
2888
2889         /*
2890          * If we don't know about the database, ignore the message, because it
2891          * may be autovacuum processing a template database.  But if the message
2892          * is for database InvalidOid, don't ignore it, because we are getting
2893          * a message from vacuuming a shared relation.
2894          */
2895         create = (msg->m_databaseid == InvalidOid);
2896
2897         dbentry = pgstat_get_db_entry(msg->m_databaseid, create);
2898         if (dbentry == NULL)
2899                 return;
2900
2901         tabentry = hash_search(dbentry->tables, &(msg->m_tableoid),
2902                                                    HASH_ENTER, &found);
2903
2904         /*
2905          * If we are creating the entry, initialize it.
2906          */
2907         if (!found)
2908         {
2909                 tabentry->numscans = 0;
2910
2911                 tabentry->tuples_returned = 0;
2912                 tabentry->tuples_fetched = 0;
2913                 tabentry->tuples_inserted = 0;
2914                 tabentry->tuples_updated = 0;
2915                 tabentry->tuples_deleted = 0;
2916
2917                 tabentry->n_live_tuples = msg->m_tuples;
2918                 tabentry->n_dead_tuples = 0;
2919
2920                 if (msg->m_analyze)
2921                         tabentry->last_anl_tuples = msg->m_tuples;
2922                 else
2923                         tabentry->last_anl_tuples = 0;
2924
2925                 tabentry->blocks_fetched = 0;
2926                 tabentry->blocks_hit = 0;
2927
2928                 tabentry->destroy = 0;
2929         }
2930         else
2931         {
2932                 tabentry->n_live_tuples = msg->m_tuples;
2933                 tabentry->n_dead_tuples = 0;
2934                 if (msg->m_analyze)
2935                         tabentry->last_anl_tuples = msg->m_tuples;
2936         }
2937 }
2938
2939 /* ----------
2940  * pgstat_recv_analyze() -
2941  *
2942  *      Process an ANALYZE message.
2943  * ----------
2944  */
2945 static void
2946 pgstat_recv_analyze(PgStat_MsgAnalyze *msg, int len)
2947 {
2948         PgStat_StatDBEntry *dbentry;
2949         PgStat_StatTabEntry *tabentry;
2950         bool            found;
2951
2952         /*
2953          * Note that we do create the database entry here, as opposed to what
2954          * we do on AutovacStart and Vacuum messages.  This is because
2955          * autovacuum never executes ANALYZE on template databases.
2956          */
2957         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
2958
2959         tabentry = hash_search(dbentry->tables, &(msg->m_tableoid),
2960                                                    HASH_ENTER, &found);
2961
2962         /*
2963          * If we are creating the entry, initialize it.
2964          */
2965         if (!found)
2966         {
2967                 tabentry->numscans = 0;
2968
2969                 tabentry->tuples_returned = 0;
2970                 tabentry->tuples_fetched = 0;
2971                 tabentry->tuples_inserted = 0;
2972                 tabentry->tuples_updated = 0;
2973                 tabentry->tuples_deleted = 0;
2974
2975                 tabentry->n_live_tuples = msg->m_live_tuples;
2976                 tabentry->n_dead_tuples = msg->m_dead_tuples;
2977                 tabentry->last_anl_tuples = msg->m_live_tuples + msg->m_dead_tuples;
2978
2979                 tabentry->blocks_fetched = 0;
2980                 tabentry->blocks_hit = 0;
2981
2982                 tabentry->destroy = 0;
2983         }
2984         else
2985         {
2986                 tabentry->n_live_tuples = msg->m_live_tuples;
2987                 tabentry->n_dead_tuples = msg->m_dead_tuples;
2988                 tabentry->last_anl_tuples = msg->m_live_tuples + msg->m_dead_tuples;
2989         }
2990 }
2991
2992 /* ----------
2993  * pgstat_recv_activity() -
2994  *
2995  *      Remember what the backend is doing.
2996  * ----------
2997  */
2998 static void
2999 pgstat_recv_activity(PgStat_MsgActivity *msg, int len)
3000 {
3001         PgStat_StatBeEntry *entry;
3002
3003         /*
3004          * Here we check explicitly for 0 return, since we don't want to
3005          * mangle the activity of an active backend by a delayed packet from a
3006          * dead one.
3007          */
3008         if (pgstat_add_backend(&msg->m_hdr) != 0)
3009                 return;
3010
3011         entry = &(pgStatBeTable[msg->m_hdr.m_backendid - 1]);
3012
3013         StrNCpy(entry->activity, msg->m_what, PGSTAT_ACTIVITY_SIZE);
3014
3015         entry->activity_start_timestamp = GetCurrentTimestamp();
3016 }
3017
3018
3019 /* ----------
3020  * pgstat_recv_tabstat() -
3021  *
3022  *      Count what the backend has done.
3023  * ----------
3024  */
3025 static void
3026 pgstat_recv_tabstat(PgStat_MsgTabstat *msg, int len)
3027 {
3028         PgStat_TableEntry *tabmsg = &(msg->m_entry[0]);
3029         PgStat_StatDBEntry *dbentry;
3030         PgStat_StatTabEntry *tabentry;
3031         int                     i;
3032         bool            found;
3033
3034         /*
3035          * Make sure the backend is counted for.
3036          */
3037         if (pgstat_add_backend(&msg->m_hdr) < 0)
3038                 return;
3039
3040         dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
3041
3042         /*
3043          * If the database is marked for destroy, this is a delayed UDP packet
3044          * and not worth being counted.
3045          */
3046         if (dbentry->destroy > 0)
3047                 return;
3048
3049         dbentry->n_xact_commit += (PgStat_Counter) (msg->m_xact_commit);
3050         dbentry->n_xact_rollback += (PgStat_Counter) (msg->m_xact_rollback);
3051
3052         /*
3053          * Process all table entries in the message.
3054          */
3055         for (i = 0; i < msg->m_nentries; i++)
3056         {
3057                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
3058                                                                                           (void *) &(tabmsg[i].t_id),
3059                                                                                                          HASH_ENTER, &found);
3060
3061                 if (!found)
3062                 {
3063                         /*
3064                          * If it's a new table entry, initialize counters to the
3065                          * values we just got.
3066                          */
3067                         tabentry->numscans = tabmsg[i].t_numscans;
3068                         tabentry->tuples_returned = tabmsg[i].t_tuples_returned;
3069                         tabentry->tuples_fetched = tabmsg[i].t_tuples_fetched;
3070                         tabentry->tuples_inserted = tabmsg[i].t_tuples_inserted;
3071                         tabentry->tuples_updated = tabmsg[i].t_tuples_updated;
3072                         tabentry->tuples_deleted = tabmsg[i].t_tuples_deleted;
3073
3074                         tabentry->n_live_tuples = tabmsg[i].t_tuples_inserted;
3075                         tabentry->n_dead_tuples = tabmsg[i].t_tuples_updated +
3076                                 tabmsg[i].t_tuples_deleted;
3077                         tabentry->last_anl_tuples = 0;
3078
3079                         tabentry->blocks_fetched = tabmsg[i].t_blocks_fetched;
3080                         tabentry->blocks_hit = tabmsg[i].t_blocks_hit;
3081
3082                         tabentry->destroy = 0;
3083                 }
3084                 else
3085                 {
3086                         /*
3087                          * Otherwise add the values to the existing entry.
3088                          */
3089                         tabentry->numscans += tabmsg[i].t_numscans;
3090                         tabentry->tuples_returned += tabmsg[i].t_tuples_returned;
3091                         tabentry->tuples_fetched += tabmsg[i].t_tuples_fetched;
3092                         tabentry->tuples_inserted += tabmsg[i].t_tuples_inserted;
3093                         tabentry->tuples_updated += tabmsg[i].t_tuples_updated;
3094                         tabentry->tuples_deleted += tabmsg[i].t_tuples_deleted;
3095
3096                         tabentry->n_live_tuples += tabmsg[i].t_tuples_inserted;
3097                         tabentry->n_dead_tuples += tabmsg[i].t_tuples_updated +
3098                                 tabmsg[i].t_tuples_deleted;
3099
3100                         tabentry->blocks_fetched += tabmsg[i].t_blocks_fetched;
3101                         tabentry->blocks_hit += tabmsg[i].t_blocks_hit;
3102                 }
3103
3104                 /*
3105                  * And add the block IO to the database entry.
3106                  */
3107                 dbentry->n_blocks_fetched += tabmsg[i].t_blocks_fetched;
3108                 dbentry->n_blocks_hit += tabmsg[i].t_blocks_hit;
3109         }
3110 }
3111
3112
3113 /* ----------
3114  * pgstat_recv_tabpurge() -
3115  *
3116  *      Arrange for dead table removal.
3117  * ----------
3118  */
3119 static void
3120 pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len)
3121 {
3122         PgStat_StatDBEntry *dbentry;
3123         PgStat_StatTabEntry *tabentry;
3124         int                     i;
3125
3126         /*
3127          * Make sure the backend is counted for.
3128          */
3129         if (pgstat_add_backend(&msg->m_hdr) < 0)
3130                 return;
3131
3132         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
3133
3134         /*
3135          * No need to purge if we don't even know the database.
3136          */
3137         if (!dbentry || !dbentry->tables)
3138                 return;
3139
3140         /*
3141          * If the database is marked for destroy, this is a delayed UDP packet
3142          * and the tables will go away at DB destruction.
3143          */
3144         if (dbentry->destroy > 0)
3145                 return;
3146
3147         /*
3148          * Process all table entries in the message.
3149          */
3150         for (i = 0; i < msg->m_nentries; i++)
3151         {
3152                 tabentry = (PgStat_StatTabEntry *) hash_search(dbentry->tables,
3153                                                                                    (void *) &(msg->m_tableid[i]),
3154                                                                                                            HASH_FIND, NULL);
3155                 if (tabentry)
3156                         tabentry->destroy = PGSTAT_DESTROY_COUNT;
3157         }
3158 }
3159
3160
3161 /* ----------
3162  * pgstat_recv_dropdb() -
3163  *
3164  *      Arrange for dead database removal
3165  * ----------
3166  */
3167 static void
3168 pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len)
3169 {
3170         PgStat_StatDBEntry *dbentry;
3171
3172         /*
3173          * Make sure the backend is counted for.
3174          */
3175         if (pgstat_add_backend(&msg->m_hdr) < 0)
3176                 return;
3177
3178         /*
3179          * Lookup the database in the hashtable.
3180          */
3181         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
3182
3183         /*
3184          * Mark the database for destruction.
3185          */
3186         if (dbentry)
3187                 dbentry->destroy = PGSTAT_DESTROY_COUNT;
3188 }
3189
3190
3191 /* ----------
3192  * pgstat_recv_resetcounter() -
3193  *
3194  *      Reset the statistics for the specified database.
3195  * ----------
3196  */
3197 static void
3198 pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len)
3199 {
3200         HASHCTL         hash_ctl;
3201         PgStat_StatDBEntry *dbentry;
3202
3203         /*
3204          * Make sure the backend is counted for.
3205          */
3206         if (pgstat_add_backend(&msg->m_hdr) < 0)
3207                 return;
3208
3209         /*
3210          * Lookup the database in the hashtable.  Nothing to do if not there.
3211          */
3212         dbentry = pgstat_get_db_entry(msg->m_databaseid, false);
3213
3214         if (!dbentry)
3215                 return;
3216
3217         /*
3218          * We simply throw away all the database's table entries by
3219          * recreating a new hash table for them.
3220          */
3221         if (dbentry->tables != NULL)
3222                 hash_destroy(dbentry->tables);
3223
3224         dbentry->tables = NULL;
3225         dbentry->n_xact_commit = 0;
3226         dbentry->n_xact_rollback = 0;
3227         dbentry->n_blocks_fetched = 0;
3228         dbentry->n_blocks_hit = 0;
3229         dbentry->destroy = 0;
3230
3231         memset(&hash_ctl, 0, sizeof(hash_ctl));
3232         hash_ctl.keysize = sizeof(Oid);
3233         hash_ctl.entrysize = sizeof(PgStat_StatTabEntry);
3234         hash_ctl.hash = oid_hash;
3235         dbentry->tables = hash_create("Per-database table",
3236                                                                   PGSTAT_TAB_HASH_SIZE,
3237                                                                   &hash_ctl,
3238                                                                   HASH_ELEM | HASH_FUNCTION);
3239 }