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