From bdb1101bc5b2fd0dd3efa6484f08fa4a1bb93b0c Mon Sep 17 00:00:00 2001 From: Andres Freund Date: Wed, 16 Mar 2016 17:28:05 -0700 Subject: [PATCH 5/6] WIP: WaitEvent API --- src/backend/libpq/be-secure.c | 24 +-- src/backend/libpq/pqcomm.c | 12 ++ src/backend/port/unix_latch.c | 317 ++++++++++++++++++++++++++++++++++++++ src/backend/utils/init/miscinit.c | 8 + src/include/libpq/libpq.h | 3 + src/include/storage/latch.h | 44 ++++++ 6 files changed, 396 insertions(+), 12 deletions(-) diff --git a/src/backend/libpq/be-secure.c b/src/backend/libpq/be-secure.c index ac709d1..c396811 100644 --- a/src/backend/libpq/be-secure.c +++ b/src/backend/libpq/be-secure.c @@ -140,13 +140,13 @@ retry: /* In blocking mode, wait until the socket is ready */ if (n < 0 && !port->noblock && (errno == EWOULDBLOCK || errno == EAGAIN)) { - int w; + WaitEvent event; Assert(waitfor); - w = WaitLatchOrSocket(MyLatch, - WL_LATCH_SET | WL_POSTMASTER_DEATH | waitfor, - port->sock, 0); + ModifyWaitEvent(FeBeWaitSet, 0, waitfor, NULL); + + WaitEventSetWait(FeBeWaitSet, 0 /* no timeout */, &event, 1); /* * If the postmaster has died, it's not safe to continue running, @@ -165,13 +165,13 @@ retry: * cycles checking for this very rare condition, and this should cause * us to exit quickly in most cases.) */ - if (w & WL_POSTMASTER_DEATH) + if (event.events & WL_POSTMASTER_DEATH) ereport(FATAL, (errcode(ERRCODE_ADMIN_SHUTDOWN), errmsg("terminating connection due to unexpected postmaster exit"))); /* Handle interrupt. */ - if (w & WL_LATCH_SET) + if (event.events & WL_LATCH_SET) { ResetLatch(MyLatch); ProcessClientReadInterrupt(true); @@ -241,22 +241,22 @@ retry: if (n < 0 && !port->noblock && (errno == EWOULDBLOCK || errno == EAGAIN)) { - int w; + WaitEvent event; Assert(waitfor); - w = WaitLatchOrSocket(MyLatch, - WL_LATCH_SET | WL_POSTMASTER_DEATH | waitfor, - port->sock, 0); + ModifyWaitEvent(FeBeWaitSet, 0, waitfor, NULL); + + WaitEventSetWait(FeBeWaitSet, 0 /* no timeout */, &event, 1); /* See comments in secure_read. */ - if (w & WL_POSTMASTER_DEATH) + if (event.events & WL_POSTMASTER_DEATH) ereport(FATAL, (errcode(ERRCODE_ADMIN_SHUTDOWN), errmsg("terminating connection due to unexpected postmaster exit"))); /* Handle interrupt. */ - if (w & WL_LATCH_SET) + if (event.events & WL_LATCH_SET) { ResetLatch(MyLatch); ProcessClientWriteInterrupt(true); diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c index 71473db..31d646d 100644 --- a/src/backend/libpq/pqcomm.c +++ b/src/backend/libpq/pqcomm.c @@ -201,6 +201,18 @@ pq_init(void) (errmsg("could not set socket to nonblocking mode: %m"))); #endif + { + MemoryContext oldcontext; + + oldcontext = MemoryContextSwitchTo(TopMemoryContext); + + FeBeWaitSet = CreateWaitEventSet(3); + AddWaitEventToSet(FeBeWaitSet, WL_SOCKET_WRITEABLE, MyProcPort->sock, NULL); + AddWaitEventToSet(FeBeWaitSet, WL_LATCH_SET, -1, MyLatch); + AddWaitEventToSet(FeBeWaitSet, WL_POSTMASTER_DEATH, -1, NULL); + + MemoryContextSwitchTo(oldcontext); + } } /* -------------------------------- diff --git a/src/backend/port/unix_latch.c b/src/backend/port/unix_latch.c index f6cb15b..9bcfe14 100644 --- a/src/backend/port/unix_latch.c +++ b/src/backend/port/unix_latch.c @@ -72,6 +72,28 @@ #error "no latch implementation available" #endif +#if defined(WAIT_USE_POLL) || defined(WAIT_USE_SELECT) +/* don't overwrite manual choice */ +#elif defined(HAVE_POLL) +#define WAIT_USE_POLL +#elif HAVE_SYS_SELECT_H +#define WAIT_USE_SELECT +#else +#error "no wait set implementation available" +#endif + +typedef struct WaitEventSet +{ + int nevents; + int nevents_space; + Latch *latch; + int latch_pos; + WaitEvent *events; +#if defined(WAIT_USE_POLL) + struct pollfd *pollfds; +#endif +} WaitEventSet; + /* Are we currently in WaitLatch? The signal handler would like to know. */ static volatile sig_atomic_t waiting = false; @@ -83,6 +105,9 @@ static int selfpipe_writefd = -1; static void sendSelfPipeByte(void); static void drainSelfPipe(void); +#if defined(WAIT_USE_POLL) +static void WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event); +#endif /* * Initialize the process-local latch infrastructure. @@ -636,6 +661,298 @@ ResetLatch(volatile Latch *latch) pg_memory_barrier(); } +WaitEventSet * +CreateWaitEventSet(int nevents) +{ + WaitEventSet *set; + char *data; + Size sz = 0; + + sz += sizeof(WaitEventSet); + sz += sizeof(WaitEvent) * nevents; + +#if defined(LATCH_USE_POLL) + sz += sizeof(struct pollfd) * nevents; +#endif + + data = (char *) palloc0(sz); + + set = (WaitEventSet *) data; + data += sizeof(WaitEventSet); + + set->events = (WaitEvent *) data; + data += sizeof(WaitEvent) * nevents; + +#if defined(WAIT_USE_POLL) + set->pollfds = (struct pollfd *) data; + data += sizeof(struct pollfd) * nevents; +#endif + + set->latch = NULL; + set->nevents_space = nevents; + + return set; +} + +int +AddWaitEventToSet(WaitEventSet *set, uint32 events, int fd, Latch *latch) +{ + WaitEvent *event; + + if (set->nevents_space <= set->nevents) + elog(ERROR, "no space for yet another event"); + + if (set->latch && latch) + elog(ERROR, "can only wait for one latch"); + if (!latch && (events & WL_LATCH_SET)) + elog(ERROR, "cannot wait on latch without latch"); + + /* FIXME: validate event mask */ + + event = &set->events[set->nevents]; + event->pos = set->nevents++; + event->fd = fd; + event->events = events; + + if (events == WL_LATCH_SET) + { + set->latch = latch; + set->latch_pos = event->pos; + event->fd = selfpipe_readfd; + } + else if (events == WL_POSTMASTER_DEATH) + { + event->fd = postmaster_alive_fds[POSTMASTER_FD_WATCH]; + } + +#if defined(WAIT_USE_POLL) + WaitEventAdjustPoll(set, event); +#endif + + return event->pos; +} + +void +ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch) +{ + WaitEvent *event; + + Assert(pos < set->nevents); + + event = &set->events[pos]; + + /* no need to perform any checks/modifications */ + if (events == event->events && !(event->events & WL_LATCH_SET)) + return; + + if (event->events & WL_LATCH_SET && + events != event->events) + { + /* we could allow to disable latch events for a while */ + elog(ERROR, "cannot modify latch event"); + } + if (event->events & WL_POSTMASTER_DEATH && + events != event->events) + { + elog(ERROR, "cannot modify postmaster death event"); + } + + /* FIXME: validate event mask */ + event->events = events; + + if (events == WL_LATCH_SET) + { + set->latch = latch; + } + +#if defined(WAIT_USE_POLL) + WaitEventAdjustPoll(set, event); +#endif +} + +#if defined(WAIT_USE_POLL) +static void +WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event) +{ + struct pollfd *pollfd = &set->pollfds[event->pos]; + + pollfd->revents = 0; + pollfd->fd = event->fd; + + /* prepare pollfd entry once */ + if (event->events == WL_LATCH_SET) + { + Assert(set->latch != NULL); + pollfd->events = POLLIN; + } + else if (event->events == WL_POSTMASTER_DEATH) + { + pollfd->events = POLLIN; + } + else + { + Assert(event->events & (WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE)); + pollfd->events = 0; + if (event->events & WL_SOCKET_READABLE) + pollfd->events |= POLLIN; + if (event->events & WL_SOCKET_WRITEABLE) + pollfd->events |= POLLOUT; + } + + Assert(event->fd >= 0); +} +#endif + +#if defined(WAIT_USE_POLL) +int +WaitEventSetWait(WaitEventSet *set, long timeout, + WaitEvent* occurred_events, int nevents) +{ + int returned_events = 0; + instr_time start_time, + cur_time; + long cur_timeout = -1; + WaitEvent *cur_event; + + struct pollfd *pfds = set->pollfds; + + Assert(nevents > 0); + + if (timeout) + { + INSTR_TIME_SET_CURRENT(start_time); + Assert(timeout >= 0 && timeout <= INT_MAX); + cur_timeout = timeout; + } + + waiting = true; + while (returned_events == 0) + { + int rc; + int pos; + struct pollfd *cur_pollfd; + + /* return immediately if latch is set */ + if (set->latch && set->latch->is_set) + { + occurred_events->fd = -1; + occurred_events->pos = set->latch_pos; + occurred_events->events = WL_LATCH_SET; + occurred_events++; + returned_events++; + + continue; + } + + /* Sleep */ + rc = poll(pfds, set->nevents, (int) cur_timeout); + + /* Check return code */ + if (rc < 0) + { + /* EINTR is okay, otherwise complain */ + if (errno != EINTR) + { + waiting = false; + ereport(ERROR, + (errcode_for_socket_access(), + errmsg("poll() failed: %m"))); + } + continue; + } + else if (rc == 0) + { + break; + } + + for (pos = 0, cur_event = set->events, cur_pollfd = set->pollfds; + pos < set->nevents && returned_events < nevents; + pos++, cur_event++, cur_pollfd++) + { + if (cur_event->events == WL_LATCH_SET && + (cur_pollfd->revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL))) + { + /* There's data in the self-pipe, clear it. */ + drainSelfPipe(); + + if (set->latch->is_set) + { + occurred_events->fd = -1; + occurred_events->pos = cur_event->pos; + occurred_events->events = WL_LATCH_SET; + occurred_events++; + } + } + else if (cur_event->events == WL_POSTMASTER_DEATH && + (cur_pollfd->revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL))) + { + /* + * According to the select(2) man page on Linux, select(2) may + * spuriously return and report a file descriptor as readable, + * when it's not; and presumably so can poll(2). It's not + * clear that the relevant cases would ever apply to the + * postmaster pipe, but since the consequences of falsely + * returning WL_POSTMASTER_DEATH could be pretty unpleasant, + * we take the trouble to positively verify EOF with + * PostmasterIsAlive(). + */ + if (!PostmasterIsAlive()) + { + occurred_events->fd = -1; + occurred_events->pos = cur_event->pos; + occurred_events->events = WL_POSTMASTER_DEATH; + occurred_events++; + returned_events++; + } + } + else if (cur_event->events & (WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE)) + { + Assert(cur_event->fd); + + occurred_events->fd = cur_event->fd; + occurred_events->pos = cur_event->pos; + occurred_events->events = 0; + + if ((cur_event->events & WL_SOCKET_READABLE) && + (cur_pollfd->revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL))) + { + occurred_events->events |= WL_SOCKET_READABLE; + } + + if ((cur_event->events & WL_SOCKET_WRITEABLE) && + (cur_pollfd->revents & (POLLOUT | POLLHUP | POLLERR | POLLNVAL))) + { + occurred_events->events |= WL_SOCKET_WRITEABLE; + occurred_events++; + returned_events++; + } + + if (occurred_events->events != 0) + { + occurred_events++; + returned_events++; + } + } + } + + + if (occurred_events == 0 && timeout != 0) + { + INSTR_TIME_SET_CURRENT(cur_time); + INSTR_TIME_SUBTRACT(cur_time, start_time); + cur_timeout = timeout - (long) INSTR_TIME_GET_MILLISEC(cur_time); + if (cur_timeout <= 0) + goto out; + } + } +out: + waiting = false; + + return returned_events; +} +#elif defined(WAIT_USE_SELECT) +#endif + /* * SetLatch uses SIGUSR1 to wake up the process waiting on the latch. * diff --git a/src/backend/utils/init/miscinit.c b/src/backend/utils/init/miscinit.c index 18f5e6f..d13355b 100644 --- a/src/backend/utils/init/miscinit.c +++ b/src/backend/utils/init/miscinit.c @@ -33,6 +33,7 @@ #include "access/htup_details.h" #include "catalog/pg_authid.h" +#include "libpq/libpq.h" #include "mb/pg_wchar.h" #include "miscadmin.h" #include "postmaster/autovacuum.h" @@ -247,6 +248,9 @@ SwitchToSharedLatch(void) MyLatch = &MyProc->procLatch; + if (FeBeWaitSet) + ModifyWaitEvent(FeBeWaitSet, 1, WL_LATCH_SET, MyLatch); + /* * Set the shared latch as the local one might have been set. This * shouldn't normally be necessary as code is supposed to check the @@ -262,6 +266,10 @@ SwitchBackToLocalLatch(void) Assert(MyProc != NULL && MyLatch == &MyProc->procLatch); MyLatch = &LocalLatchData; + + if (FeBeWaitSet) + ModifyWaitEvent(FeBeWaitSet, 1, WL_LATCH_SET, MyLatch); + SetLatch(MyLatch); } diff --git a/src/include/libpq/libpq.h b/src/include/libpq/libpq.h index 0569994..109fdf7 100644 --- a/src/include/libpq/libpq.h +++ b/src/include/libpq/libpq.h @@ -19,6 +19,7 @@ #include "lib/stringinfo.h" #include "libpq/libpq-be.h" +#include "storage/latch.h" typedef struct @@ -95,6 +96,8 @@ extern ssize_t secure_raw_write(Port *port, const void *ptr, size_t len); extern bool ssl_loaded_verify_locations; +WaitEventSet *FeBeWaitSet; + /* GUCs */ extern char *SSLCipherSuites; extern char *SSLECDHCurve; diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h index e77491e..941e2f0 100644 --- a/src/include/storage/latch.h +++ b/src/include/storage/latch.h @@ -102,6 +102,50 @@ typedef struct Latch #define WL_TIMEOUT (1 << 3) #define WL_POSTMASTER_DEATH (1 << 4) +typedef struct WaitEventSet WaitEventSet; + +typedef struct WaitEvent +{ + int pos; /* position in the event data structure */ + uint32 events; /* tripped events */ + int fd; /* fd associated with event */ +} WaitEvent; + +/* + * Create a WaitEventSet with space for nevents different events to wait for. + * + * latch may be NULL. + */ +extern WaitEventSet *CreateWaitEventSet(int nevents); + +/* --- + * Add an event to the set. Possible events are: + * - WL_LATCH_SET: Wait for the latch to be set + * - WL_POSTMASTER_DEATH: Wait for postmaster to die + * - WL_SOCKET_READABLE: Wait for socket to become readable + * can be combined in one event with WL_SOCKET_WRITEABLE + * - WL_SOCKET_WRITABLE: Wait for socket to become readable + * can be combined with WL_SOCKET_READABLE + * + * Returns the offset in WaitEventSet->events (starting from 0), which can be + * used to modify previously added wait events. + */ +extern int AddWaitEventToSet(WaitEventSet *set, uint32 events, int fd, Latch *latch); + +/* + * Change the event mask and, if applicable, the associated latch of of a + * WaitEvent. + */ +extern void ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch); + +/* + * Wait for events added to the set to happen, or until the timeout is + * reached. At most nevents occurrent events are returned. + * + * Returns the number of events occurred, or 0 if the timeout was reached. + */ +extern int WaitEventSetWait(WaitEventSet *set, long timeout, WaitEvent* occurred_events, int nevents); + /* * prototypes for functions in latch.c */ -- 2.7.0.229.g701fa7f