- do not define epoch 0

[packages/db4.6.git] / patch.4.6.21.4

*** 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);