(unnamed)
text/plain
Index: src/backend/postmaster/pgstat.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/postmaster/pgstat.c,v
retrieving revision 1.118
diff -c -c -r1.118 pgstat.c
*** src/backend/postmaster/pgstat.c 3 Jan 2006 19:54:08 -0000 1.118
--- src/backend/postmaster/pgstat.c 4 Jan 2006 23:06:26 -0000
***************
*** 109,117 ****
* ----------
*/
NON_EXEC_STATIC int pgStatSock = -1;
- NON_EXEC_STATIC int pgStatPipe[2] = {-1, -1};
static struct sockaddr_storage pgStatAddr;
- static pid_t pgStatCollectorPid = 0;
static time_t last_pgstat_start_time;
--- 109,115 ----
***************
*** 166,172 ****
NON_EXEC_STATIC void PgstatBufferMain(int argc, char *argv[]);
NON_EXEC_STATIC void PgstatCollectorMain(int argc, char *argv[]);
static void force_statwrite(SIGNAL_ARGS);
- static void pgstat_recvbuffer(void);
static void pgstat_exit(SIGNAL_ARGS);
static void pgstat_die(SIGNAL_ARGS);
static void pgstat_beshutdown_hook(int code, Datum arg);
--- 164,169 ----
***************
*** 1491,1536 ****
pgstat_parseArgs(argc, argv);
#endif
- /*
- * Start a buffering process to read from the socket, so we have a little
- * more time to process incoming messages.
- *
- * NOTE: the process structure is: postmaster is parent of buffer process
- * is parent of collector process. This way, the buffer can detect
- * collector failure via SIGCHLD, whereas otherwise it wouldn't notice
- * collector failure until it tried to write on the pipe. That would mean
- * that after the postmaster started a new collector, we'd have two buffer
- * processes competing to read from the UDP socket --- not good.
- */
- if (pgpipe(pgStatPipe) < 0)
- ereport(ERROR,
- (errcode_for_socket_access(),
- errmsg("could not create pipe for statistics buffer: %m")));
-
/* child becomes collector process */
! #ifdef EXEC_BACKEND
! pgStatCollectorPid = pgstat_forkexec(STAT_PROC_COLLECTOR);
! #else
! pgStatCollectorPid = fork();
! #endif
! switch (pgStatCollectorPid)
! {
! case -1:
! ereport(ERROR,
! (errmsg("could not fork statistics collector: %m")));
!
! #ifndef EXEC_BACKEND
! case 0:
! /* child becomes collector process */
! PgstatCollectorMain(0, NULL);
! break;
! #endif
!
! default:
! /* parent becomes buffer process */
! closesocket(pgStatPipe[0]);
! pgstat_recvbuffer();
! }
exit(0);
}
--- 1488,1495 ----
pgstat_parseArgs(argc, argv);
#endif
/* child becomes collector process */
! PgstatCollectorMain(0, NULL);
exit(0);
}
***************
*** 1548,1559 ****
PgstatCollectorMain(int argc, char *argv[])
{
PgStat_Msg msg;
- fd_set rfds;
- int readPipe;
- int len = 0;
- struct itimerval timeval;
HASHCTL hash_ctl;
bool need_timer = false;
MyProcPid = getpid(); /* reset MyProcPid */
--- 1507,1517 ----
PgstatCollectorMain(int argc, char *argv[])
{
PgStat_Msg msg;
HASHCTL hash_ctl;
bool need_timer = false;
+ struct itimerval timeval;
+ bool is_block_mode = false;
+ int loops = 0;
MyProcPid = getpid(); /* reset MyProcPid */
***************
*** 1587,1596 ****
pgstat_parseArgs(argc, argv);
#endif
- /* Close unwanted files */
- closesocket(pgStatPipe[1]);
- closesocket(pgStatSock);
-
/*
* Identify myself via ps
*/
--- 1545,1550 ----
***************
*** 1626,1791 ****
pgStatBeTable = (PgStat_StatBeEntry *)
palloc0(sizeof(PgStat_StatBeEntry) * MaxBackends);
- readPipe = pgStatPipe[0];
-
/*
* Process incoming messages and handle all the reporting stuff until
* there are no more messages.
*/
for (;;)
{
if (need_statwrite)
{
! pgstat_write_statsfile();
need_statwrite = false;
need_timer = true;
}
! /*
! * Setup the descriptor set for select(2)
! */
! FD_ZERO(&rfds);
! FD_SET(readPipe, &rfds);
!
! /*
! * Now wait for something to do.
! */
! if (select(readPipe + 1, &rfds, NULL, NULL, NULL) < 0)
{
! if (errno == EINTR)
! continue;
! ereport(ERROR,
! (errcode_for_socket_access(),
! errmsg("select() failed in statistics collector: %m")));
}
! /*
! * Check if there is a new statistics message to collect.
! */
! if (FD_ISSET(readPipe, &rfds))
! {
! /*
! * We may need to issue multiple read calls in case the buffer
! * process didn't write the message in a single write, which is
! * possible since it dumps its buffer bytewise. In any case, we'd
! * need two reads since we don't know the message length
! * initially.
! */
! int nread = 0;
! int targetlen = sizeof(PgStat_MsgHdr); /* initial */
! bool pipeEOF = false;
! while (nread < targetlen)
{
! len = piperead(readPipe, ((char *) &msg) + nread,
! targetlen - nread);
! if (len < 0)
! {
! if (errno == EINTR)
! continue;
! ereport(ERROR,
! (errcode_for_socket_access(),
! errmsg("could not read from statistics collector pipe: %m")));
! }
! if (len == 0) /* EOF on the pipe! */
{
! pipeEOF = true;
! break;
! }
! nread += len;
! if (nread == sizeof(PgStat_MsgHdr))
! {
! /* we have the header, compute actual msg length */
! targetlen = msg.msg_hdr.m_size;
! if (targetlen < (int) sizeof(PgStat_MsgHdr) ||
! targetlen > (int) sizeof(msg))
! {
! /*
! * Bogus message length implies that we got out of
! * sync with the buffer process somehow. Abort so that
! * we can restart both processes.
! */
! ereport(ERROR,
! (errmsg("invalid statistics message length")));
! }
}
}
!
! /*
! * EOF on the pipe implies that the buffer process exited. Fall
! * out of outer loop.
! */
! if (pipeEOF)
! break;
!
! /*
! * Distribute the message to the specific function handling it.
! */
! switch (msg.msg_hdr.m_type)
{
! case PGSTAT_MTYPE_DUMMY:
! break;
! case PGSTAT_MTYPE_BESTART:
! pgstat_recv_bestart((PgStat_MsgBestart *) &msg, nread);
! break;
! case PGSTAT_MTYPE_BETERM:
! pgstat_recv_beterm((PgStat_MsgBeterm *) &msg, nread);
! break;
! case PGSTAT_MTYPE_TABSTAT:
! pgstat_recv_tabstat((PgStat_MsgTabstat *) &msg, nread);
! break;
! case PGSTAT_MTYPE_TABPURGE:
! pgstat_recv_tabpurge((PgStat_MsgTabpurge *) &msg, nread);
! break;
! case PGSTAT_MTYPE_ACTIVITY:
! pgstat_recv_activity((PgStat_MsgActivity *) &msg, nread);
! break;
! case PGSTAT_MTYPE_DROPDB:
! pgstat_recv_dropdb((PgStat_MsgDropdb *) &msg, nread);
! break;
! case PGSTAT_MTYPE_RESETCOUNTER:
! pgstat_recv_resetcounter((PgStat_MsgResetcounter *) &msg,
! nread);
! break;
! case PGSTAT_MTYPE_AUTOVAC_START:
! pgstat_recv_autovac((PgStat_MsgAutovacStart *) &msg, nread);
! break;
! case PGSTAT_MTYPE_VACUUM:
! pgstat_recv_vacuum((PgStat_MsgVacuum *) &msg, nread);
! break;
! case PGSTAT_MTYPE_ANALYZE:
! pgstat_recv_analyze((PgStat_MsgAnalyze *) &msg, nread);
! break;
! default:
! break;
! }
! /*
! * Globally count messages.
! */
! pgStatNumMessages++;
! if (need_timer)
! {
! if (setitimer(ITIMER_REAL, &timeval, NULL))
! ereport(ERROR,
! (errmsg("unable to set statistics collector timer: %m")));
! need_timer = false;
! }
}
/*
* Note that we do NOT check for postmaster exit inside the loop; only
* EOF on the buffer pipe causes us to fall out. This ensures we
* don't exit prematurely if there are still a few messages in the
--- 1580,1704 ----
pgStatBeTable = (PgStat_StatBeEntry *)
palloc0(sizeof(PgStat_StatBeEntry) * MaxBackends);
/*
* Process incoming messages and handle all the reporting stuff until
* there are no more messages.
*/
for (;;)
{
+ int nread;
+
if (need_statwrite)
{
! //pgstat_write_statsfile();
need_statwrite = false;
need_timer = true;
}
! if (need_timer)
{
! if (setitimer(ITIMER_REAL, &timeval, NULL))
! ereport(ERROR,
! (errmsg("unable to set statistics collector timer: %m")));
! need_timer = false;
}
! nread = recv(pgStatSock, (char *) &msg,
! sizeof(PgStat_Msg), 0);
! if (nread > 0 && is_block_mode) /* got data */
! {
! pg_set_noblock(pgStatSock);
! is_block_mode = false;
! }
! else if (nread < 0)
! {
! if (errno == EAGAIN)
{
! if (!is_block_mode)
{
! /* no data, block mode */
! pg_set_block(pgStatSock);
! is_block_mode = true;
}
+ continue;
}
! else if (errno == EINTR)
{
! if (!PostmasterIsAlive(true))
! ereport(ERROR,
! (errmsg("stats collector exited: %m")));
! continue;
! }
! else
! ereport(ERROR,
! (errmsg("stats collector exited: %m")));
! }
! //fprintf(stderr, "nread = %d, type = %d\n", nread, msg.msg_hdr.m_type);
! if (++loops % 1000 == 0)
! fprintf(stderr, "loops = %d\n", loops);
! /*
! * Distribute the message to the specific function handling it.
! */
! switch (msg.msg_hdr.m_type)
! {
! case PGSTAT_MTYPE_DUMMY:
! break;
! case PGSTAT_MTYPE_BESTART:
! pgstat_recv_bestart((PgStat_MsgBestart *) &msg, nread);
! break;
! case PGSTAT_MTYPE_BETERM:
! pgstat_recv_beterm((PgStat_MsgBeterm *) &msg, nread);
! break;
! case PGSTAT_MTYPE_TABSTAT:
! pgstat_recv_tabstat((PgStat_MsgTabstat *) &msg, nread);
! break;
! case PGSTAT_MTYPE_TABPURGE:
! pgstat_recv_tabpurge((PgStat_MsgTabpurge *) &msg, nread);
! break;
! case PGSTAT_MTYPE_ACTIVITY:
! pgstat_recv_activity((PgStat_MsgActivity *) &msg, nread);
! break;
! case PGSTAT_MTYPE_DROPDB:
! pgstat_recv_dropdb((PgStat_MsgDropdb *) &msg, nread);
! break;
! case PGSTAT_MTYPE_RESETCOUNTER:
! pgstat_recv_resetcounter((PgStat_MsgResetcounter *) &msg,
! nread);
! break;
! case PGSTAT_MTYPE_AUTOVAC_START:
! pgstat_recv_autovac((PgStat_MsgAutovacStart *) &msg, nread);
! break;
! case PGSTAT_MTYPE_VACUUM:
! pgstat_recv_vacuum((PgStat_MsgVacuum *) &msg, nread);
! break;
! case PGSTAT_MTYPE_ANALYZE:
! pgstat_recv_analyze((PgStat_MsgAnalyze *) &msg, nread);
! break;
! default:
! break;
}
/*
+ * Globally count messages.
+ */
+ pgStatNumMessages++;
+
+
+ /*
* Note that we do NOT check for postmaster exit inside the loop; only
* EOF on the buffer pipe causes us to fall out. This ensures we
* don't exit prematurely if there are still a few messages in the
***************
*** 1813,2032 ****
}
- /* ----------
- * pgstat_recvbuffer() -
- *
- * This is the body of the separate buffering process. Its only
- * purpose is to receive messages from the UDP socket as fast as
- * possible and forward them over a pipe into the collector itself.
- * If the collector is slow to absorb messages, they are buffered here.
- * ----------
- */
- static void
- pgstat_recvbuffer(void)
- {
- fd_set rfds;
- fd_set wfds;
- struct timeval timeout;
- int writePipe = pgStatPipe[1];
- int maxfd;
- int len;
- int xfr;
- int frm;
- PgStat_Msg input_buffer;
- char *msgbuffer;
- int msg_send = 0; /* next send index in buffer */
- int msg_recv = 0; /* next receive index */
- int msg_have = 0; /* number of bytes stored */
- bool overflow = false;
-
- /*
- * Identify myself via ps
- */
- init_ps_display("stats buffer process", "", "");
- set_ps_display("");
-
- /*
- * We want to die if our child collector process does. There are two ways
- * we might notice that it has died: receive SIGCHLD, or get a write
- * failure on the pipe leading to the child. We can set SIGPIPE to kill
- * us here. Our SIGCHLD handler was already set up before we forked (must
- * do it that way, else it's a race condition).
- */
- pqsignal(SIGPIPE, SIG_DFL);
- PG_SETMASK(&UnBlockSig);
-
- /*
- * Set the write pipe to nonblock mode, so that we cannot block when the
- * collector falls behind.
- */
- if (!pg_set_noblock(writePipe))
- ereport(ERROR,
- (errcode_for_socket_access(),
- errmsg("could not set statistics collector pipe to nonblocking mode: %m")));
-
- /*
- * Allocate the message buffer
- */
- msgbuffer = (char *) palloc(PGSTAT_RECVBUFFERSZ);
-
- /*
- * Loop forever
- */
- for (;;)
- {
- FD_ZERO(&rfds);
- FD_ZERO(&wfds);
- maxfd = -1;
-
- /*
- * As long as we have buffer space we add the socket to the read
- * descriptor set.
- */
- if (msg_have <= (int) (PGSTAT_RECVBUFFERSZ - sizeof(PgStat_Msg)))
- {
- FD_SET(pgStatSock, &rfds);
- maxfd = pgStatSock;
- overflow = false;
- }
- else
- {
- if (!overflow)
- {
- ereport(LOG,
- (errmsg("statistics buffer is full")));
- overflow = true;
- }
- }
-
- /*
- * If we have messages to write out, we add the pipe to the write
- * descriptor set.
- */
- if (msg_have > 0)
- {
- FD_SET(writePipe, &wfds);
- if (writePipe > maxfd)
- maxfd = writePipe;
- }
-
- /*
- * Wait for some work to do; but not for more than 10 seconds. (This
- * determines how quickly we will shut down after an ungraceful
- * postmaster termination; so it needn't be very fast.) struct timeout
- * is modified by some operating systems.
- */
- timeout.tv_sec = 10;
- timeout.tv_usec = 0;
-
- if (select(maxfd + 1, &rfds, &wfds, NULL, &timeout) < 0)
- {
- if (errno == EINTR)
- continue;
- ereport(ERROR,
- (errcode_for_socket_access(),
- errmsg("select() failed in statistics buffer: %m")));
- }
-
- /*
- * If there is a message on the socket, read it and check for
- * validity.
- */
- if (FD_ISSET(pgStatSock, &rfds))
- {
- len = recv(pgStatSock, (char *) &input_buffer,
- sizeof(PgStat_Msg), 0);
- if (len < 0)
- ereport(ERROR,
- (errcode_for_socket_access(),
- errmsg("could not read statistics message: %m")));
-
- /*
- * We ignore messages that are smaller than our common header
- */
- if (len < sizeof(PgStat_MsgHdr))
- continue;
-
- /*
- * The received length must match the length in the header
- */
- if (input_buffer.msg_hdr.m_size != len)
- continue;
-
- /*
- * O.K. - we accept this message. Copy it to the circular
- * msgbuffer.
- */
- frm = 0;
- while (len > 0)
- {
- xfr = PGSTAT_RECVBUFFERSZ - msg_recv;
- if (xfr > len)
- xfr = len;
- Assert(xfr > 0);
- memcpy(msgbuffer + msg_recv,
- ((char *) &input_buffer) + frm,
- xfr);
- msg_recv += xfr;
- if (msg_recv == PGSTAT_RECVBUFFERSZ)
- msg_recv = 0;
- msg_have += xfr;
- frm += xfr;
- len -= xfr;
- }
- }
-
- /*
- * If the collector is ready to receive, write some data into his
- * pipe. We may or may not be able to write all that we have.
- *
- * NOTE: if what we have is less than PIPE_BUF bytes but more than the
- * space available in the pipe buffer, most kernels will refuse to
- * write any of it, and will return EAGAIN. This means we will
- * busy-loop until the situation changes (either because the collector
- * caught up, or because more data arrives so that we have more than
- * PIPE_BUF bytes buffered). This is not good, but is there any way
- * around it? We have no way to tell when the collector has caught
- * up...
- */
- if (FD_ISSET(writePipe, &wfds))
- {
- xfr = PGSTAT_RECVBUFFERSZ - msg_send;
- if (xfr > msg_have)
- xfr = msg_have;
- Assert(xfr > 0);
- len = pipewrite(writePipe, msgbuffer + msg_send, xfr);
- if (len < 0)
- {
- if (errno == EINTR || errno == EAGAIN)
- continue; /* not enough space in pipe */
- ereport(ERROR,
- (errcode_for_socket_access(),
- errmsg("could not write to statistics collector pipe: %m")));
- }
- /* NB: len < xfr is okay */
- msg_send += len;
- if (msg_send == PGSTAT_RECVBUFFERSZ)
- msg_send = 0;
- msg_have -= len;
- }
-
- /*
- * Make sure we forwarded all messages before we check for postmaster
- * termination.
- */
- if (msg_have != 0 || FD_ISSET(pgStatSock, &rfds))
- continue;
-
- /*
- * If the postmaster has terminated, we die too. (This is no longer
- * the normal exit path, however.)
- */
- if (!PostmasterIsAlive(true))
- exit(0);
- }
- }
-
/* SIGQUIT signal handler for buffer process */
static void
pgstat_exit(SIGNAL_ARGS)
--- 1726,1731 ----
***************
*** 2049,2054 ****
--- 1748,1754 ----
exit(0);
}
+
/* SIGCHLD signal handler for buffer process */
static void
pgstat_die(SIGNAL_ARGS)