granicus.if.org Git - postgresql/blob - src/backend/postmaster/pgstat.c

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