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

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