fmd/common/fmd_module.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright (c) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
 */

#include <signal.h>
#include <dirent.h>
#include <limits.h>
#include <alloca.h>
#include <unistd.h>
#include <stdio.h>

#include <fmd_string.h>
#include <fmd_alloc.h>
#include <fmd_module.h>
#include <fmd_error.h>
#include <fmd_conf.h>
#include <fmd_dispq.h>
#include <fmd_eventq.h>
#include <fmd_timerq.h>
#include <fmd_subr.h>
#include <fmd_thread.h>
#include <fmd_ustat.h>
#include <fmd_case.h>
#include <fmd_protocol.h>
#include <fmd_buf.h>
#include <fmd_ckpt.h>
#include <fmd_xprt.h>
#include <fmd_topo.h>

#include <fmd.h>

/*
 * Template for per-module statistics installed by fmd on behalf of each active
 * module.  These are used to initialize the per-module mp->mod_stats below.
 * NOTE: FMD_TYPE_STRING statistics should not be used here.  If they are
 * required in the future, the FMD_ADM_MODDSTAT service routine must change.
 */
static const fmd_modstat_t _fmd_modstat_tmpl = {
{
{ "fmd.dispatched", FMD_TYPE_UINT64, "total events dispatched to module" },
{ "fmd.dequeued", FMD_TYPE_UINT64, "total events dequeued by module" },
{ "fmd.prdequeued", FMD_TYPE_UINT64, "protocol events dequeued by module" },
{ "fmd.dropped", FMD_TYPE_UINT64, "total events dropped on queue overflow" },
{ "fmd.wcnt", FMD_TYPE_UINT32, "count of events waiting on queue" },
{ "fmd.wtime", FMD_TYPE_TIME, "total wait time on queue" },
{ "fmd.wlentime", FMD_TYPE_TIME, "total wait length * time product" },
{ "fmd.wlastupdate", FMD_TYPE_TIME, "hrtime of last wait queue update" },
{ "fmd.dtime", FMD_TYPE_TIME, "total processing time after dequeue" },
{ "fmd.dlastupdate", FMD_TYPE_TIME, "hrtime of last event dequeue completion" },
},
{ "fmd.loadtime", FMD_TYPE_TIME, "hrtime at which module was loaded" },
{ "fmd.snaptime", FMD_TYPE_TIME, "hrtime of last statistics snapshot" },
{ "fmd.accepted", FMD_TYPE_UINT64, "total events accepted by module" },
{ "fmd.debugdrop", FMD_TYPE_UINT64, "dropped debug messages" },
{ "fmd.memtotal", FMD_TYPE_SIZE, "total memory allocated by module" },
{ "fmd.memlimit", FMD_TYPE_SIZE, "limit on total memory allocated" },
{ "fmd.buftotal", FMD_TYPE_SIZE, "total buffer space used by module" },
{ "fmd.buflimit", FMD_TYPE_SIZE, "limit on total buffer space" },
{ "fmd.thrtotal", FMD_TYPE_UINT32, "total number of auxiliary threads" },
{ "fmd.thrlimit", FMD_TYPE_UINT32, "limit on number of auxiliary threads" },
{ "fmd.doorthrtotal", FMD_TYPE_UINT32, "total number of door server threads" },
{ "fmd.doorthrlimit", FMD_TYPE_UINT32, "limit on door server threads" },
{ "fmd.caseopen", FMD_TYPE_UINT64, "cases currently open by module" },
{ "fmd.casesolved", FMD_TYPE_UINT64, "total cases solved by module" },
{ "fmd.caseclosed", FMD_TYPE_UINT64, "total cases closed by module" },
{ "fmd.ckptsave", FMD_TYPE_BOOL, "save checkpoints for module" },
{ "fmd.ckptrestore", FMD_TYPE_BOOL, "restore checkpoints for module" },
{ "fmd.ckptzero", FMD_TYPE_BOOL, "zeroed checkpoint at startup" },
{ "fmd.ckptcnt", FMD_TYPE_UINT64, "number of checkpoints taken" },
{ "fmd.ckpttime", FMD_TYPE_TIME, "total checkpoint time" },
{ "fmd.xprtopen", FMD_TYPE_UINT32, "total number of open transports" },
{ "fmd.xprtlimit", FMD_TYPE_UINT32, "limit on number of open transports" },
{ "fmd.xprtqlimit", FMD_TYPE_UINT32, "limit on transport event queue length" },
};

static void
fmd_module_start(void *arg)
{
	fmd_module_t *mp = arg;
	fmd_event_t *ep;
	fmd_xprt_t *xp;

	(void) pthread_mutex_lock(&mp->mod_lock);

	if (mp->mod_ops->mop_init(mp) != 0 || mp->mod_error != 0) {
		if (mp->mod_error == 0)
			mp->mod_error = errno ? errno : EFMD_MOD_INIT;
		goto out;
	}

	if (fmd.d_mod_event != NULL)
		fmd_eventq_insert_at_head(mp->mod_queue, fmd.d_mod_event);

	ASSERT(MUTEX_HELD(&mp->mod_lock));
	mp->mod_flags |= FMD_MOD_INIT;

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	/*
	 * If the module opened any transports while executing _fmd_init(),
	 * they are suspended. Now that _fmd_init() is done, wake them up.
	 */
	for (xp = fmd_list_next(&mp->mod_transports);
	    xp != NULL; xp = fmd_list_next(xp))
		fmd_xprt_xresume(xp, FMD_XPRT_ISUSPENDED);

	/*
	 * Wait for events to arrive by checking mod_error and then sleeping in
	 * fmd_eventq_delete().  If a NULL event is returned, the eventq has
	 * been aborted and we continue on to call fini and exit the thread.
	 */
	while ((ep = fmd_eventq_delete(mp->mod_queue)) != NULL) {
		/*
		 * If the module has failed, discard the event without ever
		 * passing it to the module and go back to sleep.
		 */
		if (mp->mod_error != 0) {
			fmd_eventq_done(mp->mod_queue);
			fmd_event_rele(ep);
			continue;
		}

		mp->mod_ops->mop_dispatch(mp, ep);
		fmd_eventq_done(mp->mod_queue);

		/*
		 * Once mop_dispatch() is complete, grab the lock and perform
		 * any event-specific post-processing.  Finally, if necessary,
		 * checkpoint the state of the module after this event.
		 */
		fmd_module_lock(mp);

		if (FMD_EVENT_TYPE(ep) == FMD_EVT_CLOSE)
			fmd_case_delete(FMD_EVENT_DATA(ep));

		fmd_ckpt_save(mp);
		fmd_module_unlock(mp);
		fmd_event_rele(ep);
	}

	if (mp->mod_ops->mop_fini(mp) != 0 && mp->mod_error == 0)
		mp->mod_error = errno ? errno : EFMD_MOD_FINI;

	(void) pthread_mutex_lock(&mp->mod_lock);
	mp->mod_flags |= FMD_MOD_FINI;

out:
	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);
}

fmd_module_t *
fmd_module_create(const char *path, const fmd_modops_t *ops)
{
	fmd_module_t *mp = fmd_zalloc(sizeof (fmd_module_t), FMD_SLEEP);

	char buf[PATH_MAX], *p;
	const char *dir;
	uint32_t limit;
	int err;

	(void) strlcpy(buf, fmd_strbasename(path), sizeof (buf));
	if ((p = strrchr(buf, '.')) != NULL && strcmp(p, ".so") == 0)
		*p = '\0'; /* strip trailing .so from any module name */

	(void) pthread_mutex_init(&mp->mod_lock, NULL);
	(void) pthread_cond_init(&mp->mod_cv, NULL);
	(void) pthread_mutex_init(&mp->mod_stats_lock, NULL);

	mp->mod_name = fmd_strdup(buf, FMD_SLEEP);
	mp->mod_path = fmd_strdup(path, FMD_SLEEP);
	mp->mod_ops = ops;
	mp->mod_ustat = fmd_ustat_create();

	(void) fmd_conf_getprop(fmd.d_conf, "ckpt.dir", &dir);
	(void) snprintf(buf, sizeof (buf),
	    "%s/%s/%s", fmd.d_rootdir, dir, mp->mod_name);

	mp->mod_ckpt = fmd_strdup(buf, FMD_SLEEP);

	(void) fmd_conf_getprop(fmd.d_conf, "client.tmrlim", &limit);
	mp->mod_timerids = fmd_idspace_create(mp->mod_name, 1, limit + 1);
	mp->mod_threads = fmd_idspace_create(mp->mod_name, 0, INT_MAX);

	fmd_buf_hash_create(&mp->mod_bufs);
	fmd_serd_hash_create(&mp->mod_serds);

	mp->mod_topo_current = fmd_topo_hold();

	(void) pthread_mutex_lock(&fmd.d_mod_lock);
	fmd_list_append(&fmd.d_mod_list, mp);
	(void) pthread_mutex_unlock(&fmd.d_mod_lock);

	/*
	 * Initialize the module statistics that are kept on its behalf by fmd.
	 * These are set up using a template defined at the top of this file.
	 */
	if ((mp->mod_stats = (fmd_modstat_t *)fmd_ustat_insert(mp->mod_ustat,
	    FMD_USTAT_ALLOC, sizeof (_fmd_modstat_tmpl) / sizeof (fmd_stat_t),
	    (fmd_stat_t *)&_fmd_modstat_tmpl, NULL)) == NULL) {
		fmd_error(EFMD_MOD_INIT, "failed to initialize per-mod stats");
		fmd_module_destroy(mp);
		return (NULL);
	}

	if (nv_alloc_init(&mp->mod_nva_sleep,
	    &fmd_module_nva_ops_sleep, mp) != 0 ||
	    nv_alloc_init(&mp->mod_nva_nosleep,
	    &fmd_module_nva_ops_nosleep, mp) != 0) {
		fmd_error(EFMD_MOD_INIT, "failed to initialize nvlist "
		    "allocation routines");
		fmd_module_destroy(mp);
		return (NULL);
	}

	(void) fmd_conf_getprop(fmd.d_conf, "client.evqlim", &limit);

	mp->mod_queue = fmd_eventq_create(mp,
	    &mp->mod_stats->ms_evqstat, &mp->mod_stats_lock, limit);

	(void) fmd_conf_getprop(fmd.d_conf, "client.memlim",
	    &mp->mod_stats->ms_memlimit.fmds_value.ui64);

	(void) fmd_conf_getprop(fmd.d_conf, "client.buflim",
	    &mp->mod_stats->ms_buflimit.fmds_value.ui64);

	(void) fmd_conf_getprop(fmd.d_conf, "client.thrlim",
	    &mp->mod_stats->ms_thrlimit.fmds_value.ui32);

	(void) fmd_conf_getprop(fmd.d_conf, "client.doorthrlim",
	    &mp->mod_stats->ms_doorthrlimit.fmds_value.ui32);

	(void) fmd_conf_getprop(fmd.d_conf, "client.xprtlim",
	    &mp->mod_stats->ms_xprtlimit.fmds_value.ui32);

	(void) fmd_conf_getprop(fmd.d_conf, "client.xprtqlim",
	    &mp->mod_stats->ms_xprtqlimit.fmds_value.ui32);

	(void) fmd_conf_getprop(fmd.d_conf, "ckpt.save",
	    &mp->mod_stats->ms_ckpt_save.fmds_value.bool);

	(void) fmd_conf_getprop(fmd.d_conf, "ckpt.restore",
	    &mp->mod_stats->ms_ckpt_restore.fmds_value.bool);

	(void) fmd_conf_getprop(fmd.d_conf, "ckpt.zero",
	    &mp->mod_stats->ms_ckpt_zeroed.fmds_value.bool);

	if (mp->mod_stats->ms_ckpt_zeroed.fmds_value.bool)
		fmd_ckpt_delete(mp); /* blow away any pre-existing checkpoint */

	/*
	 * Place a hold on the module and grab the module lock before creating
	 * the module's thread to ensure that it cannot destroy the module and
	 * that it cannot call ops->mop_init() before we're done setting up.
	 * NOTE: from now on, we must use fmd_module_rele() for error paths.
	 */
	fmd_module_hold(mp);
	(void) pthread_mutex_lock(&mp->mod_lock);
	mp->mod_stats->ms_loadtime.fmds_value.ui64 = gethrtime();
	mp->mod_thread = fmd_thread_create(mp, fmd_module_start, mp);

	if (mp->mod_thread == NULL) {
		fmd_error(EFMD_MOD_THR, "failed to create thread for %s", path);
		(void) pthread_mutex_unlock(&mp->mod_lock);
		fmd_module_rele(mp);
		return (NULL);
	}

	/*
	 * At this point our module structure is nearly finished and its thread
	 * is starting execution in fmd_module_start() above, which will begin
	 * by blocking for mod_lock.  We now drop mod_lock and wait for either
	 * FMD_MOD_INIT or mod_error to be set before proceeding.
	 */
	while (!(mp->mod_flags & FMD_MOD_INIT) && mp->mod_error == 0)
		(void) pthread_cond_wait(&mp->mod_cv, &mp->mod_lock);

	/*
	 * If the module has failed to initialize, copy its errno to the errno
	 * of the caller, wait for it to unload, and then destroy it.
	 */
	if (!(mp->mod_flags & FMD_MOD_INIT)) {
		err = mp->mod_error;
		(void) pthread_mutex_unlock(&mp->mod_lock);

		if (err == EFMD_CKPT_INVAL)
			fmd_ckpt_rename(mp); /* move aside bad checkpoint */

		/*
		 * If we're in the background, keep quiet about failure to
		 * load because a handle wasn't registered: this is a module's
		 * way of telling us it didn't want to be loaded for some
		 * reason related to system configuration.  If we're in the
		 * foreground we log this too in order to inform developers.
		 */
		if (fmd.d_fg || err != EFMD_HDL_INIT) {
			fmd_error(EFMD_MOD_INIT, "failed to load %s: %s\n",
			    path, fmd_strerror(err));
		}

		fmd_module_unload(mp);
		fmd_module_rele(mp);

		(void) fmd_set_errno(err);
		return (NULL);
	}

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	fmd_dprintf(FMD_DBG_MOD, "loaded module %s\n", mp->mod_name);
	return (mp);
}

static void
fmd_module_untimeout(fmd_idspace_t *ids, id_t id, fmd_module_t *mp)
{
	void *arg = fmd_timerq_remove(fmd.d_timers, ids, id);

	/*
	 * The root module calls fmd_timerq_install() directly and must take
	 * responsibility for any cleanup of timer arguments that is required.
	 * All other modules use fmd_modtimer_t's as the arg data; free them.
	 */
	if (arg != NULL && mp != fmd.d_rmod)
		fmd_free(arg, sizeof (fmd_modtimer_t));
}

void
fmd_module_unload(fmd_module_t *mp)
{
	fmd_modtopo_t *mtp;

	(void) pthread_mutex_lock(&mp->mod_lock);

	if (mp->mod_flags & FMD_MOD_QUIT) {
		(void) pthread_mutex_unlock(&mp->mod_lock);
		return; /* module is already unloading */
	}

	ASSERT(mp->mod_thread != NULL);
	mp->mod_flags |= FMD_MOD_QUIT;

	if (mp->mod_queue != NULL)
		fmd_eventq_abort(mp->mod_queue);

	/*
	 * Wait for the module's thread to stop processing events and call
	 * _fmd_fini() and exit.  We do this by waiting for FMD_MOD_FINI to be
	 * set if INIT was set, and then attempting to join with the thread.
	 */
	while ((mp->mod_flags & (FMD_MOD_INIT | FMD_MOD_FINI)) == FMD_MOD_INIT)
		(void) pthread_cond_wait(&mp->mod_cv, &mp->mod_lock);

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	fmd_thread_destroy(mp->mod_thread, FMD_THREAD_JOIN);
	mp->mod_thread = NULL;

	/*
	 * Once the module is no longer active, clean up any data structures
	 * that are only required when the module is loaded.
	 */
	fmd_module_lock(mp);

	if (mp->mod_timerids != NULL) {
		fmd_idspace_apply(mp->mod_timerids,
		    (void (*)())fmd_module_untimeout, mp);

		fmd_idspace_destroy(mp->mod_timerids);
		mp->mod_timerids = NULL;
	}

	if (mp->mod_threads != NULL) {
		fmd_idspace_destroy(mp->mod_threads);
		mp->mod_threads = NULL;
	}

	(void) fmd_buf_hash_destroy(&mp->mod_bufs);
	fmd_serd_hash_destroy(&mp->mod_serds);

	while ((mtp = fmd_list_next(&mp->mod_topolist)) != NULL) {
		fmd_list_delete(&mp->mod_topolist, mtp);
		fmd_topo_rele(mtp->mt_topo);
		fmd_free(mtp, sizeof (fmd_modtopo_t));
	}

	fmd_module_unlock(mp);
	fmd_dprintf(FMD_DBG_MOD, "unloaded module %s\n", mp->mod_name);
}

void
fmd_module_destroy(fmd_module_t *mp)
{
	fmd_conf_formal_t *cfp = mp->mod_argv;
	int i;

	ASSERT(MUTEX_HELD(&mp->mod_lock));

	if (mp->mod_thread != NULL) {
		(void) pthread_mutex_unlock(&mp->mod_lock);
		fmd_module_unload(mp);
		(void) pthread_mutex_lock(&mp->mod_lock);
	}

	ASSERT(mp->mod_thread == NULL);
	ASSERT(mp->mod_refs == 0);

	/*
	 * Once the module's thread is dead, we can safely remove the module
	 * from global visibility and by removing it from d_mod_list.  Any
	 * modhash pointers are already gone by virtue of mod_refs being zero.
	 */
	(void) pthread_mutex_lock(&fmd.d_mod_lock);
	fmd_list_delete(&fmd.d_mod_list, mp);
	(void) pthread_mutex_unlock(&fmd.d_mod_lock);

	if (mp->mod_topo_current != NULL)
		fmd_topo_rele(mp->mod_topo_current);

	if (mp->mod_nva_sleep.nva_ops != NULL)
		nv_alloc_fini(&mp->mod_nva_sleep);
	if (mp->mod_nva_nosleep.nva_ops != NULL)
		nv_alloc_fini(&mp->mod_nva_nosleep);

	/*
	 * Once the module is no longer processing events and no longer visible
	 * through any program data structures, we can free all of its content.
	 */
	if (mp->mod_queue != NULL) {
		fmd_eventq_destroy(mp->mod_queue);
		mp->mod_queue = NULL;
	}

	if (mp->mod_ustat != NULL) {
		(void) pthread_mutex_lock(&mp->mod_stats_lock);
		fmd_ustat_destroy(mp->mod_ustat);
		mp->mod_ustat = NULL;
		mp->mod_stats = NULL;
		(void) pthread_mutex_unlock(&mp->mod_stats_lock);
	}

	for (i = 0; i < mp->mod_dictc; i++)
		fm_dc_closedict(mp->mod_dictv[i]);

	fmd_free(mp->mod_dictv, sizeof (struct fm_dc_handle *) * mp->mod_dictc);

	if (mp->mod_conf != NULL)
		fmd_conf_close(mp->mod_conf);

	for (i = 0; i < mp->mod_argc; i++, cfp++) {
		fmd_strfree((char *)cfp->cf_name);
		fmd_strfree((char *)cfp->cf_default);
	}

	fmd_free(mp->mod_argv, sizeof (fmd_conf_formal_t) * mp->mod_argc);

	fmd_strfree(mp->mod_name);
	fmd_strfree(mp->mod_path);
	fmd_strfree(mp->mod_ckpt);
	nvlist_free(mp->mod_fmri);
	fmd_strfree(mp->mod_vers);

	fmd_free(mp, sizeof (fmd_module_t));
}

/*
 * fmd_module_error() is called after the stack is unwound from a call to
 * fmd_module_abort() to indicate that the module has failed.  The mod_error
 * field is used to hold the error code of the first fatal error to the module.
 * An EFMD_MOD_FAIL event is then created and sent to fmd-self-diagnosis.
 */
static void
fmd_module_error(fmd_module_t *mp, int err)
{
	fmd_event_t *e;
	nvlist_t *nvl;
	char *class;

	ASSERT(MUTEX_HELD(&mp->mod_lock));
	ASSERT(err != 0);

	TRACE((FMD_DBG_MOD, "module aborted: err=%d", err));

	if (mp->mod_error == 0)
		mp->mod_error = err;

	if (mp == fmd.d_self)
		return; /* do not post event if fmd.d_self itself fails */

	/*
	 * Send an error indicating the module has now failed to fmd.d_self.
	 * Since the error causing the failure has already been logged by
	 * fmd_api_xerror(), we do not need to bother logging this event.
	 * It only exists for the purpose of notifying fmd.d_self that it can
	 * close the case associated with this module because mod_error is set.
	 */
	nvl = fmd_protocol_moderror(mp, EFMD_MOD_FAIL, fmd_strerror(err));
	(void) nvlist_lookup_string(nvl, FM_CLASS, &class);
	e = fmd_event_create(FMD_EVT_PROTOCOL, FMD_HRT_NOW, nvl, class);
	fmd_dispq_dispatch(fmd.d_disp, e, class);
}

void
fmd_module_dispatch(fmd_module_t *mp, fmd_event_t *e)
{
	const fmd_hdl_ops_t *ops = mp->mod_info->fmdi_ops;
	fmd_event_impl_t *ep = (fmd_event_impl_t *)e;
	fmd_hdl_t *hdl = (fmd_hdl_t *)mp;
	fmd_modtimer_t *t;
	fmd_topo_t *old_topo;
	volatile int err;

	/*
	 * Before calling the appropriate module callback, enter the module as
	 * if by fmd_module_enter() and establish mod_jmpbuf for any aborts.
	 */
	(void) pthread_mutex_lock(&mp->mod_lock);

	ASSERT(!(mp->mod_flags & FMD_MOD_BUSY));
	mp->mod_flags |= FMD_MOD_BUSY;

	if ((err = setjmp(mp->mod_jmpbuf)) != 0) {
		(void) pthread_mutex_lock(&mp->mod_lock);
		fmd_module_error(mp, err);
	}

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	/*
	 * If it's the first time through fmd_module_dispatch(), call the
	 * appropriate module callback based on the event type.  If the call
	 * triggers an fmd_module_abort(), we'll return to setjmp() above with
	 * err set to a non-zero value and then bypass this before exiting.
	 */
	if (err == 0) {
		switch (ep->ev_type) {
		case FMD_EVT_PROTOCOL:
			ops->fmdo_recv(hdl, e, ep->ev_nvl, ep->ev_data);
			break;
		case FMD_EVT_TIMEOUT:
			t = ep->ev_data;
			ASSERT(t->mt_mod == mp);
			ops->fmdo_timeout(hdl, t->mt_id, t->mt_arg);
			break;
		case FMD_EVT_CLOSE:
			ops->fmdo_close(hdl, ep->ev_data);
			break;
		case FMD_EVT_STATS:
			ops->fmdo_stats(hdl);
			fmd_modstat_publish(mp);
			break;
		case FMD_EVT_GC:
			ops->fmdo_gc(hdl);
			break;
		case FMD_EVT_PUBLISH:
			fmd_case_publish(ep->ev_data, FMD_CASE_CURRENT);
			break;
		case FMD_EVT_TOPO:
			/*
			 * Save the pointer to the old topology and update
			 * the pointer with the updated topology.
			 * With this approach, other threads that reference the
			 * topology either
			 *  - finishes with old topology since
			 *	it is released after updating
			 *	mod_topo_current.
			 *  - or is blocked while mod_topo_current is updated.
			 */
			old_topo = mp->mod_topo_current;
			fmd_module_lock(mp);
			mp->mod_topo_current = (fmd_topo_t *)ep->ev_data;
			fmd_topo_addref(mp->mod_topo_current);
			fmd_module_unlock(mp);
			fmd_topo_rele(old_topo);
			ops->fmdo_topo(hdl, mp->mod_topo_current->ft_hdl);
			break;
		}
	}

	fmd_module_exit(mp);
}

int
fmd_module_transport(fmd_module_t *mp, fmd_xprt_t *xp, fmd_event_t *e)
{
	fmd_event_impl_t *ep = (fmd_event_impl_t *)e;
	fmd_hdl_t *hdl = (fmd_hdl_t *)mp;

	ASSERT(ep->ev_type == FMD_EVT_PROTOCOL);
	return (mp->mod_info->fmdi_ops->fmdo_send(hdl, xp, e, ep->ev_nvl));
}

void
fmd_module_timeout(fmd_modtimer_t *t, id_t id, hrtime_t hrt)
{
	fmd_event_t *e;

	t->mt_id = id; /* save id in case we need to delete from eventq */
	e = fmd_event_create(FMD_EVT_TIMEOUT, hrt, NULL, t);
	fmd_eventq_insert_at_time(t->mt_mod->mod_queue, e);
}

/*
 * Garbage collection is initiated by a timer callback once per day or at the
 * request of fmadm.  Purge old SERD entries and send the module a GC event.
 */
void
fmd_module_gc(fmd_module_t *mp)
{
	fmd_hdl_info_t *info;
	fmd_event_t *e;

	if (mp->mod_error != 0)
		return; /* do not do anything if the module has failed */

	fmd_module_lock(mp);

	if ((info = mp->mod_info) != NULL) {
		fmd_serd_hash_apply(&mp->mod_serds,
		    (fmd_serd_eng_f *)fmd_serd_eng_gc, NULL);
	}

	fmd_module_unlock(mp);

	if (info != NULL) {
		e = fmd_event_create(FMD_EVT_GC, FMD_HRT_NOW, NULL, NULL);
		fmd_eventq_insert_at_head(mp->mod_queue, e);
	}
}

void
fmd_module_trygc(fmd_module_t *mp)
{
	if (fmd_module_trylock(mp)) {
		fmd_serd_hash_apply(&mp->mod_serds,
		    (fmd_serd_eng_f *)fmd_serd_eng_gc, NULL);
		fmd_module_unlock(mp);
	}
}

int
fmd_module_contains(fmd_module_t *mp, fmd_event_t *ep)
{
	fmd_case_t *cp;
	int rv = 0;

	fmd_module_lock(mp);

	for (cp = fmd_list_next(&mp->mod_cases);
	    cp != NULL; cp = fmd_list_next(cp)) {
		if ((rv = fmd_case_contains(cp, ep)) != 0)
			break;
	}

	if (rv == 0)
		rv = fmd_serd_hash_contains(&mp->mod_serds, ep);

	fmd_module_unlock(mp);
	return (rv);
}

void
fmd_module_setdirty(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);
	mp->mod_flags |= FMD_MOD_MDIRTY;
	(void) pthread_mutex_unlock(&mp->mod_lock);
}

void
fmd_module_setcdirty(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);
	mp->mod_flags |= FMD_MOD_CDIRTY;
	(void) pthread_mutex_unlock(&mp->mod_lock);
}

void
fmd_module_clrdirty(fmd_module_t *mp)
{
	fmd_case_t *cp;

	fmd_module_lock(mp);

	if (mp->mod_flags & FMD_MOD_CDIRTY) {
		for (cp = fmd_list_next(&mp->mod_cases);
		    cp != NULL; cp = fmd_list_next(cp))
			fmd_case_clrdirty(cp);
	}

	if (mp->mod_flags & FMD_MOD_MDIRTY) {
		fmd_serd_hash_apply(&mp->mod_serds,
		    (fmd_serd_eng_f *)fmd_serd_eng_clrdirty, NULL);
		fmd_buf_hash_commit(&mp->mod_bufs);
	}

	(void) pthread_mutex_lock(&mp->mod_lock);
	mp->mod_flags &= ~(FMD_MOD_MDIRTY | FMD_MOD_CDIRTY);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	fmd_module_unlock(mp);
}

void
fmd_module_commit(fmd_module_t *mp)
{
	fmd_case_t *cp;

	ASSERT(fmd_module_locked(mp));

	if (mp->mod_flags & FMD_MOD_CDIRTY) {
		for (cp = fmd_list_next(&mp->mod_cases);
		    cp != NULL; cp = fmd_list_next(cp))
			fmd_case_commit(cp);
	}

	if (mp->mod_flags & FMD_MOD_MDIRTY) {
		fmd_serd_hash_apply(&mp->mod_serds,
		    (fmd_serd_eng_f *)fmd_serd_eng_commit, NULL);
		fmd_buf_hash_commit(&mp->mod_bufs);
	}

	(void) pthread_mutex_lock(&mp->mod_lock);
	mp->mod_flags &= ~(FMD_MOD_MDIRTY | FMD_MOD_CDIRTY);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	mp->mod_gen++;
}

void
fmd_module_lock(fmd_module_t *mp)
{
	pthread_t self = pthread_self();

	(void) pthread_mutex_lock(&mp->mod_lock);

	while (mp->mod_flags & FMD_MOD_LOCK) {
		if (mp->mod_owner != self)
			(void) pthread_cond_wait(&mp->mod_cv, &mp->mod_lock);
		else
			fmd_panic("recursive module lock of %p\n", (void *)mp);
	}

	mp->mod_owner = self;
	mp->mod_flags |= FMD_MOD_LOCK;

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);
}

void
fmd_module_unlock(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);

	ASSERT(mp->mod_owner == pthread_self());
	ASSERT(mp->mod_flags & FMD_MOD_LOCK);

	mp->mod_owner = 0;
	mp->mod_flags &= ~FMD_MOD_LOCK;

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);
}

int
fmd_module_trylock(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);

	if (mp->mod_flags & FMD_MOD_LOCK) {
		(void) pthread_mutex_unlock(&mp->mod_lock);
		return (0);
	}

	mp->mod_owner = pthread_self();
	mp->mod_flags |= FMD_MOD_LOCK;

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	return (1);
}

int
fmd_module_locked(fmd_module_t *mp)
{
	return ((mp->mod_flags & FMD_MOD_LOCK) &&
	    mp->mod_owner == pthread_self());
}

int
fmd_module_enter(fmd_module_t *mp, void (*func)(fmd_hdl_t *))
{
	volatile int err;

	(void) pthread_mutex_lock(&mp->mod_lock);

	ASSERT(!(mp->mod_flags & FMD_MOD_BUSY));
	mp->mod_flags |= FMD_MOD_BUSY;

	if ((err = setjmp(mp->mod_jmpbuf)) != 0) {
		(void) pthread_mutex_lock(&mp->mod_lock);
		fmd_module_error(mp, err);
	}

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	/*
	 * If it's the first time through fmd_module_enter(), call the provided
	 * function on the module.  If no fmd_module_abort() results, we will
	 * fall through and return zero.  Otherwise we'll longjmp with an err,
	 * return to the setjmp() above, and return the error to our caller.
	 */
	if (err == 0 && func != NULL)
		(*func)((fmd_hdl_t *)mp);

	return (err);
}

void
fmd_module_exit(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);

	ASSERT(mp->mod_flags & FMD_MOD_BUSY);
	mp->mod_flags &= ~FMD_MOD_BUSY;

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);
}

/*
 * If the client.error policy has been set by a developer, stop or dump core
 * based on the policy; if we stop and are resumed we'll continue and execute
 * the default behavior to discard events in fmd_module_start().  If the caller
 * is the primary module thread, we reach this state by longjmp'ing back to
 * fmd_module_enter(), above.  If the caller is an auxiliary thread, we cancel
 * ourself and arrange for the primary thread to call fmd_module_abort().
 */
void
fmd_module_abort(fmd_module_t *mp, int err)
{
	uint_t policy = FMD_CERROR_UNLOAD;
	pthread_t tid = pthread_self();

	(void) fmd_conf_getprop(fmd.d_conf, "client.error", &policy);

	if (policy == FMD_CERROR_STOP) {
		fmd_error(err, "stopping after %s in client %s (%p)\n",
		    fmd_errclass(err), mp->mod_name, (void *)mp);
		(void) raise(SIGSTOP);
	} else if (policy == FMD_CERROR_ABORT) {
		fmd_panic("aborting due to %s in client %s (%p)\n",
		    fmd_errclass(err), mp->mod_name, (void *)mp);
	}

	/*
	 * If the caller is an auxiliary thread, cancel the current thread.  We
	 * prefer to cancel because it affords developers the option of using
	 * the pthread_cleanup* APIs.  If cancellations have been disabled,
	 * fall through to forcing the current thread to exit.  In either case
	 * we update mod_error (if zero) to enter the failed state.  Once that
	 * is set, further events received by the module will be discarded.
	 *
	 * We also set the FMD_MOD_FAIL bit, indicating an unrecoverable error.
	 * When an auxiliary thread fails, the module is left in a delicate
	 * state where it is likely not able to continue execution (even to
	 * execute its _fmd_fini() routine) because our caller may hold locks
	 * that are private to the module and can no longer be released.  The
	 * FMD_MOD_FAIL bit forces fmd_api_module_lock() to abort if any other
	 * module threads reach an API call, in an attempt to get them to exit.
	 */
	if (tid != mp->mod_thread->thr_tid) {
		(void) pthread_mutex_lock(&mp->mod_lock);

		if (mp->mod_error == 0)
			mp->mod_error = err;

		mp->mod_flags |= FMD_MOD_FAIL;
		(void) pthread_mutex_unlock(&mp->mod_lock);

		(void) pthread_cancel(tid);
		pthread_exit(NULL);
	}

	ASSERT(mp->mod_flags & FMD_MOD_BUSY);
	longjmp(mp->mod_jmpbuf, err);
}

void
fmd_module_hold(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);

	TRACE((FMD_DBG_MOD, "hold %p (%s/%u)\n",
	    (void *)mp, mp->mod_name, mp->mod_refs));

	mp->mod_refs++;
	ASSERT(mp->mod_refs != 0);

	(void) pthread_mutex_unlock(&mp->mod_lock);
}

void
fmd_module_rele(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);

	TRACE((FMD_DBG_MOD, "rele %p (%s/%u)\n",
	    (void *)mp, mp->mod_name, mp->mod_refs));

	ASSERT(mp->mod_refs != 0);

	if (--mp->mod_refs == 0)
		fmd_module_destroy(mp);
	else
		(void) pthread_mutex_unlock(&mp->mod_lock);
}

/*
 * Wrapper around libdiagcode's fm_dc_opendict() to load module dictionaries.
 * If the dictionary open is successful, the new dictionary is added to the
 * mod_dictv[] array and mod_codelen is updated with the new maximum length.
 */
int
fmd_module_dc_opendict(fmd_module_t *mp, const char *dict)
{
	struct fm_dc_handle *dcp, **dcv;
	char *dictdir, *dictnam, *p;
	size_t len;

	ASSERT(fmd_module_locked(mp));

	dictnam = strdupa(fmd_strbasename(dict));

	if ((p = strrchr(dictnam, '.')) != NULL &&
	    strcmp(p, ".dict") == 0)
		*p = '\0'; /* eliminate any trailing .dict suffix */

	/*
	 * If 'dict' is an absolute path, dictdir = $rootdir/`dirname dict`
	 * If 'dict' is not an absolute path, dictdir = $dictdir/`dirname dict`
	 */
	if (dict[0] == '/') {
		len = strlen(fmd.d_rootdir) + strlen(dict) + 1;
		dictdir = alloca(len);
		(void) snprintf(dictdir, len, "%s%s", fmd.d_rootdir, dict);
		(void) fmd_strdirname(dictdir);
	} else {
		(void) fmd_conf_getprop(fmd.d_conf, "dictdir", &p);
		len = strlen(fmd.d_rootdir) + strlen(p) + strlen(dict) + 3;
		dictdir = alloca(len);
		(void) snprintf(dictdir, len,
		    "%s/%s/%s", fmd.d_rootdir, p, dict);
		(void) fmd_strdirname(dictdir);
	}

	fmd_dprintf(FMD_DBG_MOD, "module %s opening %s -> %s/%s.dict\n",
	    mp->mod_name, dict, dictdir, dictnam);

	if ((dcp = fm_dc_opendict(FM_DC_VERSION, dictdir, dictnam)) == NULL)
		return (-1); /* errno is set for us */

	dcv = fmd_alloc(sizeof (dcp) * (mp->mod_dictc + 1), FMD_SLEEP);
	bcopy(mp->mod_dictv, dcv, sizeof (dcp) * mp->mod_dictc);
	fmd_free(mp->mod_dictv, sizeof (dcp) * mp->mod_dictc);
	mp->mod_dictv = dcv;
	mp->mod_dictv[mp->mod_dictc++] = dcp;

	len = fm_dc_codelen(dcp);
	mp->mod_codelen = MAX(mp->mod_codelen, len);

	return (0);
}

/*
 * Wrapper around libdiagcode's fm_dc_key2code() that examines all the module's
 * dictionaries.  We adhere to the libdiagcode return values and semantics.
 */
int
fmd_module_dc_key2code(fmd_module_t *mp,
    char *const keys[], char *code, size_t codelen)
{
	int i, err;

	for (i = 0; i < mp->mod_dictc; i++) {
		if ((err = fm_dc_key2code(mp->mod_dictv[i], (const char **)keys,
		    code, codelen)) == 0 || errno != ENOMSG)
			return (err);
	}

	return (fmd_set_errno(ENOMSG));
}

fmd_modhash_t *
fmd_modhash_create(void)
{
	fmd_modhash_t *mhp = fmd_alloc(sizeof (fmd_modhash_t), FMD_SLEEP);

	(void) pthread_rwlock_init(&mhp->mh_lock, NULL);
	mhp->mh_hashlen = fmd.d_str_buckets;
	mhp->mh_hash = fmd_zalloc(sizeof (void *) * mhp->mh_hashlen, FMD_SLEEP);
	mhp->mh_nelems = 0;

	return (mhp);
}

void
fmd_modhash_destroy(fmd_modhash_t *mhp)
{
	fmd_module_t *mp, *nmp;
	uint_t i;

	for (i = 0; i < mhp->mh_hashlen; i++) {
		for (mp = mhp->mh_hash[i]; mp != NULL; mp = nmp) {
			nmp = mp->mod_next;
			mp->mod_next = NULL;
			fmd_module_rele(mp);
		}
	}

	fmd_free(mhp->mh_hash, sizeof (void *) * mhp->mh_hashlen);
	(void) pthread_rwlock_destroy(&mhp->mh_lock);
	fmd_free(mhp, sizeof (fmd_modhash_t));
}

static void
fmd_modhash_loaddir(fmd_modhash_t *mhp, const char *dir,
    const fmd_modops_t *ops, const char *suffix)
{
	char path[PATH_MAX];
	struct dirent *dp;
	const char *p;
	DIR *dirp;

	if ((dirp = opendir(dir)) == NULL)
		return; /* failed to open directory; just skip it */

	while ((dp = readdir(dirp)) != NULL) {
		if (dp->d_name[0] == '.')
			continue; /* skip "." and ".." */

		p = strrchr(dp->d_name, '.');

		if (p != NULL && strcmp(p, ".conf") == 0)
			continue; /* skip .conf files */

		if (suffix != NULL && (p == NULL || strcmp(p, suffix) != 0))
			continue; /* skip files with the wrong suffix */

		(void) snprintf(path, sizeof (path), "%s/%s", dir, dp->d_name);
		(void) fmd_modhash_load(mhp, path, ops);
	}

	(void) closedir(dirp);
}

void
fmd_modhash_loadall(fmd_modhash_t *mhp, const fmd_conf_path_t *pap,
    const fmd_modops_t *ops, const char *suffix)
{
	int i;

	for (i = 0; i < pap->cpa_argc; i++)
		fmd_modhash_loaddir(mhp, pap->cpa_argv[i], ops, suffix);
}

void
fmd_modhash_apply(fmd_modhash_t *mhp, void (*func)(fmd_module_t *))
{
	fmd_module_t *mp, *np;
	uint_t i;

	(void) pthread_rwlock_rdlock(&mhp->mh_lock);

	for (i = 0; i < mhp->mh_hashlen; i++) {
		for (mp = mhp->mh_hash[i]; mp != NULL; mp = np) {
			np = mp->mod_next;
			func(mp);
		}
	}

	(void) pthread_rwlock_unlock(&mhp->mh_lock);
}

void
fmd_modhash_tryapply(fmd_modhash_t *mhp, void (*func)(fmd_module_t *))
{
	fmd_module_t *mp, *np;
	uint_t i;

	if (mhp == NULL || pthread_rwlock_tryrdlock(&mhp->mh_lock) != 0)
		return; /* not initialized or couldn't grab lock */

	for (i = 0; i < mhp->mh_hashlen; i++) {
		for (mp = mhp->mh_hash[i]; mp != NULL; mp = np) {
			np = mp->mod_next;
			func(mp);
		}
	}

	(void) pthread_rwlock_unlock(&mhp->mh_lock);
}

void
fmd_modhash_dispatch(fmd_modhash_t *mhp, fmd_event_t *ep)
{
	fmd_module_t *mp;
	uint_t i;

	fmd_event_hold(ep);
	(void) pthread_rwlock_rdlock(&mhp->mh_lock);

	for (i = 0; i < mhp->mh_hashlen; i++) {
		for (mp = mhp->mh_hash[i]; mp != NULL; mp = mp->mod_next) {
			/*
			 * If FMD_MOD_INIT is set but MOD_FINI, MOD_QUIT, and
			 * mod_error are all zero, then the module is active:
			 * enqueue the event in the corresponding event queue.
			 */
			(void) pthread_mutex_lock(&mp->mod_lock);

			if ((mp->mod_flags & (FMD_MOD_INIT | FMD_MOD_FINI |
			    FMD_MOD_QUIT)) == FMD_MOD_INIT && !mp->mod_error) {

				/*
				 * If the event we're dispatching is of type
				 * FMD_EVT_TOPO and there are already redundant
				 * FMD_EVT_TOPO events in this module's queue,
				 * then drop those before adding the new one.
				 */
				if (FMD_EVENT_TYPE(ep) == FMD_EVT_TOPO)
					fmd_eventq_drop_topo(mp->mod_queue);

				fmd_eventq_insert_at_time(mp->mod_queue, ep);

			}
			(void) pthread_mutex_unlock(&mp->mod_lock);
		}
	}

	(void) pthread_rwlock_unlock(&mhp->mh_lock);
	fmd_event_rele(ep);
}

fmd_module_t *
fmd_modhash_lookup(fmd_modhash_t *mhp, const char *name)
{
	fmd_module_t *mp;
	uint_t h;

	(void) pthread_rwlock_rdlock(&mhp->mh_lock);
	h = fmd_strhash(name) % mhp->mh_hashlen;

	for (mp = mhp->mh_hash[h]; mp != NULL; mp = mp->mod_next) {
		if (strcmp(name, mp->mod_name) == 0)
			break;
	}

	if (mp != NULL)
		fmd_module_hold(mp);
	else
		(void) fmd_set_errno(EFMD_MOD_NOMOD);

	(void) pthread_rwlock_unlock(&mhp->mh_lock);
	return (mp);
}

fmd_module_t *
fmd_modhash_load(fmd_modhash_t *mhp, const char *path, const fmd_modops_t *ops)
{
	char name[PATH_MAX], *p;
	fmd_module_t *mp;
	int tries = 0;
	uint_t h;

	(void) strlcpy(name, fmd_strbasename(path), sizeof (name));
	if ((p = strrchr(name, '.')) != NULL && strcmp(p, ".so") == 0)
		*p = '\0'; /* strip trailing .so from any module name */

	(void) pthread_rwlock_wrlock(&mhp->mh_lock);
	h = fmd_strhash(name) % mhp->mh_hashlen;

	/*
	 * First check to see if a module is already present in the hash table
	 * for this name.  If so, the module is already loaded: skip it.
	 */
	for (mp = mhp->mh_hash[h]; mp != NULL; mp = mp->mod_next) {
		if (strcmp(name, mp->mod_name) == 0)
			break;
	}

	if (mp != NULL) {
		(void) pthread_rwlock_unlock(&mhp->mh_lock);
		(void) fmd_set_errno(EFMD_MOD_LOADED);
		return (NULL);
	}

	/*
	 * fmd_module_create() will return a held (as if by fmd_module_hold())
	 * module.  We leave this hold in place to correspond to the hash-in.
	 */
	while ((mp = fmd_module_create(path, ops)) == NULL) {
		if (tries++ != 0 || errno != EFMD_CKPT_INVAL) {
			(void) pthread_rwlock_unlock(&mhp->mh_lock);
			return (NULL); /* errno is set for us */
		}
	}

	mp->mod_hash = mhp;
	mp->mod_next = mhp->mh_hash[h];

	mhp->mh_hash[h] = mp;
	mhp->mh_nelems++;

	(void) pthread_rwlock_unlock(&mhp->mh_lock);
	return (mp);
}

int
fmd_modhash_unload(fmd_modhash_t *mhp, const char *name)
{
	fmd_module_t *mp, **pp;
	uint_t h;

	(void) pthread_rwlock_wrlock(&mhp->mh_lock);
	h = fmd_strhash(name) % mhp->mh_hashlen;
	pp = &mhp->mh_hash[h];

	for (mp = *pp; mp != NULL; mp = mp->mod_next) {
		if (strcmp(name, mp->mod_name) == 0)
			break;
		else
			pp = &mp->mod_next;
	}

	if (mp == NULL) {
		(void) pthread_rwlock_unlock(&mhp->mh_lock);
		return (fmd_set_errno(EFMD_MOD_NOMOD));
	}

	*pp = mp->mod_next;
	mp->mod_next = NULL;

	ASSERT(mhp->mh_nelems != 0);
	mhp->mh_nelems--;

	(void) pthread_rwlock_unlock(&mhp->mh_lock);

	fmd_module_unload(mp);
	fmd_module_rele(mp);

	return (0);
}

void
fmd_modstat_publish(fmd_module_t *mp)
{
	(void) pthread_mutex_lock(&mp->mod_lock);

	ASSERT(mp->mod_flags & FMD_MOD_STSUB);
	mp->mod_flags |= FMD_MOD_STPUB;
	(void) pthread_cond_broadcast(&mp->mod_cv);

	while (mp->mod_flags & FMD_MOD_STPUB)
		(void) pthread_cond_wait(&mp->mod_cv, &mp->mod_lock);

	(void) pthread_mutex_unlock(&mp->mod_lock);
}

int
fmd_modstat_snapshot(fmd_module_t *mp, fmd_ustat_snap_t *uss)
{
	fmd_event_t *e;
	int err;

	/*
	 * Grab the module lock and wait for the STSUB bit to be clear.  Then
	 * set it to indicate we are a subscriber and everyone else must wait.
	 */
	(void) pthread_mutex_lock(&mp->mod_lock);

	while (mp->mod_error == 0 && (mp->mod_flags & FMD_MOD_STSUB))
		(void) pthread_cond_wait(&mp->mod_cv, &mp->mod_lock);

	if (mp->mod_error != 0) {
		(void) pthread_mutex_unlock(&mp->mod_lock);
		return (fmd_set_errno(EFMD_HDL_ABORT));
	}

	mp->mod_flags |= FMD_MOD_STSUB;
	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	/*
	 * Create a stats pseudo-event and dispatch it to the module, forcing
	 * it to next execute its custom snapshot routine (or the empty one).
	 */
	e = fmd_event_create(FMD_EVT_STATS, FMD_HRT_NOW, NULL, NULL);
	fmd_eventq_insert_at_head(mp->mod_queue, e);

	/*
	 * Grab the module lock and then wait on mod_cv for STPUB to be set,
	 * indicating the snapshot routine is completed and the module is idle.
	 */
	(void) pthread_mutex_lock(&mp->mod_lock);

	while (mp->mod_error == 0 && !(mp->mod_flags & FMD_MOD_STPUB)) {
		struct timespec tms;

		(void) pthread_cond_wait(&mp->mod_cv, &mp->mod_lock);
		(void) pthread_mutex_unlock(&mp->mod_lock);
		tms.tv_sec = 0;
		tms.tv_nsec = 10000000;
		(void) nanosleep(&tms, NULL);
		(void) pthread_mutex_lock(&mp->mod_lock);
	}

	if (mp->mod_error != 0) {
		(void) pthread_mutex_unlock(&mp->mod_lock);
		return (fmd_set_errno(EFMD_HDL_ABORT));
	}

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	/*
	 * Update ms_snaptime and take the actual snapshot of the various
	 * statistics while the module is quiescent and waiting for us.
	 */
	(void) pthread_mutex_lock(&mp->mod_stats_lock);

	if (mp->mod_stats != NULL) {
		mp->mod_stats->ms_snaptime.fmds_value.ui64 = gethrtime();
		err = fmd_ustat_snapshot(mp->mod_ustat, uss);
	} else
		err = fmd_set_errno(EFMD_HDL_ABORT);

	(void) pthread_mutex_unlock(&mp->mod_stats_lock);

	/*
	 * With the snapshot complete, grab the module lock and clear both
	 * STSUB and STPUB, permitting everyone to wake up and continue.
	 */
	(void) pthread_mutex_lock(&mp->mod_lock);

	ASSERT(mp->mod_flags & FMD_MOD_STSUB);
	ASSERT(mp->mod_flags & FMD_MOD_STPUB);
	mp->mod_flags &= ~(FMD_MOD_STSUB | FMD_MOD_STPUB);

	(void) pthread_cond_broadcast(&mp->mod_cv);
	(void) pthread_mutex_unlock(&mp->mod_lock);

	return (err);
}

struct topo_hdl *
fmd_module_topo_hold(fmd_module_t *mp)
{
	fmd_modtopo_t *mtp;

	ASSERT(fmd_module_locked(mp));

	mtp = fmd_zalloc(sizeof (fmd_modtopo_t), FMD_SLEEP);
	mtp->mt_topo = mp->mod_topo_current;
	fmd_topo_addref(mtp->mt_topo);
	fmd_list_prepend(&mp->mod_topolist, mtp);

	return (mtp->mt_topo->ft_hdl);
}

int
fmd_module_topo_rele(fmd_module_t *mp, struct topo_hdl *hdl)
{
	fmd_modtopo_t *mtp;

	ASSERT(fmd_module_locked(mp));

	for (mtp = fmd_list_next(&mp->mod_topolist); mtp != NULL;
	    mtp = fmd_list_next(mtp)) {
		if (mtp->mt_topo->ft_hdl == hdl)
			break;
	}

	if (mtp == NULL)
		return (-1);

	fmd_list_delete(&mp->mod_topolist, mtp);
	fmd_topo_rele(mtp->mt_topo);
	fmd_free(mtp, sizeof (fmd_modtopo_t));
	return (0);
}