+++ /dev/null
-*** dbinc/repmgr.h 2007-10-31 10:23:52.000000000 -0700
---- dbinc/repmgr.h 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 36,41 ****
---- 36,55 ----
- #endif
-
- /*
-+ * The (arbitrary) maximum number of outgoing messages we're willing to hold, on
-+ * a queue per connection, waiting for TCP buffer space to become available in
-+ * the kernel. Rather than exceeding this limit, we simply discard additional
-+ * messages (since this is always allowed by the replication protocol).
-+ * As a special dispensation, if a message is destined for a specific remote
-+ * site (i.e., it's not a broadcast), then we first try blocking the sending
-+ * thread, waiting for space to become available (though we only wait a limited
-+ * time). This is so as to be able to handle the immediate flood of (a
-+ * potentially large number of) outgoing messages that replication generates, in
-+ * a tight loop, when handling PAGE_REQ, LOG_REQ and ALL_REQ requests.
-+ */
-+ #define OUT_QUEUE_LIMIT 10
-+
-+ /*
- * The system value is available from sysconf(_SC_HOST_NAME_MAX).
- * Historically, the maximum host name was 256.
- */
-***************
-*** 47,52 ****
---- 61,71 ----
- #define MAX_SITE_LOC_STRING (MAXHOSTNAMELEN+20)
- typedef char SITE_STRING_BUFFER[MAX_SITE_LOC_STRING+1];
-
-+ /* Default timeout values, in seconds. */
-+ #define DB_REPMGR_DEFAULT_ACK_TIMEOUT (1 * US_PER_SEC)
-+ #define DB_REPMGR_DEFAULT_CONNECTION_RETRY (30 * US_PER_SEC)
-+ #define DB_REPMGR_DEFAULT_ELECTION_RETRY (10 * US_PER_SEC)
-+
- struct __repmgr_connection;
- typedef struct __repmgr_connection REPMGR_CONNECTION;
- struct __repmgr_queue; typedef struct __repmgr_queue REPMGR_QUEUE;
-***************
-*** 171,178 ****
- #ifdef DB_WIN32
- WSAEVENT event_object;
- #endif
-! #define CONN_CONNECTING 0x01 /* nonblocking connect in progress */
-! #define CONN_DEFUNCT 0x02 /* socket close pending */
- u_int32_t flags;
-
- /*
---- 190,198 ----
- #ifdef DB_WIN32
- WSAEVENT event_object;
- #endif
-! #define CONN_CONGESTED 0x01 /* msg thread wait has exceeded timeout */
-! #define CONN_CONNECTING 0x02 /* nonblocking connect in progress */
-! #define CONN_DEFUNCT 0x04 /* socket close pending */
- u_int32_t flags;
-
- /*
-***************
-*** 180,189 ****
- * send() function's thread. But if TCP doesn't have enough network
- * buffer space for us when we first try it, we instead allocate some
- * memory, and copy the message, and then send it as space becomes
-! * available in our main select() thread.
- */
- OUT_Q_HEADER outbound_queue;
- int out_queue_length;
-
- /*
- * Input: while we're reading a message, we keep track of what phase
---- 200,215 ----
- * send() function's thread. But if TCP doesn't have enough network
- * buffer space for us when we first try it, we instead allocate some
- * memory, and copy the message, and then send it as space becomes
-! * available in our main select() thread. In some cases, if the queue
-! * gets too long we wait until it's drained, and then append to it.
-! * This condition variable's associated mutex is the normal per-repmgr
-! * db_rep->mutex, because that mutex is always held anyway whenever the
-! * output queue is consulted.
- */
- OUT_Q_HEADER outbound_queue;
- int out_queue_length;
-+ cond_var_t drained;
-+ int blockers; /* ref count of msg threads waiting on us */
-
- /*
- * Input: while we're reading a message, we keep track of what phase
-*** dbinc_auto/int_def.in 2007-10-31 10:23:52.000000000 -0700
---- dbinc_auto/int_def.in 2007-10-31 10:23:52.000000000 -0700
-***************
-*** 1420,1425 ****
---- 1420,1428 ----
- #define __repmgr_wake_waiting_senders __repmgr_wake_waiting_senders@DB_VERSION_UNIQUE_NAME@
- #define __repmgr_await_ack __repmgr_await_ack@DB_VERSION_UNIQUE_NAME@
- #define __repmgr_compute_wait_deadline __repmgr_compute_wait_deadline@DB_VERSION_UNIQUE_NAME@
-+ #define __repmgr_await_drain __repmgr_await_drain@DB_VERSION_UNIQUE_NAME@
-+ #define __repmgr_alloc_cond __repmgr_alloc_cond@DB_VERSION_UNIQUE_NAME@
-+ #define __repmgr_free_cond __repmgr_free_cond@DB_VERSION_UNIQUE_NAME@
- #define __repmgr_init_sync __repmgr_init_sync@DB_VERSION_UNIQUE_NAME@
- #define __repmgr_close_sync __repmgr_close_sync@DB_VERSION_UNIQUE_NAME@
- #define __repmgr_net_init __repmgr_net_init@DB_VERSION_UNIQUE_NAME@
-*** dbinc_auto/repmgr_ext.h 2007-10-31 10:23:52.000000000 -0700
---- dbinc_auto/repmgr_ext.h 2007-10-31 10:23:52.000000000 -0700
-***************
-*** 21,30 ****
- int __repmgr_handle_event __P((DB_ENV *, u_int32_t, void *));
- void __repmgr_stash_generation __P((DB_ENV *));
- int __repmgr_send __P((DB_ENV *, const DBT *, const DBT *, const DB_LSN *, int, u_int32_t));
-! int __repmgr_send_one __P((DB_ENV *, REPMGR_CONNECTION *, u_int, const DBT *, const DBT *));
- int __repmgr_is_permanent __P((DB_ENV *, const DB_LSN *));
-! int __repmgr_bust_connection __P((DB_ENV *, REPMGR_CONNECTION *, int));
-! void __repmgr_cleanup_connection __P((DB_ENV *, REPMGR_CONNECTION *));
- int __repmgr_find_site __P((DB_ENV *, const char *, u_int));
- int __repmgr_pack_netaddr __P((DB_ENV *, const char *, u_int, ADDRINFO *, repmgr_netaddr_t *));
- int __repmgr_getaddr __P((DB_ENV *, const char *, u_int, int, ADDRINFO **));
---- 21,30 ----
- int __repmgr_handle_event __P((DB_ENV *, u_int32_t, void *));
- void __repmgr_stash_generation __P((DB_ENV *));
- int __repmgr_send __P((DB_ENV *, const DBT *, const DBT *, const DB_LSN *, int, u_int32_t));
-! int __repmgr_send_one __P((DB_ENV *, REPMGR_CONNECTION *, u_int, const DBT *, const DBT *, int));
- int __repmgr_is_permanent __P((DB_ENV *, const DB_LSN *));
-! int __repmgr_bust_connection __P((DB_ENV *, REPMGR_CONNECTION *));
-! int __repmgr_cleanup_connection __P((DB_ENV *, REPMGR_CONNECTION *));
- int __repmgr_find_site __P((DB_ENV *, const char *, u_int));
- int __repmgr_pack_netaddr __P((DB_ENV *, const char *, u_int, ADDRINFO *, repmgr_netaddr_t *));
- int __repmgr_getaddr __P((DB_ENV *, const char *, u_int, int, ADDRINFO **));
-***************
-*** 39,44 ****
---- 39,47 ----
- int __repmgr_wake_waiting_senders __P((DB_ENV *));
- int __repmgr_await_ack __P((DB_ENV *, const DB_LSN *));
- void __repmgr_compute_wait_deadline __P((DB_ENV*, struct timespec *, db_timeout_t));
-+ int __repmgr_await_drain __P((DB_ENV *, REPMGR_CONNECTION *, db_timeout_t));
-+ int __repmgr_alloc_cond __P((cond_var_t *));
-+ int __repmgr_free_cond __P((cond_var_t *));
- int __repmgr_init_sync __P((DB_ENV *, DB_REP *));
- int __repmgr_close_sync __P((DB_ENV *));
- int __repmgr_net_init __P((DB_ENV *, DB_REP *));
-*** repmgr/repmgr_method.c 2007-10-31 10:23:52.000000000 -0700
---- repmgr/repmgr_method.c 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 196,204 ****
- int ret;
-
- /* Set some default values. */
-! db_rep->ack_timeout = 1 * US_PER_SEC; /* 1 second */
-! db_rep->connection_retry_wait = 30 * US_PER_SEC; /* 30 seconds */
-! db_rep->election_retry_wait = 10 * US_PER_SEC; /* 10 seconds */
- db_rep->config_nsites = 0;
- db_rep->peer = DB_EID_INVALID;
- db_rep->perm_policy = DB_REPMGR_ACKS_QUORUM;
---- 196,204 ----
- int ret;
-
- /* Set some default values. */
-! db_rep->ack_timeout = DB_REPMGR_DEFAULT_ACK_TIMEOUT;
-! db_rep->connection_retry_wait = DB_REPMGR_DEFAULT_CONNECTION_RETRY;
-! db_rep->election_retry_wait = DB_REPMGR_DEFAULT_ELECTION_RETRY;
- db_rep->config_nsites = 0;
- db_rep->peer = DB_EID_INVALID;
- db_rep->perm_policy = DB_REPMGR_ACKS_QUORUM;
-***************
-*** 238,243 ****
---- 238,244 ----
- DB_ENV *dbenv;
- {
- DB_REP *db_rep;
-+ REPMGR_CONNECTION *conn;
- int ret;
-
- db_rep = dbenv->rep_handle;
-***************
-*** 254,259 ****
---- 255,266 ----
-
- if ((ret = __repmgr_signal(&db_rep->queue_nonempty)) != 0)
- goto unlock;
-+
-+ TAILQ_FOREACH(conn, &db_rep->connections, entries) {
-+ if (conn->blockers > 0 &&
-+ ((ret = __repmgr_signal(&conn->drained)) != 0))
-+ goto unlock;
-+ }
- UNLOCK_MUTEX(db_rep->mutex);
-
- return (__repmgr_wake_main_thread(dbenv));
-*** repmgr/repmgr_msg.c 2007-10-31 10:23:52.000000000 -0700
---- repmgr/repmgr_msg.c 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 183,192 ****
-
- /*
- * Acknowledges a message.
-- *
-- * !!!
-- * Note that this cannot be called from the select() thread, in case we call
-- * __repmgr_bust_connection(..., FALSE).
- */
- static int
- ack_message(dbenv, generation, lsn)
---- 183,188 ----
-***************
-*** 227,235 ****
- rec2.size = 0;
-
- conn = site->ref.conn;
- if ((ret = __repmgr_send_one(dbenv, conn, REPMGR_ACK,
-! &control2, &rec2)) == DB_REP_UNAVAIL)
-! ret = __repmgr_bust_connection(dbenv, conn, FALSE);
- }
-
- UNLOCK_MUTEX(db_rep->mutex);
---- 223,236 ----
- rec2.size = 0;
-
- conn = site->ref.conn;
-+ /*
-+ * It's hard to imagine anyone would care about a lost ack if
-+ * the path to the master is so congested as to need blocking;
-+ * so pass "blockable" argument as FALSE.
-+ */
- if ((ret = __repmgr_send_one(dbenv, conn, REPMGR_ACK,
-! &control2, &rec2, FALSE)) == DB_REP_UNAVAIL)
-! ret = __repmgr_bust_connection(dbenv, conn);
- }
-
- UNLOCK_MUTEX(db_rep->mutex);
-*** repmgr/repmgr_net.c 2007-10-31 10:23:52.000000000 -0700
---- repmgr/repmgr_net.c 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 63,69 ****
- static void setup_sending_msg
- __P((struct sending_msg *, u_int, const DBT *, const DBT *));
- static int __repmgr_send_internal
-! __P((DB_ENV *, REPMGR_CONNECTION *, struct sending_msg *));
- static int enqueue_msg
- __P((DB_ENV *, REPMGR_CONNECTION *, struct sending_msg *, size_t));
- static int flatten __P((DB_ENV *, struct sending_msg *));
---- 63,69 ----
- static void setup_sending_msg
- __P((struct sending_msg *, u_int, const DBT *, const DBT *));
- static int __repmgr_send_internal
-! __P((DB_ENV *, REPMGR_CONNECTION *, struct sending_msg *, int));
- static int enqueue_msg
- __P((DB_ENV *, REPMGR_CONNECTION *, struct sending_msg *, size_t));
- static int flatten __P((DB_ENV *, struct sending_msg *));
-***************
-*** 73,85 ****
- * __repmgr_send --
- * The send function for DB_ENV->rep_set_transport.
- *
-- * !!!
-- * This is only ever called as the replication transport call-back, which means
-- * it's either on one of our message processing threads or an application
-- * thread. It mustn't be called from the select() thread, because we might call
-- * __repmgr_bust_connection(..., FALSE) here, and that's not allowed in the
-- * select() thread.
-- *
- * PUBLIC: int __repmgr_send __P((DB_ENV *, const DBT *, const DBT *,
- * PUBLIC: const DB_LSN *, int, u_int32_t));
- */
---- 73,78 ----
-***************
-*** 126,134 ****
- }
-
- conn = site->ref.conn;
- if ((ret = __repmgr_send_one(dbenv, conn, REPMGR_REP_MESSAGE,
-! control, rec)) == DB_REP_UNAVAIL &&
-! (t_ret = __repmgr_bust_connection(dbenv, conn, FALSE)) != 0)
- ret = t_ret;
- if (ret != 0)
- goto out;
---- 119,128 ----
- }
-
- conn = site->ref.conn;
-+ /* Pass the "blockable" argument as TRUE. */
- if ((ret = __repmgr_send_one(dbenv, conn, REPMGR_REP_MESSAGE,
-! control, rec, TRUE)) == DB_REP_UNAVAIL &&
-! (t_ret = __repmgr_bust_connection(dbenv, conn)) != 0)
- ret = t_ret;
- if (ret != 0)
- goto out;
-***************
-*** 222,228 ****
- if (site->state != SITE_CONNECTED)
- return (NULL);
-
-! if (F_ISSET(site->ref.conn, CONN_CONNECTING))
- return (NULL);
- return (site);
- }
---- 216,222 ----
- if (site->state != SITE_CONNECTED)
- return (NULL);
-
-! if (F_ISSET(site->ref.conn, CONN_CONNECTING|CONN_DEFUNCT))
- return (NULL);
- return (site);
- }
-***************
-*** 235,244 ****
- *
- * !!!
- * Caller must hold dbenv->mutex.
-- *
-- * !!!
-- * Note that this cannot be called from the select() thread, in case we call
-- * __repmgr_bust_connection(..., FALSE).
- */
- static int
- __repmgr_send_broadcast(dbenv, control, rec, nsitesp, npeersp)
---- 229,234 ----
-***************
-*** 268,281 ****
- !IS_VALID_EID(conn->eid))
- continue;
-
-! if ((ret = __repmgr_send_internal(dbenv, conn, &msg)) == 0) {
- site = SITE_FROM_EID(conn->eid);
- nsites++;
- if (site->priority > 0)
- npeers++;
- } else if (ret == DB_REP_UNAVAIL) {
-! if ((ret = __repmgr_bust_connection(
-! dbenv, conn, FALSE)) != 0)
- return (ret);
- } else
- return (ret);
---- 258,277 ----
- !IS_VALID_EID(conn->eid))
- continue;
-
-! /*
-! * Broadcast messages are either application threads committing
-! * transactions, or replication status message that we can
-! * afford to lose. So don't allow blocking for them (pass
-! * "blockable" argument as FALSE).
-! */
-! if ((ret = __repmgr_send_internal(dbenv,
-! conn, &msg, FALSE)) == 0) {
- site = SITE_FROM_EID(conn->eid);
- nsites++;
- if (site->priority > 0)
- npeers++;
- } else if (ret == DB_REP_UNAVAIL) {
-! if ((ret = __repmgr_bust_connection(dbenv, conn)) != 0)
- return (ret);
- } else
- return (ret);
-***************
-*** 301,339 ****
- * intersperse writes that are part of two single messages.
- *
- * PUBLIC: int __repmgr_send_one __P((DB_ENV *, REPMGR_CONNECTION *,
-! * PUBLIC: u_int, const DBT *, const DBT *));
- */
- int
-! __repmgr_send_one(dbenv, conn, msg_type, control, rec)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
- u_int msg_type;
- const DBT *control, *rec;
- {
- struct sending_msg msg;
-
- setup_sending_msg(&msg, msg_type, control, rec);
-! return (__repmgr_send_internal(dbenv, conn, &msg));
- }
-
- /*
- * Attempts a "best effort" to send a message on the given site. If there is an
-! * excessive backlog of message already queued on the connection, we simply drop
-! * this message, and still return 0 even in this case.
- */
- static int
-! __repmgr_send_internal(dbenv, conn, msg)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
- struct sending_msg *msg;
- {
-! #define OUT_QUEUE_LIMIT 10 /* arbitrary, for now */
- REPMGR_IOVECS iovecs;
- SITE_STRING_BUFFER buffer;
- int ret;
- size_t nw;
- size_t total_written;
-
- DB_ASSERT(dbenv, !F_ISSET(conn, CONN_CONNECTING));
- if (!STAILQ_EMPTY(&conn->outbound_queue)) {
- /*
---- 297,355 ----
- * intersperse writes that are part of two single messages.
- *
- * PUBLIC: int __repmgr_send_one __P((DB_ENV *, REPMGR_CONNECTION *,
-! * PUBLIC: u_int, const DBT *, const DBT *, int));
- */
- int
-! __repmgr_send_one(dbenv, conn, msg_type, control, rec, blockable)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
- u_int msg_type;
- const DBT *control, *rec;
-+ int blockable;
- {
- struct sending_msg msg;
-
- setup_sending_msg(&msg, msg_type, control, rec);
-! return (__repmgr_send_internal(dbenv, conn, &msg, blockable));
- }
-
- /*
- * Attempts a "best effort" to send a message on the given site. If there is an
-! * excessive backlog of message already queued on the connection, what shall we
-! * do? If the caller doesn't mind blocking, we'll wait (a limited amount of
-! * time) for the queue to drain. Otherwise we'll simply drop the message. This
-! * is always allowed by the replication protocol. But in the case of a
-! * multi-message response to a request like PAGE_REQ, LOG_REQ or ALL_REQ we
-! * almost always get a flood of messages that instantly fills our queue, so
-! * blocking improves performance (by avoiding the need for the client to
-! * re-request).
-! *
-! * How long shall we wait? We could of course create a new timeout
-! * configuration type, so that the application could set it directly. But that
-! * would start to overwhelm the user with too many choices to think about. We
-! * already have an ACK timeout, which is the user's estimate of how long it
-! * should take to send a message to the client, have it be processed, and return
-! * a message back to us. We multiply that by the queue size, because that's how
-! * many messages have to be swallowed up by the client before we're able to
-! * start sending again (at least to a rough approximation).
- */
- static int
-! __repmgr_send_internal(dbenv, conn, msg, blockable)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
- struct sending_msg *msg;
-+ int blockable;
- {
-! DB_REP *db_rep;
- REPMGR_IOVECS iovecs;
- SITE_STRING_BUFFER buffer;
-+ db_timeout_t drain_to;
- int ret;
- size_t nw;
- size_t total_written;
-
-+ db_rep = dbenv->rep_handle;
-+
- DB_ASSERT(dbenv, !F_ISSET(conn, CONN_CONNECTING));
- if (!STAILQ_EMPTY(&conn->outbound_queue)) {
- /*
-***************
-*** 344,358 ****
- RPRINT(dbenv, (dbenv, "msg to %s to be queued",
- __repmgr_format_eid_loc(dbenv->rep_handle,
- conn->eid, buffer)));
- if (conn->out_queue_length < OUT_QUEUE_LIMIT)
- return (enqueue_msg(dbenv, conn, msg, 0));
- else {
- RPRINT(dbenv, (dbenv, "queue limit exceeded"));
- STAT(dbenv->rep_handle->
- region->mstat.st_msgs_dropped++);
-! return (0);
- }
- }
-
- /*
- * Send as much data to the site as we can, without blocking. Keep
---- 360,393 ----
- RPRINT(dbenv, (dbenv, "msg to %s to be queued",
- __repmgr_format_eid_loc(dbenv->rep_handle,
- conn->eid, buffer)));
-+ if (conn->out_queue_length >= OUT_QUEUE_LIMIT &&
-+ blockable && !F_ISSET(conn, CONN_CONGESTED)) {
-+ RPRINT(dbenv, (dbenv,
-+ "block msg thread, await queue space"));
-+
-+ if ((drain_to = db_rep->ack_timeout) == 0)
-+ drain_to = DB_REPMGR_DEFAULT_ACK_TIMEOUT;
-+ conn->blockers++;
-+ ret = __repmgr_await_drain(dbenv,
-+ conn, drain_to * OUT_QUEUE_LIMIT);
-+ conn->blockers--;
-+ if (db_rep->finished)
-+ return (DB_TIMEOUT);
-+ if (ret != 0)
-+ return (ret);
-+ if (STAILQ_EMPTY(&conn->outbound_queue))
-+ goto empty;
-+ }
- if (conn->out_queue_length < OUT_QUEUE_LIMIT)
- return (enqueue_msg(dbenv, conn, msg, 0));
- else {
- RPRINT(dbenv, (dbenv, "queue limit exceeded"));
- STAT(dbenv->rep_handle->
- region->mstat.st_msgs_dropped++);
-! return (blockable ? DB_TIMEOUT : 0);
- }
- }
-+ empty:
-
- /*
- * Send as much data to the site as we can, without blocking. Keep
-***************
-*** 498,521 ****
-
- /*
- * Abandons a connection, to recover from an error. Upon entry the conn struct
-! * must be on the connections list.
-! *
-! * If the 'do_close' flag is true, we do the whole job; the clean-up includes
-! * removing the struct from the list and freeing all its memory, so upon return
-! * the caller must not refer to it any further. Otherwise, we merely mark the
-! * connection for clean-up later by the main thread.
- *
- * PUBLIC: int __repmgr_bust_connection __P((DB_ENV *,
-! * PUBLIC: REPMGR_CONNECTION *, int));
- *
- * !!!
- * Caller holds mutex.
- */
- int
-! __repmgr_bust_connection(dbenv, conn, do_close)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
-- int do_close;
- {
- DB_REP *db_rep;
- int connecting, ret, eid;
---- 533,553 ----
-
- /*
- * Abandons a connection, to recover from an error. Upon entry the conn struct
-! * must be on the connections list. For now, just mark it as unusable; it will
-! * be fully cleaned up in the top-level select thread, as soon as possible.
- *
- * PUBLIC: int __repmgr_bust_connection __P((DB_ENV *,
-! * PUBLIC: REPMGR_CONNECTION *));
- *
- * !!!
- * Caller holds mutex.
-+ *
-+ * Must be idempotent
- */
- int
-! __repmgr_bust_connection(dbenv, conn)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
- {
- DB_REP *db_rep;
- int connecting, ret, eid;
-***************
-*** 526,537 ****
- DB_ASSERT(dbenv, !TAILQ_EMPTY(&db_rep->connections));
- eid = conn->eid;
- connecting = F_ISSET(conn, CONN_CONNECTING);
-! if (do_close)
-! __repmgr_cleanup_connection(dbenv, conn);
-! else {
-! F_SET(conn, CONN_DEFUNCT);
-! conn->eid = -1;
-! }
-
- /*
- * When we first accepted the incoming connection, we set conn->eid to
---- 558,566 ----
- DB_ASSERT(dbenv, !TAILQ_EMPTY(&db_rep->connections));
- eid = conn->eid;
- connecting = F_ISSET(conn, CONN_CONNECTING);
-!
-! F_SET(conn, CONN_DEFUNCT);
-! conn->eid = -1;
-
- /*
- * When we first accepted the incoming connection, we set conn->eid to
-***************
-*** 557,563 ****
- dbenv, ELECT_FAILURE_ELECTION)) != 0)
- return (ret);
- }
-! } else if (!do_close) {
- /*
- * One way or another, make sure the main thread is poked, so
- * that we do the deferred clean-up.
---- 586,592 ----
- dbenv, ELECT_FAILURE_ELECTION)) != 0)
- return (ret);
- }
-! } else {
- /*
- * One way or another, make sure the main thread is poked, so
- * that we do the deferred clean-up.
-***************
-*** 568,577 ****
- }
-
- /*
-! * PUBLIC: void __repmgr_cleanup_connection
- * PUBLIC: __P((DB_ENV *, REPMGR_CONNECTION *));
- */
-! void
- __repmgr_cleanup_connection(dbenv, conn)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
---- 597,610 ----
- }
-
- /*
-! * PUBLIC: int __repmgr_cleanup_connection
- * PUBLIC: __P((DB_ENV *, REPMGR_CONNECTION *));
-+ *
-+ * !!!
-+ * Idempotent. This can be called repeatedly as blocking message threads (of
-+ * which there could be multiples) wake up in case of error on the connection.
- */
-! int
- __repmgr_cleanup_connection(dbenv, conn)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
-***************
-*** 580,596 ****
- QUEUED_OUTPUT *out;
- REPMGR_FLAT *msg;
- DBT *dbt;
-
- db_rep = dbenv->rep_handle;
-
-! TAILQ_REMOVE(&db_rep->connections, conn, entries);
- if (conn->fd != INVALID_SOCKET) {
-! (void)closesocket(conn->fd);
- #ifdef DB_WIN32
-! (void)WSACloseEvent(conn->event_object);
- #endif
- }
-
- /*
- * Deallocate any input and output buffers we may have.
- */
---- 613,643 ----
- QUEUED_OUTPUT *out;
- REPMGR_FLAT *msg;
- DBT *dbt;
-+ int ret;
-
- db_rep = dbenv->rep_handle;
-
-! DB_ASSERT(dbenv, F_ISSET(conn, CONN_DEFUNCT) || db_rep->finished);
-!
- if (conn->fd != INVALID_SOCKET) {
-! ret = closesocket(conn->fd);
-! conn->fd = INVALID_SOCKET;
-! if (ret == SOCKET_ERROR) {
-! ret = net_errno;
-! __db_err(dbenv, ret, "closing socket");
-! }
- #ifdef DB_WIN32
-! if (!WSACloseEvent(conn->event_object) && ret != 0)
-! ret = net_errno;
- #endif
-+ if (ret != 0)
-+ return (ret);
- }
-
-+ if (conn->blockers > 0)
-+ return (__repmgr_signal(&conn->drained));
-+
-+ TAILQ_REMOVE(&db_rep->connections, conn, entries);
- /*
- * Deallocate any input and output buffers we may have.
- */
-***************
-*** 614,620 ****
---- 661,669 ----
- __os_free(dbenv, out);
- }
-
-+ ret = __repmgr_free_cond(&conn->drained);
- __os_free(dbenv, conn);
-+ return (ret);
- }
-
- static int
-***************
-*** 1063,1069 ****
-
- while (!TAILQ_EMPTY(&db_rep->connections)) {
- conn = TAILQ_FIRST(&db_rep->connections);
-! __repmgr_cleanup_connection(dbenv, conn);
- }
-
- for (i = 0; i < db_rep->site_cnt; i++) {
---- 1112,1118 ----
-
- while (!TAILQ_EMPTY(&db_rep->connections)) {
- conn = TAILQ_FIRST(&db_rep->connections);
-! (void)__repmgr_cleanup_connection(dbenv, conn);
- }
-
- for (i = 0; i < db_rep->site_cnt; i++) {
-*** repmgr/repmgr_posix.c 2007-10-31 10:23:52.000000000 -0700
---- repmgr/repmgr_posix.c 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 21,26 ****
---- 21,28 ----
- size_t __repmgr_guesstimated_max = (128 * 1024);
- #endif
-
-+ static int __repmgr_conn_work __P((DB_ENV *,
-+ REPMGR_CONNECTION *, fd_set *, fd_set *, int));
- static int finish_connecting __P((DB_ENV *, REPMGR_CONNECTION *));
-
- /*
-***************
-*** 189,194 ****
---- 191,284 ----
- }
-
- /*
-+ * PUBLIC: int __repmgr_await_drain __P((DB_ENV *,
-+ * PUBLIC: REPMGR_CONNECTION *, db_timeout_t));
-+ *
-+ * Waits for space to become available on the connection's output queue.
-+ * Various ways we can exit:
-+ *
-+ * 1. queue becomes non-full
-+ * 2. exceed time limit
-+ * 3. connection becomes defunct (due to error in another thread)
-+ * 4. repmgr is shutting down
-+ * 5. any unexpected system resource failure
-+ *
-+ * In cases #3 and #5 we return an error code. Caller is responsible for
-+ * distinguishing the remaining cases if desired.
-+ *
-+ * !!!
-+ * Caller must hold repmgr->mutex.
-+ */
-+ int
-+ __repmgr_await_drain(dbenv, conn, timeout)
-+ DB_ENV *dbenv;
-+ REPMGR_CONNECTION *conn;
-+ db_timeout_t timeout;
-+ {
-+ DB_REP *db_rep;
-+ struct timespec deadline;
-+ int ret;
-+
-+ db_rep = dbenv->rep_handle;
-+
-+ __repmgr_compute_wait_deadline(dbenv, &deadline, timeout);
-+
-+ ret = 0;
-+ while (conn->out_queue_length >= OUT_QUEUE_LIMIT) {
-+ ret = pthread_cond_timedwait(&conn->drained,
-+ &db_rep->mutex, &deadline);
-+ switch (ret) {
-+ case 0:
-+ if (db_rep->finished)
-+ goto out; /* #4. */
-+ /*
-+ * Another thread could have stumbled into an error on
-+ * the socket while we were waiting.
-+ */
-+ if (F_ISSET(conn, CONN_DEFUNCT)) {
-+ ret = DB_REP_UNAVAIL; /* #3. */
-+ goto out;
-+ }
-+ break;
-+ case ETIMEDOUT:
-+ F_SET(conn, CONN_CONGESTED);
-+ ret = 0;
-+ goto out; /* #2. */
-+ default:
-+ goto out; /* #5. */
-+ }
-+ }
-+ /* #1. */
-+
-+ out:
-+ return (ret);
-+ }
-+
-+ /*
-+ * PUBLIC: int __repmgr_alloc_cond __P((cond_var_t *));
-+ *
-+ * Initialize a condition variable (in allocated space).
-+ */
-+ int
-+ __repmgr_alloc_cond(c)
-+ cond_var_t *c;
-+ {
-+ return (pthread_cond_init(c, NULL));
-+ }
-+
-+ /*
-+ * PUBLIC: int __repmgr_free_cond __P((cond_var_t *));
-+ *
-+ * Clean up a previously initialized condition variable.
-+ */
-+ int
-+ __repmgr_free_cond(c)
-+ cond_var_t *c;
-+ {
-+ return (pthread_cond_destroy(c));
-+ }
-+
-+ /*
- * PUBLIC: int __repmgr_init_sync __P((DB_ENV *, DB_REP *));
- *
- * Allocate/initialize all data necessary for thread synchronization. This
-***************
-*** 443,449 ****
- 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;
---- 533,539 ----
- REPMGR_RETRY *retry;
- db_timespec timeout;
- fd_set reads, writes;
-! int ret, flow_control, maxfd;
- u_int8_t buf[10]; /* arbitrary size */
-
- flow_control = FALSE;
-***************
-*** 477,482 ****
---- 567,575 ----
- * each one.
- */
- TAILQ_FOREACH(conn, &db_rep->connections, entries) {
-+ if (F_ISSET(conn, CONN_DEFUNCT))
-+ continue;
-+
- if (F_ISSET(conn, CONN_CONNECTING)) {
- FD_SET((u_int)conn->fd, &reads);
- FD_SET((u_int)conn->fd, &writes);
-***************
-*** 533,616 ****
- 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;
-! }
- }
-
- /*
---- 626,650 ----
- return (ret);
- }
- }
- LOCK_MUTEX(db_rep->mutex);
-
- if ((ret = __repmgr_retry_connections(dbenv)) != 0)
- goto out;
-
- /*
-! * Examine each connection, to see what work needs to be done.
-! *
-! * 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 ((ret = __repmgr_conn_work(dbenv,
-! conn, &reads, &writes, flow_control)) != 0)
-! goto out;
- }
-
- /*
-***************
-*** 637,642 ****
---- 671,719 ----
- }
-
- static int
-+ __repmgr_conn_work(dbenv, conn, reads, writes, flow_control)
-+ DB_ENV *dbenv;
-+ REPMGR_CONNECTION *conn;
-+ fd_set *reads, *writes;
-+ int flow_control;
-+ {
-+ int ret;
-+ u_int fd;
-+
-+ if (F_ISSET(conn, CONN_DEFUNCT)) {
-+ /*
-+ * Deferred clean-up, from an error that happened in another
-+ * thread, while we were sleeping in select().
-+ */
-+ return (__repmgr_cleanup_connection(dbenv, conn));
-+ }
-+
-+ ret = 0;
-+ fd = (u_int)conn->fd;
-+
-+ if (F_ISSET(conn, CONN_CONNECTING)) {
-+ if (FD_ISSET(fd, reads) || FD_ISSET(fd, writes))
-+ ret = finish_connecting(dbenv, conn);
-+ } else {
-+ /*
-+ * Here, the site is connected, and the FD_SET's are valid.
-+ */
-+ if (FD_ISSET(fd, writes))
-+ ret = __repmgr_write_some(dbenv, conn);
-+
-+ if (ret == 0 && !flow_control && FD_ISSET(fd, reads))
-+ ret = __repmgr_read_from_site(dbenv, conn);
-+ }
-+
-+ if (ret == DB_REP_UNAVAIL) {
-+ if ((ret = __repmgr_bust_connection(dbenv, conn)) != 0)
-+ return (ret);
-+ ret = __repmgr_cleanup_connection(dbenv, conn);
-+ }
-+ return (ret);
-+ }
-+
-+ static int
- finish_connecting(dbenv, conn)
- DB_ENV *dbenv;
- REPMGR_CONNECTION *conn;
-***************
-*** 657,662 ****
---- 734,740 ----
- goto err_rpt;
- }
-
-+ DB_ASSERT(dbenv, F_ISSET(conn, CONN_CONNECTING));
- F_CLR(conn, CONN_CONNECTING);
- return (__repmgr_send_handshake(dbenv, conn));
-
-***************
-*** 671,690 ****
- "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);
---- 749,773 ----
- "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) {
-! STAT(db_rep->region->mstat.st_connect_fail++);
- 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. (Note that we don't have to worry about message threads
-! * blocking on a full output queue: that can't happen when we're only
-! * just connecting.)
- *
- * !!!
- * 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));
-! if ((ret = __repmgr_cleanup_connection(dbenv, conn)) != 0)
-! return (ret);
- ret = __repmgr_connect_site(dbenv, eid);
- DB_ASSERT(dbenv, ret != DB_REP_UNAVAIL);
- return (ret);
-*** repmgr/repmgr_sel.c 2007-10-31 10:23:52.000000000 -0700
---- repmgr/repmgr_sel.c 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 36,45 ****
-
- /*
- * 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)
---- 36,41 ----
-***************
-*** 133,139 ****
- case 0:
- return (0);
- case DB_REP_UNAVAIL:
-! return (__repmgr_bust_connection(dbenv, conn, TRUE));
- default:
- return (ret);
- }
---- 129,135 ----
- case 0:
- return (0);
- case DB_REP_UNAVAIL:
-! return (__repmgr_bust_connection(dbenv, conn));
- default:
- return (ret);
- }
-***************
-*** 254,263 ****
- * 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
---- 250,255 ----
-***************
-*** 332,338 ****
- case 0:
- break;
- case DB_REP_UNAVAIL:
-! return (__repmgr_bust_connection(dbenv, con, TRUE));
- default:
- return (ret);
- }
---- 324,330 ----
- case 0:
- break;
- case DB_REP_UNAVAIL:
-! return (__repmgr_bust_connection(dbenv, con));
- default:
- return (ret);
- }
-***************
-*** 437,443 ****
-
- DB_SET_DBT(rec, my_addr->host, strlen(my_addr->host) + 1);
-
-! return (__repmgr_send_one(dbenv, conn, REPMGR_HANDSHAKE, &cntrl, &rec));
- }
-
- /*
---- 429,443 ----
-
- DB_SET_DBT(rec, my_addr->host, strlen(my_addr->host) + 1);
-
-! /*
-! * It would of course be disastrous to block the select() thread, so
-! * pass the "blockable" argument as FALSE. Fortunately blocking should
-! * never be necessary here, because the hand-shake is always the first
-! * thing we send. Which is a good thing, because it would be almost as
-! * disastrous if we allowed ourselves to drop a handshake.
-! */
-! return (__repmgr_send_one(dbenv,
-! conn, REPMGR_HANDSHAKE, &cntrl, &rec, FALSE));
- }
-
- /*
-***************
-*** 854,859 ****
---- 854,872 ----
- conn->out_queue_length--;
- if (--msg->ref_count <= 0)
- __os_free(dbenv, msg);
-+
-+ /*
-+ * We've achieved enough movement to free up at least
-+ * one space in the outgoing queue. Wake any message
-+ * threads that may be waiting for space. Clear the
-+ * CONGESTED status so that when the queue reaches the
-+ * high-water mark again, the filling thread will be
-+ * allowed to try waiting again.
-+ */
-+ F_CLR(conn, CONN_CONGESTED);
-+ if (conn->blockers > 0 &&
-+ (ret = __repmgr_signal(&conn->drained)) != 0)
-+ return (ret);
- }
- }
-
-*** repmgr/repmgr_util.c 2007-10-31 10:23:52.000000000 -0700
---- repmgr/repmgr_util.c 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 103,108 ****
---- 103,113 ----
- db_rep = dbenv->rep_handle;
- if ((ret = __os_malloc(dbenv, sizeof(REPMGR_CONNECTION), &c)) != 0)
- return (ret);
-+ if ((ret = __repmgr_alloc_cond(&c->drained)) != 0) {
-+ __os_free(dbenv, c);
-+ return (ret);
-+ }
-+ c->blockers = 0;
-
- c->fd = s;
- c->flags = flags;
-*** repmgr/repmgr_windows.c 2007-10-31 10:23:52.000000000 -0700
---- repmgr/repmgr_windows.c 2007-10-31 10:23:53.000000000 -0700
-***************
-*** 11,16 ****
---- 11,19 ----
- #define __INCLUDE_NETWORKING 1
- #include "db_int.h"
-
-+ /* Convert time-out from microseconds to milliseconds, rounding up. */
-+ #define DB_TIMEOUT_TO_WINDOWS_TIMEOUT(t) (((t) + (US_PER_MS - 1)) / US_PER_MS)
-+
- typedef struct __ack_waiter {
- HANDLE event;
- const DB_LSN *lsnp;
-***************
-*** 120,136 ****
- {
- DB_REP *db_rep;
- ACK_WAITER *me;
-! DWORD ret;
-! DWORD timeout;
-
- db_rep = dbenv->rep_handle;
-
- if ((ret = allocate_wait_slot(dbenv, &me)) != 0)
- goto err;
-
-- /* convert time-out from microseconds to milliseconds, rounding up */
- timeout = db_rep->ack_timeout > 0 ?
-! ((db_rep->ack_timeout + (US_PER_MS - 1)) / US_PER_MS) : INFINITE;
- me->lsnp = lsnp;
- if ((ret = SignalObjectAndWait(db_rep->mutex, me->event, timeout,
- FALSE)) == WAIT_FAILED) {
---- 123,137 ----
- {
- DB_REP *db_rep;
- ACK_WAITER *me;
-! DWORD ret, timeout;
-
- db_rep = dbenv->rep_handle;
-
- if ((ret = allocate_wait_slot(dbenv, &me)) != 0)
- goto err;
-
- timeout = db_rep->ack_timeout > 0 ?
-! DB_TIMEOUT_TO_WINDOWS_TIMEOUT(db_rep->ack_timeout) : INFINITE;
- me->lsnp = lsnp;
- if ((ret = SignalObjectAndWait(db_rep->mutex, me->event, timeout,
- FALSE)) == WAIT_FAILED) {
-***************
-*** 211,216 ****
---- 212,296 ----
- db_rep->waiters->first_free = slot;
- }
-
-+ /* (See requirements described in repmgr_posix.c.) */
-+ int
-+ __repmgr_await_drain(dbenv, conn, timeout)
-+ DB_ENV *dbenv;
-+ REPMGR_CONNECTION *conn;
-+ db_timeout_t timeout;
-+ {
-+ DB_REP *db_rep;
-+ db_timespec deadline, delta, now;
-+ db_timeout_t t;
-+ DWORD duration, ret;
-+ int round_up;
-+
-+ db_rep = dbenv->rep_handle;
-+
-+ __os_gettime(dbenv, &deadline);
-+ DB_TIMEOUT_TO_TIMESPEC(timeout, &delta);
-+ timespecadd(&deadline, &delta);
-+
-+ while (conn->out_queue_length >= OUT_QUEUE_LIMIT) {
-+ if (!ResetEvent(conn->drained))
-+ return (GetLastError());
-+
-+ /* How long until the deadline? */
-+ __os_gettime(dbenv, &now);
-+ if (timespeccmp(&now, &deadline, >=)) {
-+ F_SET(conn, CONN_CONGESTED);
-+ return (0);
-+ }
-+ delta = deadline;
-+ timespecsub(&delta, &now);
-+ round_up = TRUE;
-+ DB_TIMESPEC_TO_TIMEOUT(t, &delta, round_up);
-+ duration = DB_TIMEOUT_TO_WINDOWS_TIMEOUT(t);
-+
-+ ret = SignalObjectAndWait(db_rep->mutex,
-+ conn->drained, duration, FALSE);
-+ LOCK_MUTEX(db_rep->mutex);
-+ if (ret == WAIT_FAILED)
-+ return (GetLastError());
-+ else if (ret == WAIT_TIMEOUT) {
-+ F_SET(conn, CONN_CONGESTED);
-+ return (0);
-+ } else
-+ DB_ASSERT(dbenv, ret == WAIT_OBJECT_0);
-+
-+ if (db_rep->finished)
-+ return (0);
-+ if (F_ISSET(conn, CONN_DEFUNCT))
-+ return (DB_REP_UNAVAIL);
-+ }
-+ return (0);
-+ }
-+
-+ /*
-+ * Creates a manual reset event, which is usually our best choice when we may
-+ * have multiple threads waiting on a single event.
-+ */
-+ int
-+ __repmgr_alloc_cond(c)
-+ cond_var_t *c;
-+ {
-+ HANDLE event;
-+
-+ if ((event = CreateEvent(NULL, TRUE, FALSE, NULL)) == NULL)
-+ return (GetLastError());
-+ *c = event;
-+ return (0);
-+ }
-+
-+ int
-+ __repmgr_free_cond(c)
-+ cond_var_t *c;
-+ {
-+ if (CloseHandle(*c))
-+ return (0);
-+ return (GetLastError());
-+ }
-+
- /*
- * Make resource allocation an all-or-nothing affair, outside of this and the
- * close_sync function. db_rep->waiters should be non-NULL iff all of these
-***************
-*** 488,493 ****
---- 568,576 ----
- * don't hurt anything flow-control-wise.
- */
- TAILQ_FOREACH(conn, &db_rep->connections, entries) {
-+ if (F_ISSET(conn, CONN_DEFUNCT))
-+ continue;
-+
- if (F_ISSET(conn, CONN_CONNECTING) ||
- !STAILQ_EMPTY(&conn->outbound_queue) ||
- (!flow_control || !IS_VALID_EID(conn->eid))) {
-***************
-*** 534,541 ****
- conn != NULL;
- conn = next) {
- next = TAILQ_NEXT(conn, entries);
-! if (F_ISSET(conn, CONN_DEFUNCT))
-! __repmgr_cleanup_connection(dbenv, conn);
- }
-
- /*
---- 617,626 ----
- conn != NULL;
- conn = next) {
- next = TAILQ_NEXT(conn, entries);
-! if (F_ISSET(conn, CONN_DEFUNCT) &&
-! (ret = __repmgr_cleanup_connection(dbenv,
-! conn)) != 0)
-! goto unlock;
- }
-
- /*
-***************
-*** 587,597 ****
- return (ret);
- }
-
-- /*
-- * !!!
-- * Only ever called on the select() thread, since we may call
-- * __repmgr_bust_connection(..., TRUE).
-- */
- static int
- handle_completion(dbenv, conn)
- DB_ENV *dbenv;
---- 672,677 ----
-***************
-*** 651,660 ****
- }
- }
-
-! return (0);
-!
-! err: if (ret == DB_REP_UNAVAIL)
-! return (__repmgr_bust_connection(dbenv, conn, TRUE));
- return (ret);
- }
-
---- 731,742 ----
- }
- }
-
-! err:
-! if (ret == DB_REP_UNAVAIL) {
-! if ((ret = __repmgr_bust_connection(dbenv, conn)) != 0)
-! return (ret);
-! ret = __repmgr_cleanup_connection(dbenv, conn);
-! }
- return (ret);
- }
-
-***************
-*** 708,714 ****
- }
-
- 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);
---- 790,797 ----
- }
-
- DB_ASSERT(dbenv, !TAILQ_EMPTY(&db_rep->connections));
-! if ((ret = __repmgr_cleanup_connection(dbenv, conn)) != 0)
-! return (ret);
- ret = __repmgr_connect_site(dbenv, eid);
- DB_ASSERT(dbenv, ret != DB_REP_UNAVAIL);
- return (ret);
projects / packages / db4.6.git / commitdiff