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