unix_latch.c
text/plain
/*-------------------------------------------------------------------------
*
* unix_latch.c
* Routines for inter-process latches
*
* A latch is a boolean variable, with operations that let you to sleep
* until it is set. A latch can be set from another process, or a signal
* handler within the same process.
*
* The latch interface is a reliable replacement for the common pattern of
* using pg_usleep() or select() to wait until a signal arrives, where the
* signal handler sets a global variable. Because on some platforms, an
* incoming signal doesn't interrupt sleep, and even on platforms where it
* does there is a race condition if the signal arrives just before
* entering the sleep, the common pattern must periodically wake up and
* poll the global variable. pselect() system call was invented to solve
* the problem, but it is not portable enough. Latches are designed to
* overcome these limitations, allowing you to sleep without polling and
* ensuring a quick response to signals from other processes.
*
* There are two kinds of latches: local and shared. A local latch is
* initialized by InitLatch, and can only be set from the same process.
* A local latch can be used to wait for a signal to arrive, by calling
* SetLatch in the signal handler. A shared latch resides in shared memory,
* and must be initialized at postmaster startup by InitSharedLatch. Before
* a shared latch can be waited on, it must be associated with a process
* with OwnLatch. Only the process owning the latch can wait on it, but any
* process can set it.
*
* There are three basic operations on a latch:
*
* SetLatch - Sets the latch
* ResetLatch - Clears the latch, allowing it to be set again
* WaitLatch - Waits for the latch to become set
*
* The correct pattern to wait for an event is:
*
* for (;;)
* {
* ResetLatch();
* if (work to do)
* Do Stuff();
*
* WaitLatch();
* }
*
* It's important to reset the latch *before* checking if there's work to
* do. Otherwise, if someone sets the latch between the check and the
* ResetLatch call, you will miss it and Wait will block.
*
* To wake up the waiter, you must first set a global flag or something
* else that the main loop tests in the "if (work to do)" part, and call
* SetLatch *after* that. SetLatch is designed to return quickly if the
* latch is already set.
*
*
* Implementation
* --------------
*
* The Unix implementation uses the so-called self-pipe trick to overcome
* the race condition involved with select() 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 select() in
* WaitLatch by writing a byte to a pipe. A signal by itself doesn't
* interrupt select() on all platforms, and even on platforms where it
* does, a signal that arrives just before the select() call does not
* prevent the select() from entering sleep. An incoming byte on a pipe
* however reliably interrupts the sleep, and makes select() to return
* immediately if the signal arrives just before select() begins.
*
* When SetLatch is called from the same process that owns the latch,
* SetLatch writes the byte directly to the pipe. If it's owned by another
* process, SIGUSR1 is sent and the signal handler in the waiting process
* writes the byte to the pipe on behalf of the signaling process.
*
* Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* $PostgreSQL$
*
*-------------------------------------------------------------------------
*/
#include <fcntl.h>
#include <signal.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/sem.h>
#include <poll.h>
#include "latch.h"
/* Are we currently in WaitLatch? The signal handler would like to know. */
static volatile sig_atomic_t waiting = false;
/* Read and write end of the self-pipe */
static int selfpipe_readfd = -1;
static int selfpipe_writefd = -1;
/* private function prototypes */
static void initSelfPipe(void);
static void drainSelfPipe(void);
static void sendSelfPipeByte(void);
#ifdef PIPE
static void drainLatchPipe(volatile Latch *latch);
#endif
/*
* Initialize a backend-local latch.
*/
void
InitLatch(volatile Latch *latch)
{
#if defined(LINUX_SEM)
#elif defined(PIPE)
#else
/* Initialize the self pipe if this is our first latch in the process
*/
if (selfpipe_readfd == -1)
initSelfPipe();
latch->is_set = false;
#endif
DEBUG("SetLatch %p\n", latch);
latch->owner_pid = MyProcPid;
latch->is_shared = false;
}
/*
* 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.
*
* NB: When you introduce a new shared latch, you must increase the shared
* latch count in NumSharedLatches in win32_latch.c!
*/
void
InitSharedLatch(volatile Latch *latch)
{
latch->owner_pid = 0;
latch->is_shared = true;
#if defined(LINUX_SEM)
latch->semId = -1;
#elif defined(PIPE)
{
int ret = pipe((int *)(latch->pipefds));
Assert(!ret);
if (fcntl(latch->pipefds[0], F_SETFL, O_NONBLOCK) < 0)
elog(FATAL, "fcntl() failed on read-end of self-pipe: %m");
if (fcntl(latch->pipefds[1], F_SETFL, O_NONBLOCK) < 0)
elog(FATAL, "fcntl() failed on write-end of self-pipe: %m");
}
#else
latch->is_set = false;
#endif
}
/*
* Associate a shared latch with the current process, allowing it to
* wait on it.
*
* Make sure that latch_sigusr1_handler() is called from the SIGUSR1 signal
* handler, as shared latches use SIGUSR1 to for inter-process communication.
*/
void
OwnLatch(volatile Latch *latch)
{
int i;
int s;
int ret;
union semun semun;
Assert(latch->is_shared);
/* Initialize the self pipe if this is our first latch in the process */
#if defined(LINUX_SEM)
semun.val = 0;
if (latch->semId == -1) {
for (i = 0; i < 1000; i++) {
s = semget(7500 + i, 1, IPC_CREAT | IPC_EXCL|0x1ff);
if (s < 0) {
DEBUG("semget error %d %d\n", s, errno);
if (i == 999) {
printf("Can't get sem, aborting\n");
exit (1);
}
} else {
break;
}
}
ret = semctl(s, 0, SETVAL, semun);
Assert(!ret);
latch->semId = s;
}
DEBUG("latch %p latch->semId %d\n", latch, latch->semId);
#elif defined (PIPE)
/* Nothing to do: we allocated it in initlatch. */
#else
if (selfpipe_readfd == -1)
initSelfPipe();
#endif
if (latch->owner_pid != 0)
elog(ERROR, "latch already owned");
latch->owner_pid = MyProcPid;
}
/*
* Disown a shared latch currently owned by the current process.
*/
void
DisownLatch(volatile Latch *latch)
{
Assert(latch->is_shared);
Assert(latch->owner_pid == MyProcPid);
latch->owner_pid = 0;
}
/*
* Wait for given latch to be set or until timeout is exceeded.
* If the latch is already set, the function returns immediately.
*
* The 'timeout' is given in microseconds, and -1 means wait forever.
* On some platforms, signals cause the timeout to be restarted, so beware
* that the function can sleep for several times longer than the specified
* timeout.
*
* The latch must be owned by the current process, ie. it must be a
* backend-local latch initialized with InitLatch, or a shared latch
* associated with the current process by calling OwnLatch.
*
* Returns 'true' if the latch was set, or 'false' if timeout was reached.
*/
bool
WaitLatch(volatile Latch *latch, long timeout)
{
#if defined(LINUX_SEM)
struct timespec t;
struct sembuf sops;
t.tv_sec = timeout * 1000000;
t.tv_nsec = 0;
sops.sem_op = -1;
sops.sem_flg = 0;
sops.sem_num = 0;
DEBUG("wait on latch %p latch->semId %d\n", latch, latch->semId);
do {
int errStatus = semtimedop(latch->semId, &sops, 1, &t);
if (errStatus == 0) {
return true;
} else if (errStatus == -1) {
Assert(errno == EAGAIN || errno == EINTR);
if (errno == EAGAIN) {
return false;
} else {
continue;
}
}
} while (true);
#else
return WaitLatchOrSocket(latch, PGINVALID_SOCKET, timeout) > 0;
#endif
}
/*
* Like WaitLatch, but will also return when there's data available in
* 'sock' for reading. Returns 0 if timeout was reached, 1 if the latch
* was set, or 2 if the scoket became readable.
*/
int
WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, long timeout)
{
#if defined(LINUX_SEM)
/* Nothing to do. */
#elif defined(PIPE)
struct timeval tv, *tvp = NULL;
int result = 0;
struct pollfd pfd[2];
int nfds = 1;
int rc = 0;
if (latch->owner_pid != MyProcPid)
elog(ERROR, "cannot wait on a latch owned by another process");
if (timeout > 0) {
timeout = timeout * 1000; //millis
}
pfd[0].fd = latch->pipefds[0];
pfd[0].events = POLLIN;
pfd[0].revents = 0x0;
if (sock != PGINVALID_SOCKET) {
pfd[1].fd = sock;
pfd[1].events = POLLIN;
pfd[1].revents = 0x0;
nfds = 2;
}
for (;;)
{
int hifd;
rc = poll(pfd, nfds, timeout);
if (rc < 0)
{
if (errno == EINTR)
continue;
elog(ERROR, "select() failed: %m");
}
if (rc == 0 && timeout != -1)
{
result = 0;
break;
}
if (sock != PGINVALID_SOCKET && (pfd[1].revents & POLLIN) != 0) {
result = 2;
break;
} else {
Assert((pfd[0].revents & POLLIN) != 0);
result = 1;
break;
}
}
return result;
#else
struct timeval tv, *tvp = NULL;
fd_set input_mask;
int rc;
int result = 0;
if (latch->owner_pid != MyProcPid)
elog(ERROR, "cannot wait on a latch owned by another process")
;
/* Initialize timeout */
if (timeout >= 0)
{
tv.tv_sec = timeout / 1000000L;
tv.tv_usec = timeout % 1000000L;
tvp = &tv;
}
waiting = true;
for (;;)
{
int hifd;
/*
* Clear the pipe, and check if the latch is set already. If s
omeone
* sets the latch between this and the select() below, the set
ter
* will write a byte to the pipe (or signal us and the signal
handler
* will do that), and the select() will return immediately.
*/
drainSelfPipe();
if (latch->is_set)
{
result = 1;
break;
}
FD_ZERO(&input_mask);
FD_SET(selfpipe_readfd, &input_mask);
hifd = selfpipe_readfd;
if (sock != PGINVALID_SOCKET)
{
FD_SET(sock, &input_mask);
if (sock > hifd)
hifd = sock;
}
rc = select(hifd + 1, &input_mask, NULL, NULL, tvp);
if (rc < 0)
{
if (errno == EINTR)
continue;
elog(ERROR, "select() failed: %m");
}
if (rc == 0)
{
/* timeout exceeded */
result = 0;
break;
}
if (sock != PGINVALID_SOCKET && FD_ISSET(sock, &input_mask))
{
result = 2;
break; /* data available in socket */
}
}
waiting = false;
return result;
#endif
}
/*
* Sets a latch and wakes up anyone waiting on it. Returns quickly if the
* latch is already set.
*/
void
SetLatch(volatile Latch *latch)
{
#if defined(LINUX_SEM)
int ret;
union semun semun;
struct sembuf sops;
semun.val = 1;
DEBUG("SetLatch %p\n", latch);
ret = semctl(latch->semId, 0, SETVAL, semun);
Assert(!ret);
#elif defined(PIPE)
char c;
int ret;
DEBUG("SetLatch %p\n", latch);
ret = write(latch->pipefds[1], &c, 1);
Assert(ret == 1);
#else
pid_t owner_pid;
/* Quick exit if already set */
if (latch->is_set)
return;
DEBUG("SetLatch %p\n", latch);
latch->is_set = true;
/*
* 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 to wake up the
* select() in that case. If it's another process, send a signal. * *
Fetch owner_pid only once, in case the owner simultaneously disowns
* the latch and clears owner_pid. XXX: This assumes that pid_t is *
atomic, which isn't guaranteed to be true! In practice, the effecti
ve
* range of pid_t fits in a 32 bit integer, and so should be atomic. I
n
* the worst case, we might end up signaling wrong process if the righ
t
* one disowns the latch just as we fetch owner_pid. Even then,
you're * very unlucky if a process with that bogus pid exists. */
owner_pid = latch->owner_pid; if (owner_pid == 0)
return;
else if (owner_pid == MyProcPid)
sendSelfPipeByte();
else
kill(owner_pid, SIGUSR1);
#endif
}
/*
* Clear the latch. Calling WaitLatch after this will sleep, unless
* the latch is set again before the WaitLatch call.
*/
void
ResetLatch(volatile Latch *latch)
{
#if defined(LINUX_SEM)
#elif defined (PIPE)
drainLatchPipe(latch);
#else
/* Only the owner should reset the latch */
Assert(latch->owner_pid == MyProcPid);
latch->is_set = false;
#endif
}
/*
* SetLatch uses SIGUSR1 to wake up the process waiting on the latch. Wake
* up WaitLatch.
*/
void
latch_sigusr1_handler(void)
{
#ifdef PIPE
Assert(false);
#endif
if (waiting)
sendSelfPipeByte();
}
/* initialize the self-pipe */
static void
initSelfPipe(void)
{
int pipefd[2];
/*
* Set up the self-pipe that allows a signal handler to wake up the
* select() 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.
*/
if (pipe(pipefd) < 0)
elog(FATAL, "pipe() failed: %m");
if (fcntl(pipefd[0], F_SETFL, O_NONBLOCK) < 0)
elog(FATAL, "fcntl() failed on read-end of self-pipe: %m");
if (fcntl(pipefd[1], F_SETFL, O_NONBLOCK) < 0)
elog(FATAL, "fcntl() failed on write-end of self-pipe: %m");
DEBUG("InitSelfPipe %d\n", getpid());
selfpipe_readfd = pipefd[0];
selfpipe_writefd = pipefd[1];
}
/* 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 ch
oice
* but silently ignore the error.
*/
return;
}
}
/* Read all available data from the self-pipe */
static void
drainSelfPipe(void)
{
/*
* There shouldn't normally be more than one byte in the pipe, or mayb
e
* a few more if multiple processes run SetLatch at the same instant.
*/
char buf[16];
int rc;
for (;;)
{
rc = read(selfpipe_readfd, buf, sizeof(buf));
if (rc < 0)
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
break; /* the pipe is empty */
else if (errno == EINTR)
continue; /* retry */
else
elog(ERROR, "read() on self-pipe failed: %m");
}
else if (rc == 0)
elog(ERROR, "unexpected EOF on self-pipe");
}
}
#ifdef PIPE
/* Read all available data from the self-pipe */
static void
drainLatchPipe(volatile Latch *latch)
{
/*
* There shouldn't normally be more than one byte in the pipe, or mayb
e
* a few more if multiple processes run SetLatch at the same instant.
*/
char buf[16];
int rc;
for (;;)
{
rc = read(latch->pipefds[0], buf, sizeof(buf));
if (rc < 0)
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
break; /* the pipe is empty */
else if (errno == EINTR)
continue; /* retry */
else
elog(ERROR, "read() on self-pipe failed: %m");
}
else if (rc == 0)
elog(ERROR, "unexpected EOF on latch-pipe");
}
}
#endif