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_sel.c

876 lines
22 KiB
C
Raw Blame History

/*-
* See the file LICENSE for redistribution information.
*
* Copyright (c) 2006,2007 Oracle. All rights reserved.
*
* $Id: repmgr_sel.c,v 1.36 2007/06/11 18:29:34 alanb Exp $
*/
#include "db_config.h"
#define __INCLUDE_NETWORKING 1
#include "db_int.h"
static int __repmgr_connect __P((DB_ENV*, socket_t *, REPMGR_SITE *));
static int dispatch_phase_completion __P((DB_ENV *, REPMGR_CONNECTION *));
static int notify_handshake __P((DB_ENV *, REPMGR_CONNECTION *));
static int record_ack __P((DB_ENV *, REPMGR_SITE *, DB_REPMGR_ACK *));
static int __repmgr_try_one __P((DB_ENV *, u_int));
/*
* PUBLIC: void *__repmgr_select_thread __P((void *));
*/
void *
__repmgr_select_thread(args)
void *args;
{
DB_ENV *dbenv = args;
int ret;
if ((ret = __repmgr_select_loop(dbenv)) != 0) {
__db_err(dbenv, ret, "select loop failed");
__repmgr_thread_failure(dbenv, ret);
}
return (NULL);
}
/*
* PUBLIC: int __repmgr_accept __P((DB_ENV *));
*
* !!!
* Only ever called in the select() thread, since we may call
* __repmgr_bust_connection(..., TRUE).
*/
int
__repmgr_accept(dbenv)
DB_ENV *dbenv;
{
DB_REP *db_rep;
REPMGR_CONNECTION *conn;
struct sockaddr_in siaddr;
socklen_t addrlen;
socket_t s;
int ret;
#ifdef DB_WIN32
WSAEVENT event_obj;
#endif
db_rep = dbenv->rep_handle;
addrlen = sizeof(siaddr);
if ((s = accept(db_rep->listen_fd, (struct sockaddr *)&siaddr,
&addrlen)) == -1) {
/*
* Some errors are innocuous and so should be ignored. MSDN
* Library documents the Windows ones; the Unix ones are
* advocated in Stevens' UNPv1, section 16.6; and Linux
* Application Development, p. 416.
*/
switch (ret = net_errno) {
#ifdef DB_WIN32
case WSAECONNRESET:
case WSAEWOULDBLOCK:
#else
case EINTR:
case EWOULDBLOCK:
case ECONNABORTED:
case ENETDOWN:
#ifdef EPROTO
case EPROTO:
#endif
case ENOPROTOOPT:
case EHOSTDOWN:
#ifdef ENONET
case ENONET:
#endif
case EHOSTUNREACH:
case EOPNOTSUPP:
case ENETUNREACH:
#endif
RPRINT(dbenv, (dbenv,
"accept error %d considered innocuous", ret));
return (0);
default:
__db_err(dbenv, ret, "accept error");
return (ret);
}
}
RPRINT(dbenv, (dbenv, "accepted a new connection"));
if ((ret = __repmgr_set_nonblocking(s)) != 0) {
__db_err(dbenv, ret, "can't set nonblock after accept");
(void)closesocket(s);
return (ret);
}
#ifdef DB_WIN32
if ((event_obj = WSACreateEvent()) == WSA_INVALID_EVENT) {
ret = net_errno;
__db_err(dbenv, ret, "can't create WSA event");
(void)closesocket(s);
return (ret);
}
if (WSAEventSelect(s, event_obj, FD_READ|FD_CLOSE) == SOCKET_ERROR) {
ret = net_errno;
__db_err(dbenv, ret, "can't set desired event bits");
(void)WSACloseEvent(event_obj);
(void)closesocket(s);
return (ret);
}
#endif
if ((ret = __repmgr_new_connection(dbenv, &conn, s, 0)) != 0) {
#ifdef DB_WIN32
(void)WSACloseEvent(event_obj);
#endif
(void)closesocket(s);
return (ret);
}
conn->eid = -1;
#ifdef DB_WIN32
conn->event_object = event_obj;
#endif
switch (ret = __repmgr_send_handshake(dbenv, conn)) {
case 0:
return (0);
case DB_REP_UNAVAIL:
return (__repmgr_bust_connection(dbenv, conn, TRUE));
default:
return (ret);
}
}
/*
* Initiates connection attempts for any sites on the idle list whose retry
* times have expired.
*
* PUBLIC: int __repmgr_retry_connections __P((DB_ENV *));
*
* !!!
* Assumes caller holds the mutex.
*/
int
__repmgr_retry_connections(dbenv)
DB_ENV *dbenv;
{
DB_REP *db_rep;
REPMGR_RETRY *retry;
db_timespec now;
u_int eid;
int ret;
db_rep = dbenv->rep_handle;
__os_gettime(dbenv, &now);
while (!TAILQ_EMPTY(&db_rep->retries)) {
retry = TAILQ_FIRST(&db_rep->retries);
if (timespeccmp(&retry->time, &now, >=))
break; /* since items are in time order */
TAILQ_REMOVE(&db_rep->retries, retry, entries);
eid = retry->eid;
__os_free(dbenv, retry);
if ((ret = __repmgr_try_one(dbenv, eid)) != 0)
return (ret);
}
return (0);
}
/*
* PUBLIC: int __repmgr_first_try_connections __P((DB_ENV *));
*
* !!!
* Assumes caller holds the mutex.
*/
int
__repmgr_first_try_connections(dbenv)
DB_ENV *dbenv;
{
DB_REP *db_rep;
u_int eid;
int ret;
db_rep = dbenv->rep_handle;
for (eid=0; eid<db_rep->site_cnt; eid++)
if ((ret = __repmgr_try_one(dbenv, eid)) != 0)
return (ret);
return (0);
}
/*
* Makes a best-effort attempt to connect to the indicated site. Returns a
* non-zero error indication only for disastrous failures. For re-tryable
* errors, we will have scheduled another attempt, and that can be considered
* success enough.
*/
static int
__repmgr_try_one(dbenv, eid)
DB_ENV *dbenv;
u_int eid;
{
DB_REP *db_rep;
ADDRINFO *list;
repmgr_netaddr_t *addr;
int ret;
db_rep = dbenv->rep_handle;
/*
* If have never yet successfully resolved this site's host name, try to
* do so now.
*
* Throughout all the rest of repmgr, we almost never do any sort of
* blocking operation in the select thread. This is the sole exception
* to that rule. Fortunately, it should rarely happen:
*
* - for a site that we only learned about because it connected to us:
* not only were we not configured to know about it, but we also never
* got a NEWSITE message about it. And even then only if the
* connection fails and we want to retry it from this end;
*
* - if the name look-up system (e.g., DNS) is not working (let's hope
* it's temporary), or the host name is not found.
*/
addr = &SITE_FROM_EID(eid)->net_addr;
if (ADDR_LIST_FIRST(addr) == NULL) {
if ((ret = __repmgr_getaddr(
dbenv, addr->host, addr->port, 0, &list)) == 0) {
addr->address_list = list;
(void)ADDR_LIST_FIRST(addr);
} else if (ret == DB_REP_UNAVAIL)
return (__repmgr_schedule_connection_attempt(
dbenv, eid, FALSE));
else
return (ret);
}
/* Here, when we have a valid address. */
return (__repmgr_connect_site(dbenv, eid));
}
/*
* Tries to establish a connection with the site indicated by the given eid,
* starting with the "current" element of its address list and trying as many
* addresses as necessary until the list is exhausted.
*
* !!!
* Only ever called in the select() thread, since we may call
* __repmgr_bust_connection(..., TRUE).
*
* PUBLIC: int __repmgr_connect_site __P((DB_ENV *, u_int eid));
*/
int
__repmgr_connect_site(dbenv, eid)
DB_ENV *dbenv;
u_int eid;
{
DB_REP *db_rep;
REPMGR_SITE *site;
REPMGR_CONNECTION *con;
socket_t s;
u_int32_t flags;
int ret;
#ifdef DB_WIN32
long desired_event;
WSAEVENT event_obj;
#endif
db_rep = dbenv->rep_handle;
site = SITE_FROM_EID(eid);
flags = 0;
switch (ret = __repmgr_connect(dbenv, &s, site)) {
case 0:
flags = 0;
#ifdef DB_WIN32
desired_event = FD_READ|FD_CLOSE;
#endif
break;
case INPROGRESS:
flags = CONN_CONNECTING;
#ifdef DB_WIN32
desired_event = FD_CONNECT;
#endif
break;
default:
STAT(db_rep->region->mstat.st_connect_fail++);
return (
__repmgr_schedule_connection_attempt(dbenv, eid, FALSE));
}
#ifdef DB_WIN32
if ((event_obj = WSACreateEvent()) == WSA_INVALID_EVENT) {
ret = net_errno;
__db_err(dbenv, ret, "can't create WSA event");
(void)closesocket(s);
return (ret);
}
if (WSAEventSelect(s, event_obj, desired_event) == SOCKET_ERROR) {
ret = net_errno;
__db_err(dbenv, ret, "can't set desired event bits");
(void)WSACloseEvent(event_obj);
(void)closesocket(s);
return (ret);
}
#endif
if ((ret = __repmgr_new_connection(dbenv, &con, s, flags))
!= 0) {
#ifdef DB_WIN32
(void)WSACloseEvent(event_obj);
#endif
(void)closesocket(s);
return (ret);
}
#ifdef DB_WIN32
con->event_object = event_obj;
#endif
if (flags == 0) {
switch (ret = __repmgr_send_handshake(dbenv, con)) {
case 0:
break;
case DB_REP_UNAVAIL:
return (__repmgr_bust_connection(dbenv, con, TRUE));
default:
return (ret);
}
}
con->eid = (int)eid;
site->ref.conn = con;
site->state = SITE_CONNECTED;
return (0);
}
static int
__repmgr_connect(dbenv, socket_result, site)
DB_ENV *dbenv;
socket_t *socket_result;
REPMGR_SITE *site;
{
repmgr_netaddr_t *addr;
ADDRINFO *ai;
socket_t s;
char *why;
int ret;
SITE_STRING_BUFFER buffer;
/*
* Lint doesn't know about DB_ASSERT, so it can't tell that this
* loop will always get executed at least once, giving 'why' a value.
*/
COMPQUIET(why, "");
addr = &site->net_addr;
ai = ADDR_LIST_CURRENT(addr);
DB_ASSERT(dbenv, ai != NULL);
for (; ai != NULL; ai = ADDR_LIST_NEXT(addr)) {
if ((s = socket(ai->ai_family,
ai->ai_socktype, ai->ai_protocol)) == SOCKET_ERROR) {
why = "can't create socket to connect";
continue;
}
if ((ret = __repmgr_set_nonblocking(s)) != 0) {
__db_err(dbenv,
ret, "can't make nonblock socket to connect");
(void)closesocket(s);
return (ret);
}
if (connect(s, ai->ai_addr, (socklen_t)ai->ai_addrlen) != 0)
ret = net_errno;
if (ret == 0 || ret == INPROGRESS) {
*socket_result = s;
RPRINT(dbenv, (dbenv,
"init connection to %s with result %d",
__repmgr_format_site_loc(site, buffer), ret));
return (ret);
}
why = "connection failed";
(void)closesocket(s);
}
/* We've exhausted all possible addresses. */
ret = net_errno;
__db_err(dbenv, ret, "%s to %s", why,
__repmgr_format_site_loc(site, buffer));
return (ret);
}
/*
* PUBLIC: int __repmgr_send_handshake __P((DB_ENV *, REPMGR_CONNECTION *));
*/
int
__repmgr_send_handshake(dbenv, conn)
DB_ENV *dbenv;
REPMGR_CONNECTION *conn;
{
DB_REP *db_rep;
REP *rep;
repmgr_netaddr_t *my_addr;
DB_REPMGR_HANDSHAKE buffer;
DBT cntrl, rec;
DB_ASSERT(dbenv, !F_ISSET(conn, CONN_CONNECTING | CONN_DEFUNCT));
db_rep = dbenv->rep_handle;
rep = db_rep->region;
my_addr = &db_rep->my_addr;
/*
* Remind us that we need to fix priority on the rep API not to be an
* int, if we're going to pass it across the wire.
*/
DB_ASSERT(dbenv, sizeof(u_int32_t) >= sizeof(int));
buffer.version = DB_REPMGR_VERSION;
buffer.priority = htonl((u_int32_t)rep->priority);
buffer.port = my_addr->port;
cntrl.data = &buffer;
cntrl.size = sizeof(buffer);
DB_SET_DBT(rec, my_addr->host, strlen(my_addr->host) + 1);
return (__repmgr_send_one(dbenv, conn, REPMGR_HANDSHAKE, &cntrl, &rec));
}
/*
* PUBLIC: int __repmgr_read_from_site __P((DB_ENV *, REPMGR_CONNECTION *));
*
* !!!
* Caller is assumed to hold repmgr->mutex, 'cuz we call queue_put() from here.
*/
int
__repmgr_read_from_site(dbenv, conn)
DB_ENV *dbenv;
REPMGR_CONNECTION *conn;
{
SITE_STRING_BUFFER buffer;
size_t nr;
int ret;
/*
* Keep reading pieces as long as we're making some progress, or until
* we complete the current read phase.
*/
for (;;) {
if ((ret = __repmgr_readv(conn->fd,
&conn->iovecs.vectors[conn->iovecs.offset],
conn->iovecs.count - conn->iovecs.offset, &nr)) != 0) {
switch (ret) {
#ifndef DB_WIN32
case EINTR:
continue;
#endif
case WOULDBLOCK:
return (0);
default:
(void)__repmgr_format_eid_loc(dbenv->rep_handle,
conn->eid, buffer);
__db_err(dbenv, ret,
"can't read from %s", buffer);
STAT(dbenv->rep_handle->
region->mstat.st_connection_drop++);
return (DB_REP_UNAVAIL);
}
}
if (nr > 0) {
if (__repmgr_update_consumed(&conn->iovecs, nr))
return (dispatch_phase_completion(dbenv,
conn));
} else {
(void)__repmgr_format_eid_loc(dbenv->rep_handle,
conn->eid, buffer);
__db_errx(dbenv, "EOF on connection from %s", buffer);
STAT(dbenv->rep_handle->
region->mstat.st_connection_drop++);
return (DB_REP_UNAVAIL);
}
}
}
/*
* Handles whatever needs to be done upon the completion of a reading phase on a
* given connection.
*/
static int
dispatch_phase_completion(dbenv, conn)
DB_ENV *dbenv;
REPMGR_CONNECTION *conn;
{
#define MEM_ALIGN sizeof(double)
DB_REP *db_rep;
DB_REPMGR_ACK *ack;
REPMGR_SITE *site;
DB_REPMGR_HANDSHAKE *handshake;
REPMGR_RETRY *retry;
repmgr_netaddr_t addr;
DBT *dbt;
u_int32_t control_size, rec_size;
size_t memsize, control_offset, rec_offset;
void *membase;
char *host;
u_int port;
int ret, eid;
db_rep = dbenv->rep_handle;
switch (conn->reading_phase) {
case SIZES_PHASE:
/*
* We've received the header: a message type and the lengths of
* the two pieces of the message. Set up buffers to read the
* two pieces.
*/
if (conn->msg_type != REPMGR_HANDSHAKE &&
!IS_VALID_EID(conn->eid)) {
__db_errx(dbenv,
"expected handshake as first msg from passively connected site");
return (DB_REP_UNAVAIL);
}
__repmgr_iovec_init(&conn->iovecs);
control_size = ntohl(conn->control_size_buf);
rec_size = ntohl(conn->rec_size_buf);
if (conn->msg_type == REPMGR_REP_MESSAGE) {
/*
* Allocate a block of memory large enough to hold a
* DB_REPMGR_MESSAGE wrapper, plus the (one or) two DBT
* data areas that it points to. Start by calculating
* the total memory needed, rounding up for the start of
* each DBT, to ensure possible alignment requirements.
*/
memsize = (size_t)
DB_ALIGN(sizeof(REPMGR_MESSAGE), MEM_ALIGN);
control_offset = memsize;
memsize += control_size;
if (rec_size > 0) {
memsize = (size_t)DB_ALIGN(memsize, MEM_ALIGN);
rec_offset = memsize;
memsize += rec_size;
} else
COMPQUIET(rec_offset, 0);
if ((ret = __os_malloc(dbenv, memsize, &membase)) != 0)
return (ret);
conn->input.rep_message = membase;
memset(&conn->input.rep_message->control, 0,
sizeof(DBT));
memset(&conn->input.rep_message->rec, 0, sizeof(DBT));
conn->input.rep_message->originating_eid = conn->eid;
conn->input.rep_message->control.size = control_size;
conn->input.rep_message->control.data =
(u_int8_t*)membase + control_offset;
__repmgr_add_buffer(&conn->iovecs,
conn->input.rep_message->control.data,
control_size);
conn->input.rep_message->rec.size = rec_size;
if (rec_size > 0) {
conn->input.rep_message->rec.data =
(u_int8_t*)membase + rec_offset;
__repmgr_add_buffer(&conn->iovecs,
conn->input.rep_message->rec.data,
rec_size);
} else
conn->input.rep_message->rec.data = NULL;
} else {
if (control_size == 0) {
__db_errx(
dbenv, "illegal size for non-rep msg");
return (DB_REP_UNAVAIL);
}
conn->input.repmgr_msg.cntrl.size = control_size;
conn->input.repmgr_msg.rec.size = rec_size;
dbt = &conn->input.repmgr_msg.cntrl;
dbt->size = control_size;
if ((ret = __os_malloc(dbenv, control_size,
&dbt->data)) != 0)
return (ret);
__repmgr_add_dbt(&conn->iovecs, dbt);
dbt = &conn->input.repmgr_msg.rec;
if ((dbt->size = rec_size) > 0) {
if ((ret = __os_malloc(dbenv, rec_size,
&dbt->data)) != 0) {
__os_free(dbenv,
conn->input.repmgr_msg.cntrl.data);
return (ret);
}
__repmgr_add_dbt(&conn->iovecs, dbt);
}
}
conn->reading_phase = DATA_PHASE;
break;
case DATA_PHASE:
/*
* We have a complete message, so process it. Acks and
* handshakes get processed here, in line. Regular rep messages
* get posted to a queue, to be handled by a thread from the
* message thread pool.
*/
switch (conn->msg_type) {
case REPMGR_ACK:
/*
* Extract the LSN. Save it only if it is an
* improvement over what the site has already ack'ed.
*/
ack = conn->input.repmgr_msg.cntrl.data;
if (conn->input.repmgr_msg.cntrl.size != sizeof(*ack) ||
conn->input.repmgr_msg.rec.size != 0) {
__db_errx(dbenv, "bad ack msg size");
return (DB_REP_UNAVAIL);
}
if ((ret = record_ack(dbenv, SITE_FROM_EID(conn->eid),
ack)) != 0)
return (ret);
__os_free(dbenv, conn->input.repmgr_msg.cntrl.data);
break;
case REPMGR_HANDSHAKE:
handshake = conn->input.repmgr_msg.cntrl.data;
if (conn->input.repmgr_msg.cntrl.size >=
sizeof(handshake->version) &&
handshake->version != DB_REPMGR_VERSION) {
__db_errx(dbenv,
"mismatched repmgr message protocol version (%lu)",
(u_long)handshake->version);
return (DB_REP_UNAVAIL);
}
if (conn->input.repmgr_msg.cntrl.size !=
sizeof(*handshake) ||
conn->input.repmgr_msg.rec.size == 0) {
__db_errx(dbenv, "bad handshake msg size");
return (DB_REP_UNAVAIL);
}
port = handshake->port;
host = conn->input.repmgr_msg.rec.data;
host[conn->input.repmgr_msg.rec.size-1] = '\0';
RPRINT(dbenv, (dbenv,
"got handshake %s:%u, pri %lu", host, port,
(u_long)ntohl(handshake->priority)));
if (IS_VALID_EID(conn->eid)) {
/*
* We must have initiated this as an outgoing
* connection, since we already know the EID.
* All we need from the handshake is the
* priority.
*/
site = SITE_FROM_EID(conn->eid);
RPRINT(dbenv, (dbenv,
"handshake from connection to %s:%lu",
site->net_addr.host,
(u_long)site->net_addr.port));
} else {
if (IS_VALID_EID(eid =
__repmgr_find_site(dbenv, host, port))) {
site = SITE_FROM_EID(eid);
if (site->state == SITE_IDLE) {
RPRINT(dbenv, (dbenv,
"handshake from previously idle site"));
retry = site->ref.retry;
TAILQ_REMOVE(&db_rep->retries,
retry, entries);
__os_free(dbenv, retry);
conn->eid = eid;
site->state = SITE_CONNECTED;
site->ref.conn = conn;
} else {
/*
* We got an incoming connection
* for a site we were already
* connected to; discard it.
*/
__db_errx(dbenv,
"redundant incoming connection will be ignored");
return (DB_REP_UNAVAIL);
}
} else {
RPRINT(dbenv, (dbenv,
"handshake introduces unknown site"));
if ((ret = __repmgr_pack_netaddr(
dbenv, host, port, NULL,
&addr)) != 0)
return (ret);
if ((ret = __repmgr_new_site(
dbenv, &site, &addr,
SITE_CONNECTED)) != 0) {
__repmgr_cleanup_netaddr(dbenv,
&addr);
return (ret);
}
conn->eid = EID_FROM_SITE(site);
site->ref.conn = conn;
}
}
/* TODO: change priority to be u_int32_t. */
DB_ASSERT(dbenv, sizeof(int) == sizeof(u_int32_t));
site->priority = (int)ntohl(handshake->priority);
if ((ret = notify_handshake(dbenv, conn)) != 0)
return (ret);
__os_free(dbenv, conn->input.repmgr_msg.cntrl.data);
__os_free(dbenv, conn->input.repmgr_msg.rec.data);
break;
case REPMGR_REP_MESSAGE:
if ((ret = __repmgr_queue_put(dbenv,
conn->input.rep_message)) != 0)
return (ret);
/*
* The queue has taken over responsibility for the
* rep_message buffer, and will free it later.
*/
break;
default:
__db_errx(dbenv, "unknown msg type rcvd: %d",
(int)conn->msg_type);
return (DB_REP_UNAVAIL);
}
__repmgr_reset_for_reading(conn);
break;
default:
DB_ASSERT(dbenv, FALSE);
}
return (0);
}
/*
* Performs any processing needed upon the receipt of a handshake.
*
* !!!
* Caller must hold mutex.
*/
static int
notify_handshake(dbenv, conn)
DB_ENV *dbenv;
REPMGR_CONNECTION *conn;
{
DB_REP *db_rep;
COMPQUIET(conn, NULL);
db_rep = dbenv->rep_handle;
/*
* If we're moping around wishing we knew who the master was, then
* getting in touch with another site might finally provide sufficient
* connectivity to find out. But just do this once, because otherwise
* we get messages while the subsequent rep_start operations are going
* on, and rep tosses them in that case.
*/
if (db_rep->master_eid == DB_EID_INVALID && !db_rep->done_one) {
db_rep->done_one = TRUE;
RPRINT(dbenv, (dbenv,
"handshake with no known master to wake election thread"));
return (__repmgr_init_election(dbenv, ELECT_REPSTART));
}
return (0);
}
static int
record_ack(dbenv, site, ack)
DB_ENV *dbenv;
REPMGR_SITE *site;
DB_REPMGR_ACK *ack;
{
DB_REP *db_rep;
SITE_STRING_BUFFER buffer;
int ret;
db_rep = dbenv->rep_handle;
/* Ignore stale acks. */
if (ack->generation < db_rep->generation) {
RPRINT(dbenv, (dbenv,
"ignoring stale ack (%lu<%lu), from %s",
(u_long)ack->generation, (u_long)db_rep->generation,
__repmgr_format_site_loc(site, buffer)));
return (0);
}
RPRINT(dbenv, (dbenv,
"got ack [%lu][%lu](%lu) from %s", (u_long)ack->lsn.file,
(u_long)ack->lsn.offset, (u_long)ack->generation,
__repmgr_format_site_loc(site, buffer)));
if (ack->generation == db_rep->generation &&
log_compare(&ack->lsn, &site->max_ack) == 1) {
memcpy(&site->max_ack, &ack->lsn, sizeof(DB_LSN));
if ((ret = __repmgr_wake_waiting_senders(dbenv)) != 0)
return (ret);
}
return (0);
}
/*
* PUBLIC: int __repmgr_write_some __P((DB_ENV *, REPMGR_CONNECTION *));
*/
int
__repmgr_write_some(dbenv, conn)
DB_ENV *dbenv;
REPMGR_CONNECTION *conn;
{
REPMGR_FLAT *msg;
QUEUED_OUTPUT *output;
int bytes, ret;
while (!STAILQ_EMPTY(&conn->outbound_queue)) {
output = STAILQ_FIRST(&conn->outbound_queue);
msg = output->msg;
if ((bytes = send(conn->fd, &msg->data[output->offset],
(size_t)msg->length - output->offset, 0)) == SOCKET_ERROR) {
if ((ret = net_errno) == WOULDBLOCK)
return (0);
else {
__db_err(dbenv, ret, "writing data");
STAT(dbenv->rep_handle->
region->mstat.st_connection_drop++);
return (DB_REP_UNAVAIL);
}
}
if ((output->offset += (size_t)bytes) >= msg->length) {
STAILQ_REMOVE_HEAD(&conn->outbound_queue, entries);
__os_free(dbenv, output);
conn->out_queue_length--;
if (--msg->ref_count <= 0)
__os_free(dbenv, msg);
}
}
#ifdef DB_WIN32
/*
* With the queue now empty, it's time to relinquish ownership of this
* connection again, so that the next call to send() can write the
* message in line, instead of posting it to the queue for us.
*/
if (WSAEventSelect(conn->fd, conn->event_object, FD_READ|FD_CLOSE)
== SOCKET_ERROR) {
ret = net_errno;
__db_err(dbenv, ret, "can't remove FD_WRITE event bit");
return (ret);
}
#endif
return (0);
}
Reference in New Issue View Git Blame Copy Permalink