text: /*-
   * Copyright (c) 2014-2016 MongoDB, Inc.
   * Copyright (c) 2008-2014 WiredTiger, Inc.
   *	All rights reserved.
   *
   * See the file LICENSE for redistribution information.
   */
  
  #include "wt_internal.h"
  
  static int __lsm_manager_run_server(WT_SESSION_IMPL *);
  
  static WT_THREAD_RET __lsm_worker_manager(void *);
  
  /*
   * __wt_lsm_manager_config --
   *	Configure the LSM manager.
   */
  int
  __wt_lsm_manager_config(WT_SESSION_IMPL *session, const char **cfg)
  {
  	WT_CONNECTION_IMPL *conn;
  	WT_CONFIG_ITEM cval;
  
  	conn = S2C(session);
  
  	WT_RET(__wt_config_gets(session, cfg, "lsm_manager.merge", &cval));
  	if (cval.val)
  		F_SET(conn, WT_CONN_LSM_MERGE);
  	WT_RET(__wt_config_gets(
  	    session, cfg, "lsm_manager.worker_thread_max", &cval));
  	if (cval.val)
  		conn->lsm_manager.lsm_workers_max = (uint32_t)cval.val;
  	return (0);
  }
  
  /*
   * __lsm_general_worker_start --
   *	Start up all of the general LSM worker threads.
   */
  static int
  __lsm_general_worker_start(WT_SESSION_IMPL *session)
  {
  	WT_CONNECTION_IMPL *conn;
  	WT_LSM_MANAGER *manager;
  	WT_LSM_WORKER_ARGS *worker_args;
  
  	conn = S2C(session);
  	manager = &conn->lsm_manager;
  
  	/*
  	 * Start the worker threads or new worker threads if called via
  	 * reconfigure. The LSM manager is worker[0].
  	 * This should get more sophisticated in the future - only launching
  	 * as many worker threads as are required to keep up with demand.
  	 */
  	WT_ASSERT(session, manager->lsm_workers > 0);
  	WT_ASSERT(session, manager->lsm_workers < manager->lsm_workers_max);
  	for (; manager->lsm_workers < manager->lsm_workers_max;
  	    manager->lsm_workers++) {
  		worker_args =
  		    &manager->lsm_worker_cookies[manager->lsm_workers];
  		worker_args->work_cond = manager->work_cond;
  		worker_args->id = manager->lsm_workers;
  		/*
  		 * The first worker only does switch and drop operations as
  		 * these are both short operations and it is essential
  		 * that switches are responsive to avoid introducing
  		 * throttling stalls.
  		 */
  		if (manager->lsm_workers == 1)
  			worker_args->type =
  			    WT_LSM_WORK_DROP | WT_LSM_WORK_SWITCH;
  		else {
  			worker_args->type =
  			    WT_LSM_WORK_BLOOM |
  			    WT_LSM_WORK_DROP |
  			    WT_LSM_WORK_FLUSH |
  			    WT_LSM_WORK_SWITCH;
  			/*
  			 * Only allow half of the threads to run merges to
  			 * avoid all all workers getting stuck in long-running
  			 * merge operations. Make sure the first worker is
  			 * allowed, so that there is at least one thread
  			 * capable of running merges.  We know the first
  			 * worker is id 2, so set merges on even numbered
  			 * workers.
  			 */
  			if (manager->lsm_workers % 2 == 0)
  				FLD_SET(worker_args->type, WT_LSM_WORK_MERGE);
  		}
  		WT_RET(__wt_lsm_worker_start(session, worker_args));
  	}
  
  	/*
  	 * Setup the first worker properly - if there are only a minimal
  	 * number of workers allow the first worker to flush. Otherwise a
  	 * single merge can lead to switched chunks filling up the cache.
  	 * This is separate to the main loop so that it is applied on startup
  	 * and reconfigure.
  	 */
  	if (manager->lsm_workers_max == WT_LSM_MIN_WORKERS)
  		FLD_SET(manager->lsm_worker_cookies[1].type, WT_LSM_WORK_FLUSH);
  	else
  		FLD_CLR(manager->lsm_worker_cookies[1].type, WT_LSM_WORK_FLUSH);
  
  	return (0);
  }
  
  /*
   * __lsm_stop_workers --
   *	Stop worker threads until the number reaches the configured amount.
   */
  static int
  __lsm_stop_workers(WT_SESSION_IMPL *session)
  {
  	WT_LSM_MANAGER *manager;
  	WT_LSM_WORKER_ARGS *worker_args;
  
  	manager = &S2C(session)->lsm_manager;
  	/*
  	 * Start at the end of the list of threads and stop them until we have
  	 * the desired number. We want to keep all active threads packed at the
  	 * front of the worker array.
  	 */
  	WT_ASSERT(session, manager->lsm_workers > manager->lsm_workers_max);
  	for (; manager->lsm_workers > manager->lsm_workers_max;
  	    manager->lsm_workers--) {
  		worker_args =
  		    &manager->lsm_worker_cookies[manager->lsm_workers - 1];
  		WT_ASSERT(session, worker_args->tid_set);
  
  		WT_RET(__wt_lsm_worker_stop(session, worker_args));
  		worker_args->type = 0;
  
  		/*
  		 * We do not clear the other fields because they are allocated
  		 * statically when the connection was opened.
  		 */
  	}
  
  	/*
  	 * Setup the first worker properly - if there are only a minimal
  	 * number of workers it should flush. Since the number of threads
  	 * is being reduced the field can't already be set.
  	 */
  	if (manager->lsm_workers_max == WT_LSM_MIN_WORKERS)
  		FLD_SET(manager->lsm_worker_cookies[1].type, WT_LSM_WORK_FLUSH);
  
  	return (0);
  }
  
  /*
   * __wt_lsm_manager_reconfig --
   *	Re-configure the LSM manager.
   */
  int
  __wt_lsm_manager_reconfig(WT_SESSION_IMPL *session, const char **cfg)
  {
  	WT_LSM_MANAGER *manager;
  	uint32_t orig_workers;
  
  	manager = &S2C(session)->lsm_manager;
  	orig_workers = manager->lsm_workers_max;
  
  	WT_RET(__wt_lsm_manager_config(session, cfg));
  	/*
  	 * If LSM hasn't started yet, we simply reconfigured the settings
  	 * and we'll let the normal code path start the threads.
  	 */
  	if (manager->lsm_workers_max == 0)
  		return (0);
  	if (manager->lsm_workers == 0)
  		return (0);
  	/*
  	 * If the number of workers has not changed, we're done.
  	 */
  	if (orig_workers == manager->lsm_workers_max)
  		return (0);
  	/*
  	 * If we want more threads, start them.
  	 */
  	if (manager->lsm_workers_max > orig_workers)
  		return (__lsm_general_worker_start(session));
  
  	/*
  	 * Otherwise we want to reduce the number of workers.
  	 */
  	WT_ASSERT(session, manager->lsm_workers_max < orig_workers);
  	WT_RET(__lsm_stop_workers(session));
  	return (0);
  }
  
  /*
   * __wt_lsm_manager_start --
   *	Start the LSM management infrastructure. Our queues and locks were
   *	initialized when the connection was initialized.
   */
  int
  __wt_lsm_manager_start(WT_SESSION_IMPL *session)
  {
  	WT_CONNECTION_IMPL *conn;
  	WT_DECL_RET;
  	WT_LSM_MANAGER *manager;
  	WT_SESSION_IMPL *worker_session;
  	uint32_t i;
  
  	conn = S2C(session);
  	manager = &conn->lsm_manager;
  
  	if (F_ISSET(conn, WT_CONN_READONLY)) {
  		manager->lsm_workers = 0;
  		return (0);
  	}
  	/*
  	 * We need at least a manager, a switch thread and a generic
  	 * worker.
  	 */
  	WT_ASSERT(session, manager->lsm_workers_max > 2);
  
  	/*
  	 * Open sessions for all potential worker threads here - it's not
  	 * safe to have worker threads open/close sessions themselves.
  	 * All the LSM worker threads do their operations on read-only
  	 * files. Use read-uncommitted isolation to avoid keeping
  	 * updates in cache unnecessarily.
  	 */
  	for (i = 0; i < WT_LSM_MAX_WORKERS; i++) {
  		WT_ERR(__wt_open_internal_session(
  		    conn, "lsm-worker", false, 0, &worker_session));
  		worker_session->isolation = WT_ISO_READ_UNCOMMITTED;
  		manager->lsm_worker_cookies[i].session = worker_session;
  	}
  
  	F_SET(conn, WT_CONN_SERVER_LSM);
  
  	/* Start the LSM manager thread. */
  	WT_ERR(__wt_thread_create(session, &manager->lsm_worker_cookies[0].tid,
  	    __lsm_worker_manager, &manager->lsm_worker_cookies[0]));
  
  	if (0) {
  err:		for (i = 0;
  		    (worker_session =
  		    manager->lsm_worker_cookies[i].session) != NULL;
  		    i++)
  			WT_TRET((&worker_session->iface)->close(
  			    &worker_session->iface, NULL));
  	}
  	return (ret);
  }
  
  /*
   * __wt_lsm_manager_free_work_unit --
   *	Release an LSM tree work unit.
   */
  void
  __wt_lsm_manager_free_work_unit(
      WT_SESSION_IMPL *session, WT_LSM_WORK_UNIT *entry)
  {
  	if (entry != NULL) {
  		WT_ASSERT(session, entry->lsm_tree->queue_ref > 0);
  
  		(void)__wt_atomic_sub32(&entry->lsm_tree->queue_ref, 1);
  		__wt_free(session, entry);
  	}
  }
  
  /*
   * __wt_lsm_manager_destroy --
   *	Destroy the LSM manager threads and subsystem.
   */
  int
  __wt_lsm_manager_destroy(WT_SESSION_IMPL *session)
  {
  	WT_CONNECTION_IMPL *conn;
  	WT_DECL_RET;
  	WT_LSM_MANAGER *manager;
  	WT_LSM_WORK_UNIT *current;
  	WT_SESSION *wt_session;
  	uint32_t i;
  	uint64_t removed;
  
  	conn = S2C(session);
  	manager = &conn->lsm_manager;
  	removed = 0;
  
  	/*
  	 * Clear the LSM server flag and flush to ensure running threads see
  	 * the state change.
  	 */
  	F_CLR(conn, WT_CONN_SERVER_LSM);
  	WT_FULL_BARRIER();
  
  	WT_ASSERT(session, !F_ISSET(conn, WT_CONN_READONLY) ||
  	    manager->lsm_workers == 0);
  	if (manager->lsm_workers > 0) {
  		/* Wait for the main LSM manager thread to finish. */
  		while (!F_ISSET(manager, WT_LSM_MANAGER_SHUTDOWN))
  			__wt_yield();
  
  		/* Clean up open LSM handles. */
  		ret = __wt_lsm_tree_close_all(session);
  
  		WT_TRET(__wt_thread_join(
  		    session, manager->lsm_worker_cookies[0].tid));
  
  		/* Release memory from any operations left on the queue. */
  		while ((current = TAILQ_FIRST(&manager->switchqh)) != NULL) {
  			TAILQ_REMOVE(&manager->switchqh, current, q);
  			++removed;
  			__wt_lsm_manager_free_work_unit(session, current);
  		}
  		while ((current = TAILQ_FIRST(&manager->appqh)) != NULL) {
  			TAILQ_REMOVE(&manager->appqh, current, q);
  			++removed;
  			__wt_lsm_manager_free_work_unit(session, current);
  		}
  		while ((current = TAILQ_FIRST(&manager->managerqh)) != NULL) {
  			TAILQ_REMOVE(&manager->managerqh, current, q);
  			++removed;
  			__wt_lsm_manager_free_work_unit(session, current);
  		}
  
  		/* Close all LSM worker sessions. */
  		for (i = 0; i < WT_LSM_MAX_WORKERS; i++) {
  			wt_session =
  			    &manager->lsm_worker_cookies[i].session->iface;
  			WT_TRET(wt_session->close(wt_session, NULL));
  		}
  	}
  	WT_STAT_CONN_INCRV(session, lsm_work_units_discarded, removed);
  
  	/* Free resources that are allocated in connection initialize */
  	__wt_spin_destroy(session, &manager->switch_lock);
  	__wt_spin_destroy(session, &manager->app_lock);
  	__wt_spin_destroy(session, &manager->manager_lock);
  	__wt_cond_destroy(session, &manager->work_cond);
  
  	return (ret);
  }
  
  /*
   * __lsm_manager_worker_shutdown --
   *	Shutdown the LSM worker threads.
   */
  static int
  __lsm_manager_worker_shutdown(WT_SESSION_IMPL *session)
  {
  	WT_DECL_RET;
  	WT_LSM_MANAGER *manager;
  	u_int i;
  
  	manager = &S2C(session)->lsm_manager;
  
  	/*
  	 * Wait for the rest of the LSM workers to shutdown. Start at index
  	 * one - since we (the manager) are at index 0.
  	 */
  	for (i = 1; i < manager->lsm_workers; i++) {
  		WT_ASSERT(session, manager->lsm_worker_cookies[i].tid_set);
  		WT_TRET(__wt_lsm_worker_stop(
  		    session, &manager->lsm_worker_cookies[i]));
  	}
  	return (ret);
  }
  
  /*
   * __lsm_manager_run_server --
   *	Run manager thread operations.
   */
  static int
  __lsm_manager_run_server(WT_SESSION_IMPL *session)
  {
  	WT_CONNECTION_IMPL *conn;
  	WT_DECL_RET;
  	WT_LSM_TREE *lsm_tree;
  	struct timespec now;
  	uint64_t fillms, pushms;
  	bool dhandle_locked;
  
  	conn = S2C(session);
  	dhandle_locked = false;
  
  	while (F_ISSET(conn, WT_CONN_SERVER_LSM)) {
  		__wt_sleep(0, 10000);
  		if (TAILQ_EMPTY(&conn->lsmqh))
  			continue;
  		__wt_readlock(session, &conn->dhandle_lock);
  		F_SET(session, WT_SESSION_LOCKED_HANDLE_LIST_READ);
  		dhandle_locked = true;
  		TAILQ_FOREACH(lsm_tree, &S2C(session)->lsmqh, q) {
  			if (!lsm_tree->active)
  				continue;
  			__wt_epoch(session, &now);
  			pushms = lsm_tree->work_push_ts.tv_sec == 0 ? 0 :
  			    WT_TIMEDIFF_MS(now, lsm_tree->work_push_ts);
  			fillms = 3 * lsm_tree->chunk_fill_ms;
  			if (fillms == 0)
  				fillms = 10000;
  			/*
  			 * If the tree appears to not be triggering enough
  			 * LSM maintenance, help it out. Additional work units
  			 * don't hurt, and can be necessary if some work
  			 * units aren't completed for some reason.
  			 * If the tree hasn't been modified, and there are
  			 * more than 1 chunks - try to get the tree smaller
  			 * so queries run faster.
  			 * If we are getting aggressive - ensure there are
  			 * enough work units that we can get chunks merged.
  			 * If we aren't pushing enough work units, compared
  			 * to how often new chunks are being created add some
  			 * more.
  			 */
  			if (lsm_tree->queue_ref >= LSM_TREE_MAX_QUEUE)
  				WT_STAT_CONN_INCR(session,
  				    lsm_work_queue_max);
  			else if ((!lsm_tree->modified &&
  			    lsm_tree->nchunks > 1) ||
  			    (lsm_tree->queue_ref == 0 &&
  			    lsm_tree->nchunks > 1) ||
  			    (lsm_tree->merge_aggressiveness >
  			    WT_LSM_AGGRESSIVE_THRESHOLD &&
  			     !F_ISSET(lsm_tree, WT_LSM_TREE_COMPACTING)) ||
  			    pushms > fillms) {
  				WT_ERR(__wt_lsm_manager_push_entry(
  				    session, WT_LSM_WORK_SWITCH, 0, lsm_tree));
  				WT_ERR(__wt_lsm_manager_push_entry(
  				    session, WT_LSM_WORK_DROP, 0, lsm_tree));
  				WT_ERR(__wt_lsm_manager_push_entry(
  				    session, WT_LSM_WORK_FLUSH, 0, lsm_tree));
  				WT_ERR(__wt_lsm_manager_push_entry(
  				    session, WT_LSM_WORK_BLOOM, 0, lsm_tree));
  				__wt_verbose(session,
  				    WT_VERB_LSM_MANAGER,
  				    "MGR %s: queue %" PRIu32 " mod %d "
  				    "nchunks %" PRIu32
  				    " flags %#" PRIx32 " aggressive %" PRIu32
  				    " pushms %" PRIu64
  				    " fillms %" PRIu64,
  				    lsm_tree->name, lsm_tree->queue_ref,
  				    lsm_tree->modified, lsm_tree->nchunks,
  				    lsm_tree->flags,
  				    lsm_tree->merge_aggressiveness,
  				    pushms, fillms);
  				WT_ERR(__wt_lsm_manager_push_entry(
  				    session, WT_LSM_WORK_MERGE, 0, lsm_tree));
  			}
  		}
  		__wt_readunlock(session, &conn->dhandle_lock);
  		F_CLR(session, WT_SESSION_LOCKED_HANDLE_LIST_READ);
  		dhandle_locked = false;
  	}
  
  err:	if (dhandle_locked) {
  		__wt_readunlock(session, &conn->dhandle_lock);
  		F_CLR(session, WT_SESSION_LOCKED_HANDLE_LIST_READ);
  	}
  	return (ret);
  }
  
  /*
   * __lsm_worker_manager --
   *	A thread that manages all open LSM trees, and the shared LSM worker
   *	threads.
   */
  static WT_THREAD_RET
  __lsm_worker_manager(void *arg)
  {
  	WT_DECL_RET;
  	WT_LSM_MANAGER *manager;
  	WT_LSM_WORKER_ARGS *cookie;
  	WT_SESSION_IMPL *session;
  
  	cookie = (WT_LSM_WORKER_ARGS *)arg;
  	session = cookie->session;
  	manager = &S2C(session)->lsm_manager;
  
  	WT_ERR(__lsm_general_worker_start(session));
  	WT_ERR(__lsm_manager_run_server(session));
  	WT_ERR(__lsm_manager_worker_shutdown(session));
  
  	if (ret != 0) {
  err:		WT_PANIC_MSG(session, ret, "LSM worker manager thread error");
  	}
  
  	/* Connection close waits on us to shutdown, let it know we're done. */
  	F_SET(manager, WT_LSM_MANAGER_SHUTDOWN);
  	WT_FULL_BARRIER();
  
  	return (WT_THREAD_RET_VALUE);
  }
  
  /*
   * __wt_lsm_manager_clear_tree --
   *	Remove all entries for a tree from the LSM manager queues. This
   *	introduces an inefficiency if LSM trees are being opened and closed
   *	regularly.
   */
  void
  __wt_lsm_manager_clear_tree(WT_SESSION_IMPL *session, WT_LSM_TREE *lsm_tree)
  {
  	WT_LSM_MANAGER *manager;
  	WT_LSM_WORK_UNIT *current, *next;
  	uint64_t removed;
  
  	manager = &S2C(session)->lsm_manager;
  	removed = 0;
  
  	/* Clear out the tree from the switch queue */
  	__wt_spin_lock(session, &manager->switch_lock);
  
  	/* Structure the loop so that it's safe to free as we iterate */
  	for (current = TAILQ_FIRST(&manager->switchqh);
  	    current != NULL; current = next) {
  		next = TAILQ_NEXT(current, q);
  		if (current->lsm_tree != lsm_tree)
  			continue;
  		++removed;
  		TAILQ_REMOVE(&manager->switchqh, current, q);
  		__wt_lsm_manager_free_work_unit(session, current);
  	}
  	__wt_spin_unlock(session, &manager->switch_lock);
  	/* Clear out the tree from the application queue */
  	__wt_spin_lock(session, &manager->app_lock);
  	for (current = TAILQ_FIRST(&manager->appqh);
  	    current != NULL; current = next) {
  		next = TAILQ_NEXT(current, q);
  		if (current->lsm_tree != lsm_tree)
  			continue;
  		++removed;
  		TAILQ_REMOVE(&manager->appqh, current, q);
  		__wt_lsm_manager_free_work_unit(session, current);
  	}
  	__wt_spin_unlock(session, &manager->app_lock);
  	/* Clear out the tree from the manager queue */
  	__wt_spin_lock(session, &manager->manager_lock);
  	for (current = TAILQ_FIRST(&manager->managerqh);
  	    current != NULL; current = next) {
  		next = TAILQ_NEXT(current, q);
  		if (current->lsm_tree != lsm_tree)
  			continue;
  		++removed;
  		TAILQ_REMOVE(&manager->managerqh, current, q);
  		__wt_lsm_manager_free_work_unit(session, current);
  	}
  	__wt_spin_unlock(session, &manager->manager_lock);
  	WT_STAT_CONN_INCRV(session, lsm_work_units_discarded, removed);
  }
  
  /*
   * We assume this is only called from __wt_lsm_manager_pop_entry and we
   * have session, entry and type available to use.  If the queue is empty
   * we may return from the macro.
   */
  #define	LSM_POP_ENTRY(qh, qlock, qlen) do {				\
  	if (TAILQ_EMPTY(qh))						\
  		return (0);						\
  	__wt_spin_lock(session, qlock);					\
  	TAILQ_FOREACH(entry, (qh), q) {					\
  		if (FLD_ISSET(type, entry->type)) {			\
  			TAILQ_REMOVE(qh, entry, q);			\
  			WT_STAT_CONN_DECR(session, qlen);		\
  			break;						\
  		}							\
  	}								\
  	__wt_spin_unlock(session, (qlock));				\
  } while (0)
  
  /*
   * __wt_lsm_manager_pop_entry --
   *	Retrieve the head of the queue, if it matches the requested work
   *	unit type.
   */
  int
  __wt_lsm_manager_pop_entry(
      WT_SESSION_IMPL *session, uint32_t type, WT_LSM_WORK_UNIT **entryp)
  {
  	WT_LSM_MANAGER *manager;
  	WT_LSM_WORK_UNIT *entry;
  
  	manager = &S2C(session)->lsm_manager;
  	*entryp = NULL;
  	entry = NULL;
  
  	/*
  	 * Pop the entry off the correct queue based on our work type.
  	 */
  	if (type == WT_LSM_WORK_SWITCH)
  		LSM_POP_ENTRY(&manager->switchqh,
  		    &manager->switch_lock, lsm_work_queue_switch);
  	else if (type == WT_LSM_WORK_MERGE)
  		LSM_POP_ENTRY(&manager->managerqh,
  		    &manager->manager_lock, lsm_work_queue_manager);
  	else
  		LSM_POP_ENTRY(&manager->appqh,
  		    &manager->app_lock, lsm_work_queue_app);
  	if (entry != NULL)
  		WT_STAT_CONN_INCR(session, lsm_work_units_done);
  	*entryp = entry;
  	return (0);
  }
  
  /*
   * Push a work unit onto the appropriate queue.  This macro assumes we are
   * called from __wt_lsm_manager_push_entry and we have session and entry
   * available for use.
   */
  #define	LSM_PUSH_ENTRY(qh, qlock, qlen) do {				\
  	__wt_spin_lock(session, qlock);					\
  	TAILQ_INSERT_TAIL((qh), entry, q);				\
  	WT_STAT_CONN_INCR(session, qlen);				\
  	__wt_spin_unlock(session, qlock);				\
  } while (0)
  
  /*
   * __wt_lsm_manager_push_entry --
   *	Add an entry to the end of the switch queue.
   */
  int
  __wt_lsm_manager_push_entry(WT_SESSION_IMPL *session,
      uint32_t type, uint32_t flags, WT_LSM_TREE *lsm_tree)
  {
  	WT_LSM_MANAGER *manager;
  	WT_LSM_WORK_UNIT *entry;
  
  	manager = &S2C(session)->lsm_manager;
  
  	WT_ASSERT(session, !F_ISSET(S2C(session), WT_CONN_READONLY));
  	/*
  	 * Don't add merges or bloom filter creates if merges
  	 * or bloom filters are disabled in the tree.
  	 */
  	switch (type) {
  	case WT_LSM_WORK_BLOOM:
  		if (FLD_ISSET(lsm_tree->bloom, WT_LSM_BLOOM_OFF))
  			return (0);
  		break;
  	case WT_LSM_WORK_MERGE:
  		if (!F_ISSET(lsm_tree, WT_LSM_TREE_MERGES))
  			return (0);
  		break;
  	}
  
  	/*
  	 * Don't allow any work units unless a tree is active, this avoids
  	 * races on shutdown between clearing out queues and pushing new
  	 * work units.
  	 *
  	 * Increment the queue reference before checking the flag since
  	 * on close, the flag is cleared and then the queue reference count
  	 * is checked.
  	 */
  	(void)__wt_atomic_add32(&lsm_tree->queue_ref, 1);
  	if (!lsm_tree->active) {
  		(void)__wt_atomic_sub32(&lsm_tree->queue_ref, 1);
  		return (0);
  	}
  
  	__wt_epoch(session, &lsm_tree->work_push_ts);
  	WT_RET(__wt_calloc_one(session, &entry));
  	entry->type = type;
  	entry->flags = flags;
  	entry->lsm_tree = lsm_tree;
  	WT_STAT_CONN_INCR(session, lsm_work_units_created);
  
  	if (type == WT_LSM_WORK_SWITCH)
  		LSM_PUSH_ENTRY(&manager->switchqh,
  		    &manager->switch_lock, lsm_work_queue_switch);
  	else if (type == WT_LSM_WORK_MERGE)
  		LSM_PUSH_ENTRY(&manager->managerqh,
  		    &manager->manager_lock, lsm_work_queue_manager);
  	else
  		LSM_PUSH_ENTRY(&manager->appqh,
  		    &manager->app_lock, lsm_work_queue_app);
  
  	__wt_cond_signal(session, manager->work_cond);
  	return (0);
  }
  