waiteventset.c File Reference

#include "postgres.h"
#include <fcntl.h>
#include <limits.h>
#include <signal.h>
#include <unistd.h>
#include <poll.h>
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "port/atomics.h"
#include "portability/instr_time.h"
#include "postmaster/postmaster.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/pmsignal.h"
#include "storage/latch.h"
#include "storage/waiteventset.h"
#include "utils/memutils.h"
#include "utils/resowner.h"

Include dependency graph for waiteventset.c:

Go to the source code of this file.

Data Structures
struct	WaitEventSet

Macros
#define	WAIT_USE_POLL
#define	WAIT_USE_SELF_PIPE

Functions
static void	latch_sigurg_handler (SIGNAL_ARGS)
static void	sendSelfPipeByte (void)
static void	drain (void)
static void	WaitEventAdjustPoll (WaitEventSet *set, WaitEvent *event)
static int	WaitEventSetWaitBlock (WaitEventSet *set, int cur_timeout, WaitEvent *occurred_events, int nevents)
static void	ResOwnerReleaseWaitEventSet (Datum res)
static void	ResourceOwnerRememberWaitEventSet (ResourceOwner owner, WaitEventSet *set)
static void	ResourceOwnerForgetWaitEventSet (ResourceOwner owner, WaitEventSet *set)
void	InitializeWaitEventSupport (void)
WaitEventSet *	CreateWaitEventSet (ResourceOwner resowner, int nevents)
void	FreeWaitEventSet (WaitEventSet *set)
void	FreeWaitEventSetAfterFork (WaitEventSet *set)
int	AddWaitEventToSet (WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch, void *user_data)
void	ModifyWaitEvent (WaitEventSet *set, int pos, uint32 events, Latch *latch)
int	WaitEventSetWait (WaitEventSet *set, long timeout, WaitEvent *occurred_events, int nevents, uint32 wait_event_info)
bool	WaitEventSetCanReportClosed (void)
int	GetNumRegisteredWaitEvents (WaitEventSet *set)
void	WakeupMyProc (void)
void	WakeupOtherProc (int pid)

Variables
static volatile sig_atomic_t	waiting = false
static int	selfpipe_readfd = -1
static int	selfpipe_writefd = -1
static int	selfpipe_owner_pid = 0
static const ResourceOwnerDesc	wait_event_set_resowner_desc

Macro Definition Documentation

WAIT_USE_POLL

#define WAIT_USE_POLL

Definition at line 95 of file waiteventset.c.

WAIT_USE_SELF_PIPE

#define WAIT_USE_SELF_PIPE

Definition at line 112 of file waiteventset.c.

Function Documentation

AddWaitEventToSet()

int AddWaitEventToSet	(	WaitEventSet *	set,
		uint32	events,
		pgsocket	fd,
		Latch *	latch,
		void *	user_data
	)

Definition at line 570 of file waiteventset.c.

{
    WaitEvent  *event;
 
    /* not enough space */
    Assert(set->nevents < set->nevents_space);
 
    if (events == WL_EXIT_ON_PM_DEATH)
    {
        events = WL_POSTMASTER_DEATH;
        set->exit_on_postmaster_death = true;
    }
 
    if (latch)
    {
        if (latch->owner_pid != MyProcPid)
            elog(ERROR, "cannot wait on a latch owned by another process");
        if (set->latch)
            elog(ERROR, "cannot wait on more than one latch");
        if ((events & WL_LATCH_SET) != WL_LATCH_SET)
            elog(ERROR, "latch events only support being set");
    }
    else
    {
        if (events & WL_LATCH_SET)
            elog(ERROR, "cannot wait on latch without a specified latch");
    }
 
    /* waiting for socket readiness without a socket indicates a bug */
    if (fd == PGINVALID_SOCKET && (events & WL_SOCKET_MASK))
        elog(ERROR, "cannot wait on socket event without a socket");
 
    event = &set->events[set->nevents];
    event->pos = set->nevents++;
    event->fd = fd;
    event->events = events;
    event->user_data = user_data;
#ifdef WIN32
    event->reset = false;
#endif
 
    if (events == WL_LATCH_SET)
    {
        set->latch = latch;
        set->latch_pos = event->pos;
#if defined(WAIT_USE_SELF_PIPE)
        event->fd = selfpipe_readfd;
#elif defined(WAIT_USE_SIGNALFD)
        event->fd = signal_fd;
#else
        event->fd = PGINVALID_SOCKET;
#ifdef WAIT_USE_EPOLL
        return event->pos;
#endif
#endif
    }
    else if (events == WL_POSTMASTER_DEATH)
    {
#ifndef WIN32
        event->fd = postmaster_alive_fds[POSTMASTER_FD_WATCH];
#endif
    }
 
    /* perform wait primitive specific initialization, if needed */
#if defined(WAIT_USE_EPOLL)
    WaitEventAdjustEpoll(set, event, EPOLL_CTL_ADD);
#elif defined(WAIT_USE_KQUEUE)
    WaitEventAdjustKqueue(set, event, 0);
#elif defined(WAIT_USE_POLL)
    WaitEventAdjustPoll(set, event);
#elif defined(WAIT_USE_WIN32)
    WaitEventAdjustWin32(set, event);
#endif
 
    return event->pos;
}

References Assert(), elog, ERROR, WaitEventSet::events, WaitEventSet::exit_on_postmaster_death, fd(), WaitEventSet::latch, WaitEventSet::latch_pos, MyProcPid, WaitEventSet::nevents, WaitEventSet::nevents_space, Latch::owner_pid, PGINVALID_SOCKET, WaitEvent::pos, postmaster_alive_fds, POSTMASTER_FD_WATCH, selfpipe_readfd, WaitEventAdjustPoll(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_POSTMASTER_DEATH, and WL_SOCKET_MASK.

Referenced by ConfigurePostmasterWaitSet(), ExecAppendAsyncEventWait(), InitializeLatchWaitSet(), postgresForeignAsyncConfigureWait(), pq_init(), SysLoggerMain(), and WaitLatchOrSocket().

CreateWaitEventSet()

WaitEventSet * CreateWaitEventSet	(	ResourceOwner	resowner,
		int	nevents
	)

Definition at line 364 of file waiteventset.c.

{
    WaitEventSet *set;
    char       *data;
    Size        sz = 0;
 
    /*
     * Use MAXALIGN size/alignment to guarantee that later uses of memory are
     * aligned correctly. E.g. epoll_event might need 8 byte alignment on some
     * platforms, but earlier allocations like WaitEventSet and WaitEvent
     * might not be sized to guarantee that when purely using sizeof().
     */
    sz += MAXALIGN(sizeof(WaitEventSet));
    sz += MAXALIGN(sizeof(WaitEvent) * nevents);
 
#if defined(WAIT_USE_EPOLL)
    sz += MAXALIGN(sizeof(struct epoll_event) * nevents);
#elif defined(WAIT_USE_KQUEUE)
    sz += MAXALIGN(sizeof(struct kevent) * nevents);
#elif defined(WAIT_USE_POLL)
    sz += MAXALIGN(sizeof(struct pollfd) * nevents);
#elif defined(WAIT_USE_WIN32)
    /* need space for the pgwin32_signal_event */
    sz += MAXALIGN(sizeof(HANDLE) * (nevents + 1));
#endif
 
    if (resowner != NULL)
        ResourceOwnerEnlarge(resowner);
 
    data = (char *) MemoryContextAllocZero(TopMemoryContext, sz);
 
    set = (WaitEventSet *) data;
    data += MAXALIGN(sizeof(WaitEventSet));
 
    set->events = (WaitEvent *) data;
    data += MAXALIGN(sizeof(WaitEvent) * nevents);
 
#if defined(WAIT_USE_EPOLL)
    set->epoll_ret_events = (struct epoll_event *) data;
    data += MAXALIGN(sizeof(struct epoll_event) * nevents);
#elif defined(WAIT_USE_KQUEUE)
    set->kqueue_ret_events = (struct kevent *) data;
    data += MAXALIGN(sizeof(struct kevent) * nevents);
#elif defined(WAIT_USE_POLL)
    set->pollfds = (struct pollfd *) data;
    data += MAXALIGN(sizeof(struct pollfd) * nevents);
#elif defined(WAIT_USE_WIN32)
    set->handles = (HANDLE) data;
    data += MAXALIGN(sizeof(HANDLE) * nevents);
#endif
 
    set->latch = NULL;
    set->nevents_space = nevents;
    set->exit_on_postmaster_death = false;
 
    if (resowner != NULL)
    {
        ResourceOwnerRememberWaitEventSet(resowner, set);
        set->owner = resowner;
    }
 
#if defined(WAIT_USE_EPOLL)
    if (!AcquireExternalFD())
        elog(ERROR, "AcquireExternalFD, for epoll_create1, failed: %m");
    set->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
    if (set->epoll_fd < 0)
    {
        ReleaseExternalFD();
        elog(ERROR, "epoll_create1 failed: %m");
    }
#elif defined(WAIT_USE_KQUEUE)
    if (!AcquireExternalFD())
        elog(ERROR, "AcquireExternalFD, for kqueue, failed: %m");
    set->kqueue_fd = kqueue();
    if (set->kqueue_fd < 0)
    {
        ReleaseExternalFD();
        elog(ERROR, "kqueue failed: %m");
    }
    if (fcntl(set->kqueue_fd, F_SETFD, FD_CLOEXEC) == -1)
    {
        int         save_errno = errno;
 
        close(set->kqueue_fd);
        ReleaseExternalFD();
        errno = save_errno;
        elog(ERROR, "fcntl(F_SETFD) failed on kqueue descriptor: %m");
    }
    set->report_postmaster_not_running = false;
#elif defined(WAIT_USE_WIN32)
 
    /*
     * To handle signals while waiting, we need to add a win32 specific event.
     * We accounted for the additional event at the top of this routine. See
     * port/win32/signal.c for more details.
     *
     * Note: pgwin32_signal_event should be first to ensure that it will be
     * reported when multiple events are set.  We want to guarantee that
     * pending signals are serviced.
     */
    set->handles[0] = pgwin32_signal_event;
    StaticAssertStmt(WSA_INVALID_EVENT == NULL, "");
#endif
 
    return set;
}

References AcquireExternalFD(), close, data, elog, ERROR, WaitEventSet::events, WaitEventSet::exit_on_postmaster_death, WaitEventSet::latch, MAXALIGN, MemoryContextAllocZero(), WaitEventSet::nevents_space, WaitEventSet::owner, pgwin32_signal_event, WaitEventSet::pollfds, ReleaseExternalFD(), ResourceOwnerEnlarge(), ResourceOwnerRememberWaitEventSet(), StaticAssertStmt, and TopMemoryContext.

Referenced by ConfigurePostmasterWaitSet(), ExecAppendAsyncEventWait(), InitializeLatchWaitSet(), pq_init(), SysLoggerMain(), and WaitLatchOrSocket().

drain()

static void drain ( void  )

static

Definition at line 1945 of file waiteventset.c.

{
    char        buf[1024];
    int         rc;
    int         fd;
 
#ifdef WAIT_USE_SELF_PIPE
    fd = selfpipe_readfd;
#else
    fd = signal_fd;
#endif
 
    for (;;)
    {
        rc = read(fd, buf, sizeof(buf));
        if (rc < 0)
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                break;          /* the descriptor is empty */
            else if (errno == EINTR)
                continue;       /* retry */
            else
            {
                waiting = false;
#ifdef WAIT_USE_SELF_PIPE
                elog(ERROR, "read() on self-pipe failed: %m");
#else
                elog(ERROR, "read() on signalfd failed: %m");
#endif
            }
        }
        else if (rc == 0)
        {
            waiting = false;
#ifdef WAIT_USE_SELF_PIPE
            elog(ERROR, "unexpected EOF on self-pipe");
#else
            elog(ERROR, "unexpected EOF on signalfd");
#endif
        }
        else if (rc < sizeof(buf))
        {
            /* we successfully drained the pipe; no need to read() again */
            break;
        }
        /* else buffer wasn't big enough, so read again */
    }
}

References buf, EAGAIN, EINTR, elog, ERROR, EWOULDBLOCK, fd(), read, selfpipe_readfd, and waiting.

Referenced by WaitEventSetWaitBlock().

FreeWaitEventSet()

void FreeWaitEventSet

(

WaitEventSet *

set

)

Definition at line 481 of file waiteventset.c.

{
    if (set->owner)
    {
        ResourceOwnerForgetWaitEventSet(set->owner, set);
        set->owner = NULL;
    }
 
#if defined(WAIT_USE_EPOLL)
    close(set->epoll_fd);
    ReleaseExternalFD();
#elif defined(WAIT_USE_KQUEUE)
    close(set->kqueue_fd);
    ReleaseExternalFD();
#elif defined(WAIT_USE_WIN32)
    for (WaitEvent *cur_event = set->events;
         cur_event < (set->events + set->nevents);
         cur_event++)
    {
        if (cur_event->events & WL_LATCH_SET)
        {
            /* uses the latch's HANDLE */
        }
        else if (cur_event->events & WL_POSTMASTER_DEATH)
        {
            /* uses PostmasterHandle */
        }
        else
        {
            /* Clean up the event object we created for the socket */
            WSAEventSelect(cur_event->fd, NULL, 0);
            WSACloseEvent(set->handles[cur_event->pos + 1]);
        }
    }
#endif
 
    pfree(set);
}

References close, WaitEventSet::events, WaitEventSet::nevents, WaitEventSet::owner, pfree(), ReleaseExternalFD(), ResourceOwnerForgetWaitEventSet(), WL_LATCH_SET, and WL_POSTMASTER_DEATH.

Referenced by ConfigurePostmasterWaitSet(), ExecAppendAsyncEventWait(), ResOwnerReleaseWaitEventSet(), and WaitLatchOrSocket().

FreeWaitEventSetAfterFork()

void FreeWaitEventSetAfterFork

(

WaitEventSet *

set

)

Definition at line 524 of file waiteventset.c.

{
#if defined(WAIT_USE_EPOLL)
    close(set->epoll_fd);
    ReleaseExternalFD();
#elif defined(WAIT_USE_KQUEUE)
    /* kqueues are not normally inherited by child processes */
    ReleaseExternalFD();
#endif
 
    pfree(set);
}

References close, pfree(), and ReleaseExternalFD().

Referenced by ClosePostmasterPorts().

GetNumRegisteredWaitEvents()

int GetNumRegisteredWaitEvents

(

WaitEventSet *

set

)

Definition at line 1883 of file waiteventset.c.

{
    return set->nevents;
}

References WaitEventSet::nevents.

Referenced by ExecAppendAsyncEventWait(), and postgresForeignAsyncConfigureWait().

InitializeWaitEventSupport()

void InitializeWaitEventSupport

(

void

)

Definition at line 241 of file waiteventset.c.

{
#if defined(WAIT_USE_SELF_PIPE)
    int         pipefd[2];
 
    if (IsUnderPostmaster)
    {
        /*
         * We might have inherited connections to a self-pipe created by the
         * postmaster.  It's critical that child processes create their own
         * self-pipes, of course, and we really want them to close the
         * inherited FDs for safety's sake.
         */
        if (selfpipe_owner_pid != 0)
        {
            /* Assert we go through here but once in a child process */
            Assert(selfpipe_owner_pid != MyProcPid);
            /* Release postmaster's pipe FDs; ignore any error */
            (void) close(selfpipe_readfd);
            (void) close(selfpipe_writefd);
            /* Clean up, just for safety's sake; we'll set these below */
            selfpipe_readfd = selfpipe_writefd = -1;
            selfpipe_owner_pid = 0;
            /* Keep fd.c's accounting straight */
            ReleaseExternalFD();
            ReleaseExternalFD();
        }
        else
        {
            /*
             * Postmaster didn't create a self-pipe ... or else we're in an
             * EXEC_BACKEND build, in which case it doesn't matter since the
             * postmaster's pipe FDs were closed by the action of FD_CLOEXEC.
             * fd.c won't have state to clean up, either.
             */
            Assert(selfpipe_readfd == -1);
        }
    }
    else
    {
        /* In postmaster or standalone backend, assert we do this but once */
        Assert(selfpipe_readfd == -1);
        Assert(selfpipe_owner_pid == 0);
    }
 
    /*
     * Set up the self-pipe that allows a signal handler to wake up the
     * poll()/epoll_wait() in WaitLatch. Make the write-end non-blocking, so
     * that SetLatch won't block if the event has already been set many times
     * filling the kernel buffer. Make the read-end non-blocking too, so that
     * we can easily clear the pipe by reading until EAGAIN or EWOULDBLOCK.
     * Also, make both FDs close-on-exec, since we surely do not want any
     * child processes messing with them.
     */
    if (pipe(pipefd) < 0)
        elog(FATAL, "pipe() failed: %m");
    if (fcntl(pipefd[0], F_SETFL, O_NONBLOCK) == -1)
        elog(FATAL, "fcntl(F_SETFL) failed on read-end of self-pipe: %m");
    if (fcntl(pipefd[1], F_SETFL, O_NONBLOCK) == -1)
        elog(FATAL, "fcntl(F_SETFL) failed on write-end of self-pipe: %m");
    if (fcntl(pipefd[0], F_SETFD, FD_CLOEXEC) == -1)
        elog(FATAL, "fcntl(F_SETFD) failed on read-end of self-pipe: %m");
    if (fcntl(pipefd[1], F_SETFD, FD_CLOEXEC) == -1)
        elog(FATAL, "fcntl(F_SETFD) failed on write-end of self-pipe: %m");
 
    selfpipe_readfd = pipefd[0];
    selfpipe_writefd = pipefd[1];
    selfpipe_owner_pid = MyProcPid;
 
    /* Tell fd.c about these two long-lived FDs */
    ReserveExternalFD();
    ReserveExternalFD();
 
    pqsignal(SIGURG, latch_sigurg_handler);
#endif
 
#ifdef WAIT_USE_SIGNALFD
    sigset_t    signalfd_mask;
 
    if (IsUnderPostmaster)
    {
        /*
         * It would probably be safe to re-use the inherited signalfd since
         * signalfds only see the current process's pending signals, but it
         * seems less surprising to close it and create our own.
         */
        if (signal_fd != -1)
        {
            /* Release postmaster's signal FD; ignore any error */
            (void) close(signal_fd);
            signal_fd = -1;
            ReleaseExternalFD();
        }
    }
 
    /* Block SIGURG, because we'll receive it through a signalfd. */
    sigaddset(&UnBlockSig, SIGURG);
 
    /* Set up the signalfd to receive SIGURG notifications. */
    sigemptyset(&signalfd_mask);
    sigaddset(&signalfd_mask, SIGURG);
    signal_fd = signalfd(-1, &signalfd_mask, SFD_NONBLOCK | SFD_CLOEXEC);
    if (signal_fd < 0)
        elog(FATAL, "signalfd() failed");
    ReserveExternalFD();
#endif
 
#ifdef WAIT_USE_KQUEUE
    /* Ignore SIGURG, because we'll receive it via kqueue. */
    pqsignal(SIGURG, SIG_IGN);
#endif
}

References Assert(), close, elog, FATAL, IsUnderPostmaster, latch_sigurg_handler(), MyProcPid, pqsignal, ReleaseExternalFD(), ReserveExternalFD(), selfpipe_owner_pid, selfpipe_readfd, selfpipe_writefd, and UnBlockSig.

Referenced by InitPostmasterChild(), InitStandaloneProcess(), and PostmasterMain().

latch_sigurg_handler()

static void latch_sigurg_handler ( SIGNAL_ARGS  )

static

Definition at line 1896 of file waiteventset.c.

{
    if (waiting)
        sendSelfPipeByte();
}

References sendSelfPipeByte(), and waiting.

Referenced by InitializeWaitEventSupport().

ModifyWaitEvent()

void ModifyWaitEvent	(	WaitEventSet *	set,
		int	pos,
		uint32	events,
		Latch *	latch
	)

Definition at line 656 of file waiteventset.c.

{
    WaitEvent  *event;
#if defined(WAIT_USE_KQUEUE)
    int         old_events;
#endif
 
    Assert(pos < set->nevents);
 
    event = &set->events[pos];
#if defined(WAIT_USE_KQUEUE)
    old_events = event->events;
#endif
 
    /*
     * Allow switching between WL_POSTMASTER_DEATH and WL_EXIT_ON_PM_DEATH.
     *
     * Note that because WL_EXIT_ON_PM_DEATH is mapped to WL_POSTMASTER_DEATH
     * in AddWaitEventToSet(), this needs to be checked before the fast-path
     * below that checks if 'events' has changed.
     */
    if (event->events == WL_POSTMASTER_DEATH)
    {
        if (events != WL_POSTMASTER_DEATH && events != WL_EXIT_ON_PM_DEATH)
            elog(ERROR, "cannot remove postmaster death event");
        set->exit_on_postmaster_death = ((events & WL_EXIT_ON_PM_DEATH) != 0);
        return;
    }
 
    /*
     * If neither the event mask nor the associated latch changes, return
     * early. That's an important optimization for some sockets, where
     * ModifyWaitEvent is frequently used to switch from waiting for reads to
     * waiting on writes.
     */
    if (events == event->events &&
        (!(event->events & WL_LATCH_SET) || set->latch == latch))
        return;
 
    if (event->events & WL_LATCH_SET && events != event->events)
        elog(ERROR, "cannot modify latch event");
 
    /* FIXME: validate event mask */
    event->events = events;
 
    if (events == WL_LATCH_SET)
    {
        if (latch && latch->owner_pid != MyProcPid)
            elog(ERROR, "cannot wait on a latch owned by another process");
        set->latch = latch;
 
        /*
         * On Unix, we don't need to modify the kernel object because the
         * underlying pipe (if there is one) is the same for all latches so we
         * can return immediately.  On Windows, we need to update our array of
         * handles, but we leave the old one in place and tolerate spurious
         * wakeups if the latch is disabled.
         */
#if defined(WAIT_USE_WIN32)
        if (!latch)
            return;
#else
        return;
#endif
    }
 
#if defined(WAIT_USE_EPOLL)
    WaitEventAdjustEpoll(set, event, EPOLL_CTL_MOD);
#elif defined(WAIT_USE_KQUEUE)
    WaitEventAdjustKqueue(set, event, old_events);
#elif defined(WAIT_USE_POLL)
    WaitEventAdjustPoll(set, event);
#elif defined(WAIT_USE_WIN32)
    WaitEventAdjustWin32(set, event);
#endif
}

References Assert(), elog, ERROR, WaitEventSet::events, WaitEvent::events, WaitEventSet::exit_on_postmaster_death, WaitEventSet::latch, MyProcPid, Latch::owner_pid, WaitEventAdjustPoll(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_POSTMASTER_DEATH.

Referenced by pq_check_connection(), secure_read(), secure_write(), SwitchBackToLocalLatch(), SwitchToSharedLatch(), WaitLatch(), and WalSndWait().

ResourceOwnerForgetWaitEventSet()

static void ResourceOwnerForgetWaitEventSet ( ResourceOwner  owner, WaitEventSet *  set  )

inlinestatic

Definition at line 228 of file waiteventset.c.

{
    ResourceOwnerForget(owner, PointerGetDatum(set), &wait_event_set_resowner_desc);
}

References PointerGetDatum(), ResourceOwnerForget(), and wait_event_set_resowner_desc.

Referenced by FreeWaitEventSet().

ResourceOwnerRememberWaitEventSet()

static void ResourceOwnerRememberWaitEventSet ( ResourceOwner  owner, WaitEventSet *  set  )

inlinestatic

Definition at line 223 of file waiteventset.c.

{
    ResourceOwnerRemember(owner, PointerGetDatum(set), &wait_event_set_resowner_desc);
}

References PointerGetDatum(), ResourceOwnerRemember(), and wait_event_set_resowner_desc.

Referenced by CreateWaitEventSet().

ResOwnerReleaseWaitEventSet()

static void ResOwnerReleaseWaitEventSet ( Datum  res )

static

Definition at line 1997 of file waiteventset.c.

{
    WaitEventSet *set = (WaitEventSet *) DatumGetPointer(res);
 
    Assert(set->owner != NULL);
    set->owner = NULL;
    FreeWaitEventSet(set);
}

References Assert(), DatumGetPointer(), FreeWaitEventSet(), and WaitEventSet::owner.

sendSelfPipeByte()

static void sendSelfPipeByte ( void  )

static

Definition at line 1904 of file waiteventset.c.

{
    int         rc;
    char        dummy = 0;
 
retry:
    rc = write(selfpipe_writefd, &dummy, 1);
    if (rc < 0)
    {
        /* If interrupted by signal, just retry */
        if (errno == EINTR)
            goto retry;
 
        /*
         * If the pipe is full, we don't need to retry, the data that's there
         * already is enough to wake up WaitLatch.
         */
        if (errno == EAGAIN || errno == EWOULDBLOCK)
            return;
 
        /*
         * Oops, the write() failed for some other reason. We might be in a
         * signal handler, so it's not safe to elog(). We have no choice but
         * silently ignore the error.
         */
        return;
    }
}

References EAGAIN, EINTR, EWOULDBLOCK, selfpipe_writefd, and write.

Referenced by latch_sigurg_handler(), and WakeupMyProc().

WaitEventAdjustPoll()

static void WaitEventAdjustPoll ( WaitEventSet *  set, WaitEvent *  event  )

static

Definition at line 790 of file waiteventset.c.

{
    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 |
                                WL_SOCKET_CLOSED));
        pollfd->events = 0;
        if (event->events & WL_SOCKET_READABLE)
            pollfd->events |= POLLIN;
        if (event->events & WL_SOCKET_WRITEABLE)
            pollfd->events |= POLLOUT;
#ifdef POLLRDHUP
        if (event->events & WL_SOCKET_CLOSED)
            pollfd->events |= POLLRDHUP;
#endif
    }
 
    Assert(event->fd != PGINVALID_SOCKET);
}

References Assert(), WaitEvent::events, WaitEvent::fd, WaitEventSet::latch, PGINVALID_SOCKET, WaitEventSet::pollfds, WL_LATCH_SET, WL_POSTMASTER_DEATH, WL_SOCKET_CLOSED, WL_SOCKET_READABLE, and WL_SOCKET_WRITEABLE.

Referenced by AddWaitEventToSet(), and ModifyWaitEvent().

WaitEventSetCanReportClosed()

bool WaitEventSetCanReportClosed

(

void

)

Definition at line 1868 of file waiteventset.c.

{
#if (defined(WAIT_USE_POLL) && defined(POLLRDHUP)) || \
    defined(WAIT_USE_EPOLL) || \
    defined(WAIT_USE_KQUEUE)
    return true;
#else
    return false;
#endif
}

Referenced by check_client_connection_check_interval().

WaitEventSetWait()

int WaitEventSetWait	(	WaitEventSet *	set,
		long	timeout,
		WaitEvent *	occurred_events,
		int	nevents,
		uint32	wait_event_info
	)

Definition at line 1038 of file waiteventset.c.

{
    int         returned_events = 0;
    instr_time  start_time;
    instr_time  cur_time;
    long        cur_timeout = -1;
 
    Assert(nevents > 0);
 
    /*
     * Initialize timeout if requested.  We must record the current time so
     * that we can determine the remaining timeout if interrupted.
     */
    if (timeout >= 0)
    {
        INSTR_TIME_SET_CURRENT(start_time);
        Assert(timeout >= 0 && timeout <= INT_MAX);
        cur_timeout = timeout;
    }
    else
        INSTR_TIME_SET_ZERO(start_time);
 
    pgstat_report_wait_start(wait_event_info);
 
#ifndef WIN32
    waiting = true;
#else
    /* Ensure that signals are serviced even if latch is already set */
    pgwin32_dispatch_queued_signals();
#endif
    while (returned_events == 0)
    {
        int         rc;
 
        /*
         * Check if the latch is set already first.  If so, we either exit
         * immediately or ask the kernel for further events available right
         * now without waiting, depending on how many events the caller wants.
         *
         * If someone sets the latch between this and the
         * WaitEventSetWaitBlock() below, the setter will write a byte to the
         * pipe (or signal us and the signal handler will do that), and the
         * readiness routine will return immediately.
         *
         * On unix, If there's a pending byte in the self pipe, we'll notice
         * whenever blocking. Only clearing the pipe in that case avoids
         * having to drain it every time WaitLatchOrSocket() is used. Should
         * the pipe-buffer fill up we're still ok, because the pipe is in
         * nonblocking mode. It's unlikely for that to happen, because the
         * self pipe isn't filled unless we're blocking (waiting = true), or
         * from inside a signal handler in latch_sigurg_handler().
         *
         * On windows, we'll also notice if there's a pending event for the
         * latch when blocking, but there's no danger of anything filling up,
         * as "Setting an event that is already set has no effect.".
         *
         * Note: we assume that the kernel calls involved in latch management
         * will provide adequate synchronization on machines with weak memory
         * ordering, so that we cannot miss seeing is_set if a notification
         * has already been queued.
         */
        if (set->latch && !set->latch->is_set)
        {
            /* about to sleep on a latch */
            set->latch->maybe_sleeping = true;
            pg_memory_barrier();
            /* and recheck */
        }
 
        if (set->latch && set->latch->is_set)
        {
            occurred_events->fd = PGINVALID_SOCKET;
            occurred_events->pos = set->latch_pos;
            occurred_events->user_data =
                set->events[set->latch_pos].user_data;
            occurred_events->events = WL_LATCH_SET;
            occurred_events++;
            returned_events++;
 
            /* could have been set above */
            set->latch->maybe_sleeping = false;
 
            if (returned_events == nevents)
                break;          /* output buffer full already */
 
            /*
             * Even though we already have an event, we'll poll just once with
             * zero timeout to see what non-latch events we can fit into the
             * output buffer at the same time.
             */
            cur_timeout = 0;
            timeout = 0;
        }
 
        /*
         * Wait for events using the readiness primitive chosen at the top of
         * this file. If -1 is returned, a timeout has occurred, if 0 we have
         * to retry, everything >= 1 is the number of returned events.
         */
        rc = WaitEventSetWaitBlock(set, cur_timeout,
                                   occurred_events, nevents - returned_events);
 
        if (set->latch &&
            set->latch->maybe_sleeping)
            set->latch->maybe_sleeping = false;
 
        if (rc == -1)
            break;              /* timeout occurred */
        else
            returned_events += rc;
 
        /* If we're not done, update cur_timeout for next iteration */
        if (returned_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)
                break;
        }
    }
#ifndef WIN32
    waiting = false;
#endif
 
    pgstat_report_wait_end();
 
    return returned_events;
}

References Assert(), WaitEventSet::events, WaitEvent::events, WaitEvent::fd, INSTR_TIME_GET_MILLISEC, INSTR_TIME_SET_CURRENT, INSTR_TIME_SET_ZERO, INSTR_TIME_SUBTRACT, Latch::is_set, WaitEventSet::latch, WaitEventSet::latch_pos, Latch::maybe_sleeping, pg_memory_barrier, PGINVALID_SOCKET, pgstat_report_wait_end(), pgstat_report_wait_start(), pgwin32_dispatch_queued_signals(), WaitEvent::pos, start_time, WaitEvent::user_data, WaitEventSetWaitBlock(), waiting, and WL_LATCH_SET.

Referenced by ExecAppendAsyncEventWait(), pq_check_connection(), secure_read(), secure_write(), ServerLoop(), SysLoggerMain(), WaitLatch(), WaitLatchOrSocket(), and WalSndWait().

WaitEventSetWaitBlock()

static int WaitEventSetWaitBlock ( WaitEventSet *  set, int  cur_timeout, WaitEvent *  occurred_events, int  nevents  )

inlinestatic

Definition at line 1471 of file waiteventset.c.

{
    int         returned_events = 0;
    int         rc;
    WaitEvent  *cur_event;
    struct pollfd *cur_pollfd;
 
    /* Sleep */
    rc = poll(set->pollfds, 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("%s() failed: %m",
                            "poll")));
        }
        return 0;
    }
    else if (rc == 0)
    {
        /* timeout exceeded */
        return -1;
    }
 
    for (cur_event = set->events, cur_pollfd = set->pollfds;
         cur_event < (set->events + set->nevents) &&
         returned_events < nevents;
         cur_event++, cur_pollfd++)
    {
        /* no activity on this FD, skip */
        if (cur_pollfd->revents == 0)
            continue;
 
        occurred_events->pos = cur_event->pos;
        occurred_events->user_data = cur_event->user_data;
        occurred_events->events = 0;
 
        if (cur_event->events == WL_LATCH_SET &&
            (cur_pollfd->revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL)))
        {
            /* There's data in the self-pipe, clear it. */
            drain();
 
            if (set->latch && set->latch->maybe_sleeping && set->latch->is_set)
            {
                occurred_events->fd = PGINVALID_SOCKET;
                occurred_events->events = WL_LATCH_SET;
                occurred_events++;
                returned_events++;
            }
        }
        else if (cur_event->events == WL_POSTMASTER_DEATH &&
                 (cur_pollfd->revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL)))
        {
            /*
             * We expect an POLLHUP when the remote end is closed, but because
             * we don't expect the pipe to become readable or to have any
             * errors either, treat those cases as postmaster death, too.
             *
             * Be paranoid about a spurious event signaling the postmaster as
             * being dead.  There have been reports about that happening with
             * older primitives (select(2) to be specific), and a spurious
             * WL_POSTMASTER_DEATH event would be painful. Re-checking doesn't
             * cost much.
             */
            if (!PostmasterIsAliveInternal())
            {
                if (set->exit_on_postmaster_death)
                    proc_exit(1);
                occurred_events->fd = PGINVALID_SOCKET;
                occurred_events->events = WL_POSTMASTER_DEATH;
                occurred_events++;
                returned_events++;
            }
        }
        else if (cur_event->events & (WL_SOCKET_READABLE |
                                      WL_SOCKET_WRITEABLE |
                                      WL_SOCKET_CLOSED))
        {
            int         errflags = POLLHUP | POLLERR | POLLNVAL;
 
            Assert(cur_event->fd >= PGINVALID_SOCKET);
 
            if ((cur_event->events & WL_SOCKET_READABLE) &&
                (cur_pollfd->revents & (POLLIN | errflags)))
            {
                /* data available in socket, or EOF */
                occurred_events->events |= WL_SOCKET_READABLE;
            }
 
            if ((cur_event->events & WL_SOCKET_WRITEABLE) &&
                (cur_pollfd->revents & (POLLOUT | errflags)))
            {
                /* writeable, or EOF */
                occurred_events->events |= WL_SOCKET_WRITEABLE;
            }
 
#ifdef POLLRDHUP
            if ((cur_event->events & WL_SOCKET_CLOSED) &&
                (cur_pollfd->revents & (POLLRDHUP | errflags)))
            {
                /* remote peer closed, or error */
                occurred_events->events |= WL_SOCKET_CLOSED;
            }
#endif
 
            if (occurred_events->events != 0)
            {
                occurred_events->fd = cur_event->fd;
                occurred_events++;
                returned_events++;
            }
        }
    }
    return returned_events;
}

References Assert(), drain(), EINTR, ereport, errcode_for_socket_access(), errmsg(), ERROR, WaitEventSet::events, WaitEvent::events, WaitEventSet::exit_on_postmaster_death, WaitEvent::fd, Latch::is_set, WaitEventSet::latch, Latch::maybe_sleeping, WaitEventSet::nevents, PGINVALID_SOCKET, WaitEventSet::pollfds, WaitEvent::pos, PostmasterIsAliveInternal(), proc_exit(), WaitEvent::user_data, waiting, WL_LATCH_SET, WL_POSTMASTER_DEATH, WL_SOCKET_CLOSED, WL_SOCKET_READABLE, and WL_SOCKET_WRITEABLE.

Referenced by WaitEventSetWait().

WakeupMyProc()

void WakeupMyProc

(

void

)

Definition at line 2020 of file waiteventset.c.

{
#if defined(WAIT_USE_SELF_PIPE)
    if (waiting)
        sendSelfPipeByte();
#else
    if (waiting)
        kill(MyProcPid, SIGURG);
#endif
}

References kill, MyProcPid, sendSelfPipeByte(), and waiting.

Referenced by SetLatch().

WakeupOtherProc()

void WakeupOtherProc

(

int

pid

)

Definition at line 2033 of file waiteventset.c.

{
    kill(pid, SIGURG);
}

References kill.

Referenced by SetLatch().

Variable Documentation

selfpipe_owner_pid

int selfpipe_owner_pid = 0

static

Definition at line 185 of file waiteventset.c.

Referenced by InitializeWaitEventSupport().

selfpipe_readfd

int selfpipe_readfd = -1

static

Definition at line 181 of file waiteventset.c.

Referenced by AddWaitEventToSet(), drain(), and InitializeWaitEventSupport().

selfpipe_writefd

int selfpipe_writefd = -1

static

Definition at line 182 of file waiteventset.c.

Referenced by InitializeWaitEventSupport(), and sendSelfPipeByte().

wait_event_set_resowner_desc

const ResourceOwnerDesc wait_event_set_resowner_desc

static

Initial value:

=
{
    .name = "WaitEventSet",
    .release_phase = RESOURCE_RELEASE_AFTER_LOCKS,
    .release_priority = RELEASE_PRIO_WAITEVENTSETS,
    .ReleaseResource = ResOwnerReleaseWaitEventSet,
    .DebugPrint = NULL
}

Definition at line 212 of file waiteventset.c.

Referenced by ResourceOwnerForgetWaitEventSet(), and ResourceOwnerRememberWaitEventSet().

waiting

volatile sig_atomic_t waiting = false

static

Definition at line 171 of file waiteventset.c.

Referenced by drain(), latch_sigurg_handler(), LockBufferForCleanup(), ResolveRecoveryConflictWithVirtualXIDs(), run_permutation(), try_complete_step(), try_complete_steps(), WaitEventSetWait(), WaitEventSetWaitBlock(), and WakeupMyProc().