folkslinux
/
rpm
mirror of https://github.com/rpm-software-management/rpm.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
rpm/db/repmgr/repmgr_posix.c

692 lines
16 KiB
C
Raw Blame History

/*-
* See the file LICENSE for redistribution information.
*
* Copyright (c) 2005,2007 Oracle. All rights reserved.
*
* $Id: repmgr_posix.c,v 1.29 2007/06/11 18:29:34 alanb Exp $
*/
#include "db_config.h"
#define __INCLUDE_NETWORKING 1
#define __INCLUDE_SELECT_H 1
#include "db_int.h"
/*
* A very rough guess at the maximum stack space one of our threads could ever
* need, which we hope is plenty conservative. This can be patched in the field
* if necessary.
*/
#ifdef _POSIX_THREAD_ATTR_STACKSIZE
size_t __repmgr_guesstimated_max = (128 * 1024);
#endif
static int finish_connecting __P((DB_ENV *, REPMGR_CONNECTION *));
/*
* Starts the thread described in the argument, and stores the resulting thread
* ID therein.
*
* PUBLIC: int __repmgr_thread_start __P((DB_ENV *, REPMGR_RUNNABLE *));
*/
int
__repmgr_thread_start(dbenv, runnable)
DB_ENV *dbenv;
REPMGR_RUNNABLE *runnable;
{
pthread_attr_t *attrp;
#ifdef _POSIX_THREAD_ATTR_STACKSIZE
pthread_attr_t attributes;
size_t size;
int ret;
#endif
runnable->finished = FALSE;
#ifdef _POSIX_THREAD_ATTR_STACKSIZE
attrp = &attributes;
if ((ret = pthread_attr_init(&attributes)) != 0) {
__db_err(dbenv,
ret, "pthread_attr_init in repmgr_thread_start");
return (ret);
}
/*
* On a 64-bit machine it seems reasonable that we could need twice as
* much stack space as we did on a 32-bit machine.
*/
size = __repmgr_guesstimated_max;
if (sizeof(size_t) > 4)
size *= 2;
#ifdef PTHREAD_STACK_MIN
if (size < PTHREAD_STACK_MIN)
size = PTHREAD_STACK_MIN;
#endif
if ((ret = pthread_attr_setstacksize(&attributes, size)) != 0) {
__db_err(dbenv,
ret, "pthread_attr_setstacksize in repmgr_thread_start");
return (ret);
}
#else
attrp = NULL;
#endif
return (pthread_create(&runnable->thread_id, attrp,
runnable->run, dbenv));
}
/*
* PUBLIC: int __repmgr_thread_join __P((REPMGR_RUNNABLE *));
*/
int
__repmgr_thread_join(thread)
REPMGR_RUNNABLE *thread;
{
return (pthread_join(thread->thread_id, NULL));
}
/*
* PUBLIC: int __repmgr_set_nonblocking __P((socket_t));
*/
int
__repmgr_set_nonblocking(fd)
socket_t fd;
{
int flags;
if ((flags = fcntl(fd, F_GETFL, 0)) < 0)
return (errno);
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0)
return (errno);
return (0);
}
/*
* PUBLIC: int __repmgr_wake_waiting_senders __P((DB_ENV *));
*
* Wake any send()-ing threads waiting for an acknowledgement.
*
* !!!
* Caller must hold the db_rep->mutex, if this thread synchronization is to work
* properly.
*/
int
__repmgr_wake_waiting_senders(dbenv)
DB_ENV *dbenv;
{
return (pthread_cond_broadcast(&dbenv->rep_handle->ack_condition));
}
/*
* PUBLIC: int __repmgr_await_ack __P((DB_ENV *, const DB_LSN *));
*
* Waits (a limited time) for configured number of remote sites to ack the given
* LSN.
*
* !!!
* Caller must hold repmgr->mutex.
*/
int
__repmgr_await_ack(dbenv, lsnp)
DB_ENV *dbenv;
const DB_LSN *lsnp;
{
DB_REP *db_rep;
struct timespec deadline;
int ret, timed;
db_rep = dbenv->rep_handle;
if ((timed = (db_rep->ack_timeout > 0)))
__repmgr_compute_wait_deadline(dbenv, &deadline,
db_rep->ack_timeout);
else
COMPQUIET(deadline.tv_sec, 0);
while (!__repmgr_is_permanent(dbenv, lsnp)) {
if (timed)
ret = pthread_cond_timedwait(&db_rep->ack_condition,
&db_rep->mutex, &deadline);
else
ret = pthread_cond_wait(&db_rep->ack_condition,
&db_rep->mutex);
if (db_rep->finished)
return (DB_REP_UNAVAIL);
if (ret != 0)
return (ret);
}
return (0);
}
/*
* Computes a deadline time a certain distance into the future, in a form
* suitable for the pthreads timed wait operation. Curiously, that call uses
* nano-second resolution; elsewhere we use microseconds.
*
* PUBLIC: void __repmgr_compute_wait_deadline __P((DB_ENV*,
* PUBLIC: struct timespec *, db_timeout_t));
*/
void
__repmgr_compute_wait_deadline(dbenv, result, wait)
DB_ENV *dbenv;
struct timespec *result;
db_timeout_t wait;
{
db_timespec v;
/*
* Start with "now"; then add the "wait" offset.
*
* A db_timespec is the same as a "struct timespec" so we can pass
* result directly to the underlying Berkeley DB OS routine.
*/
__os_gettime(dbenv, (db_timespec *)result);
/* Convert microsecond wait to a timespec. */
DB_TIMEOUT_TO_TIMESPEC(wait, &v);
timespecadd(result, &v);
}
/*
* PUBLIC: int __repmgr_init_sync __P((DB_ENV *, DB_REP *));
*
* Allocate/initialize all data necessary for thread synchronization. This
* should be an all-or-nothing affair. Other than here and in _close_sync there
* should never be a time when these resources aren't either all allocated or
* all freed. If that's true, then we can safely use the values of the file
* descriptor(s) to keep track of which it is.
*/
int
__repmgr_init_sync(dbenv, db_rep)
DB_ENV *dbenv;
DB_REP *db_rep;
{
int ret, mutex_inited, ack_inited, elect_inited, queue_inited,
file_desc[2];
COMPQUIET(dbenv, NULL);
mutex_inited = ack_inited = elect_inited = queue_inited = FALSE;
if ((ret = pthread_mutex_init(&db_rep->mutex, NULL)) != 0)
goto err;
mutex_inited = TRUE;
if ((ret = pthread_cond_init(&db_rep->ack_condition, NULL)) != 0)
goto err;
ack_inited = TRUE;
if ((ret = pthread_cond_init(&db_rep->check_election, NULL)) != 0)
goto err;
elect_inited = TRUE;
if ((ret = pthread_cond_init(&db_rep->queue_nonempty, NULL)) != 0)
goto err;
queue_inited = TRUE;
if ((ret = pipe(file_desc)) == -1) {
ret = errno;
goto err;
}
db_rep->read_pipe = file_desc[0];
db_rep->write_pipe = file_desc[1];
return (0);
err:
if (queue_inited)
(void)pthread_cond_destroy(&db_rep->queue_nonempty);
if (elect_inited)
(void)pthread_cond_destroy(&db_rep->check_election);
if (ack_inited)
(void)pthread_cond_destroy(&db_rep->ack_condition);
if (mutex_inited)
(void)pthread_mutex_destroy(&db_rep->mutex);
db_rep->read_pipe = db_rep->write_pipe = -1;
return (ret);
}
/*
* PUBLIC: int __repmgr_close_sync __P((DB_ENV *));
*
* Frees the thread synchronization data within a repmgr struct, in a
* platform-specific way.
*/
int
__repmgr_close_sync(dbenv)
DB_ENV *dbenv;
{
DB_REP *db_rep;
int ret, t_ret;
db_rep = dbenv->rep_handle;
if (!(REPMGR_SYNC_INITED(db_rep)))
return (0);
ret = pthread_cond_destroy(&db_rep->queue_nonempty);
if ((t_ret = pthread_cond_destroy(&db_rep->check_election)) != 0 &&
ret == 0)
ret = t_ret;
if ((t_ret = pthread_cond_destroy(&db_rep->ack_condition)) != 0 &&
ret == 0)
ret = t_ret;
if ((t_ret = pthread_mutex_destroy(&db_rep->mutex)) != 0 &&
ret == 0)
ret = t_ret;
if (close(db_rep->read_pipe) == -1 && ret == 0)
ret = errno;
if (close(db_rep->write_pipe) == -1 && ret == 0)
ret = errno;
db_rep->read_pipe = db_rep->write_pipe = -1;
return (ret);
}
/*
* Performs net-related resource initialization other than memory initialization
* and allocation. A valid db_rep->listen_fd acts as the "all-or-nothing"
* sentinel signifying that these resources are allocated.
*
* PUBLIC: int __repmgr_net_init __P((DB_ENV *, DB_REP *));
*/
int
__repmgr_net_init(dbenv, db_rep)
DB_ENV *dbenv;
DB_REP *db_rep;
{
int ret;
struct sigaction sigact;
if ((ret = __repmgr_listen(dbenv)) != 0)
return (ret);
/*
* Make sure we're not ignoring SIGPIPE, 'cuz otherwise we'd be killed
* just for trying to write onto a socket that had been reset.
*/
if (sigaction(SIGPIPE, NULL, &sigact) == -1) {
ret = errno;
__db_err(dbenv, ret, "can't access signal handler");
goto err;
}
/*
* If we need to change the sig handler, do so, and also set a flag so
* that we remember we did.
*/
if ((db_rep->chg_sig_handler = (sigact.sa_handler == SIG_DFL))) {
sigact.sa_handler = SIG_IGN;
sigact.sa_flags = 0;
if (sigaction(SIGPIPE, &sigact, NULL) == -1) {
ret = errno;
__db_err(dbenv, ret, "can't access signal handler");
goto err;
}
}
return (0);
err:
(void)closesocket(db_rep->listen_fd);
db_rep->listen_fd = INVALID_SOCKET;
return (ret);
}
/*
* PUBLIC: int __repmgr_lock_mutex __P((mgr_mutex_t *));
*/
int
__repmgr_lock_mutex(mutex)
mgr_mutex_t *mutex;
{
return (pthread_mutex_lock(mutex));
}
/*
* PUBLIC: int __repmgr_unlock_mutex __P((mgr_mutex_t *));
*/
int
__repmgr_unlock_mutex(mutex)
mgr_mutex_t *mutex;
{
return (pthread_mutex_unlock(mutex));
}
/*
* Signals a condition variable.
*
* !!!
* Caller must hold mutex.
*
* PUBLIC: int __repmgr_signal __P((cond_var_t *));
*/
int
__repmgr_signal(v)
cond_var_t *v;
{
return (pthread_cond_broadcast(v));
}
/*
* PUBLIC: int __repmgr_wake_main_thread __P((DB_ENV*));
*/
int
__repmgr_wake_main_thread(dbenv)
DB_ENV *dbenv;
{
DB_REP *db_rep;
u_int8_t any_value;
COMPQUIET(any_value, 0);
db_rep = dbenv->rep_handle;
/*
* It doesn't matter what byte value we write. Just the appearance of a
* byte in the stream is enough to wake up the select() thread reading
* the pipe.
*/
if (write(db_rep->write_pipe, &any_value, 1) == -1)
return (errno);
return (0);
}
/*
* PUBLIC: int __repmgr_writev __P((socket_t, db_iovec_t *, int, size_t *));
*/
int
__repmgr_writev(fd, iovec, buf_count, byte_count_p)
socket_t fd;
db_iovec_t *iovec;
int buf_count;
size_t *byte_count_p;
{
int nw;
if ((nw = writev(fd, iovec, buf_count)) == -1)
return (errno);
*byte_count_p = (size_t)nw;
return (0);
}
/*
* PUBLIC: int __repmgr_readv __P((socket_t, db_iovec_t *, int, size_t *));
*/
int
__repmgr_readv(fd, iovec, buf_count, byte_count_p)
socket_t fd;
db_iovec_t *iovec;
int buf_count;
size_t *byte_count_p;
{
ssize_t nw;
if ((nw = readv(fd, iovec, buf_count)) == -1)
return (errno);
*byte_count_p = (size_t)nw;
return (0);
}
/*
* PUBLIC: int __repmgr_select_loop __P((DB_ENV *));
*/
int
__repmgr_select_loop(dbenv)
DB_ENV *dbenv;
{
struct timeval select_timeout, *select_timeout_p;
DB_REP *db_rep;
REPMGR_CONNECTION *conn, *next;
REPMGR_RETRY *retry;
db_timespec timeout;
fd_set reads, writes;
int ret, flow_control, maxfd, nready;
u_int8_t buf[10]; /* arbitrary size */
flow_control = FALSE;
db_rep = dbenv->rep_handle;
/*
* Almost this entire thread operates while holding the mutex. But note
* that it never blocks, except in the call to select() (which is the
* one place we relinquish the mutex).
*/
LOCK_MUTEX(db_rep->mutex);
if ((ret = __repmgr_first_try_connections(dbenv)) != 0)
goto out;
for (;;) {
FD_ZERO(&reads);
FD_ZERO(&writes);
/*
* Always ask for input on listening socket and signalling
* pipe.
*/
FD_SET((u_int)db_rep->listen_fd, &reads);
maxfd = db_rep->listen_fd;
FD_SET((u_int)db_rep->read_pipe, &reads);
if (db_rep->read_pipe > maxfd)
maxfd = db_rep->read_pipe;
/*
* Examine all connections to see what sort of I/O to ask for on
* each one.
*/
TAILQ_FOREACH(conn, &db_rep->connections, entries) {
if (F_ISSET(conn, CONN_CONNECTING)) {
FD_SET((u_int)conn->fd, &reads);
FD_SET((u_int)conn->fd, &writes);
if (conn->fd > maxfd)
maxfd = conn->fd;
continue;
}
if (!STAILQ_EMPTY(&conn->outbound_queue)) {
FD_SET((u_int)conn->fd, &writes);
if (conn->fd > maxfd)
maxfd = conn->fd;
}
/*
* If we haven't yet gotten site's handshake, then read
* from it even if we're flow-controlling.
*/
if (!flow_control || !IS_VALID_EID(conn->eid)) {
FD_SET((u_int)conn->fd, &reads);
if (conn->fd > maxfd)
maxfd = conn->fd;
}
}
/*
* Decide how long to wait based on when it will next be time to
* retry an idle connection. (List items are in order, so we
* only have to examine the first one.)
*/
if (TAILQ_EMPTY(&db_rep->retries))
select_timeout_p = NULL;
else {
retry = TAILQ_FIRST(&db_rep->retries);
__repmgr_timespec_diff_now(
dbenv, &retry->time, &timeout);
/* Convert the timespec to a timeval. */
select_timeout.tv_sec = timeout.tv_sec;
select_timeout.tv_usec = timeout.tv_nsec / NS_PER_US;
select_timeout_p = &select_timeout;
}
UNLOCK_MUTEX(db_rep->mutex);
if ((ret = select(maxfd + 1,
&reads, &writes, NULL, select_timeout_p)) == -1) {
switch (ret = errno) {
case EINTR:
case EWOULDBLOCK:
LOCK_MUTEX(db_rep->mutex);
continue; /* simply retry */
default:
__db_err(dbenv, ret, "select");
return (ret);
}
}
nready = ret;
LOCK_MUTEX(db_rep->mutex);
/*
* The first priority thing we must do is to clean up any
* pending defunct connections. Otherwise, if they have any
* lingering pending input, we get very confused if we try to
* process it.
*
* The TAILQ_FOREACH macro would be suitable here, except that
* it doesn't allow unlinking the current element, which is
* needed for cleanup_connection.
*/
for (conn = TAILQ_FIRST(&db_rep->connections);
conn != NULL;
conn = next) {
next = TAILQ_NEXT(conn, entries);
if (F_ISSET(conn, CONN_DEFUNCT))
__repmgr_cleanup_connection(dbenv, conn);
}
if ((ret = __repmgr_retry_connections(dbenv)) != 0)
goto out;
if (nready == 0)
continue;
/*
* Traverse the linked list. (Again, like TAILQ_FOREACH, except
* that we need the ability to unlink an element along the way.)
*/
for (conn = TAILQ_FIRST(&db_rep->connections);
conn != NULL;
conn = next) {
next = TAILQ_NEXT(conn, entries);
if (F_ISSET(conn, CONN_CONNECTING)) {
if (FD_ISSET((u_int)conn->fd, &reads) ||
FD_ISSET((u_int)conn->fd, &writes)) {
if ((ret = finish_connecting(dbenv,
conn)) == DB_REP_UNAVAIL) {
if ((ret =
__repmgr_bust_connection(
dbenv, conn, TRUE)) != 0)
goto out;
} else if (ret != 0)
goto out;
}
continue;
}
/*
* Here, the site is connected, and the FD_SET's are
* valid.
*/
if (FD_ISSET((u_int)conn->fd, &writes)) {
if ((ret = __repmgr_write_some(
dbenv, conn)) == DB_REP_UNAVAIL) {
if ((ret =
__repmgr_bust_connection(dbenv,
conn, TRUE)) != 0)
goto out;
continue;
} else if (ret != 0)
goto out;
}
if (!flow_control &&
FD_ISSET((u_int)conn->fd, &reads)) {
if ((ret = __repmgr_read_from_site(dbenv, conn))
== DB_REP_UNAVAIL) {
if ((ret =
__repmgr_bust_connection(dbenv,
conn, TRUE)) != 0)
goto out;
continue;
} else if (ret != 0)
goto out;
}
}
/*
* Read any bytes in the signalling pipe. Note that we don't
* actually need to do anything with them; they're just there to
* wake us up when necessary.
*/
if (FD_ISSET((u_int)db_rep->read_pipe, &reads)) {
if (read(db_rep->read_pipe, buf, sizeof(buf)) <= 0) {
ret = errno;
goto out;
} else if (db_rep->finished) {
ret = 0;
goto out;
}
}
if (FD_ISSET((u_int)db_rep->listen_fd, &reads) &&
(ret = __repmgr_accept(dbenv)) != 0)
goto out;
}
out:
UNLOCK_MUTEX(db_rep->mutex);
return (ret);
}
static int
finish_connecting(dbenv, conn)
DB_ENV *dbenv;
REPMGR_CONNECTION *conn;
{
DB_REP *db_rep;
REPMGR_SITE *site;
socklen_t len;
SITE_STRING_BUFFER buffer;
u_int eid;
int error, ret;
len = sizeof(error);
if (getsockopt(
conn->fd, SOL_SOCKET, SO_ERROR, (sockopt_t)&error, &len) < 0)
goto err_rpt;
if (error) {
errno = error;
goto err_rpt;
}
F_CLR(conn, CONN_CONNECTING);
return (__repmgr_send_handshake(dbenv, conn));
err_rpt:
db_rep = dbenv->rep_handle;
DB_ASSERT(dbenv, IS_VALID_EID(conn->eid));
eid = (u_int)conn->eid;
site = SITE_FROM_EID(eid);
__db_err(dbenv, errno,
"connecting to %s", __repmgr_format_site_loc(site, buffer));
/* If we've exhausted the list of possible addresses, give up. */
if (ADDR_LIST_NEXT(&site->net_addr) == NULL)
return (DB_REP_UNAVAIL);
/*
* This is just like a little mini-"bust_connection", except that we
* don't reschedule for later, 'cuz we're just about to try again right
* now.
*
* !!!
* Which means this must only be called on the select() thread, since
* only there are we allowed to actually close a connection.
*/
DB_ASSERT(dbenv, !TAILQ_EMPTY(&db_rep->connections));
__repmgr_cleanup_connection(dbenv, conn);
ret = __repmgr_connect_site(dbenv, eid);
DB_ASSERT(dbenv, ret != DB_REP_UNAVAIL);
return (ret);
}
Reference in New Issue View Git Blame Copy Permalink