/*------------------------------------------------------------------------- * * latch.c * Routines for inter-process latches * * The poll() implementation uses the so-called self-pipe trick to overcome the * race condition involved with poll() and setting a global flag in the signal * handler. When a latch is set and the current process is waiting for it, the * signal handler wakes up the poll() in WaitLatch by writing a byte to a pipe. * A signal by itself doesn't interrupt poll() on all platforms, and even on * platforms where it does, a signal that arrives just before the poll() call * does not prevent poll() from entering sleep. An incoming byte on a pipe * however reliably interrupts the sleep, and causes poll() to return * immediately even if the signal arrives before poll() begins. * * The epoll() implementation overcomes the race with a different technique: it * keeps SIGURG blocked and consumes from a signalfd() descriptor instead. We * don't need to register a signal handler or create our own self-pipe. We * assume that any system that has Linux epoll() also has Linux signalfd(). * * The kqueue() implementation waits for SIGURG with EVFILT_SIGNAL. * * The Windows implementation uses Windows events that are inherited by all * postmaster child processes. There's no need for the self-pipe trick there. * * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * IDENTIFICATION * src/backend/storage/ipc/latch.c * *------------------------------------------------------------------------- */ #include "postgres.h" #include #include #include #include #ifdef HAVE_SYS_EPOLL_H #include #endif #ifdef HAVE_SYS_EVENT_H #include #endif #ifdef HAVE_SYS_SIGNALFD_H #include #endif #ifdef HAVE_POLL_H #include #endif #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/latch.h" #include "storage/pmsignal.h" #include "storage/shmem.h" #include "utils/memutils.h" /* * Select the fd readiness primitive to use. Normally the "most modern" * primitive supported by the OS will be used, but for testing it can be * useful to manually specify the used primitive. If desired, just add a * define somewhere before this block. */ #if defined(WAIT_USE_EPOLL) || defined(WAIT_USE_POLL) || \ defined(WAIT_USE_KQUEUE) || defined(WAIT_USE_WIN32) /* don't overwrite manual choice */ #elif defined(HAVE_SYS_EPOLL_H) #define WAIT_USE_EPOLL #elif defined(HAVE_KQUEUE) #define WAIT_USE_KQUEUE #elif defined(HAVE_POLL) #define WAIT_USE_POLL #elif WIN32 #define WAIT_USE_WIN32 #else #error "no wait set implementation available" #endif /* * By default, we use a self-pipe with poll() and a signalfd with epoll(), if * available. We avoid signalfd on illumos for now based on problem reports. * For testing the choice can also be manually specified. */ #if defined(WAIT_USE_POLL) || defined(WAIT_USE_EPOLL) #if defined(WAIT_USE_SELF_PIPE) || defined(WAIT_USE_SIGNALFD) /* don't overwrite manual choice */ #elif defined(WAIT_USE_EPOLL) && defined(HAVE_SYS_SIGNALFD_H) && \ !defined(__illumos__) #define WAIT_USE_SIGNALFD #else #define WAIT_USE_SELF_PIPE #endif #endif /* typedef in latch.h */ struct WaitEventSet { int nevents; /* number of registered events */ int nevents_space; /* maximum number of events in this set */ /* * Array, of nevents_space length, storing the definition of events this * set is waiting for. */ WaitEvent *events; /* * If WL_LATCH_SET is specified in any wait event, latch is a pointer to * said latch, and latch_pos the offset in the ->events array. This is * useful because we check the state of the latch before performing doing * syscalls related to waiting. */ Latch *latch; int latch_pos; /* * WL_EXIT_ON_PM_DEATH is converted to WL_POSTMASTER_DEATH, but this flag * is set so that we'll exit immediately if postmaster death is detected, * instead of returning. */ bool exit_on_postmaster_death; #if defined(WAIT_USE_EPOLL) int epoll_fd; /* epoll_wait returns events in a user provided arrays, allocate once */ struct epoll_event *epoll_ret_events; #elif defined(WAIT_USE_KQUEUE) int kqueue_fd; /* kevent returns events in a user provided arrays, allocate once */ struct kevent *kqueue_ret_events; bool report_postmaster_not_running; #elif defined(WAIT_USE_POLL) /* poll expects events to be waited on every poll() call, prepare once */ struct pollfd *pollfds; #elif defined(WAIT_USE_WIN32) /* * Array of windows events. The first element always contains * pgwin32_signal_event, so the remaining elements are offset by one (i.e. * event->pos + 1). */ HANDLE *handles; #endif }; /* A common WaitEventSet used to implement WatchLatch() */ static WaitEventSet *LatchWaitSet; /* The position of the latch in LatchWaitSet. */ #define LatchWaitSetLatchPos 0 #ifndef WIN32 /* Are we currently in WaitLatch? The signal handler would like to know. */ static volatile sig_atomic_t waiting = false; #endif #ifdef WAIT_USE_SIGNALFD /* On Linux, we'll receive SIGURG via a signalfd file descriptor. */ static int signal_fd = -1; #endif #ifdef WAIT_USE_SELF_PIPE /* Read and write ends of the self-pipe */ static int selfpipe_readfd = -1; static int selfpipe_writefd = -1; /* Process owning the self-pipe --- needed for checking purposes */ static int selfpipe_owner_pid = 0; /* Private function prototypes */ static void latch_sigurg_handler(SIGNAL_ARGS); static void sendSelfPipeByte(void); #endif #if defined(WAIT_USE_SELF_PIPE) || defined(WAIT_USE_SIGNALFD) static void drain(void); #endif #if defined(WAIT_USE_EPOLL) static void WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action); #elif defined(WAIT_USE_KQUEUE) static void WaitEventAdjustKqueue(WaitEventSet *set, WaitEvent *event, int old_events); #elif defined(WAIT_USE_POLL) static void WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event); #elif defined(WAIT_USE_WIN32) static void WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event); #endif static inline int WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, WaitEvent *occurred_events, int nevents); /* * Initialize the process-local latch infrastructure. * * This must be called once during startup of any process that can wait on * latches, before it issues any InitLatch() or OwnLatch() calls. */ void InitializeLatchSupport(void) { #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; /* 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 } void InitializeLatchWaitSet(void) { int latch_pos PG_USED_FOR_ASSERTS_ONLY; Assert(LatchWaitSet == NULL); /* Set up the WaitEventSet used by WaitLatch(). */ LatchWaitSet = CreateWaitEventSet(TopMemoryContext, 2); latch_pos = AddWaitEventToSet(LatchWaitSet, WL_LATCH_SET, PGINVALID_SOCKET, MyLatch, NULL); if (IsUnderPostmaster) AddWaitEventToSet(LatchWaitSet, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET, NULL, NULL); Assert(latch_pos == LatchWaitSetLatchPos); } void ShutdownLatchSupport(void) { #if defined(WAIT_USE_POLL) pqsignal(SIGURG, SIG_IGN); #endif if (LatchWaitSet) { FreeWaitEventSet(LatchWaitSet); LatchWaitSet = NULL; } #if defined(WAIT_USE_SELF_PIPE) close(selfpipe_readfd); close(selfpipe_writefd); selfpipe_readfd = -1; selfpipe_writefd = -1; selfpipe_owner_pid = InvalidPid; #endif #if defined(WAIT_USE_SIGNALFD) close(signal_fd); signal_fd = -1; #endif } /* * Initialize a process-local latch. */ void InitLatch(Latch *latch) { latch->is_set = false; latch->maybe_sleeping = false; latch->owner_pid = MyProcPid; latch->is_shared = false; #if defined(WAIT_USE_SELF_PIPE) /* Assert InitializeLatchSupport has been called in this process */ Assert(selfpipe_readfd >= 0 && selfpipe_owner_pid == MyProcPid); #elif defined(WAIT_USE_SIGNALFD) /* Assert InitializeLatchSupport has been called in this process */ Assert(signal_fd >= 0); #elif defined(WAIT_USE_WIN32) latch->event = CreateEvent(NULL, TRUE, FALSE, NULL); if (latch->event == NULL) elog(ERROR, "CreateEvent failed: error code %lu", GetLastError()); #endif /* WIN32 */ } /* * Initialize a shared latch that can be set from other processes. The latch * is initially owned by no-one; use OwnLatch to associate it with the * current process. * * InitSharedLatch needs to be called in postmaster before forking child * processes, usually right after allocating the shared memory block * containing the latch with ShmemInitStruct. (The Unix implementation * doesn't actually require that, but the Windows one does.) Because of * this restriction, we have no concurrency issues to worry about here. * * Note that other handles created in this module are never marked as * inheritable. Thus we do not need to worry about cleaning up child * process references to postmaster-private latches or WaitEventSets. */ void InitSharedLatch(Latch *latch) { #ifdef WIN32 SECURITY_ATTRIBUTES sa; /* * Set up security attributes to specify that the events are inherited. */ ZeroMemory(&sa, sizeof(sa)); sa.nLength = sizeof(sa); sa.bInheritHandle = TRUE; latch->event = CreateEvent(&sa, TRUE, FALSE, NULL); if (latch->event == NULL) elog(ERROR, "CreateEvent failed: error code %lu", GetLastError()); #endif latch->is_set = false; latch->maybe_sleeping = false; latch->owner_pid = 0; latch->is_shared = true; } /* * Associate a shared latch with the current process, allowing it to * wait on the latch. * * Although there is a sanity check for latch-already-owned, we don't do * any sort of locking here, meaning that we could fail to detect the error * if two processes try to own the same latch at about the same time. If * there is any risk of that, caller must provide an interlock to prevent it. */ void OwnLatch(Latch *latch) { int owner_pid; /* Sanity checks */ Assert(latch->is_shared); #if defined(WAIT_USE_SELF_PIPE) /* Assert InitializeLatchSupport has been called in this process */ Assert(selfpipe_readfd >= 0 && selfpipe_owner_pid == MyProcPid); #elif defined(WAIT_USE_SIGNALFD) /* Assert InitializeLatchSupport has been called in this process */ Assert(signal_fd >= 0); #endif owner_pid = latch->owner_pid; if (owner_pid != 0) elog(PANIC, "latch already owned by PID %d", owner_pid); latch->owner_pid = MyProcPid; } /* * Disown a shared latch currently owned by the current process. */ void DisownLatch(Latch *latch) { Assert(latch->is_shared); Assert(latch->owner_pid == MyProcPid); latch->owner_pid = 0; } /* * Wait for a given latch to be set, or for postmaster death, or until timeout * is exceeded. 'wakeEvents' is a bitmask that specifies which of those events * to wait for. If the latch is already set (and WL_LATCH_SET is given), the * function returns immediately. * * The "timeout" is given in milliseconds. It must be >= 0 if WL_TIMEOUT flag * is given. Although it is declared as "long", we don't actually support * timeouts longer than INT_MAX milliseconds. Note that some extra overhead * is incurred when WL_TIMEOUT is given, so avoid using a timeout if possible. * * The latch must be owned by the current process, ie. it must be a * process-local latch initialized with InitLatch, or a shared latch * associated with the current process by calling OwnLatch. * * Returns bit mask indicating which condition(s) caused the wake-up. Note * that if multiple wake-up conditions are true, there is no guarantee that * we return all of them in one call, but we will return at least one. */ int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info) { WaitEvent event; /* Postmaster-managed callers must handle postmaster death somehow. */ Assert(!IsUnderPostmaster || (wakeEvents & WL_EXIT_ON_PM_DEATH) || (wakeEvents & WL_POSTMASTER_DEATH)); /* * Some callers may have a latch other than MyLatch, or no latch at all, * or want to handle postmaster death differently. It's cheap to assign * those, so just do it every time. */ if (!(wakeEvents & WL_LATCH_SET)) latch = NULL; ModifyWaitEvent(LatchWaitSet, LatchWaitSetLatchPos, WL_LATCH_SET, latch); LatchWaitSet->exit_on_postmaster_death = ((wakeEvents & WL_EXIT_ON_PM_DEATH) != 0); if (WaitEventSetWait(LatchWaitSet, (wakeEvents & WL_TIMEOUT) ? timeout : -1, &event, 1, wait_event_info) == 0) return WL_TIMEOUT; else return event.events; } /* * Like WaitLatch, but with an extra socket argument for WL_SOCKET_* * conditions. * * When waiting on a socket, EOF and error conditions always cause the socket * to be reported as readable/writable/connected, so that the caller can deal * with the condition. * * wakeEvents must include either WL_EXIT_ON_PM_DEATH for automatic exit * if the postmaster dies or WL_POSTMASTER_DEATH for a flag set in the * return value if the postmaster dies. The latter is useful for rare cases * where some behavior other than immediate exit is needed. * * NB: These days this is just a wrapper around the WaitEventSet API. When * using a latch very frequently, consider creating a longer living * WaitEventSet instead; that's more efficient. */ int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info) { int ret = 0; int rc; WaitEvent event; WaitEventSet *set = CreateWaitEventSet(CurrentMemoryContext, 3); if (wakeEvents & WL_TIMEOUT) Assert(timeout >= 0); else timeout = -1; if (wakeEvents & WL_LATCH_SET) AddWaitEventToSet(set, WL_LATCH_SET, PGINVALID_SOCKET, latch, NULL); /* Postmaster-managed callers must handle postmaster death somehow. */ Assert(!IsUnderPostmaster || (wakeEvents & WL_EXIT_ON_PM_DEATH) || (wakeEvents & WL_POSTMASTER_DEATH)); if ((wakeEvents & WL_POSTMASTER_DEATH) && IsUnderPostmaster) AddWaitEventToSet(set, WL_POSTMASTER_DEATH, PGINVALID_SOCKET, NULL, NULL); if ((wakeEvents & WL_EXIT_ON_PM_DEATH) && IsUnderPostmaster) AddWaitEventToSet(set, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET, NULL, NULL); if (wakeEvents & WL_SOCKET_MASK) { int ev; ev = wakeEvents & WL_SOCKET_MASK; AddWaitEventToSet(set, ev, sock, NULL, NULL); } rc = WaitEventSetWait(set, timeout, &event, 1, wait_event_info); if (rc == 0) ret |= WL_TIMEOUT; else { ret |= event.events & (WL_LATCH_SET | WL_POSTMASTER_DEATH | WL_SOCKET_MASK); } FreeWaitEventSet(set); return ret; } /* * Sets a latch and wakes up anyone waiting on it. * * This is cheap if the latch is already set, otherwise not so much. * * NB: when calling this in a signal handler, be sure to save and restore * errno around it. (That's standard practice in most signal handlers, of * course, but we used to omit it in handlers that only set a flag.) * * NB: this function is called from critical sections and signal handlers so * throwing an error is not a good idea. */ void SetLatch(Latch *latch) { #ifndef WIN32 pid_t owner_pid; #else HANDLE handle; #endif /* * The memory barrier has to be placed here to ensure that any flag * variables possibly changed by this process have been flushed to main * memory, before we check/set is_set. */ pg_memory_barrier(); /* Quick exit if already set */ if (latch->is_set) return; latch->is_set = true; pg_memory_barrier(); if (!latch->maybe_sleeping) return; #ifndef WIN32 /* * See if anyone's waiting for the latch. It can be the current process if * we're in a signal handler. We use the self-pipe or SIGURG to ourselves * to wake up WaitEventSetWaitBlock() without races in that case. If it's * another process, send a signal. * * Fetch owner_pid only once, in case the latch is concurrently getting * owned or disowned. XXX: This assumes that pid_t is atomic, which isn't * guaranteed to be true! In practice, the effective range of pid_t fits * in a 32 bit integer, and so should be atomic. In the worst case, we * might end up signaling the wrong process. Even then, you're very * unlucky if a process with that bogus pid exists and belongs to * Postgres; and PG database processes should handle excess SIGUSR1 * interrupts without a problem anyhow. * * Another sort of race condition that's possible here is for a new * process to own the latch immediately after we look, so we don't signal * it. This is okay so long as all callers of ResetLatch/WaitLatch follow * the standard coding convention of waiting at the bottom of their loops, * not the top, so that they'll correctly process latch-setting events * that happen before they enter the loop. */ owner_pid = latch->owner_pid; if (owner_pid == 0) return; else if (owner_pid == MyProcPid) { #if defined(WAIT_USE_SELF_PIPE) if (waiting) sendSelfPipeByte(); #else if (waiting) kill(MyProcPid, SIGURG); #endif } else kill(owner_pid, SIGURG); #else /* * See if anyone's waiting for the latch. It can be the current process if * we're in a signal handler. * * Use a local variable here just in case somebody changes the event field * concurrently (which really should not happen). */ handle = latch->event; if (handle) { SetEvent(handle); /* * Note that we silently ignore any errors. We might be in a signal * handler or other critical path where it's not safe to call elog(). */ } #endif } /* * Clear the latch. Calling WaitLatch after this will sleep, unless * the latch is set again before the WaitLatch call. */ void ResetLatch(Latch *latch) { /* Only the owner should reset the latch */ Assert(latch->owner_pid == MyProcPid); Assert(latch->maybe_sleeping == false); latch->is_set = false; /* * Ensure that the write to is_set gets flushed to main memory before we * examine any flag variables. Otherwise a concurrent SetLatch might * falsely conclude that it needn't signal us, even though we have missed * seeing some flag updates that SetLatch was supposed to inform us of. */ pg_memory_barrier(); } /* * Create a WaitEventSet with space for nevents different events to wait for. * * These events can then be efficiently waited upon together, using * WaitEventSetWait(). */ WaitEventSet * CreateWaitEventSet(MemoryContext context, int nevents) { 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 data = (char *) MemoryContextAllocZero(context, 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 defined(WAIT_USE_EPOLL) if (!AcquireExternalFD()) { /* treat this as though epoll_create1 itself returned EMFILE */ elog(ERROR, "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()) { /* treat this as though kqueue itself returned EMFILE */ elog(ERROR, "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; } /* * Free a previously created WaitEventSet. * * Note: preferably, this shouldn't have to free any resources that could be * inherited across an exec(). If it did, we'd likely leak those resources in * many scenarios. For the epoll case, we ensure that by setting EPOLL_CLOEXEC * when the FD is created. For the Windows case, we assume that the handles * involved are non-inheritable. */ void FreeWaitEventSet(WaitEventSet *set) { #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) WaitEvent *cur_event; for (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); } /* --- * 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 other WL_SOCKET_* events * - WL_SOCKET_WRITEABLE: Wait for socket to become writeable, * can be combined with other WL_SOCKET_* events * - WL_SOCKET_CONNECTED: Wait for socket connection to be established, * can be combined with other WL_SOCKET_* events (on non-Windows * platforms, this is the same as WL_SOCKET_WRITEABLE) * - WL_SOCKET_CLOSED: Wait for socket to be closed by remote peer. * - WL_EXIT_ON_PM_DEATH: Exit immediately if the postmaster dies * * Returns the offset in WaitEventSet->events (starting from 0), which can be * used to modify previously added wait events using ModifyWaitEvent(). * * In the WL_LATCH_SET case the latch must be owned by the current process, * i.e. it must be a process-local latch initialized with InitLatch, or a * shared latch associated with the current process by calling OwnLatch. * * In the WL_SOCKET_READABLE/WRITEABLE/CONNECTED cases, EOF and error * conditions cause the socket to be reported as readable/writable/connected, * so that the caller can deal with the condition. * * The user_data pointer specified here will be set for the events returned * by WaitEventSetWait(), allowing to easily associate additional data with * events. */ int AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch, void *user_data) { 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; } /* * Change the event mask and, in the WL_LATCH_SET case, the latch associated * with the WaitEvent. The latch may be changed to NULL to disable the latch * temporarily, and then set back to a latch later. * * 'pos' is the id returned by AddWaitEventToSet. */ void ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch) { 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 /* * 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"); } if (event->events & WL_POSTMASTER_DEATH) { elog(ERROR, "cannot modify postmaster death 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 } #if defined(WAIT_USE_EPOLL) /* * action can be one of EPOLL_CTL_ADD | EPOLL_CTL_MOD | EPOLL_CTL_DEL */ static void WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action) { struct epoll_event epoll_ev; int rc; /* pointer to our event, returned by epoll_wait */ epoll_ev.data.ptr = event; /* always wait for errors */ epoll_ev.events = EPOLLERR | EPOLLHUP; /* prepare pollfd entry once */ if (event->events == WL_LATCH_SET) { Assert(set->latch != NULL); epoll_ev.events |= EPOLLIN; } else if (event->events == WL_POSTMASTER_DEATH) { epoll_ev.events |= EPOLLIN; } else { Assert(event->fd != PGINVALID_SOCKET); Assert(event->events & (WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE | WL_SOCKET_CLOSED)); if (event->events & WL_SOCKET_READABLE) epoll_ev.events |= EPOLLIN; if (event->events & WL_SOCKET_WRITEABLE) epoll_ev.events |= EPOLLOUT; if (event->events & WL_SOCKET_CLOSED) epoll_ev.events |= EPOLLRDHUP; } /* * Even though unused, we also pass epoll_ev as the data argument if * EPOLL_CTL_DEL is passed as action. There used to be an epoll bug * requiring that, and actually it makes the code simpler... */ rc = epoll_ctl(set->epoll_fd, action, event->fd, &epoll_ev); if (rc < 0) ereport(ERROR, (errcode_for_socket_access(), errmsg("%s() failed: %m", "epoll_ctl"))); } #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 | 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); } #endif #if defined(WAIT_USE_KQUEUE) /* * On most BSD family systems, the udata member of struct kevent is of type * void *, so we could directly convert to/from WaitEvent *. Unfortunately, * NetBSD has it as intptr_t, so here we wallpaper over that difference with * an lvalue cast. */ #define AccessWaitEvent(k_ev) (*((WaitEvent **)(&(k_ev)->udata))) static inline void WaitEventAdjustKqueueAdd(struct kevent *k_ev, int filter, int action, WaitEvent *event) { k_ev->ident = event->fd; k_ev->filter = filter; k_ev->flags = action; k_ev->fflags = 0; k_ev->data = 0; AccessWaitEvent(k_ev) = event; } static inline void WaitEventAdjustKqueueAddPostmaster(struct kevent *k_ev, WaitEvent *event) { /* For now postmaster death can only be added, not removed. */ k_ev->ident = PostmasterPid; k_ev->filter = EVFILT_PROC; k_ev->flags = EV_ADD; k_ev->fflags = NOTE_EXIT; k_ev->data = 0; AccessWaitEvent(k_ev) = event; } static inline void WaitEventAdjustKqueueAddLatch(struct kevent *k_ev, WaitEvent *event) { /* For now latch can only be added, not removed. */ k_ev->ident = SIGURG; k_ev->filter = EVFILT_SIGNAL; k_ev->flags = EV_ADD; k_ev->fflags = 0; k_ev->data = 0; AccessWaitEvent(k_ev) = event; } /* * old_events is the previous event mask, used to compute what has changed. */ static void WaitEventAdjustKqueue(WaitEventSet *set, WaitEvent *event, int old_events) { int rc; struct kevent k_ev[2]; int count = 0; bool new_filt_read = false; bool old_filt_read = false; bool new_filt_write = false; bool old_filt_write = false; if (old_events == event->events) return; Assert(event->events != WL_LATCH_SET || set->latch != NULL); Assert(event->events == WL_LATCH_SET || event->events == WL_POSTMASTER_DEATH || (event->events & (WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE | WL_SOCKET_CLOSED))); if (event->events == WL_POSTMASTER_DEATH) { /* * Unlike all the other implementations, we detect postmaster death * using process notification instead of waiting on the postmaster * alive pipe. */ WaitEventAdjustKqueueAddPostmaster(&k_ev[count++], event); } else if (event->events == WL_LATCH_SET) { /* We detect latch wakeup using a signal event. */ WaitEventAdjustKqueueAddLatch(&k_ev[count++], event); } else { /* * We need to compute the adds and deletes required to get from the * old event mask to the new event mask, since kevent treats readable * and writable as separate events. */ if (old_events & (WL_SOCKET_READABLE | WL_SOCKET_CLOSED)) old_filt_read = true; if (event->events & (WL_SOCKET_READABLE | WL_SOCKET_CLOSED)) new_filt_read = true; if (old_events & WL_SOCKET_WRITEABLE) old_filt_write = true; if (event->events & WL_SOCKET_WRITEABLE) new_filt_write = true; if (old_filt_read && !new_filt_read) WaitEventAdjustKqueueAdd(&k_ev[count++], EVFILT_READ, EV_DELETE, event); else if (!old_filt_read && new_filt_read) WaitEventAdjustKqueueAdd(&k_ev[count++], EVFILT_READ, EV_ADD, event); if (old_filt_write && !new_filt_write) WaitEventAdjustKqueueAdd(&k_ev[count++], EVFILT_WRITE, EV_DELETE, event); else if (!old_filt_write && new_filt_write) WaitEventAdjustKqueueAdd(&k_ev[count++], EVFILT_WRITE, EV_ADD, event); } /* For WL_SOCKET_READ -> WL_SOCKET_CLOSED, no change needed. */ if (count == 0) return; Assert(count <= 2); rc = kevent(set->kqueue_fd, &k_ev[0], count, NULL, 0, NULL); /* * When adding the postmaster's pid, we have to consider that it might * already have exited and perhaps even been replaced by another process * with the same pid. If so, we have to defer reporting this as an event * until the next call to WaitEventSetWaitBlock(). */ if (rc < 0) { if (event->events == WL_POSTMASTER_DEATH && (errno == ESRCH || errno == EACCES)) set->report_postmaster_not_running = true; else ereport(ERROR, (errcode_for_socket_access(), errmsg("%s() failed: %m", "kevent"))); } else if (event->events == WL_POSTMASTER_DEATH && PostmasterPid != getppid() && !PostmasterIsAlive()) { /* * The extra PostmasterIsAliveInternal() check prevents false alarms * on systems that give a different value for getppid() while being * traced by a debugger. */ set->report_postmaster_not_running = true; } } #endif #if defined(WAIT_USE_WIN32) static void WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event) { HANDLE *handle = &set->handles[event->pos + 1]; if (event->events == WL_LATCH_SET) { Assert(set->latch != NULL); *handle = set->latch->event; } else if (event->events == WL_POSTMASTER_DEATH) { *handle = PostmasterHandle; } else { int flags = FD_CLOSE; /* always check for errors/EOF */ if (event->events & WL_SOCKET_READABLE) flags |= FD_READ; if (event->events & WL_SOCKET_WRITEABLE) flags |= FD_WRITE; if (event->events & WL_SOCKET_CONNECTED) flags |= FD_CONNECT; if (*handle == WSA_INVALID_EVENT) { *handle = WSACreateEvent(); if (*handle == WSA_INVALID_EVENT) elog(ERROR, "failed to create event for socket: error code %d", WSAGetLastError()); } if (WSAEventSelect(event->fd, *handle, flags) != 0) elog(ERROR, "failed to set up event for socket: error code %d", WSAGetLastError()); Assert(event->fd != PGINVALID_SOCKET); } } #endif /* * Wait for events added to the set to happen, or until the timeout is * reached. At most nevents occurred events are returned. * * If timeout = -1, block until an event occurs; if 0, check sockets for * readiness, but don't block; if > 0, block for at most timeout milliseconds. * * Returns the number of events occurred, or 0 if the timeout was reached. * * Returned events will have the fd, pos, user_data fields set to the * values associated with the registered event. */ int WaitEventSetWait(WaitEventSet *set, long timeout, WaitEvent *occurred_events, int nevents, uint32 wait_event_info) { 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; } 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. If so, leave the loop * immediately, avoid blocking again. We don't attempt to report any * other events that might also be satisfied. * * 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; break; } /* * 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); if (set->latch) { Assert(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; } #if defined(WAIT_USE_EPOLL) /* * Wait using linux's epoll_wait(2). * * This is the preferable wait method, as several readiness notifications are * delivered, without having to iterate through all of set->events. The return * epoll_event struct contain a pointer to our events, making association * easy. */ static inline int WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, WaitEvent *occurred_events, int nevents) { int returned_events = 0; int rc; WaitEvent *cur_event; struct epoll_event *cur_epoll_event; /* Sleep */ rc = epoll_wait(set->epoll_fd, set->epoll_ret_events, Min(nevents, set->nevents_space), 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", "epoll_wait"))); } return 0; } else if (rc == 0) { /* timeout exceeded */ return -1; } /* * At least one event occurred, iterate over the returned epoll events * until they're either all processed, or we've returned all the events * the caller desired. */ for (cur_epoll_event = set->epoll_ret_events; cur_epoll_event < (set->epoll_ret_events + rc) && returned_events < nevents; cur_epoll_event++) { /* epoll's data pointer is set to the associated WaitEvent */ cur_event = (WaitEvent *) cur_epoll_event->data.ptr; 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_epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP)) { /* Drain the signalfd. */ drain(); if (set->latch && 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_epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP)) { /* * We expect an EPOLLHUP 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)) { Assert(cur_event->fd != PGINVALID_SOCKET); if ((cur_event->events & WL_SOCKET_READABLE) && (cur_epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP))) { /* data available in socket, or EOF */ occurred_events->events |= WL_SOCKET_READABLE; } if ((cur_event->events & WL_SOCKET_WRITEABLE) && (cur_epoll_event->events & (EPOLLOUT | EPOLLERR | EPOLLHUP))) { /* writable, or EOF */ occurred_events->events |= WL_SOCKET_WRITEABLE; } if ((cur_event->events & WL_SOCKET_CLOSED) && (cur_epoll_event->events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP))) { /* remote peer shut down, or error */ occurred_events->events |= WL_SOCKET_CLOSED; } if (occurred_events->events != 0) { occurred_events->fd = cur_event->fd; occurred_events++; returned_events++; } } } return returned_events; } #elif defined(WAIT_USE_KQUEUE) /* * Wait using kevent(2) on BSD-family systems and macOS. * * For now this mirrors the epoll code, but in future it could modify the fd * set in the same call to kevent as it uses for waiting instead of doing that * with separate system calls. */ static int WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, WaitEvent *occurred_events, int nevents) { int returned_events = 0; int rc; WaitEvent *cur_event; struct kevent *cur_kqueue_event; struct timespec timeout; struct timespec *timeout_p; if (cur_timeout < 0) timeout_p = NULL; else { timeout.tv_sec = cur_timeout / 1000; timeout.tv_nsec = (cur_timeout % 1000) * 1000000; timeout_p = &timeout; } /* * Report postmaster events discovered by WaitEventAdjustKqueue() or an * earlier call to WaitEventSetWait(). */ if (unlikely(set->report_postmaster_not_running)) { if (set->exit_on_postmaster_death) proc_exit(1); occurred_events->fd = PGINVALID_SOCKET; occurred_events->events = WL_POSTMASTER_DEATH; return 1; } /* Sleep */ rc = kevent(set->kqueue_fd, NULL, 0, set->kqueue_ret_events, Min(nevents, set->nevents_space), timeout_p); /* 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", "kevent"))); } return 0; } else if (rc == 0) { /* timeout exceeded */ return -1; } /* * At least one event occurred, iterate over the returned kqueue events * until they're either all processed, or we've returned all the events * the caller desired. */ for (cur_kqueue_event = set->kqueue_ret_events; cur_kqueue_event < (set->kqueue_ret_events + rc) && returned_events < nevents; cur_kqueue_event++) { /* kevent's udata points to the associated WaitEvent */ cur_event = AccessWaitEvent(cur_kqueue_event); 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_kqueue_event->filter == EVFILT_SIGNAL) { if (set->latch && 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_kqueue_event->filter == EVFILT_PROC && (cur_kqueue_event->fflags & NOTE_EXIT) != 0) { /* * The kernel will tell this kqueue object only once about the * exit of the postmaster, so let's remember that for next time so * that we provide level-triggered semantics. */ set->report_postmaster_not_running = true; 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)) { Assert(cur_event->fd >= 0); if ((cur_event->events & WL_SOCKET_READABLE) && (cur_kqueue_event->filter == EVFILT_READ)) { /* readable, or EOF */ occurred_events->events |= WL_SOCKET_READABLE; } if ((cur_event->events & WL_SOCKET_CLOSED) && (cur_kqueue_event->filter == EVFILT_READ) && (cur_kqueue_event->flags & EV_EOF)) { /* the remote peer has shut down */ occurred_events->events |= WL_SOCKET_CLOSED; } if ((cur_event->events & WL_SOCKET_WRITEABLE) && (cur_kqueue_event->filter == EVFILT_WRITE)) { /* writable, or EOF */ occurred_events->events |= WL_SOCKET_WRITEABLE; } if (occurred_events->events != 0) { occurred_events->fd = cur_event->fd; occurred_events++; returned_events++; } } } return returned_events; } #elif defined(WAIT_USE_POLL) /* * Wait using poll(2). * * This allows to receive readiness notifications for several events at once, * but requires iterating through all of set->pollfds. */ static inline int WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, WaitEvent *occurred_events, int nevents) { 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->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; } #elif defined(WAIT_USE_WIN32) /* * Wait using Windows' WaitForMultipleObjects(). * * Unfortunately this will only ever return a single readiness notification at * a time. Note that while the official documentation for * WaitForMultipleObjects is ambiguous about multiple events being "consumed" * with a single bWaitAll = FALSE call, * https://blogs.msdn.microsoft.com/oldnewthing/20150409-00/?p=44273 confirms * that only one event is "consumed". */ static inline int WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, WaitEvent *occurred_events, int nevents) { int returned_events = 0; DWORD rc; WaitEvent *cur_event; /* Reset any wait events that need it */ for (cur_event = set->events; cur_event < (set->events + set->nevents); cur_event++) { if (cur_event->reset) { WaitEventAdjustWin32(set, cur_event); cur_event->reset = false; } /* * Windows does not guarantee to log an FD_WRITE network event * indicating that more data can be sent unless the previous send() * failed with WSAEWOULDBLOCK. While our caller might well have made * such a call, we cannot assume that here. Therefore, if waiting for * write-ready, force the issue by doing a dummy send(). If the dummy * send() succeeds, assume that the socket is in fact write-ready, and * return immediately. Also, if it fails with something other than * WSAEWOULDBLOCK, return a write-ready indication to let our caller * deal with the error condition. */ if (cur_event->events & WL_SOCKET_WRITEABLE) { char c; WSABUF buf; DWORD sent; int r; buf.buf = &c; buf.len = 0; r = WSASend(cur_event->fd, &buf, 1, &sent, 0, NULL, NULL); if (r == 0 || WSAGetLastError() != WSAEWOULDBLOCK) { occurred_events->pos = cur_event->pos; occurred_events->user_data = cur_event->user_data; occurred_events->events = WL_SOCKET_WRITEABLE; occurred_events->fd = cur_event->fd; return 1; } } } /* * Sleep. * * Need to wait for ->nevents + 1, because signal handle is in [0]. */ rc = WaitForMultipleObjects(set->nevents + 1, set->handles, FALSE, cur_timeout); /* Check return code */ if (rc == WAIT_FAILED) elog(ERROR, "WaitForMultipleObjects() failed: error code %lu", GetLastError()); else if (rc == WAIT_TIMEOUT) { /* timeout exceeded */ return -1; } if (rc == WAIT_OBJECT_0) { /* Service newly-arrived signals */ pgwin32_dispatch_queued_signals(); return 0; /* retry */ } /* * With an offset of one, due to the always present pgwin32_signal_event, * the handle offset directly corresponds to a wait event. */ cur_event = (WaitEvent *) &set->events[rc - WAIT_OBJECT_0 - 1]; 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) { /* * We cannot use set->latch->event to reset the fired event if we * aren't waiting on this latch now. */ if (!ResetEvent(set->handles[cur_event->pos + 1])) elog(ERROR, "ResetEvent failed: error code %lu", GetLastError()); if (set->latch && 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) { /* * Postmaster apparently died. Since the consequences of falsely * returning WL_POSTMASTER_DEATH could be pretty unpleasant, we take * the trouble to positively verify this with PostmasterIsAlive(), * even though there is no known reason to think that the event could * be falsely set on Windows. */ 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_MASK) { WSANETWORKEVENTS resEvents; HANDLE handle = set->handles[cur_event->pos + 1]; Assert(cur_event->fd); occurred_events->fd = cur_event->fd; ZeroMemory(&resEvents, sizeof(resEvents)); if (WSAEnumNetworkEvents(cur_event->fd, handle, &resEvents) != 0) elog(ERROR, "failed to enumerate network events: error code %d", WSAGetLastError()); if ((cur_event->events & WL_SOCKET_READABLE) && (resEvents.lNetworkEvents & FD_READ)) { /* data available in socket */ occurred_events->events |= WL_SOCKET_READABLE; /*------ * WaitForMultipleObjects doesn't guarantee that a read event will * be returned if the latch is set at the same time. Even if it * did, the caller might drop that event expecting it to reoccur * on next call. So, we must force the event to be reset if this * WaitEventSet is used again in order to avoid an indefinite * hang. Refer https://msdn.microsoft.com/en-us/library/windows/desktop/ms741576(v=vs.85).aspx * for the behavior of socket events. *------ */ cur_event->reset = true; } if ((cur_event->events & WL_SOCKET_WRITEABLE) && (resEvents.lNetworkEvents & FD_WRITE)) { /* writeable */ occurred_events->events |= WL_SOCKET_WRITEABLE; } if ((cur_event->events & WL_SOCKET_CONNECTED) && (resEvents.lNetworkEvents & FD_CONNECT)) { /* connected */ occurred_events->events |= WL_SOCKET_CONNECTED; } if (resEvents.lNetworkEvents & FD_CLOSE) { /* EOF/error, so signal all caller-requested socket flags */ occurred_events->events |= (cur_event->events & WL_SOCKET_MASK); } if (occurred_events->events != 0) { occurred_events++; returned_events++; } } return returned_events; } #endif /* * Return whether the current build options can report WL_SOCKET_CLOSED. */ bool WaitEventSetCanReportClosed(void) { #if (defined(WAIT_USE_POLL) && defined(POLLRDHUP)) || \ defined(WAIT_USE_EPOLL) || \ defined(WAIT_USE_KQUEUE) return true; #else return false; #endif } /* * Get the number of wait events registered in a given WaitEventSet. */ int GetNumRegisteredWaitEvents(WaitEventSet *set) { return set->nevents; } #if defined(WAIT_USE_SELF_PIPE) /* * SetLatch uses SIGURG to wake up the process waiting on the latch. * * Wake up WaitLatch, if we're waiting. */ static void latch_sigurg_handler(SIGNAL_ARGS) { int save_errno = errno; if (waiting) sendSelfPipeByte(); errno = save_errno; } /* Send one byte to the self-pipe, to wake up WaitLatch */ static void sendSelfPipeByte(void) { 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; } } #endif #if defined(WAIT_USE_SELF_PIPE) || defined(WAIT_USE_SIGNALFD) /* * Read all available data from self-pipe or signalfd. * * Note: this is only called when waiting = true. If it fails and doesn't * return, it must reset that flag first (though ideally, this will never * happen). */ static void drain(void) { 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 */ } } #endif