Imported from /home/lorry/working-area/delta_berkeleydb/db-6.1.23.tar.gz.HEADdb-6.1.23master

1 files changed, 664 insertions, 62 deletions

diff --git a/src/repmgr/repmgr_sel.c b/src/repmgr/repmgr_sel.c
index ba14368f..c32dad25 100644
--- a/src/repmgr/repmgr_sel.c
+++ b/src/repmgr/repmgr_sel.c
@@ -1,7 +1,7 @@
 /*-
  * See the file LICENSE for redistribution information.
  *
- * Copyright (c) 2006, 2012 Oracle and/or its affiliates.  All rights reserved.
+ * Copyright (c) 2006, 2015 Oracle and/or its affiliates.  All rights reserved.
  *
  * $Id$
  */
@@ -12,7 +12,7 @@
 
 typedef int (*HEARTBEAT_ACTION) __P((ENV *));
 
-static int accept_handshake __P((ENV *, REPMGR_CONNECTION *, char *));
+static int accept_handshake __P((ENV *, REPMGR_CONNECTION *, char *, int *));
 static int accept_v1_handshake __P((ENV *, REPMGR_CONNECTION *, char *));
 static void check_min_log_file __P((ENV *));
 static int dispatch_msgin __P((ENV *, REPMGR_CONNECTION *));
@@ -23,13 +23,18 @@ static int process_parameters __P((ENV *,
 static int read_version_response __P((ENV *, REPMGR_CONNECTION *));
 static int record_permlsn __P((ENV *, REPMGR_CONNECTION *));
 static int __repmgr_call_election __P((ENV *));
+static int __repmgr_check_listener __P((ENV *));
+static int __repmgr_check_master_listener __P((ENV *));
 static int __repmgr_connector_main __P((ENV *, REPMGR_RUNNABLE *));
 static void *__repmgr_connector_thread __P((void *));
 static int __repmgr_next_timeout __P((ENV *,
     db_timespec *, HEARTBEAT_ACTION *));
+static int __repmgr_reset_last_rcvd __P((ENV *));
 static int __repmgr_retry_connections __P((ENV *));
 static int __repmgr_send_heartbeat __P((ENV *));
-static int __repmgr_try_one __P((ENV *, int));
+static int __repmgr_start_takeover __P((ENV *));
+static void *__repmgr_takeover_thread __P((void *));
+static int __repmgr_try_one __P((ENV *, int, int));
 static int resolve_collision __P((ENV *, REPMGR_SITE *, REPMGR_CONNECTION *));
 static int send_version_response __P((ENV *, REPMGR_CONNECTION *));
 
@@ -49,17 +54,24 @@ void *
 __repmgr_select_thread(argsp)
 	void *argsp;
 {
-	REPMGR_RUNNABLE *args;
 	ENV *env;
+	DB_THREAD_INFO *ip;
 	int ret;
+	REPMGR_RUNNABLE *args;
 
 	args = argsp;
 	env = args->env;
+	ip = NULL;
+	ret = 0;
 
-	if ((ret = __repmgr_select_loop(env))  != 0) {
+	ENV_ENTER_RET(env, ip, ret);
+	if (ret != 0 || (ret = __repmgr_select_loop(env)) != 0) {
 		__db_err(env, ret, DB_STR("3614", "select loop failed"));
+		ENV_LEAVE(env, ip);
 		(void)__repmgr_thread_failure(env, ret);
 	}
+	if (ret == 0)
+		ENV_LEAVE(env, ip);
 	return (NULL);
 }
 
@@ -71,12 +83,19 @@ __repmgr_bow_out(env)
 	ENV *env;
 {
 	DB_REP *db_rep;
+	REP *rep;
 	int ret;
 
 	db_rep = env->rep_handle;
+	rep = db_rep->region;
 	LOCK_MUTEX(db_rep->mutex);
 	ret = __repmgr_stop_threads(env);
 	UNLOCK_MUTEX(db_rep->mutex);
+	/*
+	 * Reset sites_avail so that it will be calculated correctly if this
+	 * site rejoins the group in the future.
+	 */
+	rep->sites_avail = 0;
 	DB_EVENT(env, DB_EVENT_REP_LOCAL_SITE_REMOVED, NULL);
 	return (ret);
 }
@@ -187,23 +206,53 @@ __repmgr_compute_timeout(env, timeout)
 	db_rep = env->rep_handle;
 
 	/*
-	 * There are two factors to consider: are heartbeats in use?  and, do we
+	 * There are four factors to consider: are heartbeats in use? do we
 	 * have any sites with broken connections that we ought to retry?
+	 * is there a listener process running locally? do we need to call
+	 * an election if no master listener exists?
 	 */
 	have_timeout = __repmgr_next_timeout(env, &t, NULL);
 
 	/* List items are in order, so we only have to examine the first one. */
 	if (!TAILQ_EMPTY(&db_rep->retries)) {
 		retry = TAILQ_FIRST(&db_rep->retries);
-		if (have_timeout) {
+		if (have_timeout)
 			/* Choose earliest timeout deadline. */
 			t = timespeccmp(&retry->time, &t, <) ? retry->time : t;
-		} else {
+		else {
 			t = retry->time;
 			have_timeout = TRUE;
 		}
 	}
 
+	/* Check listener every timeout in subordinate rep-aware process. */
+	if (IS_LISTENER_CAND(db_rep)) {
+		if (!timespecisset(&db_rep->l_listener_chk)) {
+			__os_gettime(env, &now, 1);
+			TIMESPEC_ADD_DB_TIMEOUT(&now, db_rep->l_listener_wait);
+			db_rep->l_listener_chk = now;
+		}
+		if (have_timeout)
+			t = timespeccmp(&db_rep->l_listener_chk, &t, <) ?
+			    db_rep->l_listener_chk : t;
+		else {
+			t = db_rep->l_listener_chk;
+			have_timeout = TRUE;
+		}
+	}
+
+	/* Check master listener if needed. */
+	if (FLD_ISSET(db_rep->region->config, REP_C_AUTOTAKEOVER) &&
+	    timespecisset(&db_rep->m_listener_chk)) {
+		if (have_timeout)
+			t = timespeccmp(&db_rep->m_listener_chk, &t, <) ?
+			    db_rep->m_listener_chk : t;
+		else {
+			t = db_rep->m_listener_chk;
+			have_timeout = TRUE;
+		}
+	}
+
 	if (have_timeout) {
 		__os_gettime(env, &now, 1);
 		if (timespeccmp(&now, &t, >=))
@@ -242,7 +291,17 @@ __repmgr_next_timeout(env, deadline, action)
 
 	if (rep->master_id == db_rep->self_eid &&
 	    rep->heartbeat_frequency > 0) {
-		t = db_rep->last_bcast;
+		/*
+		 * A temporary master in preferred master mode must send
+		 * regular heartbeats regardless of other activity because
+		 * the preferred master requires a heartbeat to take over as
+		 * master after it has synced with the temporary master.
+		 */
+		if (IS_PREFMAS_MODE(env) &&
+		    FLD_ISSET(rep->config, REP_C_PREFMAS_CLIENT))
+			t = db_rep->last_hbeat;
+		else
+			t = db_rep->last_bcast;
 		TIMESPEC_ADD_DB_TIMEOUT(&t, rep->heartbeat_frequency);
 		my_action = __repmgr_send_heartbeat;
 	} else if ((master = __repmgr_connected_master(env)) != NULL &&
@@ -301,6 +360,24 @@ __repmgr_send_heartbeat(env)
 
 	db_rep = env->rep_handle;
 	rep = db_rep->region;
+	ret = 0;
+
+	/*
+	 * Check test hook preventing heartbeats and connection attempts.
+	 * This is used to create and maintain a dupmaster condition in
+	 * a test until the test hook is rescinded.
+	 */
+	DB_TEST_SET(env->test_abort, DB_TEST_REPMGR_HEARTBEAT);
+
+	/*
+	 * Track last heartbeat for temporary master in preferred master
+	 * mode so that it will send regular heartbeats regardless of
+	 * other activity.
+	 */
+	if (IS_PREFMAS_MODE(env) &&
+	    FLD_ISSET(rep->config, REP_C_PREFMAS_CLIENT) &&
+	    rep->master_id == db_rep->self_eid)
+		__os_gettime(env, &db_rep->last_hbeat, 1);
 
 	permlsn.generation = rep->gen;
 	if ((ret = __rep_get_maxpermlsn(env, &permlsn.lsn)) != 0)
@@ -310,8 +387,11 @@ __repmgr_send_heartbeat(env)
 	control.size = __REPMGR_PERMLSN_SIZE;
 
 	DB_INIT_DBT(rec, NULL, 0);
-	return (__repmgr_send_broadcast(env,
-	    REPMGR_HEARTBEAT, &control, &rec, &unused1, &unused2, &unused3));
+	ret =__repmgr_send_broadcast(env,
+	    REPMGR_HEARTBEAT, &control, &rec, &unused1, &unused2, &unused3);
+
+DB_TEST_RECOVERY_LABEL
+	return (ret);
 }
 
 /*
@@ -373,6 +453,8 @@ __repmgr_check_timeouts(env)
 	HEARTBEAT_ACTION action;
 	int ret;
 
+	ret = 0;
+
 	/*
 	 * Figure out the next heartbeat-related thing to be done.  Then, if
 	 * it's time to do it, do so.
@@ -384,7 +466,342 @@ __repmgr_check_timeouts(env)
 			return (ret);
 	}
 
-	return (__repmgr_retry_connections(env));
+	/* Check the existence of local listener. */
+	if ((ret = __repmgr_check_listener(env)) != 0)
+		return (ret);
+
+	/* Check the existence of master listener. */
+	if ((ret = __repmgr_check_master_listener(env)) != 0)
+		return (ret);
+
+	/*
+	 * Check test hook preventing heartbeats and connection attempts.
+	 * This is used to create and maintain a dupmaster condition in
+	 * a test until the test hook is rescinded.
+	 */
+	DB_TEST_SET(env->test_abort, DB_TEST_REPMGR_HEARTBEAT);
+
+	ret = __repmgr_retry_connections(env);
+
+DB_TEST_RECOVERY_LABEL
+	return (ret);
+}
+
+/*
+ * Check the existence of the listener process on the local site.  If one
+ * does not exist and the current process is a subordinate rep-aware process,
+ * then start a takeover thread to covert this process to the listener process.
+ */
+static int
+__repmgr_check_listener(env)
+	ENV *env;
+{
+	DB_REP *db_rep;
+	REP *rep;
+	SITEINFO *sites;
+	db_timespec t;
+	int ret;
+
+	db_rep = env->rep_handle;
+	rep = db_rep->region;
+	ret = 0;
+
+	/*
+	 * Only subordinate rep-aware process can take over listener role, so
+	 * no need to check listener in listener process or rep unaware process.
+	 */
+	if (!IS_LISTENER_CAND(db_rep))
+		return (0);
+
+	/*
+	 * If the listener quits due to site removal, no subordinate process
+	 * should take over as listener as the current site is not expected
+	 * to be active in the group.  Check the status from the site array
+	 * in the shared region instead of that in the GMDB.  We do this
+	 * because the GMDB doesn't apply the change yet when replication
+	 * is stopped on the removed site.
+	 */
+	sites = R_ADDR(env->reginfo, rep->siteinfo_off);
+	if (sites[rep->self_eid].status == SITE_DELETING)
+		return (0);
+
+	/*
+	 * Check the listener after timeout.  If there is no listener, we
+	 * take over.  During takeover, we will refresh all connections.
+	 * A subordinate process does not have an up-to-date site list, so sync
+	 * up addresses from the in-memory site array before takeover.
+	 */
+	__os_gettime(env, &t, 1);
+	if (timespeccmp(&t, &db_rep->l_listener_chk, >=)) {
+		/* Compute the next timeout. */
+		TIMESPEC_ADD_DB_TIMEOUT(&t, db_rep->l_listener_wait);
+		db_rep->l_listener_chk = t;
+
+		/* Check if site address information needs to be refreshed. */
+		if ((rep->siteinfo_seq > db_rep->siteinfo_seq) &&
+		    (ret = __repmgr_sync_siteaddr(env)) != 0)
+			return (ret);
+
+		if (rep->listener == 0)
+			ret = __repmgr_start_takeover(env);
+	}
+	return (ret);
+}
+
+/*
+ * Start a thread to take over the listener role in the current subordinate
+ * process.
+ */
+static int
+__repmgr_start_takeover(env)
+	ENV *env;
+{
+	DB_REP *db_rep;
+	REPMGR_RUNNABLE *th;
+	int ret;
+
+	db_rep = env->rep_handle;
+	th = db_rep->takeover_thread;
+	if (th == NULL) {
+		if ((ret = __os_calloc(env, 1, sizeof(REPMGR_RUNNABLE),
+		    &th)) != 0)
+			return (ret);
+		db_rep->takeover_thread = th;
+	} else if (th->finished) {
+		if ((ret = __repmgr_thread_join(th)) != 0)
+			return (ret);
+	} else {
+		RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+		    "takeover thread still running"));
+		return (0);
+	}
+	th->run = __repmgr_takeover_thread;
+	if ((ret = __repmgr_thread_start(env, th)) != 0) {
+		__os_free(env, th);
+		db_rep->takeover_thread = NULL;
+	}
+	return (ret);
+}
+
+/*
+ * Take over listener role in the current subordinate process.
+ */
+static void *
+__repmgr_takeover_thread(argsp)
+	void *argsp;
+{
+	DB_REP *db_rep;
+	DB_THREAD_INFO *ip;
+	ENV *env;
+	REP *rep;
+	REPMGR_RUNNABLE *th;
+	int nthreads, ret, save_policy;
+
+	th = argsp;
+	env = th->env;
+	db_rep = env->rep_handle;
+	ip = NULL;
+	rep = db_rep->region;
+	ret = 0;
+
+	ENV_ENTER_RET(env, ip, ret);
+	if (ret != 0)
+		goto out;
+	RPRINT(env, (env, DB_VERB_REPMGR_MISC, "starting takeover thread"));
+	/*
+	 * It is likely that there is an old heartbeat ready to expire
+	 * immediately upon restarting repmgr, leading to an unnecessary
+	 * election.  Reset the expiration countdown here to avoid this.
+	 */
+	if ((ret = __repmgr_reset_last_rcvd(env)) != 0)
+		goto out;
+	/*
+	 * If nthreads is set to be 0 in the current subordinate process, use
+	 * the value in the last listener. The nthreads should be larger than
+	 * 0 in listener.
+	 */
+	nthreads = db_rep->config_nthreads == 0 ? (int)rep->listener_nthreads :
+	    db_rep->config_nthreads;
+	/*
+	 * It is possible that this subordinate process does not have intact
+	 * connections to the other sites.  For most ack policies, restarting
+	 * repmgr will wait for acks when it commits its transaction to reload
+	 * the gmdb.  Temporarily set the ack policy to NONE for the takeover
+	 * so that it is not delayed waiting for acks that can never come.
+	 */
+	save_policy = rep->perm_policy;
+	rep->perm_policy = DB_REPMGR_ACKS_NONE;
+	/*
+	 * Restart the repmgr as listener.  If DB_REP_IGNORE is returned,
+	 * the current process has become listener.  If DB_REP_UNAVAIL is
+	 * returned, the site has been removed from the group and no listener
+	 * should be started.  For any other error, if the replication is
+	 * stopped because of the takeover thread, we will notify the
+	 * application.
+	 */
+	ret = __repmgr_start_int(env, nthreads, F_ISSET(rep, REP_F_MASTER) ?
+	    DB_REP_MASTER : DB_REP_CLIENT);
+	if (ret == 0 && !IS_SUBORDINATE(db_rep) &&
+	    db_rep->repmgr_status == running) {
+		STAT(rep->mstat.st_takeovers++);
+		RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+		    "finished takeover and became listener"));
+	} else if (ret != 0 && db_rep->repmgr_status == stopped) {
+		RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+		    "failed to take over, repmgr was stopped"));
+		DB_EVENT(env, DB_EVENT_REP_AUTOTAKEOVER_FAILED, NULL);
+	} else {
+		/* The current process is not changed to listener. */
+		RPRINT(env, (env, DB_VERB_REPMGR_MISC, "failed to take over"));
+	}
+	rep->perm_policy = save_policy;
+	RPRINT(env, (env, DB_VERB_REPMGR_MISC, "takeover thread is exiting"));
+	ENV_LEAVE(env, ip);
+out:	th->finished = TRUE;
+	return (NULL);
+}
+
+/*
+ * Reset the last_rcvd_timestamp to restart the wait for a heartbeat
+ * monitor expiration.
+ */
+static int
+__repmgr_reset_last_rcvd(env)
+	ENV *env;
+{
+	DB_REP *db_rep;
+	REPMGR_SITE *master;
+
+	db_rep = env->rep_handle;
+
+	LOCK_MUTEX(db_rep->mutex);
+	if ((master = __repmgr_connected_master(env)) != NULL)
+		__os_gettime(env, &master->last_rcvd_timestamp, 1);
+	UNLOCK_MUTEX(db_rep->mutex);
+	return (0);
+}
+
+/*
+ * Monitor the connection to master listener.  When the master listener is
+ * disconnected and some other master process might take over as listener
+ * soon, we will delay the election.  After the delay if there is still no
+ * connection from master listener, call an election then.
+ */
+static int
+__repmgr_check_master_listener(env)
+	ENV *env;
+{
+	DB_REP *db_rep;
+	REP *rep;
+	REPMGR_SITE *master;
+	db_timespec t;
+	u_int32_t flags;
+	int ret;
+
+	db_rep = env->rep_handle;
+	rep = db_rep->region;
+	ret = 0;
+
+	/*
+	 * We only check for a master listener if m_listener_chk is set.
+	 * The field is only set when __repmgr_bust_connection() previously
+	 * detected the loss of our connection to the master listener.
+	 * If rep->master_id is invalid, wait until it is ready to check.
+	 */
+	if (!FLD_ISSET((db_rep)->region->config, REP_C_AUTOTAKEOVER) ||
+	    !timespecisset(&db_rep->m_listener_chk) ||
+	    !IS_VALID_EID(rep->master_id))
+		return (0);
+
+	__os_gettime(env, &t, 1);
+	if (timespeccmp(&t, &db_rep->m_listener_chk, >=)) {
+		master = SITE_FROM_EID(db_rep->region->master_id);
+		if (master->ref.conn.out == NULL &&
+		    master->ref.conn.in == NULL) {
+			flags = ELECT_F_EVENT_NOTIFY;
+			if (FLD_ISSET(db_rep->region->config, REP_C_ELECTIONS))
+				LF_SET(ELECT_F_IMMED | ELECT_F_FAST);
+			else
+				RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+				    "Master failure, but no elections"));
+
+			/*
+			 * In preferred master mode, a client that has lost its
+			 * connection to the master uses an election thread to
+			 * restart as master.
+			 */
+			if (IS_PREFMAS_MODE(env)) {
+				RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+"check_master_listener setting preferred master temp master"));
+				db_rep->prefmas_pending = start_temp_master;
+			}
+
+			ret = __repmgr_init_election(env, flags);
+		}
+		/*
+		 * If the delay has expired reset m_listener_chk.  We reset
+		 * it whether or not the master listener process comes back
+		 * so that we will not continue checking for a master listener
+		 * indefinitely.
+		 */
+		timespecclear(&db_rep->m_listener_chk);
+	}
+	return (ret);
+}
+
+/*
+ * Wake up I/O waiting in selector thread, refresh connections to all connected
+ * and present sites.
+ *
+ * PUBLIC: int __repmgr_refresh_selector __P((ENV *));
+ */
+int
+__repmgr_refresh_selector(env)
+	ENV *env;
+{
+	DB_REP *db_rep;
+	REP *rep;
+	REPMGR_RETRY *retry;
+	REPMGR_SITE *site;
+	SITEINFO *sites;
+	int eid, ret;
+
+	db_rep = env->rep_handle;
+	rep = db_rep->region;
+
+	if ((ret = __repmgr_wake_main_thread(env)) != 0)
+		return (ret);
+
+	FOR_EACH_REMOTE_SITE_INDEX(eid) {
+		SET_LISTENER_CAND(1, = 0);
+		site = SITE_FROM_EID(eid);
+
+		/*
+		 * It is possible some sites were left in a paused state
+		 * during the switch, so they have to be removed from the
+		 * retry list.
+		 */
+		if (site->state == SITE_PAUSING) {
+			retry = site->ref.retry;
+			if (retry != NULL) {
+				RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+				    "Removing site from retry list eid %lu",
+				    (u_long)eid));
+				TAILQ_REMOVE(&db_rep->retries, retry, entries);
+				__os_free(env, retry);
+				site->ref.retry = NULL;
+			}
+
+		}
+		/*
+		 * Try to connect to any site that is now PRESENT after
+		 * rereading the gmdb.
+		 */
+		if (site->membership == SITE_PRESENT &&
+		    (ret = __repmgr_try_one(env, eid, TRUE)) != 0)
+			return (ret);
+	}
+	return (0);
 }
 
 /*
@@ -415,10 +832,11 @@ __repmgr_retry_connections(env)
 		__os_free(env, retry);
 		DB_ASSERT(env, IS_VALID_EID(eid));
 		site = SITE_FROM_EID(eid);
+		site->ref.retry = NULL;
 		DB_ASSERT(env, site->state == SITE_PAUSING);
 
 		if (site->membership == SITE_PRESENT) {
-			if ((ret = __repmgr_try_one(env, eid)) != 0)
+			if ((ret = __repmgr_try_one(env, eid, FALSE)) != 0)
 				return (ret);
 		} else
 			site->state = SITE_IDLE;
@@ -437,11 +855,23 @@ __repmgr_first_try_connections(env)
 	ENV *env;
 {
 	DB_REP *db_rep;
+	REP *rep;
 	REPMGR_SITE *site;
+	SITEINFO *sites;
 	int eid, ret;
 
 	db_rep = env->rep_handle;
+	rep = db_rep->region;
+
+	/*
+	 * Check test hook preventing heartbeats and connection attempts.
+	 * This is used to create and maintain a dupmaster condition in
+	 * a test until the test hook is rescinded.
+	 */
+	DB_TEST_SET(env->test_abort, DB_TEST_REPMGR_HEARTBEAT);
+
 	FOR_EACH_REMOTE_SITE_INDEX(eid) {
+		SET_LISTENER_CAND(1, = 0);
 		site = SITE_FROM_EID(eid);
 		/*
 		 * Normally all sites would be IDLE here.  But if a user thread
@@ -453,19 +883,22 @@ __repmgr_first_try_connections(env)
 		 */
 		if (site->state == SITE_IDLE &&
 		    site->membership == SITE_PRESENT &&
-		    (ret = __repmgr_try_one(env, eid)) != 0)
+		    (ret = __repmgr_try_one(env, eid, FALSE)) != 0)
 			return (ret);
 	}
+DB_TEST_RECOVERY_LABEL
 	return (0);
 }
 
 /*
- * Starts a thread to open a connection to the site at the given EID.
+ * Starts a thread to open a connection to the site at the given EID.  We might
+ * have no connection to the site, or an existing connection to be replaced.
  */
 static int
-__repmgr_try_one(env, eid)
+__repmgr_try_one(env, eid, refresh)
 	ENV *env;
 	int eid;
+	int refresh;
 {
 	DB_REP *db_rep;
 	REPMGR_SITE *site;
@@ -488,13 +921,22 @@ __repmgr_try_one(env, eid)
 		  "eid %lu previous connector thread still running; will retry",
 		    (u_long)eid));
 		return (__repmgr_schedule_connection_attempt(env,
-			eid, FALSE));
+			eid, refresh));
 	}
 
 	site->state = SITE_CONNECTING;
 
 	th->run = __repmgr_connector_thread;
-	th->args.eid = eid;
+	th->args.conn_th.eid = eid;
+	/*
+	 * The flag CONNECT_F_REFRESH indicates an immediate connection attempt
+	 * should be scheduled if the current connection attempt fails.  It is
+	 * turned on before the first attempt to refresh the connection but
+	 * turned off if the first attempt fails.  In this way, when refreshing
+	 * the connection, there will be at most two immediate connection
+	 * attempts, after that, retry as usual.
+	 */
+	th->args.conn_th.flags = refresh ? CONNECT_F_REFRESH : 0;
 	if ((ret = __repmgr_thread_start(env, th)) != 0) {
 		__os_free(env, th);
 		site->connector = NULL;
@@ -506,21 +948,33 @@ static void *
 __repmgr_connector_thread(argsp)
 	void *argsp;
 {
-	REPMGR_RUNNABLE *th;
 	ENV *env;
+	DB_THREAD_INFO *ip;
+	REPMGR_RUNNABLE *th;
 	int ret;
 
 	th = argsp;
 	env = th->env;
+	ip = NULL;
+	ret = 0;
 
-	RPRINT(env, (env, DB_VERB_REPMGR_MISC,
-	    "starting connector thread, eid %u", th->args.eid));
-	if ((ret = __repmgr_connector_main(env, th)) != 0) {
+	ENV_ENTER_RET(env, ip, ret);
+	if (ret == 0)
+		RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+		    "starting connector thread, eid %u",
+		    th->args.conn_th.eid));
+	if (ret != 0 || (ret = __repmgr_connector_main(env, th)) != 0) {
 		__db_err(env, ret, DB_STR("3617", "connector thread failed"));
+		RPRINT(env, (env,
+		    DB_VERB_REPMGR_MISC, "connector thread is exiting"));
+		ENV_LEAVE(env, ip);
 		(void)__repmgr_thread_failure(env, ret);
 	}
-	RPRINT(env, (env, DB_VERB_REPMGR_MISC, "connector thread is exiting"));
-
+	if (ret == 0) {
+		RPRINT(env, (env,
+		    DB_VERB_REPMGR_MISC, "connector thread is exiting"));
+		ENV_LEAVE(env, ip);
+	}
 	th->finished = TRUE;
 	return (NULL);
 }
@@ -542,8 +996,8 @@ __repmgr_connector_main(env, th)
 	ret = 0;
 
 	LOCK_MUTEX(db_rep->mutex);
-	DB_ASSERT(env, IS_VALID_EID(th->args.eid));
-	site = SITE_FROM_EID(th->args.eid);
+	DB_ASSERT(env, IS_VALID_EID(th->args.conn_th.eid));
+	site = SITE_FROM_EID(th->args.conn_th.eid);
 	if (site->state != SITE_CONNECTING && db_rep->repmgr_status == stopped)
 		goto unlock;
 
@@ -563,7 +1017,8 @@ __repmgr_connector_main(env, th)
 	UNLOCK_MUTEX(db_rep->mutex);
 
 	if ((ret = __repmgr_connect(env, &netaddr, &conn, &err)) == 0) {
-		DB_EVENT(env,  DB_EVENT_REP_CONNECT_ESTD, &th->args.eid);
+		DB_EVENT(env,
+		    DB_EVENT_REP_CONNECT_ESTD, &th->args.conn_th.eid);
 		LOCK_MUTEX(db_rep->mutex);
 		if ((ret = __repmgr_set_nonblock_conn(conn)) != 0) {
 			__db_err(env, ret, DB_STR("3618",
@@ -571,33 +1026,53 @@ __repmgr_connector_main(env, th)
 			goto cleanup;
 		}
 		conn->type = REP_CONNECTION;
-		site = SITE_FROM_EID(th->args.eid);
+		site = SITE_FROM_EID(th->args.conn_th.eid);
 		if (site->state != SITE_CONNECTING ||
 		    db_rep->repmgr_status == stopped)
 			goto cleanup;
 
-		conn->eid = th->args.eid;
-		site = SITE_FROM_EID(th->args.eid);
-		site->ref.conn.out = conn;
+		conn->eid = th->args.conn_th.eid;
+		site = SITE_FROM_EID(th->args.conn_th.eid);
+		/*
+		 * If there is an existing outgoing connection, disable it and
+		 * replace it with a new connection.  The sites for a formerly
+		 * subordinate handle that is now taking over might still be
+		 * SITE_CONNECTING.  Set to SITE_CONNECTED before disabling
+		 * connection so that sites_avail is correctly maintained.
+		 */
 		site->state = SITE_CONNECTED;
+		if (site->ref.conn.out != NULL)
+			(void)__repmgr_disable_connection(env,
+			    site->ref.conn.out);
+		site->ref.conn.out = conn;
 		__os_gettime(env, &site->last_rcvd_timestamp, 1);
 		ret = __repmgr_wake_main_thread(env);
 	} else if (ret == DB_REP_UNAVAIL) {
 		/* Retryable error while trying to connect: retry later. */
-		info.eid = th->args.eid;
+		info.eid = th->args.conn_th.eid;
 		info.error = err;
 		DB_EVENT(env, DB_EVENT_REP_CONNECT_TRY_FAILED, &info);
 		STAT(db_rep->region->mstat.st_connect_fail++);
 
 		LOCK_MUTEX(db_rep->mutex);
-		site = SITE_FROM_EID(th->args.eid);
+		site = SITE_FROM_EID(th->args.conn_th.eid);
 		if (site->state != SITE_CONNECTING ||
 		    db_rep->repmgr_status == stopped) {
 			ret = 0;
 			goto unlock;
 		}
+		/*
+		 * If it fails to create a new outgoing connection to replace
+		 * the existing one in the first attempt, schedule another
+		 * immediate attempt.  If it is our second attempt, disable
+		 * the existing connections and retry as normal.
+		 */
+		if (site->ref.conn.out != NULL && th->args.conn_th.flags == 0)
+			(void)__repmgr_disable_connection(env,
+			    site->ref.conn.out);
 		ret = __repmgr_schedule_connection_attempt(env,
-		    th->args.eid, FALSE);
+		    th->args.conn_th.eid,
+		    th->args.conn_th.flags == CONNECT_F_REFRESH);
 	} else
 		goto out;
 
@@ -842,6 +1317,7 @@ prepare_input(env, conn)
 		if ((ret = __os_malloc(env, memsize, &membase)) != 0)
 			return (ret);
 		conn->input.rep_message = membase;
+		conn->input.rep_message->size = memsize;
 		conn->input.rep_message->msg_hdr = msg_hdr;
 		conn->input.rep_message->v.repmsg.originating_eid = conn->eid;
 
@@ -876,6 +1352,7 @@ prepare_input(env, conn)
 		if ((ret = __os_malloc(env, memsize, &membase)) != 0)
 			return (ret);
 		conn->input.rep_message = membase;
+		conn->input.rep_message->size = memsize;
 		conn->input.rep_message->msg_hdr = msg_hdr;
 		conn->input.rep_message->v.appmsg.conn = conn;
 
@@ -891,6 +1368,7 @@ prepare_input(env, conn)
 		if ((ret = __os_malloc(env, size, &membase)) != 0)
 			return (ret);
 		conn->input.rep_message = membase;
+		conn->input.rep_message->size = size;
 		conn->input.rep_message->msg_hdr = msg_hdr;
 
 		/*
@@ -1065,16 +1543,18 @@ dispatch_msgin(env, conn)
 	ENV *env;
 	REPMGR_CONNECTION *conn;
 {
+	DBT *dbt;
 	DB_REP *db_rep;
-	REPMGR_SITE *site;
-	REPMGR_RUNNABLE *th;
+	REP *rep;
 	REPMGR_RESPONSE *resp;
-	DBT *dbt;
+	REPMGR_RUNNABLE *th;
+	REPMGR_SITE *site;
 	char *hostname;
-	int eid, ret;
+	int eid, ret, subord;
 
 	DB_ASSERT(env, conn->reading_phase == DATA_PHASE);
 	db_rep = env->rep_handle;
+	rep = db_rep->region;
 
 	switch (conn->state) {
 	case CONN_CONNECTED:
@@ -1129,9 +1609,22 @@ dispatch_msgin(env, conn)
 			dbt = &conn->input.repmgr_msg.rec;
 			hostname = dbt->data;
 			hostname[dbt->size-1] = '\0';
-			if ((ret = accept_handshake(env, conn, hostname)) != 0)
+			if ((ret = accept_handshake(env,
+			    conn, hostname, &subord)) != 0)
 				return (ret);
 			conn->state = CONN_READY;
+			site = SITE_FROM_EID(conn->eid);
+			/*
+			 * Do not increase sites_avail redundantly for an
+			 * incoming subordinate connection.
+			 */
+			if (conn->type == REP_CONNECTION &&
+			    site->state == SITE_CONNECTED && !subord) {
+				rep->sites_avail++;
+				RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+    "msgin: EID %lu CONNECTED, READY.  sites_avail %lu",
+    (u_long)conn->eid, (u_long)rep->sites_avail));
+			}
 			break;
 		case REPMGR_OWN_MSG:
 			/*
@@ -1279,9 +1772,11 @@ process_own_msg(env, conn)
 	REPMGR_SITE *site;
 	REPMGR_MESSAGE *msg;
 	__repmgr_connect_reject_args reject;
+	__repmgr_v4connect_reject_args v4reject;
 	__repmgr_parm_refresh_args parms;
 	int ret;
 
+	db_rep = env->rep_handle;
 	ret = 0;
 	/*
 	 * Set "msg" to point to the message struct.  If we do all necessary
@@ -1293,28 +1788,61 @@ process_own_msg(env, conn)
 	switch (REPMGR_OWN_MSG_TYPE((msg = conn->input.rep_message)->msg_hdr)) {
 	case REPMGR_CONNECT_REJECT:
 		dbt = &msg->v.gmdb_msg.request;
-		if ((ret = __repmgr_connect_reject_unmarshal(env,
-		    &reject, dbt->data, dbt->size, NULL)) != 0)
-			return (DB_REP_UNAVAIL);
+		if (conn->version < 5) {
+			if ((ret = __repmgr_v4connect_reject_unmarshal(env,
+			    &v4reject, dbt->data, dbt->size, NULL)) != 0)
+				return (DB_REP_UNAVAIL);
+			reject.version = v4reject.version;
+			reject.gen = v4reject.gen;
+			reject.status = 0;
+		} else {
+			if ((ret = __repmgr_connect_reject_unmarshal(env,
+			    &reject, dbt->data, dbt->size, NULL)) != 0)
+				return (DB_REP_UNAVAIL);
+		}
 
 		/*
 		 * If we're being rejected by someone who has more up-to-date
-		 * membership information than we do, it means we have been
-		 * removed from the group.  If we've just gotten started, we can
-		 * make one attempt at automatically rejoining; otherwise we bow
-		 * out gracefully.
+		 * membership information than we do, it means we are not in
+		 * the group.  If we've just gotten started, or our status is
+		 * adding, we can make one attempt at automatically rejoining;
+		 * otherwise we bow out gracefully.
 		 */
 		RPRINT(env, (env, DB_VERB_REPMGR_MISC,
-			"got rejection msg citing version %lu/%lu",
-			(u_long)reject.gen, (u_long)reject.version));
+	"got rejection msg citing version %lu/%lu mine %lu/%lu membership %lu",
+			(u_long)reject.gen, (u_long)reject.version,
+			(u_long)db_rep->member_version_gen,
+			(u_long)db_rep->membership_version,
+			(u_long)reject.status));
 
 		if (__repmgr_gmdb_version_cmp(env,
 		    reject.gen, reject.version) > 0) {
-			if (env->rep_handle->seen_repmsg)
+			if (db_rep->seen_repmsg && reject.status != SITE_ADDING)
 				ret = DB_DELETED;
-			else if ((ret = __repmgr_defer_op(env,
-			    REPMGR_REJOIN)) == 0)
-				ret = DB_REP_UNAVAIL;
+			else {
+				/*
+				 * If 2SITE_STRICT is off, we are likely to
+				 * win an election with our own vote before
+				 * discovering there is already a master.
+				 * Set indicator to defer the election until
+				 * after rejoining group.
+				 *
+				 * In preferred master mode, either site
+				 * should defer the election (which
+				 * executes the preferred master startup
+				 * code and only calls an election if it is
+				 * safe) and also avoid scheduling an extra
+				 * reconnect attempt in bust_connection()
+				 * by setting the indicator.
+				 */
+				if (!FLD_ISSET(db_rep->region->config,
+				    REP_C_2SITE_STRICT) ||
+				    IS_PREFMAS_MODE(env))
+					db_rep->rejoin_pending = TRUE;
+				if ((ret = __repmgr_defer_op(env,
+				    REPMGR_REJOIN)) == 0)
+					ret = DB_REP_UNAVAIL;
+			}
 		} else
 			ret = DB_REP_UNAVAIL;
 		DB_ASSERT(env, ret != 0);
@@ -1332,7 +1860,6 @@ process_own_msg(env, conn)
 		if ((ret = __repmgr_parm_refresh_unmarshal(env,
 		    &parms, dbt->data, dbt->size, NULL)) != 0)
 			return (DB_REP_UNAVAIL);
-		db_rep = env->rep_handle;
 		DB_ASSERT(env, conn->type == REP_CONNECTION &&
 		    IS_KNOWN_REMOTE_SITE(conn->eid));
 		site = SITE_FROM_EID(conn->eid);
@@ -1348,8 +1875,15 @@ process_own_msg(env, conn)
 	case REPMGR_GM_FORWARD:
 	case REPMGR_JOIN_REQUEST:
 	case REPMGR_JOIN_SUCCESS:
+	case REPMGR_LSNHIST_REQUEST:
+	case REPMGR_LSNHIST_RESPONSE:
+	case REPMGR_PREFMAS_FAILURE:
+	case REPMGR_PREFMAS_SUCCESS:
+	case REPMGR_READONLY_MASTER:
+	case REPMGR_READONLY_RESPONSE:
 	case REPMGR_REMOVE_REQUEST:
 	case REPMGR_RESOLVE_LIMBO:
+	case REPMGR_RESTART_CLIENT:
 	default:
 		__db_errx(env, DB_STR_A("3677",
 		    "unexpected msg type %lu in process_own_msg", "%lu"),
@@ -1482,6 +2016,8 @@ __repmgr_send_handshake(env, conn, opt, optlen, flags)
 		cntrl_len = __REPMGR_V3HANDSHAKE_SIZE;
 		break;
 	case 4:
+	case 5:
+	case 6:
 		cntrl_len = __REPMGR_HANDSHAKE_SIZE;
 		break;
 	default:
@@ -1513,6 +2049,8 @@ __repmgr_send_handshake(env, conn, opt, optlen, flags)
 		__repmgr_v3handshake_marshal(env, &v3hs, p);
 		break;
 	case 4:
+	case 5:
+	case 6:
 		hs.port = my_addr->port;
 		hs.alignment = MEM_ALIGN;
 		hs.ack_policy = (u_int32_t)rep->perm_policy;
@@ -1551,11 +2089,14 @@ read_version_response(env, conn)
 	DB_REP *db_rep;
 	__repmgr_version_confirmation_args conf;
 	DBT vi;
+	REP *rep;
+	REPMGR_SITE *site;
 	char *hostname;
 	u_int32_t flags;
-	int ret;
+	int ret, subord;
 
 	db_rep = env->rep_handle;
+	rep = db_rep->region;
 
 	if ((ret = __repmgr_find_version_info(env, conn, &vi)) != 0)
 		return (ret);
@@ -1581,14 +2122,37 @@ read_version_response(env, conn)
 			return (DB_REP_UNAVAIL);
 		}
 
-		if ((ret = accept_handshake(env, conn, hostname)) != 0)
+		if ((ret = accept_handshake(env, conn, hostname, &subord)) != 0)
 			return (ret);
-		flags = IS_SUBORDINATE(db_rep) ? REPMGR_SUBORDINATE : 0;
+		if (!IS_SUBORDINATE(db_rep))
+			flags = 0;
+		else {
+			flags = REPMGR_SUBORDINATE;
+			if (FLD_ISSET(rep->config, REP_C_AUTOTAKEOVER) &&
+			    db_rep->repmgr_status == running)
+				/*
+				 * Takeover is enabled in rep-aware subordinate
+				 * process.
+				 */
+				flags |= REPMGR_AUTOTAKEOVER;
+		}
 		if ((ret = __repmgr_send_handshake(env,
 		    conn, NULL, 0, flags)) != 0)
 			return (ret);
 	}
 	conn->state = CONN_READY;
+	site = SITE_FROM_EID(conn->eid);
+	/*
+	 * Do not increase sites_avail redundantly for a new outgoing
+	 * connection from a subordinate process.
+	 */
+	if (conn->type == REP_CONNECTION &&
+	    site->state == SITE_CONNECTED && !IS_SUBORDINATE(db_rep)) {
+		rep->sites_avail++;
+		RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+		    "vers_resp: EID %lu CONNECTED, READY.  sites_avail %lu",
+		    (u_long)conn->eid, (u_long)rep->sites_avail));
+	}
 	return (ret);
 }
 
@@ -1641,10 +2205,11 @@ __repmgr_find_version_info(env, conn, vi)
 }
 
 static int
-accept_handshake(env, conn, hostname)
+accept_handshake(env, conn, hostname, subordinate)
 	ENV *env;
 	REPMGR_CONNECTION *conn;
 	char *hostname;
+	int *subordinate;
 {
 	__repmgr_handshake_args hs;
 	__repmgr_v2handshake_args hs2;
@@ -1653,6 +2218,7 @@ accept_handshake(env, conn, hostname)
 	u_int32_t ack, flags;
 	int electable;
 
+	*subordinate = 0;
 	switch (conn->version) {
 	case 2:
 		if (__repmgr_v2handshake_unmarshal(env, &hs2,
@@ -1674,6 +2240,8 @@ accept_handshake(env, conn, hostname)
 		ack = 0;
 		break;
 	case 4:
+	case 5:
+	case 6:
 		if (__repmgr_handshake_unmarshal(env, &hs,
 		   conn->input.repmgr_msg.cntrl.data,
 		   conn->input.repmgr_msg.cntrl.size, NULL) != 0)
@@ -1682,6 +2250,8 @@ accept_handshake(env, conn, hostname)
 		electable = F_ISSET(&hs, ELECTABLE_SITE);
 		flags = hs.flags;
 		ack = hs.ack_policy;
+		if (LF_ISSET(REPMGR_SUBORDINATE))
+			*subordinate = 1;
 		break;
 	default:
 		__db_errx(env, DB_STR_A("3679",
@@ -1729,13 +2299,17 @@ process_parameters(env, conn, host, port, ack, electable, flags)
 	u_int32_t ack, flags;
 {
 	DB_REP *db_rep;
+	REP *rep;
 	REPMGR_RETRY *retry;
 	REPMGR_SITE *site;
+	SITEINFO *sites;
 	__repmgr_connect_reject_args reject;
+	__repmgr_v4connect_reject_args v4reject;
 	u_int8_t reject_buf[__REPMGR_CONNECT_REJECT_SIZE];
 	int eid, ret;
 
 	db_rep = env->rep_handle;
+	rep = db_rep->region;
 
 	/* Connection state can be used to discern incoming versus outgoing. */
 	if (conn->state == CONN_CONNECTED) {
@@ -1785,6 +2359,13 @@ process_parameters(env, conn, host, port, ack, electable, flags)
 				TAILQ_INSERT_TAIL(&site->sub_conns,
 				    conn, entries);
 				conn->eid = eid;
+				conn->auto_takeover =
+				    LF_ISSET(REPMGR_AUTOTAKEOVER) ? 1 : 0;
+				SET_LISTENER_CAND(conn->auto_takeover, ++);
+				RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+		"handshake from subordinate %sconnection at site %s:%u EID %u",
+				    LF_ISSET(REPMGR_AUTOTAKEOVER)?
+				    "takeover ": "", host, port, eid));
 			} else {
 				DB_EVENT(env,
 				    DB_EVENT_REP_CONNECT_ESTD, &eid);
@@ -1797,6 +2378,7 @@ process_parameters(env, conn, host, port, ack, electable, flags)
 					TAILQ_REMOVE(&db_rep->retries,
 					    retry, entries);
 					__os_free(env, retry);
+					site->ref.retry = NULL;
 					break;
 				case SITE_CONNECTED:
 					/*
@@ -1821,6 +2403,16 @@ process_parameters(env, conn, host, port, ack, electable, flags)
 					 * don't have to do anything else here.
 					 */
 					break;
+				case SITE_IDLE:
+					/*
+					 * This can occur after the heartbeat
+					 * test hook artificially kept this
+					 * site from first trying to connect.
+					 */
+					RPRINT(env, (env, DB_VERB_REPMGR_MISC,
+				      "handshake from idle site %s:%u EID %u",
+					    host, port, eid));
+					break;
 				default:
 					DB_ASSERT(env, FALSE);
 				}
@@ -1834,10 +2426,18 @@ process_parameters(env, conn, host, port, ack, electable, flags)
 			RPRINT(env, (env, DB_VERB_REPMGR_MISC,
 		  "rejecting connection from unknown or provisional site %s:%u",
 			    host, port));
-			reject.version = db_rep->membership_version;
-			reject.gen = db_rep->member_version_gen;
-			__repmgr_connect_reject_marshal(env,
-			    &reject, reject_buf);
+			if (conn->version < 5) {
+				v4reject.version = db_rep->membership_version;
+				v4reject.gen = db_rep->member_version_gen;
+				__repmgr_v4connect_reject_marshal(env,
+				    &v4reject, reject_buf);
+			} else {
+				reject.version = db_rep->membership_version;
+				reject.gen = db_rep->member_version_gen;
+				reject.status = (site) ? site->membership : 0;
+				__repmgr_connect_reject_marshal(env,
+				    &reject, reject_buf);
+			}
 
 			if ((ret = __repmgr_send_own_msg(env, conn,
 			    REPMGR_CONNECT_REJECT, reject_buf,
@@ -1867,7 +2467,8 @@ process_parameters(env, conn, host, port, ack, electable, flags)
 	 */
 	if (!IS_SUBORDINATE(db_rep) && /* us */
 	    !__repmgr_master_is_known(env) &&
-	    !LF_ISSET(REPMGR_SUBORDINATE)) { /* the remote site */
+	    !LF_ISSET(REPMGR_SUBORDINATE) && /* the remote site */
+	    !IS_PREFMAS_MODE(env)) {
 		RPRINT(env, (env, DB_VERB_REPMGR_MISC,
 		    "handshake with no known master to wake election thread"));
 		db_rep->new_connection = TRUE;
@@ -1980,6 +2581,7 @@ record_permlsn(env, conn)
 		 */
 		if (ackp->lsn.file > site->max_ack.file)
 			do_log_check = 1;
+		site->max_ack_gen = ackp->generation;
 		memcpy(&site->max_ack, &ackp->lsn, sizeof(DB_LSN));
 		if (do_log_check)
 			check_min_log_file(env);