/*
 * __wt_log_truncate_files --
 *	Truncate log files via archive once. Requires that the server is not
 *	currently running.
 */
int
__wt_log_truncate_files(
    WT_SESSION_IMPL *session, WT_CURSOR *cursor, const char *cfg[])
{
	WT_CONNECTION_IMPL *conn;
	WT_DECL_RET;
	WT_LOG *log;
	uint32_t backup_file;

	WT_UNUSED(cfg);
	conn = S2C(session);
	if (!FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED))
		return (0);
	if (F_ISSET(conn, WT_CONN_SERVER_RUN) &&
	    FLD_ISSET(conn->log_flags, WT_CONN_LOG_ARCHIVE))
		WT_RET_MSG(session, EINVAL,
		    "Attempt to archive manually while a server is running");

	log = conn->log;

	backup_file = 0;
	if (cursor != NULL)
		backup_file = WT_CURSOR_BACKUP_ID(cursor);
	WT_ASSERT(session, backup_file <= log->alloc_lsn.l.file);
	__wt_verbose(session, WT_VERB_LOG,
	    "log_truncate_files: Archive once up to %" PRIu32, backup_file);

	__wt_writelock(session, log->log_archive_lock);
	ret = __log_archive_once(session, backup_file);
	__wt_writeunlock(session, log->log_archive_lock);
	return (ret);
}

/*
 * __wt_btcur_range_truncate --
 *	Discard a cursor range from the tree.
 */
int
__wt_btcur_range_truncate(WT_CURSOR_BTREE *start, WT_CURSOR_BTREE *stop)
{
	WT_BTREE *btree;
	WT_CURSOR_BTREE *cbt;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;

	cbt = (start != NULL) ? start : stop;
	session = (WT_SESSION_IMPL *)cbt->iface.session;
	btree = cbt->btree;

	/*
	 * For recovery, we log the start and stop keys for a truncate
	 * operation, not the individual records removed.  On the other hand,
	 * for rollback we need to keep track of all the in-memory operations.
	 *
	 * We deal with this here by logging the truncate range first, then (in
	 * the logging code) disabling writing of the in-memory remove records
	 * to disk.
	 */
	if (FLD_ISSET(S2C(session)->log_flags, WT_CONN_LOG_ENABLED))
		WT_RET(__wt_txn_truncate_log(session, start, stop));

	switch (btree->type) {
	case BTREE_COL_FIX:
		WT_ERR(__cursor_truncate_fix(
		    session, start, stop, __cursor_col_modify));
		break;
	case BTREE_COL_VAR:
		WT_ERR(__cursor_truncate(
		    session, start, stop, __cursor_col_modify));
		break;
	case BTREE_ROW:
		/*
		 * The underlying cursor comparison routine requires cursors be
		 * fully instantiated when truncating row-store objects because
		 * it's comparing page and/or skiplist positions, not keys. (Key
		 * comparison would work, it's only that a key comparison would
		 * be relatively expensive.  Column-store objects have record
		 * number keys, so the key comparison is cheap.)  Cursors may
		 * have only had their keys set, so we must ensure the cursors
		 * are positioned in the tree.
		 */
		if (start != NULL)
			WT_ERR(__wt_btcur_search(start));
		if (stop != NULL)
			WT_ERR(__wt_btcur_search(stop));
		WT_ERR(__cursor_truncate(
		    session, start, stop, __cursor_row_modify));
		break;
	}

err:	if (FLD_ISSET(S2C(session)->log_flags, WT_CONN_LOG_ENABLED))
		WT_TRET(__wt_txn_truncate_end(session));
	return (ret);
}

/*
 * __wt_meta_track_off --
 *	Turn off metadata operation tracking, unrolling on error.
 */
int
__wt_meta_track_off(WT_SESSION_IMPL *session, int need_sync, int unroll)
{
	WT_DECL_RET;
	WT_META_TRACK *trk, *trk_orig;

	WT_ASSERT(session,
	    WT_META_TRACKING(session) && session->meta_track_nest > 0);

	trk_orig = session->meta_track;
	trk = session->meta_track_next;

	/* If it was a nested transaction, there is nothing to do. */
	if (--session->meta_track_nest != 0)
		return (0);

	/* Turn off tracking for unroll. */
	session->meta_track_next = session->meta_track_sub = NULL;

	/*
	 * If there were no operations logged, return now and avoid unnecessary
	 * metadata checkpoints.  For example, this happens if attempting to
	 * create a data source that already exists (or drop one that doesn't).
	 */
	if (trk == trk_orig)
		return (0);

	while (--trk >= trk_orig)
		WT_TRET(__meta_track_apply(session, trk, unroll));

	/*
	 * Unroll operations don't need to flush the metadata.
	 *
	 * Also, if we don't have the metadata handle (e.g, we're in the
	 * process of creating the metadata), we can't sync it.
	 */
	if (unroll || ret != 0 || !need_sync || session->meta_dhandle == NULL)
		return (ret);

	/* If we're logging, make sure the metadata update was flushed. */
	if (FLD_ISSET(S2C(session)->log_flags, WT_CONN_LOG_ENABLED)) {
		if (!FLD_ISSET(S2C(session)->txn_logsync,
		    WT_LOG_DSYNC | WT_LOG_FSYNC))
			WT_WITH_DHANDLE(session, session->meta_dhandle,
			    ret = __wt_txn_checkpoint_log(session,
			    0, WT_TXN_LOG_CKPT_SYNC, NULL));
	} else {
		WT_WITH_DHANDLE(session, session->meta_dhandle,
		    ret = __wt_checkpoint(session, NULL));
		WT_RET(ret);
		WT_WITH_DHANDLE(session, session->meta_dhandle,
		    ret = __wt_checkpoint_sync(session, NULL));
	}

	return (ret);
}

/*
 * __wt_btcur_range_truncate --
 *	Discard a cursor range from the tree.
 */
int
__wt_btcur_range_truncate(WT_CURSOR_BTREE *start, WT_CURSOR_BTREE *stop)
{
	WT_BTREE *btree;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;

	session = (WT_SESSION_IMPL *)start->iface.session;
	btree = start->btree;
	WT_STAT_DATA_INCR(session, cursor_truncate);

	/*
	 * For recovery, log the start and stop keys for a truncate operation,
	 * not the individual records removed.  On the other hand, for rollback
	 * we need to keep track of all the in-memory operations.
	 *
	 * We deal with this here by logging the truncate range first, then (in
	 * the logging code) disabling writing of the in-memory remove records
	 * to disk.
	 */
	if (FLD_ISSET(S2C(session)->log_flags, WT_CONN_LOG_ENABLED))
		WT_RET(__wt_txn_truncate_log(session, start, stop));

	switch (btree->type) {
	case BTREE_COL_FIX:
		WT_ERR(__cursor_truncate_fix(
		    session, start, stop, __cursor_col_modify));
		break;
	case BTREE_COL_VAR:
		WT_ERR(__cursor_truncate(
		    session, start, stop, __cursor_col_modify));
		break;
	case BTREE_ROW:
		/*
		 * The underlying cursor comparison routine requires cursors be
		 * fully instantiated when truncating row-store objects because
		 * it's comparing page and/or skiplist positions, not keys. (Key
		 * comparison would work, it's only that a key comparison would
		 * be relatively expensive, especially with custom collators.
		 * Column-store objects have record number keys, so the key
		 * comparison is cheap.)  The session truncate code did cursor
		 * searches when setting up the truncate so we're good to go: if
		 * that ever changes, we'd need to do something here to ensure a
		 * fully instantiated cursor.
		 */
		WT_ERR(__cursor_truncate(
		    session, start, stop, __cursor_row_modify));
		break;
	}

err:	if (FLD_ISSET(S2C(session)->log_flags, WT_CONN_LOG_ENABLED))
		__wt_txn_truncate_end(session);
	return (ret);
}

/*
 * __wt_verbose_dump_log --
 *	Dump information about the logging subsystem.
 */
int
__wt_verbose_dump_log(WT_SESSION_IMPL *session)
{
	WT_CONNECTION_IMPL *conn;
	WT_LOG *log;

	conn = S2C(session);
	log = conn->log;

	WT_RET(__wt_msg(session, "%s", WT_DIVIDER));
	WT_RET(__wt_msg(session, "Logging subsystem: Enabled: %s",
	    FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED) ? "yes" : "no"));
	if (!FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED))
		return (0);
	/*
	 * Logging is enabled, print out the other information.
	 */
	WT_RET(__wt_msg(session, "Archiving: %s",
	    FLD_ISSET(conn->log_flags, WT_CONN_LOG_ARCHIVE) ? "yes" : "no"));
	WT_RET(__wt_msg(session, "Running downgraded: %s",
	    FLD_ISSET(conn->log_flags, WT_CONN_LOG_DOWNGRADED) ? "yes" : "no"));
	WT_RET(__wt_msg(session, "Zero fill files: %s",
	    FLD_ISSET(conn->log_flags, WT_CONN_LOG_ZERO_FILL) ? "yes" : "no"));
	WT_RET(__wt_msg(session, "Pre-allocate files: %s",
	    conn->log_prealloc > 0 ? "yes" : "no"));
	WT_RET(__wt_msg(session, "Logging directory: %s", conn->log_path));
	WT_RET(__wt_msg(session, "Logging maximum file size: %" PRId64,
	    (int64_t)conn->log_file_max));
	WT_RET(__wt_msg(session, "Log sync setting: %s",
	    !FLD_ISSET(conn->txn_logsync, WT_LOG_SYNC_ENABLED) ? "none" :
	    FLD_ISSET(conn->txn_logsync, WT_LOG_DSYNC) ? "dsync" :
	    FLD_ISSET(conn->txn_logsync, WT_LOG_FLUSH) ? "write to OS" :
	    FLD_ISSET(conn->txn_logsync, WT_LOG_FSYNC) ?
	    "fsync to disk": "unknown sync setting"));
	WT_RET(__wt_msg(session, "Log record allocation alignment: %" PRIu32,
	    log->allocsize));
	WT_RET(__wt_msg(session, "Current log file number: %" PRIu32,
	    log->fileid));
	WT_RET(__wt_msg(session, "Current log version number: %" PRIu16,
	    log->log_version));
	WT_RET(WT_LSN_MSG(&log->alloc_lsn, "Next allocation"));
	WT_RET(WT_LSN_MSG(&log->bg_sync_lsn, "Last background sync"));
	WT_RET(WT_LSN_MSG(&log->ckpt_lsn, "Last checkpoint"));
	WT_RET(WT_LSN_MSG(&log->sync_dir_lsn, "Last directory sync"));
	WT_RET(WT_LSN_MSG(&log->sync_lsn, "Last sync"));
	WT_RET(WT_LSN_MSG(&log->trunc_lsn, "Recovery truncate"));
	WT_RET(WT_LSN_MSG(&log->write_lsn, "Last written"));
	WT_RET(WT_LSN_MSG(&log->write_start_lsn, "Start of last written"));
	/*
	 * If we wanted a dump of the slots, it would go here. Walking
	 * the slot pool may not require a lock since they're statically
	 * allocated, but output could be inconsistent without it.
	 */

	return (0);
}

/*
 * __curlog_close --
 *	WT_CURSOR.close method for the log cursor type.
 */
static int
__curlog_close(WT_CURSOR *cursor)
{
	WT_CONNECTION_IMPL *conn;
	WT_CURSOR_LOG *cl;
	WT_DECL_RET;
	WT_LOG *log;
	WT_SESSION_IMPL *session;

	CURSOR_API_CALL(cursor, session, close, NULL);
	cl = (WT_CURSOR_LOG *)cursor;
	conn = S2C(session);
	WT_ASSERT(session, FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED));
	log = conn->log;
	WT_TRET(__wt_readunlock(session, log->log_archive_lock));
	WT_TRET(__curlog_reset(cursor));
	__wt_free(session, cl->cur_lsn);
	__wt_free(session, cl->next_lsn);
	__wt_scr_free(session, &cl->logrec);
	__wt_scr_free(session, &cl->opkey);
	__wt_scr_free(session, &cl->opvalue);
	__wt_free(session, cl->packed_key);
	__wt_free(session, cl->packed_value);
	WT_TRET(__wt_cursor_close(cursor));

err:	API_END_RET(session, ret);
}

/*
 * __wt_direct_io_size_check --
 *	Return a size from the configuration, complaining if it's insufficient
 * for direct I/O.
 */
int
__wt_direct_io_size_check(WT_SESSION_IMPL *session,
    const char **cfg, const char *config_name, uint32_t *allocsizep)
{
	WT_CONFIG_ITEM cval;
	WT_CONNECTION_IMPL *conn;
	int64_t align;

	*allocsizep = 0;

	conn = S2C(session);

	WT_RET(__wt_config_gets(session, cfg, config_name, &cval));

	/*
	 * This function exists as a place to hang this comment: if direct I/O
	 * is configured, page sizes must be at least as large as any buffer
	 * alignment as well as a multiple of the alignment.  Linux gets unhappy
	 * if you configure direct I/O and then don't do I/O in alignments and
	 * units of its happy place.
	 */
	if (FLD_ISSET(conn->direct_io,
	   WT_FILE_TYPE_CHECKPOINT | WT_FILE_TYPE_DATA)) {
		align = (int64_t)conn->buffer_alignment;
		if (align != 0 && (cval.val < align || cval.val % align != 0))
			WT_RET_MSG(session, EINVAL,
			    "when direct I/O is configured, the %s size must "
			    "be at least as large as the buffer alignment as "
			    "well as a multiple of the buffer alignment",
			    config_name);
	}
	*allocsizep = (uint32_t)cval.val;
	return (0);
}

/*
 * __ckpt_server_config --
 *	Parse and setup the checkpoint server options.
 */
static int
__ckpt_server_config(WT_SESSION_IMPL *session, const char **cfg, bool *startp)
{
	WT_CONFIG_ITEM cval;
	WT_CONNECTION_IMPL *conn;

	*startp = false;

	conn = S2C(session);

	WT_RET(__wt_config_gets(session, cfg, "checkpoint.wait", &cval));
	conn->ckpt_usecs = (uint64_t)cval.val * WT_MILLION;

	WT_RET(__wt_config_gets(session, cfg, "checkpoint.log_size", &cval));
	conn->ckpt_logsize = (wt_off_t)cval.val;

	/*
	 * The checkpoint configuration requires a wait time and/or a log size,
	 * if neither is set, we're not running at all. Checkpoints based on log
	 * size also require logging be enabled.
	 */
	if (conn->ckpt_usecs != 0 ||
	    (conn->ckpt_logsize != 0 &&
	    FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED))) {
		/*
		 * If checkpointing based on log data, use a minimum of the
		 * log file size.  The logging subsystem has already been
		 * initialized.
		 */
		if (conn->ckpt_logsize != 0 &&
		    FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED))
			conn->ckpt_logsize = WT_MAX(
			    conn->ckpt_logsize, conn->log_file_max);
		/* Checkpoints are incompatible with in-memory configuration */
		WT_RET(__wt_config_gets(session, cfg, "in_memory", &cval));
		if (cval.val != 0)
			WT_RET_MSG(session, EINVAL,
			    "checkpoint configuration incompatible with "
			    "in-memory configuration");

		__wt_log_written_reset(session);

		*startp = true;
	}

	return (0);
}

/*
 * __logmgr_version --
 *	Set up the versions in the log manager.
 */
static int
__logmgr_version(WT_SESSION_IMPL *session, bool reconfig)
{
	WT_CONNECTION_IMPL *conn;
	WT_LOG *log;
	bool downgrade;
	uint32_t first_record, lognum;
	uint16_t new_version;

	conn = S2C(session);
	log = conn->log;
	if (log == NULL)
		return (0);

	/*
	 * Set the log file format versions based on compatibility versions
	 * set in the connection.  We must set this before we call log_open
	 * to open or create a log file.
	 *
	 * Since the log version changed at a major release number we only need
	 * to check the major number, not the minor number in the compatibility
	 * setting.
	 */
	if (conn->compat_major < WT_LOG_V2) {
		new_version = 1;
		first_record = WT_LOG_END_HEADER;
		downgrade = true;
	} else {
		new_version = WT_LOG_VERSION;
		first_record = WT_LOG_END_HEADER + log->allocsize;
		downgrade = false;
	}

	/*
	 * If the version is the same, there is nothing to do.
	 */
	if (log->log_version == new_version)
		return (0);
	/*
	 * If we are reconfiguring and at a new version we need to force
	 * the log file to advance so that we write out a log file at the
	 * correct version.  When we are downgrading we must force a checkpoint
	 * and finally archive, even if disabled, so that all new version log
	 * files are gone.
	 *
	 * All of the version changes must be handled with locks on reconfigure
	 * because other threads may be changing log files, using pre-allocated
	 * files.
	 */
	/*
	 * Set the version.  If it is a live change the logging subsystem will
	 * do other work as well to move to a new log file.
	 */
	WT_RET(__wt_log_set_version(session, new_version,
	    first_record, downgrade, reconfig, &lognum));
	if (reconfig && FLD_ISSET(conn->log_flags, WT_CONN_LOG_DOWNGRADED))
		WT_RET(__logmgr_force_ckpt(session, lognum));
	return (0);
}

/*
 * __wt_logmgr_create --
 *	Initialize the log subsystem (before running recovery).
 */
int
__wt_logmgr_create(WT_SESSION_IMPL *session, const char *cfg[])
{
	WT_CONNECTION_IMPL *conn;
	WT_LOG *log;
	bool run;

	conn = S2C(session);

	/* Handle configuration. */
	WT_RET(__logmgr_config(session, cfg, &run, false));

	/* If logging is not configured, we're done. */
	if (!run)
		return (0);

	FLD_SET(conn->log_flags, WT_CONN_LOG_ENABLED);
	/*
	 * Logging is on, allocate the WT_LOG structure and open the log file.
	 */
	WT_RET(__wt_calloc_one(session, &conn->log));
	log = conn->log;
	WT_RET(__wt_spin_init(session, &log->log_lock, "log"));
	WT_RET(__wt_spin_init(session, &log->log_slot_lock, "log slot"));
	WT_RET(__wt_spin_init(session, &log->log_sync_lock, "log sync"));
	WT_RET(__wt_spin_init(session, &log->log_writelsn_lock,
	    "log write LSN"));
	WT_RET(__wt_rwlock_alloc(session,
	    &log->log_archive_lock, "log archive lock"));
	if (FLD_ISSET(conn->direct_io, WT_FILE_TYPE_LOG))
		log->allocsize =
		    WT_MAX((uint32_t)conn->buffer_alignment, WT_LOG_ALIGN);
	else
		log->allocsize = WT_LOG_ALIGN;
	WT_INIT_LSN(&log->alloc_lsn);
	WT_INIT_LSN(&log->ckpt_lsn);
	WT_INIT_LSN(&log->first_lsn);
	WT_INIT_LSN(&log->sync_lsn);
	/*
	 * We only use file numbers for directory sync, so this needs to
	 * initialized to zero.
	 */
	WT_ZERO_LSN(&log->sync_dir_lsn);
	WT_INIT_LSN(&log->trunc_lsn);
	WT_INIT_LSN(&log->write_lsn);
	WT_INIT_LSN(&log->write_start_lsn);
	log->fileid = 0;
	WT_RET(__wt_cond_alloc(
	    session, "log sync", false, &log->log_sync_cond));
	WT_RET(__wt_cond_alloc(
	    session, "log write", false, &log->log_write_cond));
	WT_RET(__wt_log_open(session));
	WT_RET(__wt_log_slot_init(session));

	return (0);
}

/*
 * __lsm_worker_general_op --
 *	Execute a single bloom, drop or flush work unit.
 */
static int
__lsm_worker_general_op(
    WT_SESSION_IMPL *session, WT_LSM_WORKER_ARGS *cookie, int *completed)
{
	WT_DECL_RET;
	WT_LSM_CHUNK *chunk;
	WT_LSM_WORK_UNIT *entry;
	int force;

	*completed = 0;
	/*
	 * Return if this thread cannot process a bloom, drop or flush.
	 */
	if (!FLD_ISSET(cookie->type,
	    WT_LSM_WORK_BLOOM | WT_LSM_WORK_DROP | WT_LSM_WORK_FLUSH))
		return (WT_NOTFOUND);

	if ((ret = __wt_lsm_manager_pop_entry(session,
	    cookie->type, &entry)) != 0 || entry == NULL)
		return (ret);

	if (entry->type == WT_LSM_WORK_FLUSH) {
		force = F_ISSET(entry, WT_LSM_WORK_FORCE);
		F_CLR(entry, WT_LSM_WORK_FORCE);
		WT_ERR(__wt_lsm_get_chunk_to_flush(session,
		    entry->lsm_tree, force, &chunk));
		/*
		 * If we got a chunk to flush, checkpoint it.
		 */
		if (chunk != NULL) {
			WT_ERR(__wt_verbose(session, WT_VERB_LSM,
			    "Flush%s chunk %d %s",
			    force ? " w/ force" : "",
			    chunk->id, chunk->uri));
			ret = __wt_lsm_checkpoint_chunk(
			    session, entry->lsm_tree, chunk);
			WT_ASSERT(session, chunk->refcnt > 0);
			(void)WT_ATOMIC_SUB4(chunk->refcnt, 1);
			WT_ERR(ret);
		}
	} else if (entry->type == WT_LSM_WORK_DROP)
		WT_ERR(__wt_lsm_free_chunks(session, entry->lsm_tree));
	else if (entry->type == WT_LSM_WORK_BLOOM)
		WT_ERR(__wt_lsm_work_bloom(session, entry->lsm_tree));
	*completed = 1;

err:	__wt_lsm_manager_free_work_unit(session, entry);
	return (ret);
}

/*
 * __wt_txn_log_op --
 *	Write the last logged operation into the in-memory buffer.
 */
int
__wt_txn_log_op(WT_SESSION_IMPL *session, WT_CURSOR_BTREE *cbt)
{
	WT_DECL_RET;
	WT_ITEM *logrec;
	WT_TXN *txn;
	WT_TXN_OP *op;

	txn = &session->txn;

	if (!FLD_ISSET(S2C(session)->log_flags, WT_CONN_LOG_ENABLED) ||
	    F_ISSET(session, WT_SESSION_NO_LOGGING) ||
	    F_ISSET(S2BT(session), WT_BTREE_NO_LOGGING))
		return (0);

	/* We'd better have a transaction. */
	WT_ASSERT(session,
	    F_ISSET(txn, WT_TXN_RUNNING) && F_ISSET(txn, WT_TXN_HAS_ID));

	WT_ASSERT(session, txn->mod_count > 0);
	op = txn->mod + txn->mod_count - 1;

	WT_RET(__txn_logrec_init(session));
	logrec = txn->logrec;

	switch (op->type) {
	case WT_TXN_OP_BASIC:
	case WT_TXN_OP_BASIC_TS:
		ret = __txn_op_log(session, logrec, op, cbt);
		break;
	case WT_TXN_OP_INMEM:
	case WT_TXN_OP_REF:
		/* Nothing to log, we're done. */
		break;
	case WT_TXN_OP_TRUNCATE_COL:
		ret = __wt_logop_col_truncate_pack(session, logrec,
		    op->fileid,
		    op->u.truncate_col.start, op->u.truncate_col.stop);
		break;
	case WT_TXN_OP_TRUNCATE_ROW:
		ret = __wt_logop_row_truncate_pack(session, txn->logrec,
		    op->fileid,
		    &op->u.truncate_row.start, &op->u.truncate_row.stop,
		    (uint32_t)op->u.truncate_row.mode);
		break;
	}
	return (ret);
}

/*
 * __log_server --
 *	The log server thread.
 */
static WT_THREAD_RET
__log_server(void *arg)
{
	WT_CONNECTION_IMPL *conn;
	WT_DECL_RET;
	WT_LOG *log;
	WT_SESSION_IMPL *session;
	u_int locked;

	session = arg;
	conn = S2C(session);
	log = conn->log;
	locked = 0;
	while (F_ISSET(conn, WT_CONN_LOG_SERVER_RUN)) {
		/*
		 * Perform log pre-allocation.
		 */
		if (conn->log_prealloc > 0)
			WT_ERR(__log_prealloc_once(session));

		/*
		 * Perform the archive.
		 */
		if (FLD_ISSET(conn->log_flags, WT_CONN_LOG_ARCHIVE)) {
			if (__wt_try_writelock(
			    session, log->log_archive_lock) == 0) {
				locked = 1;
				WT_ERR(__log_archive_once(session, 0));
				WT_ERR(	__wt_writeunlock(
				    session, log->log_archive_lock));
				locked = 0;
			} else
				WT_ERR(__wt_verbose(session, WT_VERB_LOG,
				    "log_archive: Blocked due to open log "
				    "cursor holding archive lock"));
		}
		/* Wait until the next event. */
		WT_ERR(__wt_cond_wait(session, conn->log_cond, WT_MILLION));
	}

	if (0) {
err:		__wt_err(session, ret, "log server error");
	}
	if (locked)
		(void)__wt_writeunlock(session, log->log_archive_lock);
	return (WT_THREAD_RET_VALUE);
}

/*
 * __wt_lsm_start_worker --
 *	Start the worker thread for an LSM tree.
 */
static int
__lsm_tree_start_worker(WT_SESSION_IMPL *session, WT_LSM_TREE *lsm_tree)
{
	WT_CONNECTION *wt_conn;
	WT_LSM_WORKER_ARGS *wargs;
	WT_SESSION *wt_session;
	WT_SESSION_IMPL *s;
	uint32_t i;

	wt_conn = &S2C(session)->iface;

	WT_RET(wt_conn->open_session(wt_conn, NULL, NULL, &wt_session));
	lsm_tree->ckpt_session = (WT_SESSION_IMPL *)wt_session;
	F_SET(lsm_tree->ckpt_session, WT_SESSION_INTERNAL);

	F_SET(lsm_tree, WT_LSM_TREE_WORKING);
	/* The new thread will rely on the WORKING value being visible. */
	WT_FULL_BARRIER();
	if (F_ISSET(S2C(session), WT_CONN_LSM_MERGE))
		for (i = 0; i < lsm_tree->merge_threads; i++) {
			WT_RET(wt_conn->open_session(
			    wt_conn, NULL, NULL, &wt_session));
			s = (WT_SESSION_IMPL *)wt_session;
			F_SET(s, WT_SESSION_INTERNAL);
			lsm_tree->worker_sessions[i] = s;

			WT_RET(__wt_calloc_def(session, 1, &wargs));
			wargs->lsm_tree = lsm_tree;
			wargs->id = i;
			WT_RET(__wt_thread_create(session,
			    &lsm_tree->worker_tids[i],
			    __wt_lsm_merge_worker, wargs));
		}
	if (FLD_ISSET(lsm_tree->bloom, WT_LSM_BLOOM_NEWEST)) {
		WT_RET(wt_conn->open_session(wt_conn, NULL, NULL, &wt_session));
		lsm_tree->bloom_session = (WT_SESSION_IMPL *)wt_session;
		F_SET(lsm_tree->bloom_session, WT_SESSION_INTERNAL);

		WT_RET(__wt_thread_create(session,
		    &lsm_tree->bloom_tid, __wt_lsm_bloom_worker, lsm_tree));
	}
	WT_RET(__wt_thread_create(session,
	    &lsm_tree->ckpt_tid, __wt_lsm_checkpoint_worker, lsm_tree));

	return (0);
}

/*
 * __wt_lsm_tree_worker --
 *	Run a schema worker operation on each level of a LSM tree.
 */
int
__wt_lsm_tree_worker(WT_SESSION_IMPL *session,
   const char *uri,
   int (*file_func)(WT_SESSION_IMPL *, const char *[]),
   int (*name_func)(WT_SESSION_IMPL *, const char *, int *),
   const char *cfg[], uint32_t open_flags)
{
	WT_DECL_RET;
	WT_LSM_CHUNK *chunk;
	WT_LSM_TREE *lsm_tree;
	u_int i;
	int exclusive, locked;

	locked = 0;
	exclusive = FLD_ISSET(open_flags, WT_DHANDLE_EXCLUSIVE) ? 1 : 0;
	WT_RET(__wt_lsm_tree_get(session, uri, exclusive, &lsm_tree));

	/*
	 * We mark that we're busy using the tree to coordinate
	 * with merges so that merging doesn't change the chunk
	 * array out from underneath us.
	 */
	WT_ERR(exclusive ?
	    __wt_lsm_tree_writelock(session, lsm_tree) :
	    __wt_lsm_tree_readlock(session, lsm_tree));
	locked = 1;
	for (i = 0; i < lsm_tree->nchunks; i++) {
		chunk = lsm_tree->chunk[i];
		if (file_func == __wt_checkpoint &&
		    F_ISSET(chunk, WT_LSM_CHUNK_ONDISK))
			continue;
		WT_ERR(__wt_schema_worker(session, chunk->uri,
		    file_func, name_func, cfg, open_flags));
		if (name_func == __wt_backup_list_uri_append &&
		    F_ISSET(chunk, WT_LSM_CHUNK_BLOOM))
			WT_ERR(__wt_schema_worker(session, chunk->bloom_uri,
			    file_func, name_func, cfg, open_flags));
	}
err:	if (locked)
		WT_TRET(exclusive ?
		    __wt_lsm_tree_writeunlock(session, lsm_tree) :
		    __wt_lsm_tree_readunlock(session, lsm_tree));
	__wt_lsm_tree_release(session, lsm_tree);
	return (ret);
}

/*
 * __wt_las_stats_update --
 *	Update the lookaside table statistics for return to the application.
 */
void
__wt_las_stats_update(WT_SESSION_IMPL *session)
{
	WT_CONNECTION_IMPL *conn;
	WT_CONNECTION_STATS **cstats;
	WT_DSRC_STATS **dstats;
	int64_t v;

	conn = S2C(session);

	/*
	 * Lookaside table statistics are copied from the underlying lookaside
	 * table data-source statistics. If there's no lookaside table, values
	 * remain 0.
	 */
	if (!F_ISSET(conn, WT_CONN_LAS_OPEN))
		return;

	/*
	 * We have a cursor, and we need the underlying data handle; we can get
	 * to it by way of the underlying btree handle, but it's a little ugly.
	 */
	cstats = conn->stats;
	dstats = ((WT_CURSOR_BTREE *)
	    conn->las_session->las_cursor)->btree->dhandle->stats;

	v = WT_STAT_READ(dstats, cursor_insert);
	WT_STAT_SET(session, cstats, cache_lookaside_insert, v);
	v = WT_STAT_READ(dstats, cursor_remove);
	WT_STAT_SET(session, cstats, cache_lookaside_remove, v);
	/*
	 * If we're clearing stats we need to clear the cursor values we just
	 * read.  This does not clear the rest of the statistics in the
	 * lookaside data source stat cursor, but we own that namespace so we
	 * don't have to worry about users seeing inconsistent data source
	 * information.
	 */
	if (FLD_ISSET(conn->stat_flags, WT_CONN_STAT_CLEAR)) {
		WT_STAT_SET(session, dstats, cursor_insert, 0);
		WT_STAT_SET(session, dstats, cursor_remove, 0);
	}
}

/*
 * __wt_meta_track_init --
 *	Initialize metadata tracking.
 */
int
__wt_meta_track_init(WT_SESSION_IMPL *session)
{
	WT_CONNECTION_IMPL *conn;

	conn = S2C(session);
	if (!FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED)) {
		WT_RET(__wt_open_internal_session(conn,
		    "metadata-ckpt", false, WT_SESSION_NO_DATA_HANDLES,
		    &conn->meta_ckpt_session));

		/*
		 * Sessions default to read-committed isolation, we rely on
		 * that for the correctness of metadata checkpoints.
		 */
		WT_ASSERT(session, conn->meta_ckpt_session->txn.isolation ==
		    WT_ISO_READ_COMMITTED);
	}

	return (0);
}

/*
 * __wt_lsm_tree_worker --
 *	Run a schema worker operation on each level of a LSM tree.
 */
int
__wt_lsm_tree_worker(WT_SESSION_IMPL *session,
   const char *uri,
   int (*func)(WT_SESSION_IMPL *, const char *[]),
   const char *cfg[], uint32_t open_flags)
{
	WT_DECL_RET;
	WT_LSM_CHUNK *chunk;
	WT_LSM_TREE *lsm_tree;
	u_int i;

	WT_RET(__wt_lsm_tree_get(session, uri,
	    FLD_ISSET(open_flags, WT_BTREE_EXCLUSIVE) ? 1 : 0, &lsm_tree));
	for (i = 0; i < lsm_tree->nchunks; i++) {
		chunk = lsm_tree->chunk[i];
		if (func == __wt_checkpoint &&
		    F_ISSET(chunk, WT_LSM_CHUNK_ONDISK))
			continue;
		WT_ERR(__wt_schema_worker(
		    session, chunk->uri, func, cfg, open_flags));
	}
err:	__wt_lsm_tree_release(session, lsm_tree);
	return (ret);
}

/*
 * __wt_logmgr_destroy --
 *	Destroy the log archiving server thread and logging subsystem.
 */
int
__wt_logmgr_destroy(WT_SESSION_IMPL *session)
{
	WT_CONNECTION_IMPL *conn;
	WT_DECL_RET;
	WT_SESSION *wt_session;

	conn = S2C(session);

	if (!FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED)) {
		/*
		 * We always set up the log_path so printlog can work without
		 * recovery. Therefore, always free it, even if logging isn't
		 * on.
		 */
		__wt_free(session, conn->log_path);
		return (0);
	}
	if (conn->log_tid_set) {
		WT_TRET(__wt_cond_signal(session, conn->log_cond));
		WT_TRET(__wt_thread_join(session, conn->log_tid));
		conn->log_tid_set = 0;
	}
	if (conn->log_file_tid_set) {
		WT_TRET(__wt_cond_signal(session, conn->log_file_cond));
		WT_TRET(__wt_thread_join(session, conn->log_file_tid));
		conn->log_file_tid_set = 0;
	}
	if (conn->log_file_session != NULL) {
		wt_session = &conn->log_file_session->iface;
		WT_TRET(wt_session->close(wt_session, NULL));
		conn->log_file_session = NULL;
	}
	if (conn->log_wrlsn_tid_set) {
		WT_TRET(__wt_cond_signal(session, conn->log_wrlsn_cond));
		WT_TRET(__wt_thread_join(session, conn->log_wrlsn_tid));
		conn->log_wrlsn_tid_set = 0;
	}
	if (conn->log_wrlsn_session != NULL) {
		wt_session = &conn->log_wrlsn_session->iface;
		WT_TRET(wt_session->close(wt_session, NULL));
		conn->log_wrlsn_session = NULL;
	}

	WT_TRET(__wt_log_slot_destroy(session));
	WT_TRET(__wt_log_close(session));

	/* Close the server thread's session. */
	if (conn->log_session != NULL) {
		wt_session = &conn->log_session->iface;
		WT_TRET(wt_session->close(wt_session, NULL));
		conn->log_session = NULL;
	}

	/* Destroy the condition variables now that all threads are stopped */
	WT_TRET(__wt_cond_destroy(session, &conn->log_cond));
	WT_TRET(__wt_cond_destroy(session, &conn->log_file_cond));
	WT_TRET(__wt_cond_destroy(session, &conn->log_wrlsn_cond));

	WT_TRET(__wt_cond_destroy(session, &conn->log->log_sync_cond));
	WT_TRET(__wt_cond_destroy(session, &conn->log->log_write_cond));
	WT_TRET(__wt_rwlock_destroy(session, &conn->log->log_archive_lock));
	__wt_spin_destroy(session, &conn->log->log_lock);
	__wt_spin_destroy(session, &conn->log->log_slot_lock);
	__wt_spin_destroy(session, &conn->log->log_sync_lock);
	__wt_spin_destroy(session, &conn->log->log_writelsn_lock);
	__wt_free(session, conn->log_path);
	__wt_free(session, conn->log);
	return (ret);
}

/*
 * __wt_logmgr_open --
 *	Start the log service threads.
 */
int
__wt_logmgr_open(WT_SESSION_IMPL *session)
{
	WT_CONNECTION_IMPL *conn;

	conn = S2C(session);

	/* If no log thread services are configured, we're done. */ 
	if (!FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED))
		return (0);

	/*
	 * Start the log close thread.  It is not configurable.
	 * If logging is enabled, this thread runs.
	 */
	WT_RET(__wt_open_internal_session(
	    conn, "log-close-server", 0, 0, &conn->log_file_session));
	WT_RET(__wt_cond_alloc(conn->log_file_session,
	    "log close server", 0, &conn->log_file_cond));

	/*
	 * Start the log file close thread.
	 */
	WT_RET(__wt_thread_create(conn->log_file_session,
	    &conn->log_file_tid, __log_file_server, conn->log_file_session));
	conn->log_file_tid_set = 1;

	/*
	 * Start the log write LSN thread.  It is not configurable.
	 * If logging is enabled, this thread runs.
	 */
	WT_RET(__wt_open_internal_session(
	    conn, "log-wrlsn-server", 0, 0, &conn->log_wrlsn_session));
	WT_RET(__wt_cond_alloc(conn->log_wrlsn_session,
	    "log write lsn server", 0, &conn->log_wrlsn_cond));
	WT_RET(__wt_thread_create(conn->log_wrlsn_session,
	    &conn->log_wrlsn_tid, __log_wrlsn_server, conn->log_wrlsn_session));
	conn->log_wrlsn_tid_set = 1;

	/* If no log thread services are configured, we're done. */ 
	if (!FLD_ISSET(conn->log_flags,
	    (WT_CONN_LOG_ARCHIVE | WT_CONN_LOG_PREALLOC)))
		return (0);

	/*
	 * If a log server thread exists, the user may have reconfigured
	 * archiving or pre-allocation.  Signal the thread.  Otherwise the
	 * user wants archiving and/or allocation and we need to start up
	 * the thread.
	 */
	if (conn->log_session != NULL) {
		WT_ASSERT(session, conn->log_cond != NULL);
		WT_ASSERT(session, conn->log_tid_set != 0);
		WT_RET(__wt_cond_signal(session, conn->log_cond));
	} else {
		/* The log server gets its own session. */
		WT_RET(__wt_open_internal_session(
		    conn, "log-server", 0, 0, &conn->log_session));
		WT_RET(__wt_cond_alloc(conn->log_session,
		    "log server", 0, &conn->log_cond));

		/*
		 * Start the thread.
		 */
		WT_RET(__wt_thread_create(conn->log_session,
		    &conn->log_tid, __log_server, conn->log_session));
		conn->log_tid_set = 1;
	}

	return (0);
}