7712 mandoc -Tlint does always exit with error code 0

[unleashed.git] / usr / src / uts / common / io / lvm / mirror / mirror_resync.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) 1992, 2010, Oracle and/or its affiliates. All rights reserved.
 */

#include <sys/param.h>
#include <sys/systm.h>
#include <sys/conf.h>
#include <sys/file.h>
#include <sys/user.h>
#include <sys/uio.h>
#include <sys/t_lock.h>
#include <sys/buf.h>
#include <sys/dkio.h>
#include <sys/vtoc.h>
#include <sys/kmem.h>
#include <vm/page.h>
#include <sys/cmn_err.h>
#include <sys/sysmacros.h>
#include <sys/types.h>
#include <sys/mkdev.h>
#include <sys/stat.h>
#include <sys/open.h>
#include <sys/disp.h>
#include <sys/lvm/md_mirror.h>
#include <sys/modctl.h>
#include <sys/ddi.h>
#include <sys/sunddi.h>
#include <sys/debug.h>
#include <sys/callb.h>

#include <sys/sysevent/eventdefs.h>
#include <sys/sysevent/svm.h>
#include <sys/lvm/mdmn_commd.h>

extern int              md_status;
extern kmutex_t         md_status_mx;
extern kmutex_t         md_mx;

extern unit_t           md_nunits;
extern set_t            md_nsets;
extern md_set_t         md_set[];
extern major_t          md_major;

extern md_ops_t         mirror_md_ops;
extern kmem_cache_t     *mirror_child_cache; /* mirror child memory pool */
extern mdq_anchor_t     md_mto_daemon;
extern daemon_request_t mirror_timeout;
extern md_resync_t      md_cpr_resync;
extern clock_t          md_hz;
extern int              md_mtioctl_cnt;

extern kmem_cache_t     *mirror_parent_cache;
#ifdef DEBUG
extern int              mirror_debug_flag;
#endif

/*
 * Tunable resync thread timeout. This is used as the time interval for updating
 * the resync progress to the mddb. This allows restartable resyncs to be
 * continued across a system reboot.
 * Default is to update the resync progress every 5 minutes.
 */
int md_mirror_resync_update_intvl = MD_DEF_MIRROR_RESYNC_INTVL;

/*
 * Settable mirror resync buffer size.  Specified in 512 byte
 * blocks.  This is set to MD_DEF_RESYNC_BUF_SIZE by default.
 */
int md_resync_bufsz = MD_DEF_RESYNC_BUF_SIZE;

/*
 * Tunables for dirty region processing when
 * closing down a mirror.
 *
 * Dirty region processing during close of a
 * mirror is basically monitoring the state
 * of the resync region bitmaps and the number
 * of outstanding i/o's per submirror to
 * determine that there are no more dirty
 * regions left over.
 *
 * The approach taken is a retry logic over
 * md_mirror_rr_cleans iterations to monitor
 * the progress.
 *
 * There are two methods of polling the progress
 * on dirty bitmap processing: busy-waits and
 * non-busy-waits.
 *
 * Busy-waits are used at the beginning to
 * determine the final state as quick as
 * possible; md_mirror_rr_polls defines the
 * number of busy-waits.
 *
 * In case the number of busy-waits got exhausted
 * with dirty regions left over, the retry logic
 * switches over to non-busy-waits, thus giving
 * relief to an obviously heavily loaded system.
 * The timeout value is defined by the tunable
 * md_mirror_rr_sleep_timo in seconds.
 *
 * The number of non-busy-waits is given by:
 * md_mirror_rr_cleans - md_mirror_rr_polls.
 *
 * The values were found by testing on a
 * 'typical' system and may require tuning
 * to meet specific customer's requirements.
 */

int md_mirror_rr_cleans = 13;
int md_mirror_rr_polls = 3;
int md_mirror_rr_sleep_timo = 1;

/*
 * The value is not #defined because it will be computed
 * in the future.
 */
int md_max_xfer_bufsz = 2048;

/*
 * mirror_generate_rr_bitmap:
 * -------------------
 * Generate a compressed bitmap md_mn_msg_rr_clean_t for the given clean
 * bitmap associated with mirror 'un'
 *
 * Input:
 *      un      - mirror unit to get bitmap data from
 *      *msgp   - location to return newly allocated md_mn_msg_rr_clean_t
 *      *activep- location to return # of active i/os
 *
 * Returns:
 *      1 => dirty bits cleared from un_dirty_bm and DRL flush required
 *          *msgp contains bitmap of to-be-cleared bits
 *      0 => no bits cleared
 *          *msgp == NULL
 */
static int
mirror_generate_rr_bitmap(mm_unit_t *un, md_mn_msg_rr_clean_t **msgp,
    int *activep)
{
        unsigned int    i, next_bit, data_bytes, start_bit;
        int             cleared_dirty = 0;

        /* Skip any initial 0s. */
retry_dirty_scan:
        if ((start_bit = un->un_rr_clean_start_bit) >= un->un_rrd_num)
                un->un_rr_clean_start_bit = start_bit = 0;

        /*
         * Handle case where NO bits are set in PERNODE_DIRTY but the
         * un_dirty_bm[] map does have entries set (after a 1st resync)
         */
        for (; start_bit < un->un_rrd_num &&
            !IS_PERNODE_DIRTY(md_mn_mynode_id, start_bit, un) &&
            (un->un_pernode_dirty_sum[start_bit] != (uchar_t)0); start_bit++)
                ;

        if (start_bit >= un->un_rrd_num) {
                if (un->un_rr_clean_start_bit == 0) {
                        return (0);
                } else {
                        un->un_rr_clean_start_bit = 0;
                        goto retry_dirty_scan;
                }
        }

        /* how much to fit into this message */
        data_bytes = MIN(howmany(un->un_rrd_num - start_bit, NBBY),
            MDMN_MSG_RR_CLEAN_DATA_MAX_BYTES);

        (*msgp) = kmem_zalloc(MDMN_MSG_RR_CLEAN_SIZE_DATA(data_bytes),
            KM_SLEEP);

        (*msgp)->rr_nodeid = md_mn_mynode_id;
        (*msgp)->rr_mnum = MD_SID(un);
        MDMN_MSG_RR_CLEAN_START_SIZE_SET(*msgp, start_bit, data_bytes);

        next_bit = MIN(start_bit + data_bytes * NBBY, un->un_rrd_num);

        for (i = start_bit; i < next_bit; i++) {
                if (un->c.un_status & MD_UN_KEEP_DIRTY && IS_KEEPDIRTY(i, un)) {
                        continue;
                }
                if (!IS_REGION_DIRTY(i, un)) {
                        continue;
                }
                if (un->un_outstanding_writes[i] != 0) {
                        (*activep)++;
                        continue;
                }

                /*
                 * Handle the case where a resync has completed and we still
                 * have the un_dirty_bm[] entries marked as dirty (these are
                 * the most recent DRL re-read from the replica). They need
                 * to be cleared from our un_dirty_bm[] but they will not have
                 * corresponding un_pernode_dirty[] entries set unless (and
                 * until) further write()s have been issued to the area.
                 * This handles the case where only the un_dirty_bm[] entry is
                 * set. Without this we'd not clear this region until a local
                 * write is issued to the affected area.
                 */
                if (IS_PERNODE_DIRTY(md_mn_mynode_id, i, un) ||
                    (un->un_pernode_dirty_sum[i] == (uchar_t)0)) {
                        if (!IS_GOING_CLEAN(i, un)) {
                                SET_GOING_CLEAN(i, un);
                                (*activep)++;
                                continue;
                        }
                        /*
                         * Now we've got a flagged pernode_dirty, _or_ a clean
                         * bitmap entry to process. Update the bitmap to flush
                         * the REGION_DIRTY / GOING_CLEAN bits when we send the
                         * cross-cluster message.
                         */
                        cleared_dirty++;
                        setbit(MDMN_MSG_RR_CLEAN_DATA(*msgp), i - start_bit);
                } else {
                        /*
                         * Not marked as active in the pernode bitmap, so skip
                         * any update to this. We just increment the 0 count
                         * and adjust the active count by any outstanding
                         * un_pernode_dirty_sum[] entries. This means we don't
                         * leave the mirror permanently dirty.
                         */
                        (*activep) += (int)un->un_pernode_dirty_sum[i];
                }
        }
        if (!cleared_dirty) {
                kmem_free(*msgp, MDMN_MSG_RR_CLEAN_SIZE_DATA(data_bytes));
                *msgp = NULL;
        }
        un->un_rr_clean_start_bit = next_bit;
        return (cleared_dirty);
}

/*
 * There are three paths into here:
 *
 * md_daemon -> check_resync_regions -> prr
 * mirror_internal_close -> mirror_process_unit_resync -> prr
 * mirror_set_capability -> mirror_process_unit_resync -> prr
 *
 * The first one is a kernel daemon, the other two result from system calls.
 * Thus, only the first case needs to deal with kernel CPR activity.  This
 * is indicated by the cprinfop being non-NULL for kernel daemon calls, and
 * NULL for system call paths.
 */
static int
process_resync_regions_non_owner(mm_unit_t *un, callb_cpr_t *cprinfop)
{
        int                     i, start, end;
        int                     cleared_dirty = 0;
        /* Number of reasons why we can not proceed shutting down the mirror. */
        int                     active = 0;
        set_t                   setno = MD_UN2SET(un);
        md_mn_msg_rr_clean_t    *rmsg;
        md_mn_kresult_t         *kres;
        int                     rval;
        minor_t                 mnum = MD_SID(un);
        mdi_unit_t              *ui = MDI_UNIT(mnum);
        md_mn_nodeid_t          owner_node;

        /*
         * We drop the readerlock here to assist lock ordering with
         * update_resync.  Once we have the un_rrp_inflight_mx, we
         * can re-acquire it.
         */
        md_unit_readerexit(ui);

        /*
         * Resync region processing must be single threaded. We can't use
         * un_resync_mx for this purpose since this mutex gets released
         * when blocking on un_resync_cv.
         */
        mutex_enter(&un->un_rrp_inflight_mx);

        (void) md_unit_readerlock(ui);

        mutex_enter(&un->un_resync_mx);

        rw_enter(&un->un_pernode_dirty_mx[md_mn_mynode_id - 1], RW_READER);
        cleared_dirty = mirror_generate_rr_bitmap(un, &rmsg, &active);
        rw_exit(&un->un_pernode_dirty_mx[md_mn_mynode_id - 1]);

        if (cleared_dirty) {
                owner_node = un->un_mirror_owner;
                mutex_exit(&un->un_resync_mx);

                /*
                 * Transmit the 'to-be-cleared' bitmap to all cluster nodes.
                 * Receipt of the message will cause the mirror owner to
                 * update the on-disk DRL.
                 */

                kres = kmem_alloc(sizeof (md_mn_kresult_t), KM_SLEEP);

                /* release readerlock before sending message */
                md_unit_readerexit(ui);

                if (cprinfop) {
                        mutex_enter(&un->un_prr_cpr_mx);
                        CALLB_CPR_SAFE_BEGIN(cprinfop);
                }

                rval = mdmn_ksend_message(setno, MD_MN_MSG_RR_CLEAN,
                    MD_MSGF_NO_LOG|MD_MSGF_BLK_SIGNAL|MD_MSGF_KSEND_NORETRY|
                    MD_MSGF_DIRECTED, un->un_mirror_owner,
                    (char *)rmsg, MDMN_MSG_RR_CLEAN_MSG_SIZE(rmsg), kres);

                if (cprinfop) {
                        CALLB_CPR_SAFE_END(cprinfop, &un->un_prr_cpr_mx);
                        mutex_exit(&un->un_prr_cpr_mx);
                }

                /* reacquire readerlock after message */
                (void) md_unit_readerlock(ui);

                if ((!MDMN_KSEND_MSG_OK(rval, kres)) &&
                    (kres->kmmr_comm_state != MDMNE_NOT_JOINED)) {
                        /* if commd is gone, no point in printing a message */
                        if (md_mn_is_commd_present())
                                mdmn_ksend_show_error(rval, kres, "RR_CLEAN");
                        kmem_free(kres, sizeof (md_mn_kresult_t));
                        kmem_free(rmsg, MDMN_MSG_RR_CLEAN_MSG_SIZE(rmsg));
                        mutex_exit(&un->un_rrp_inflight_mx);
                        return (active);
                }
                kmem_free(kres, sizeof (md_mn_kresult_t));

                /*
                 * If ownership changed while we were sending, we probably
                 * sent the message to the wrong node.  Leave fixing that for
                 * the next cycle.
                 */
                if (un->un_mirror_owner != owner_node) {
                        mutex_exit(&un->un_rrp_inflight_mx);
                        return (active);
                }

                /*
                 * Now that we've sent the message, clear them from the
                 * pernode_dirty arrays.  These are ONLY cleared on a
                 * successful send, and failure has no impact.
                 */
                cleared_dirty = 0;
                start = MDMN_MSG_RR_CLEAN_START_BIT(rmsg);
                end = start + MDMN_MSG_RR_CLEAN_DATA_BYTES(rmsg) * NBBY;
                mutex_enter(&un->un_resync_mx);
                rw_enter(&un->un_pernode_dirty_mx[md_mn_mynode_id - 1],
                    RW_READER);
                for (i = start; i < end; i++) {
                        if (isset(MDMN_MSG_RR_CLEAN_DATA(rmsg),
                            i - start)) {
                                if (IS_PERNODE_DIRTY(md_mn_mynode_id, i, un)) {
                                        un->un_pernode_dirty_sum[i]--;
                                        CLR_PERNODE_DIRTY(md_mn_mynode_id, i,
                                            un);
                                }
                                if (IS_REGION_DIRTY(i, un)) {
                                        cleared_dirty++;
                                        CLR_REGION_DIRTY(i, un);
                                        CLR_GOING_CLEAN(i, un);
                                }
                        }
                }
                rw_exit(&un->un_pernode_dirty_mx[md_mn_mynode_id - 1]);

                kmem_free(rmsg, MDMN_MSG_RR_CLEAN_MSG_SIZE(rmsg));
        }
        mutex_exit(&un->un_resync_mx);

        mutex_exit(&un->un_rrp_inflight_mx);

        return (active);
}

static int
process_resync_regions_owner(mm_unit_t *un)
{
        int                     i, start, end;
        int                     cleared_dirty = 0;
        /* Number of reasons why we can not proceed shutting down the mirror. */
        int                     active = 0;
        set_t                   setno = MD_UN2SET(un);
        int                     mnset = MD_MNSET_SETNO(setno);
        md_mn_msg_rr_clean_t    *rmsg;
        minor_t                 mnum = MD_SID(un);
        mdi_unit_t              *ui = MDI_UNIT(mnum);

        /*
         * We drop the readerlock here to assist lock ordering with
         * update_resync.  Once we have the un_rrp_inflight_mx, we
         * can re-acquire it.
         */
        md_unit_readerexit(ui);

        /*
         * Resync region processing must be single threaded. We can't use
         * un_resync_mx for this purpose since this mutex gets released
         * when blocking on un_resync_cv.
         */
        mutex_enter(&un->un_rrp_inflight_mx);

        (void) md_unit_readerlock(ui);

        mutex_enter(&un->un_resync_mx);
        un->un_waiting_to_clear++;
        while (un->un_resync_flg & MM_RF_STALL_CLEAN)
                cv_wait(&un->un_resync_cv, &un->un_resync_mx);
        un->un_waiting_to_clear--;

        if (mnset) {
                rw_enter(&un->un_pernode_dirty_mx[md_mn_mynode_id - 1],
                    RW_READER);
                cleared_dirty = mirror_generate_rr_bitmap(un, &rmsg, &active);

                if (cleared_dirty) {
                        /*
                         * Clear the bits from the pernode_dirty arrays.
                         * If that results in any being cleared from the
                         * un_dirty_bm, commit it.
                         */
                        cleared_dirty = 0;
                        start = MDMN_MSG_RR_CLEAN_START_BIT(rmsg);
                        end = start + MDMN_MSG_RR_CLEAN_DATA_BYTES(rmsg) * NBBY;
                        for (i = start; i < end; i++) {
                                if (isset(MDMN_MSG_RR_CLEAN_DATA(rmsg),
                                    i - start)) {
                                        if (IS_PERNODE_DIRTY(md_mn_mynode_id, i,
                                            un)) {
                                                un->un_pernode_dirty_sum[i]--;
                                                CLR_PERNODE_DIRTY(
                                                    md_mn_mynode_id, i, un);
                                        }
                                        if (un->un_pernode_dirty_sum[i] == 0) {
                                                cleared_dirty++;
                                                CLR_REGION_DIRTY(i, un);
                                                CLR_GOING_CLEAN(i, un);
                                        }
                                }
                        }
                        kmem_free(rmsg, MDMN_MSG_RR_CLEAN_MSG_SIZE(rmsg));
                }
                rw_exit(&un->un_pernode_dirty_mx[md_mn_mynode_id - 1]);
        } else {
                for (i = 0; i < un->un_rrd_num; i++) {
                        if (un->c.un_status & MD_UN_KEEP_DIRTY)
                                if (IS_KEEPDIRTY(i, un))
                                        continue;

                        if (!IS_REGION_DIRTY(i, un))
                                continue;
                        if (un->un_outstanding_writes[i] != 0) {
                                active++;
                                continue;
                        }

                        if (!IS_GOING_CLEAN(i, un)) {
                                SET_GOING_CLEAN(i, un);
                                active++;
                                continue;
                        }
                        CLR_REGION_DIRTY(i, un);
                        CLR_GOING_CLEAN(i, un);
                        cleared_dirty++;
                }
        }

        if (cleared_dirty) {
                un->un_resync_flg |= MM_RF_GATECLOSED;
                mutex_exit(&un->un_resync_mx);
                mddb_commitrec_wrapper(un->un_rr_dirty_recid);
                mutex_enter(&un->un_resync_mx);
                un->un_resync_flg &= ~MM_RF_GATECLOSED;

                if (un->un_waiting_to_mark != 0 ||
                    un->un_waiting_to_clear != 0) {
                        active++;
                        cv_broadcast(&un->un_resync_cv);
                }
        }
        mutex_exit(&un->un_resync_mx);

        mutex_exit(&un->un_rrp_inflight_mx);

        return (active);
}

static int
process_resync_regions(mm_unit_t *un, callb_cpr_t *cprinfop)
{
        int     mnset = MD_MNSET_SETNO(MD_UN2SET(un));
        /*
         * For a mirror we can only update the on-disk resync-record if we
         * currently own the mirror. If we are called and there is no owner we
         * bail out before scanning the outstanding_writes[] array.
         * NOTE: we only need to check here (before scanning the array) as we
         *      are called with the readerlock held. This means that a change
         *      of ownership away from us will block until this resync check
         *      has completed.
         */
        if (mnset && (MD_MN_NO_MIRROR_OWNER(un) ||
            (!MD_MN_MIRROR_OWNER(un) && !md_mn_is_commd_present_lite()))) {
                return (0);
        } else if (mnset && !MD_MN_MIRROR_OWNER(un)) {
                return (process_resync_regions_non_owner(un, cprinfop));
        } else {
                return (process_resync_regions_owner(un));
        }
}

/*
 * Function that is callable from other modules to provide
 * ability to cleanup dirty region bitmap on demand. Used
 * on last close of a unit to avoid massive device resyncs
 * when coming back after rolling large amounts of data to
 * a mirror (e.g. at umount with logging).
 */

void
mirror_process_unit_resync(mm_unit_t *un)
{
        int     cleans = 0;

        while (process_resync_regions(un, NULL)) {

                cleans++;
                if (cleans >= md_mirror_rr_cleans) {
                        cmn_err(CE_NOTE,
                            "Could not clean resync regions\n");
                        break;
                }
                if (cleans > md_mirror_rr_polls) {
                        /*
                         * We did not make it with md_mirror_rr_polls
                         * iterations. Give the system relief and
                         * switch over to non-busy-wait.
                         */
                        delay(md_mirror_rr_sleep_timo * md_hz);
                }
        }
}

static void
check_resync_regions(daemon_request_t *timeout)
{
        mdi_unit_t      *ui;
        mm_unit_t       *un;
        md_link_t       *next;
        callb_cpr_t     cprinfo;

        rw_enter(&mirror_md_ops.md_link_rw.lock, RW_READER);
        for (next = mirror_md_ops.md_head; next != NULL; next = next->ln_next) {

                if (md_get_setstatus(next->ln_setno) & MD_SET_STALE)
                        continue;

                un = MD_UNIT(next->ln_id);

                /*
                 * Register this resync thread with the CPR mechanism. This
                 * allows us to detect when the system is suspended and so
                 * keep track of the RPC failure condition.
                 */
                CALLB_CPR_INIT(&cprinfo, &un->un_prr_cpr_mx, callb_md_mrs_cpr,
                    "check_resync_regions");

                ui = MDI_UNIT(next->ln_id);
                (void) md_unit_readerlock(ui);

                /*
                 * Do not clean up resync regions if it is an ABR
                 * mirror, or if a submirror is offline (we will use the resync
                 * region to resync when back online) or if there is only one
                 * submirror.
                 */
                if ((ui->ui_tstate & MD_ABR_CAP) ||
                    (un->c.un_status & MD_UN_OFFLINE_SM) || (un->un_nsm < 2)) {
                        md_unit_readerexit(ui);
                        /* Remove this thread from the CPR callback table. */
                        mutex_enter(&un->un_prr_cpr_mx);
                        CALLB_CPR_EXIT(&cprinfo);
                        continue;
                }

                (void) process_resync_regions(un, &cprinfo);

                md_unit_readerexit(ui);

                /* Remove this thread from the CPR callback table. */
                mutex_enter(&un->un_prr_cpr_mx);
                CALLB_CPR_EXIT(&cprinfo);
        }

        rw_exit(&mirror_md_ops.md_link_rw.lock);

        /* We are done */
        mutex_enter(&mirror_timeout.dr_mx);
        timeout->dr_pending = 0;
        mutex_exit(&mirror_timeout.dr_mx);
}

static void
md_mirror_timeout(void *throwaway)
{

        mutex_enter(&mirror_timeout.dr_mx);
        if (!mirror_timeout.dr_pending) {
                mirror_timeout.dr_pending = 1;
                daemon_request(&md_mto_daemon, check_resync_regions,
                    (daemon_queue_t *)&mirror_timeout, REQ_OLD);
        }

        if (mirror_md_ops.md_head != NULL)
                mirror_timeout.dr_timeout_id = timeout(md_mirror_timeout,
                    throwaway, (int)MD_MDELAY*hz);
        else
                mirror_timeout.dr_timeout_id = 0;

        mutex_exit(&mirror_timeout.dr_mx);
}

void
resync_start_timeout(set_t setno)
{
        if (md_get_setstatus(setno) & MD_SET_STALE)
                return;

        mutex_enter(&mirror_timeout.dr_mx);
        if (mirror_timeout.dr_timeout_id == 0)
                mirror_timeout.dr_timeout_id = timeout(md_mirror_timeout,
                    (void *)NULL, (int)MD_MDELAY*hz);
        mutex_exit(&mirror_timeout.dr_mx);
}

static void
offlined_to_attached(mm_unit_t *un)
{
        int             i;
        int             changed = 0;

        if (md_get_setstatus(MD_UN2SET(un)) & MD_SET_STALE)
                return;

        for (i = 0; i < NMIRROR; i++) {
                if (SMS_BY_INDEX_IS(un, i, SMS_OFFLINE)) {
                        mirror_set_sm_state(&un->un_sm[i],
                            &un->un_smic[i], SMS_ATTACHED, 1);
                        changed++;
                }
                if (SMS_BY_INDEX_IS(un, i, SMS_OFFLINE_RESYNC)) {
                        mirror_set_sm_state(&un->un_sm[i],
                            &un->un_smic[i], SMS_ATTACHED_RESYNC, 1);
                        changed++;
                }
        }

        if (changed != 0) {
                un->c.un_status &= ~MD_UN_OFFLINE_SM;
                mddb_setrecprivate(un->c.un_record_id, MD_PRV_PENDCOM);
        }
}

static void
get_unit_resync(mm_unit_t *un)
{
        mddb_recstatus_t        status;
        struct optim_resync     *orp;

        if (un->un_rr_dirty_recid == 0) {
                offlined_to_attached(un);
                return;
        }

        status = mddb_getrecstatus(un->un_rr_dirty_recid);
        if ((status == MDDB_NORECORD) || (status == MDDB_NODATA)) {
                un->un_rr_dirty_recid = 0;
                offlined_to_attached(un);
                return;
        }

        mddb_setrecprivate(un->un_rr_dirty_recid, MD_PRV_GOTIT);
        orp = (struct optim_resync *)mddb_getrecaddr(un->un_rr_dirty_recid);
        un->un_dirty_bm = orp->or_rr;
}

static int
create_unit_resync(mm_unit_t *un, int snarfing)
{
        diskaddr_t      tb;
        int             i;
        int             blksize;        /* rr size in blocks */
        int             num_rr;
        mddb_recid_t    recid;
        size_t          size;   /* bitmap size */
        optim_resync_t  *orp;
        mddb_type_t     typ1;
        set_t           setno;

        tb = un->c.un_total_blocks;

        if (((tb + MD_MIN_RR_SIZE)/ MD_MIN_RR_SIZE) > MD_DEF_NUM_RR) {
                blksize = (int)(tb / MD_DEF_NUM_RR);
                num_rr = (int)((tb + (blksize)) / (blksize));
        } else {
                blksize = MD_MIN_RR_SIZE;
                num_rr = (int)((tb + MD_MIN_RR_SIZE) / MD_MIN_RR_SIZE);
        }

        size = howmany(num_rr, NBBY) + sizeof (*orp) - sizeof (orp->or_rr);

        setno = MD_UN2SET(un);

        typ1 = (mddb_type_t)md_getshared_key(setno,
            mirror_md_ops.md_driver.md_drivername);

        recid =  mddb_createrec(size, typ1, RESYNC_REC,
            MD_CRO_OPTIMIZE|MD_CRO_32BIT, setno);
        if (recid < 0) {
                if (snarfing && !(md_get_setstatus(setno) & MD_SET_STALE)) {
                        md_set_setstatus(setno, MD_SET_STALE);
                        cmn_err(CE_WARN, "md: state database is stale");
                }
                return (-1);
        }

        un->un_rr_dirty_recid = recid;
        orp = (optim_resync_t *)mddb_getrecaddr(recid);
        orp->or_magic = OR_MAGIC;
        orp->or_blksize = blksize;
        orp->or_num = num_rr;

        un->un_rrd_blksize = blksize;
        un->un_rrd_num  = num_rr;
        un->un_dirty_bm = orp->or_rr;

        if (snarfing)
                for (i = 0; i < howmany(num_rr, NBBY); i++)
                        orp->or_rr[i] = 0xFF;

        if (!snarfing) {
                mddb_commitrec_wrapper(recid);
                mirror_commit(un, NO_SUBMIRRORS, 0);
                return (0);
        }
        mddb_setrecprivate(recid, MD_PRV_PENDCOM);
        mddb_setrecprivate(un->c.un_record_id, MD_PRV_PENDCOM);
        return (0);
}

int
unit_setup_resync(mm_unit_t *un, int snarfing)
{
        int err;
        int syncable;
        int i;
        mdi_unit_t      *ui = MDI_UNIT(MD_SID(un));
        int nonABR = 1;         /* only set if ABR marked in ui_tstate */

        un->un_dirty_bm = NULL;
        un->un_rs_buffer = NULL;

        mutex_init(&un->un_rrp_inflight_mx, "rrp mx", MUTEX_DEFAULT, NULL);

        mutex_init(&un->un_resync_mx, NULL, MUTEX_DEFAULT, NULL);
        cv_init(&un->un_resync_cv, NULL, CV_DEFAULT, NULL);
        un->un_resync_flg = 0;
        un->un_waiting_to_mark = 0;
        un->un_waiting_to_commit = 0;
        un->un_waiting_to_clear = 0;

        un->un_goingclean_bm = NULL;
        un->un_goingdirty_bm = NULL;
        un->un_outstanding_writes = NULL;
        un->un_resync_bm = NULL;

        if (snarfing)
                get_unit_resync(un);

        if (un->un_rr_dirty_recid == 0) {
                /*
                 * If a MN diskset and snarfing and this node is not the
                 * master, do not delete any records on snarf of the
                 * mirror records (create_unit_resync deletes records).
                 *
                 * Master node should have already handled this case.
                 */
                if (MD_MNSET_SETNO(MD_UN2SET(un)) && snarfing &&
                    md_set[MD_UN2SET(un)].s_am_i_master == 0) {
#ifdef DEBUG
                        cmn_err(CE_NOTE, "unit_setup_resync: no rr for %s on"
                            " nodeid %d\n", md_shortname(MD_SID(un)),
                            md_set[MD_UN2SET(un)].s_nodeid);
#endif
                        return (-1);
                }
                if ((err = create_unit_resync(un, snarfing)) != 0)
                        return (err);
        }

        un->un_goingclean_bm = (uchar_t *)kmem_zalloc((uint_t)(howmany(
            un->un_rrd_num, NBBY)), KM_SLEEP);
        un->un_goingdirty_bm = (uchar_t *)kmem_zalloc((uint_t)(howmany(
            un->un_rrd_num, NBBY)), KM_SLEEP);
        un->un_outstanding_writes = (short *)kmem_zalloc(
            (uint_t)un->un_rrd_num * sizeof (short), KM_SLEEP);
        un->un_resync_bm = (uchar_t *)kmem_zalloc((uint_t)(howmany(
            un->un_rrd_num, NBBY)), KM_SLEEP);

        /*
         * Allocate pernode bitmap for this node. All other nodes' maps will
         * be created 'on-the-fly' in the ioctl message handler
         */
        if (MD_MNSET_SETNO(MD_UN2SET(un))) {
                un->un_pernode_dirty_sum =
                    (uchar_t *)kmem_zalloc(un->un_rrd_num, KM_SLEEP);
                if (md_mn_mynode_id > 0) {
                        un->un_pernode_dirty_bm[md_mn_mynode_id-1] = (uchar_t *)
                            kmem_zalloc((uint_t)(howmany(un->un_rrd_num, NBBY)),
                            KM_SLEEP);
                }

                /*
                 * Allocate taskq to process deferred (due to locking) RR_CLEAN
                 * requests.
                 */
                un->un_drl_task = (ddi_taskq_t *)md_create_taskq(MD_UN2SET(un),
                    MD_SID(un));
        }

        if (md_get_setstatus(MD_UN2SET(un)) & MD_SET_STALE)
                return (0);

        /*
         * Only mark mirror which has an associated DRL as requiring a resync.
         * For ABR mirrors we need not set the resync record bitmap up.
         */
        if (ui && (ui->ui_tstate & MD_ABR_CAP))
                nonABR = 0;

        for (i = 0, syncable = 0; i < NMIRROR; i++) {
                if (nonABR) {
                        if ((SUBMIRROR_IS_READABLE(un, i) ||
                            SMS_BY_INDEX_IS(un, i,
                            (SMS_OFFLINE | SMS_OFFLINE_RESYNC))))
                                syncable++;
                }
        }

        if (snarfing && un->un_pass_num && (syncable > 1)) {
                bcopy((caddr_t)un->un_dirty_bm, (caddr_t)un->un_resync_bm,
                    howmany(un->un_rrd_num, NBBY));

                un->c.un_status |= (MD_UN_OPT_NOT_DONE | MD_UN_WAR);
                un->c.un_status &= ~MD_UN_OFFLINE_SM;
                for (i = 0; i < NMIRROR; i++) {
                        if ((SUBMIRROR_IS_READABLE(un, i)) ||
                            SMS_BY_INDEX_IS(un, i, SMS_OFFLINE_RESYNC))
                                un->un_sm[i].sm_flags |= MD_SM_RESYNC_TARGET;

                        if (SMS_BY_INDEX_IS(un, i, SMS_OFFLINE)) {
                                un->un_sm[i].sm_flags |= MD_SM_RESYNC_TARGET;
                                mirror_set_sm_state(&un->un_sm[i],
                                    &un->un_smic[i], SMS_OFFLINE_RESYNC, 1);
                                mddb_setrecprivate(un->c.un_record_id,
                                    MD_PRV_PENDCOM);
                        }
                }
        }
        return (0);
}

/*
 * resync_kill_pending:
 * -------------------
 * Determine if the resync thread has been requested to terminate.
 * Block if MD_RI_BLOCK or MD_RI_BLOCK_OWNER is set in un->un_rs_thread_flags.
 * MD_RI_BLOCK is only set as a result of a user-initiated ioctl via metasync.
 * MD_RI_BLOCK_OWNER is set by the ownership change of a multi-node  mirror.
 *
 * Returns:
 *      0       Kill not pending
 *      1       Kill requested  (set MD_UN_RESYNC_CANCEL in un->c.un_status)
 *
 * Note: this routine may block
 *       the writerlock for <ui> will be dropped and reacquired if <mx_type>
 *       is set to MD_WRITER_HELD.
 *       the readerlock for <ui> will be dropped and reacquired if <mx_type>
 *       is set to MD_READER_HELD.
 */
static int
resync_kill_pending(
        mm_unit_t *un,
        mdi_unit_t *ui,
        uint_t mx_type)
{
        int     retval = 0;

        /* Ensure that we don't block with any mutex held */
        if (mx_type == MD_WRITER_HELD) {
                md_unit_writerexit(ui);
        } else if (mx_type == MD_READER_HELD) {
                md_unit_readerexit(ui);
        }
        mutex_enter(&un->un_rs_thread_mx);
        while (un->un_rs_thread_flags & (MD_RI_BLOCK|MD_RI_BLOCK_OWNER)) {
                cv_wait(&un->un_rs_thread_cv, &un->un_rs_thread_mx);
                if (un->un_rs_thread_flags & (MD_RI_KILL|MD_RI_SHUTDOWN))
                        break;
        }
        /* Determine if we've been asked to abort or shutdown gracefully */
        if (un->un_rs_thread_flags & MD_RI_KILL) {
                un->c.un_status |= MD_UN_RESYNC_CANCEL;
                retval = 1;
        } else if (un->un_rs_thread_flags & MD_RI_SHUTDOWN) {
                retval = 1;
        }
        mutex_exit(&un->un_rs_thread_mx);

        /* Reacquire mutex if dropped on entry */
        if (mx_type == MD_WRITER_HELD) {
                (void) md_unit_writerlock(ui);
        } else if (mx_type == MD_READER_HELD) {
                (void) md_unit_readerlock(ui);
        }
        return (retval);
}

/*
 * resync_read_buffer:
 * ------------------
 * Issue the resync source read for the specified start block and size.
 * This will cause the mirror strategy routine to issue a write-after-read
 * once this request completes successfully.
 * If 'flag_err' is set we expect to see a write error flagged in the b_error
 * field of the buffer created for this i/o request. If clear we do not expect
 * to see the error flagged for write failures.
 * Read failures will always set the B_ERROR bit which will stop the resync
 * immediately.
 */
static int
resync_read_buffer(mm_unit_t *un, diskaddr_t blk, size_t cnt, int flag_err)
{
        md_mcs_t        *sp;
        buf_t           *bp;
        int             ret = 0;

        sp = kmem_cache_alloc(mirror_child_cache, MD_ALLOCFLAGS);
        mirror_child_init(sp);

        bp = &sp->cs_buf;
        bp->b_edev = makedevice(md_major, MD_SID(un));
        bp->b_flags = B_READ;
        bp->b_lblkno = blk;
        bp->b_bcount = dbtob(cnt);
        bp->b_un.b_addr = un->un_rs_buffer;
        md_unit_readerexit(MDI_UNIT(MD_SID(un)));

        (void) md_mirror_strategy(bp, MD_STR_NOTTOP | MD_STR_MAPPED |
            MD_STR_WAR | (flag_err ? MD_STR_FLAG_ERR : 0), NULL);

        (void) biowait(bp);

        (void) md_unit_readerlock(MDI_UNIT(MD_SID(un)));
        if (bp->b_flags & B_ERROR) {
                ret = 1;
        }
        kmem_cache_free(mirror_child_cache, sp);
        return (ret);
}

/*
 * send_mn_resync_done_message
 *
 * At the end of a resync, send a message to all nodes to indicate that
 * the resync is complete. The argument, flags, has the following values
 *
 * RESYNC_ERR - if an error occurred that terminated the resync
 * CLEAR_OPT_NOT_DONE   - Just need to clear the OPT_NOT_DONE flag
 *
 * unit writerlock set on entry
 * Only send the message if the thread is not marked as shutting down:
 * [un_rs_thread_flags & MD_RI_SHUTDOWN] or being killed:
 * [un->c.un_status & MD_UN_RESYNC_CANCEL]
 * or if there has been an error that terminated the resync:
 *      flags & RESYNC_ERR
 *
 */
static void
send_mn_resync_done_message(
        mm_unit_t       *un,
        int             flags
)
{
        md_mn_msg_resync_t      *rmsg = un->un_rs_msg;
        set_t                   setno;
        mdi_unit_t              *ui = MDI_UNIT(MD_SID(un));
        md_mn_kresult_t         *kres;
        int                     dont_send = 0;
        int                     rval;
        int                     nretries = 0;

        rmsg = (md_mn_msg_resync_t *)un->un_rs_msg;

        /*
         * Only send the message if this resync thread is still active. This
         * handles the case where ownership changes to different nodes during
         * a resync can cause multiple spurious resync_done messages to occur
         * when the resync completes. This happens because only one node is
         * the resync owner but other nodes will have their resync_unit thread
         * blocked in 'resync_kill_pending'
         */
        mutex_enter(&un->un_rs_thread_mx);
        dont_send = (un->un_rs_thread_flags & (MD_RI_KILL|MD_RI_SHUTDOWN)) ? 1
            : 0;
        mutex_exit(&un->un_rs_thread_mx);
        dont_send |= (un->c.un_status & MD_UN_RESYNC_CANCEL) ? 1 : 0;

        /*
         * Always send a message if we've encountered an error that terminated
         * the resync.
         */
        if (flags & RESYNC_ERR)
                dont_send = 0;

        if (dont_send) {
#ifdef DEBUG
                if (mirror_debug_flag) {
                        printf("Don't send resync done message, mnum = %x,"
                            " type = %x, flags = %d\n", MD_SID(un),
                            un->un_rs_type, flags);
                }
#endif  /* DEBUG */
                return;
        }

#ifdef DEBUG
        if (mirror_debug_flag) {
                printf("send resync done message, mnum = %x, type = %x\n",
                    MD_SID(un), un->un_rs_type);
        }
#endif

        rmsg->msg_resync_mnum = MD_SID(un);
        rmsg->msg_resync_type = un->un_rs_type;
        rmsg->msg_originator = md_mn_mynode_id;
        rmsg->msg_resync_flags = 0;
        if (flags & RESYNC_ERR)
                rmsg->msg_resync_flags |= MD_MN_RS_ERR;
        if (flags & CLEAR_OPT_NOT_DONE)
                rmsg->msg_resync_flags |= MD_MN_RS_CLEAR_OPT_NOT_DONE;

        setno = MD_MIN2SET(MD_SID(un));
        md_unit_writerexit(ui);
        kres = kmem_alloc(sizeof (md_mn_kresult_t), KM_SLEEP);

smrd_msg:
        mutex_enter(&un->un_rs_cpr_mx);
        CALLB_CPR_SAFE_BEGIN(&un->un_rs_cprinfo);

        rval = mdmn_ksend_message(setno, MD_MN_MSG_RESYNC_PHASE_DONE,
            MD_MSGF_NO_LOG, 0, (char *)rmsg, sizeof (md_mn_msg_resync_t), kres);

        CALLB_CPR_SAFE_END(&un->un_rs_cprinfo, &un->un_rs_cpr_mx);
        mutex_exit(&un->un_rs_cpr_mx);

        /* if the node hasn't yet joined, it's Ok. */
        if ((!MDMN_KSEND_MSG_OK(rval, kres)) &&
            (kres->kmmr_comm_state !=  MDMNE_NOT_JOINED)) {
                mdmn_ksend_show_error(rval, kres, "RESYNC_PHASE_DONE");
                /* If we're shutting down already, pause things here. */
                if (kres->kmmr_comm_state == MDMNE_RPC_FAIL) {
                        while (!md_mn_is_commd_present()) {
                                delay(md_hz);
                        }
                        /*
                         * commd is now available again. Retry the message once.
                         * If this fails we panic as the system is in an
                         * unexpected state.
                         */
                        if (nretries++ == 0)
                                goto smrd_msg;
                }
                cmn_err(CE_PANIC, "ksend_message failure: RESYNC_PHASE_DONE");
        }
        kmem_free(kres, sizeof (md_mn_kresult_t));
        (void) md_unit_writerlock(ui);
}

/*
 * send_mn_resync_next_message
 *
 * Sent a message to all nodes indicating the next region to be resynced.
 * The message contains the region to be resynced and the current position in
 * the resync as denoted by un_rs_resync_done and un_rs_resync_2_do.
 * On entry the unit readerlock is held.
 */
static void
send_mn_resync_next_message(
        mm_unit_t       *un,
        diskaddr_t      currentblk,
        size_t          rsize,
        int             flags
)
{
        md_mn_msg_resync_t      *rmsg = un->un_rs_msg;
        set_t                   setno;
        md_mn_kresult_t         *kres;
        mdi_unit_t              *ui = MDI_UNIT(MD_SID(un));
        int                     rval;
        md_mps_t                *ps;
        mm_submirror_t          *sm;
        int                     smi;
        int                     nretries = 0;

        ASSERT(rmsg != NULL);
#ifdef DEBUG
        if (mirror_debug_flag) {
                printf("send resync next message, mnum = %x, start=%lld, "
                    "size=%ld, type=%x, done=%lld, 2_do=%lld\n",
                    MD_SID(un), currentblk, rsize, un->un_rs_type,
                    un->un_rs_resync_done, un->un_rs_resync_2_do);
        }
#endif
        rmsg->msg_resync_mnum = MD_SID(un);
        rmsg->msg_resync_type = un->un_rs_type;
        rmsg->msg_resync_start = currentblk;
        rmsg->msg_resync_rsize = rsize;
        rmsg->msg_resync_done = un->un_rs_resync_done;
        rmsg->msg_resync_2_do = un->un_rs_resync_2_do;
        rmsg->msg_originator = md_mn_mynode_id;
        if (flags & MD_FIRST_RESYNC_NEXT)
                rmsg->msg_resync_flags = MD_MN_RS_FIRST_RESYNC_NEXT;

        /*
         * Copy current submirror state and flags into message. This provides
         * a means of keeping all nodes that are currently active in the cluster
         * synchronised with regards to their submirror state settings. If we
         * did not pass this information here, the only time every node gets
         * submirror state updated is at the end of a resync phase. This can be
         * a significant amount of time for large metadevices.
         */
        for (smi = 0; smi < NMIRROR; smi++) {
                sm = &un->un_sm[smi];
                rmsg->msg_sm_state[smi] = sm->sm_state;
                rmsg->msg_sm_flags[smi] = sm->sm_flags;
        }
        setno = MD_MIN2SET(MD_SID(un));
        md_unit_readerexit(ui);
        kres = kmem_alloc(sizeof (md_mn_kresult_t), KM_SLEEP);

smrn_msg:
        mutex_enter(&un->un_rs_cpr_mx);
        CALLB_CPR_SAFE_BEGIN(&un->un_rs_cprinfo);

        rval = mdmn_ksend_message(setno, MD_MN_MSG_RESYNC_NEXT, MD_MSGF_NO_LOG,
            0, (char *)rmsg, sizeof (md_mn_msg_resync_t), kres);

        CALLB_CPR_SAFE_END(&un->un_rs_cprinfo, &un->un_rs_cpr_mx);
        mutex_exit(&un->un_rs_cpr_mx);

        if (!MDMN_KSEND_MSG_OK(rval, kres)) {
                mdmn_ksend_show_error(rval, kres, "RESYNC_NEXT");
                /* If we're shutting down already, pause things here. */
                if (kres->kmmr_comm_state == MDMNE_RPC_FAIL) {
                        while (!md_mn_is_commd_present()) {
                                delay(md_hz);
                        }
                        /*
                         * commd is now available again. Retry the message once.
                         * If this fails we panic as the system is in an
                         * unexpected state.
                         */
                        if (nretries++ == 0)
                                goto smrn_msg;
                }
                cmn_err(CE_PANIC, "ksend_message failure: RESYNC_NEXT");
        }
        kmem_free(kres, sizeof (md_mn_kresult_t));
        (void) md_unit_readerlock(ui);
        ps = un->un_rs_prev_overlap;

        /* Allocate previous overlap reference if needed */
        if (ps == NULL) {
                ps = kmem_cache_alloc(mirror_parent_cache, MD_ALLOCFLAGS);
                ps->ps_un = un;
                ps->ps_ui = ui;
                ps->ps_firstblk = 0;
                ps->ps_lastblk = 0;
                ps->ps_flags = 0;
                md_unit_readerexit(ui);
                (void) md_unit_writerlock(ui);
                un->un_rs_prev_overlap = ps;
                md_unit_writerexit(ui);
                (void) md_unit_readerlock(ui);
        }

        ps->ps_firstblk = currentblk;
        ps->ps_lastblk = currentblk + rsize - 1;
}

static int
resync_read_blk_range(
        mm_unit_t *un,
        diskaddr_t currentblk,
        diskaddr_t stopbefore,
        uint_t type,
        int     flags
)
{
        size_t copysize;        /* limited by max xfer buf size */
        size_t rsize;           /* size of resync block (for MN) */
        set_t           setno;
        diskaddr_t      newstop;
        diskaddr_t      rs_startblk;
        uint_t          rs_type;
        int             flags1 = flags & MD_FIRST_RESYNC_NEXT;

        rs_type = un->un_rs_type;
        rs_startblk = currentblk;
        if (stopbefore > un->c.un_total_blocks)
                stopbefore = un->c.un_total_blocks;
        if (currentblk < un->un_resync_startbl)
                currentblk = un->un_resync_startbl;

        copysize = un->un_rs_copysize;
        rsize = MD_DEF_RESYNC_BLK_SZ;

        setno = MD_MIN2SET(MD_SID(un));
        while (currentblk < stopbefore) {
                /*
                 * Split the block up into units of MD_DEF_RESYNC_BLK_SZ and
                 * if a MN device and sendflag is set, send a RESYNC_MESSAGE
                 * to all nodes.
                 */
                if ((currentblk + MD_DEF_RESYNC_BLK_SZ) > stopbefore)
                        rsize = stopbefore - currentblk;
                if (MD_MNSET_SETNO(setno) && (flags & MD_SEND_MESS_XMIT)) {
                        un->un_resync_startbl = currentblk;
                        rs_startblk = currentblk;
                        send_mn_resync_next_message(un, currentblk, rsize,
                            flags1);
                        if (flags1)
                                flags1 = 0;
                        /* check to see if we've been asked to terminate */
                        if (resync_kill_pending(un, MDI_UNIT(MD_SID(un)), type))
                                return ((un->c.un_status & MD_UN_RESYNC_CANCEL)
                                    ? 1:0);
                        /*
                         * Check to see if another node has completed this
                         * block, if so either the type or the resync region
                         * will have changed. If the resync type has changed,
                         * just exit.
                         * If the resync region has changed, reset currentblk
                         * to the start of the current resync region and
                         * continue.
                         */
                        if (un->un_rs_type != rs_type)
                                return (0);
                        if (un->un_rs_prev_overlap->ps_firstblk >
                            rs_startblk) {
                                currentblk =
                                    un->un_rs_prev_overlap->ps_firstblk;
                                continue;
                        }
                }
                newstop = currentblk + rsize;
                while (currentblk < newstop) {
                        if ((currentblk + copysize) > stopbefore)
                                copysize = (size_t)(stopbefore - currentblk);
                        if (resync_read_buffer(un, currentblk, copysize,
                            (flags & MD_RESYNC_FLAG_ERR)))
                                return (1);

                        /* resync_read_buffer releases/grabs a new lock */
                        un = (mm_unit_t *)MD_UNIT(MD_SID(un));
                        currentblk += copysize;

                        /* check to see if we've been asked to terminate */
                        if (resync_kill_pending(un, MDI_UNIT(MD_SID(un)), type))
                                return ((un->c.un_status & MD_UN_RESYNC_CANCEL)
                                    ? 1:0);
                        if (MD_MNSET_SETNO(setno)) {
                                /*
                                 * Check to see if another node has completed
                                 * this block, see above
                                 */
                                if (un->un_rs_type != rs_type)
                                        return (0);
                                if (un->un_rs_prev_overlap->ps_firstblk >
                                    rs_startblk)
                                        currentblk =
                                            un->un_rs_prev_overlap->ps_firstblk;
                        }
                }
        }
        return (0);
}

static void
optimized_resync(mm_unit_t *un)
{
        mdi_unit_t      *ui;
        minor_t         mnum;
        int             rr, smi;
        int             resync_regions;
        uchar_t         *dirtyregions;
        diskaddr_t      first, stopbefore;
        int             err;
        int             cnt;
        sm_state_t      state;
        int             broke_out = 0;
        set_t           setno;
        uint_t          old_rs_type = un->un_rs_type;
        uint_t          old_rs_done;
        uint_t          flags1 = MD_FIRST_RESYNC_NEXT|MD_RESYNC_FLAG_ERR;
        size_t          start_rr;

        mnum = MD_SID(un);
        ui = MDI_UNIT(mnum);
        setno = MD_UN2SET(un);

        if (!(un->c.un_status & MD_UN_OPT_NOT_DONE)) {
                /*
                 * We aren't marked as needing a resync so for multi-node
                 * sets we flag the completion so that all nodes see the same
                 * metadevice state. This is a problem when a new node joins
                 * an existing set as it has to perform a 'metasync -r' and
                 * we have to step through all of the resync phases. If we
                 * don't do this the nodes that were already in the set will
                 * have the metadevices marked as 'Okay' but the joining node
                 * will have 'Needs Maintenance' which is unclearable.
                 */
                if (MD_MNSET_SETNO(setno)) {
                        send_mn_resync_done_message(un, CLEAR_OPT_NOT_DONE);
                }
                return;
        }

        /*
         * No need for optimized resync if ABR set, clear rs_type and flags
         * and exit
         */
        if (ui->ui_tstate & MD_ABR_CAP) {
                un->un_rs_type = MD_RS_NONE;
                un->c.un_status &= ~(MD_UN_OPT_NOT_DONE | MD_UN_WAR);
                return;
        }

        un->un_rs_dropped_lock = 1;
        un->c.un_status |= MD_UN_WAR;
        resync_regions = un->un_rrd_num;
        dirtyregions = un->un_resync_bm;
        md_unit_writerexit(ui);

        /* For MN sets, resync NOTIFY is done when processing resync messages */
        if (!MD_MNSET_SETNO(setno)) {
                SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_START,
                    SVM_TAG_METADEVICE, setno, MD_SID(un));
        }
        un = (mm_unit_t *)md_unit_readerlock(ui);

        /* check to see if we've been asked to terminate */
        if (resync_kill_pending(un, MDI_UNIT(MD_SID(un)), MD_READER_HELD)) {
                if (un->c.un_status & MD_UN_RESYNC_CANCEL)
                        broke_out = RESYNC_ERR;
        }
        /*
         * Check that we are still performing an optimized
         * resync. If not, another node must have completed it
         * so we have no more work to do.
         */
        if (un->un_rs_type != old_rs_type) {
                md_unit_readerexit(ui);
                (void) md_unit_writerlock(ui);
                return;
        }
        /*
         * If rs_resync_done is non-zero, we must be completing an optimized
         * resync that has already been partially done on another node.
         * Therefore clear the bits in resync_bm for the resync regions
         * already done. If resync_startbl is zero, calculate 2_do.
         */
        if (un->un_rs_resync_done > 0) {
                BLK_TO_RR(start_rr, un->un_resync_startbl, un);
                for (rr = 0; rr < start_rr && rr < resync_regions; rr++)
                        CLR_KEEPDIRTY(rr, un);
        } else {
                un->un_rs_resync_2_do = 0;
                for (rr = 0; rr < resync_regions; rr++)
                        if (isset(dirtyregions, rr))
                                un->un_rs_resync_2_do++;
        }

        for (rr = 0; (rr < resync_regions) && (broke_out != RESYNC_ERR); rr++) {
                if (isset(dirtyregions, rr)) {
                        RR_TO_BLK(first, rr, un);
                        RR_TO_BLK(stopbefore, rr+1, un);
                        old_rs_type = un->un_rs_type;
                        old_rs_done = un->un_rs_resync_done;
                        err = resync_read_blk_range(un, first, stopbefore,
                            MD_READER_HELD, MD_SEND_MESS_XMIT | flags1);
                        flags1 = MD_RESYNC_FLAG_ERR;

                        /* resync_read_blk_range releases/grabs a new lock */
                        un = (mm_unit_t *)MD_UNIT(mnum);

                        if (err) {
                                broke_out = RESYNC_ERR;
                                break;
                        }

                        /*
                         * Check that we are still performing an optimized
                         * resync. If not, another node must have completed it
                         * so we have no more work to do.
                         */
                        if (un->un_rs_type != old_rs_type) {
                                md_unit_readerexit(ui);
                                (void) md_unit_writerlock(ui);
                                return;
                        }

                        /*
                         * If resync_done has increased, we must have
                         * blocked in resync_read_blk_range while another node
                         * continued with the resync. Therefore clear resync_bm
                         * for the blocks that have been resynced on another
                         * node and update rr to the next RR to be done.
                         */
                        if (old_rs_done < un->un_rs_resync_done) {
                                int i;
                                BLK_TO_RR(start_rr, un->un_resync_startbl - 1,
                                    un);
                                for (i = rr; i < start_rr; i++)
                                        CLR_KEEPDIRTY(i, un);
                                rr = start_rr;
                        } else
                                un->un_rs_resync_done++;

                        for (smi = 0, cnt = 0; smi < NMIRROR; smi++)
                                if (SUBMIRROR_IS_WRITEABLE(un, smi) &&
                                    !(SMS_BY_INDEX_IS(un, smi, SMS_ALL_ERRED)))
                                        cnt++;
                        if (cnt < 2) {
                                broke_out = RESYNC_ERR;
                                break;
                        }
                        CLR_KEEPDIRTY(rr, un);
                        /* Check to see if we've completed the resync cleanly */
                        if (un->un_rs_thread_flags & MD_RI_SHUTDOWN)
                                break;

                        /*
                         * Check that we haven't exceeded un_rs_resync_2_do. If
                         * we have we've completed the resync.
                         */
                        if (un->un_rs_resync_done > un->un_rs_resync_2_do)
                                break;
                }
        }
        md_unit_readerexit(ui);
        un = (mm_unit_t *)md_unit_writerlock(ui);

        /*
         * If MN set send message to all nodes to indicate resync
         * phase is complete. The processing of the message will update the
         * mirror state
         */
        if (MD_MNSET_SETNO(setno)) {
                send_mn_resync_done_message(un, broke_out);
        } else {

                if (!broke_out)
                        un->c.un_status &= ~MD_UN_WAR;

                un->c.un_status &= ~MD_UN_KEEP_DIRTY;

                setno = MD_UN2SET(un);
                for (smi = 0; smi < NMIRROR; smi++) {
                        un->un_sm[smi].sm_flags &= ~MD_SM_RESYNC_TARGET;
                        if (SMS_BY_INDEX_IS(un, smi, SMS_OFFLINE_RESYNC)) {
                                state = (broke_out ? SMS_OFFLINE : SMS_RUNNING);
                                mirror_set_sm_state(&un->un_sm[smi],
                                    &un->un_smic[smi], state, broke_out);
                                mirror_commit(un, NO_SUBMIRRORS, 0);
                        }
                        if (SMS_BY_INDEX_IS(un, smi, SMS_OFFLINE))
                                un->c.un_status |= MD_UN_OFFLINE_SM;
                }
        }

        /* For MN sets, resync NOTIFY is done when processing resync messages */
        if (!MD_MNSET_SETNO(setno)) {
                if (broke_out) {
                        SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_FAILED,
                            SVM_TAG_METADEVICE, MD_UN2SET(un), MD_SID(un));
                } else {
                        SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_DONE,
                            SVM_TAG_METADEVICE, MD_UN2SET(un), MD_SID(un));
                }
        }
}

/*
 * recalc_resync_done
 *
 * This function deals with a change in value of un_rs_resync_2_do in a
 * component resync. This may change if we are restarting a component
 * resync on a single node having rebooted with a different value of
 * md_resync_bufsz or if we are running in a multi-node with nodes having
 * different values of md_resync_bufsz.
 * If there is a change in un_rs_resync_2_do, we need to recalculate
 * the value of un_rs_resync_done given the new value for resync_2_do.
 * We have to calculate a new value for resync_done to be either
 * if un_resync_startbl is set, (un_resync_startbl - initblock)/(blksize + skip)
 * or if it is not set, we need to calculate it from un_rs_resync_done,
 * (un_rs_resync_done/un_rs_resync_2_do) * resync_2_do
 * In addition we need to deal with the overflow case by using a factor to
 * prevent overflow
 */

static void
recalc_resync_done(mm_unit_t *un, size_t resync_2_do, diskaddr_t initblock,
    u_longlong_t blk_size, u_longlong_t skip)
{
        diskaddr_t              x;
        uint_t                  factor = 1;

        /*
         * If resync_2_do has not yet been calculated, no need to modify
         * resync_done
         */
        if (un->un_rs_resync_2_do == 0) {
                return;
        }
        if (un->un_rs_resync_2_do == resync_2_do)
                return; /* No change, so nothing to do */
        /*
         * If un_rs_startbl is set, another node must have already started
         * this resync and hence we can calculate resync_done from
         * resync_startbl
         */
        if (un->un_resync_startbl) {
                un->un_rs_resync_done = (un->un_resync_startbl - initblock) /
                    (blk_size + skip);
                return;
        }
        /*
         * un_resync_startbl is not set so we must calculate it from
         * un_rs_resync_done.
         * If the larger of the two values of resync_2_do is greater than 32
         * bits, calculate a factor to divide by to ensure that we don't
         * overflow 64 bits when calculating the new value for resync_done
         */
        x = (un->un_rs_resync_2_do > resync_2_do) ? un->un_rs_resync_2_do :
            resync_2_do;
        while (x > INT32_MAX) {
                x = x >> 1;
                factor = factor << 1;
        }
        un->un_rs_resync_done = ((un->un_rs_resync_done/factor) *
            (resync_2_do/factor)) /
            ((un->un_rs_resync_2_do + (factor * factor) - 1)/
            (factor * factor));
}

static void
check_comp_4_resync(mm_unit_t *un, int smi, int ci)
{
        mdi_unit_t              *ui;
        minor_t                 mnum;
        mm_submirror_t          *sm;
        mm_submirror_ic_t       *smic;
        size_t                  count;
        u_longlong_t            skip;
        u_longlong_t            size;
        u_longlong_t            blk_size;
        diskaddr_t              initblock;
        diskaddr_t              block;
        diskaddr_t              frag = 0;
        md_m_shared_t           *shared;
        int                     err;
        set_t                   setno;
        int                     broke_out = 0;
        int                     blks;
        uint_t                  old_rs_type = un->un_rs_type;
        diskaddr_t              old_rs_done;
        uint_t                  flags1 = MD_FIRST_RESYNC_NEXT;
        diskaddr_t              resync_2_do;

        mnum = MD_SID(un);
        ui = MDI_UNIT(mnum);
        sm = &un->un_sm[smi];
        smic = &un->un_smic[smi];
        setno = MD_UN2SET(un);

        shared = (md_m_shared_t *)(*(smic->sm_shared_by_indx))
            (sm->sm_dev, sm, ci);

        if (shared->ms_state != CS_RESYNC) {
                SET_RS_TYPE_NONE(un->un_rs_type);
                return;
        }

        if (shared->ms_flags & MDM_S_RS_TRIED) {
                SET_RS_TYPE_NONE(un->un_rs_type);
                return;
        }

        (void) (*(smic->sm_get_bcss))
            (sm->sm_dev, sm, ci, &initblock, &count, &skip, &size);

        if ((count == 1) && (skip == 0)) {
                count = (size_t)(size / un->un_rs_copysize);
                if ((frag = (size - (count * un->un_rs_copysize))) != 0)
                        count++;
                size = (u_longlong_t)un->un_rs_copysize;
        }
        blk_size = size; /* Save block size for this resync */

        ASSERT(count >= 1);
        resync_2_do = count;
        /*
         * If part way through a resync, un_rs_resync_done/un_rs_resync_2_do
         * gives the proportion of the resync that has already been done.
         * If un_rs_copysize has changed since this previous partial resync,
         * either because this node has been rebooted with a different value
         * for md_resync_bufsz or because another node with a different value
         * for md_resync_bufsz performed the previous resync, we need to
         * recalculate un_rs_resync_done as a proportion of our value of
         * resync_2_do.
         */
        recalc_resync_done(un, resync_2_do, initblock, blk_size, skip);

        /*
         * For MN mirrors we need to send a message to all nodes indicating
         * the next region to be resynced. For a component resync, the size of
         * the contiguous region that is processed by resync_read_blk_range()
         * may be small if there is the interleave size.
         * Therefore, rather than sending the message within
         * resync_read_blk_range(), we will send a message every
         * MD_DEF_RESYNC_BLK_SZ blocks. Calculate the frequency in terms of
         * the number of blocks. Then, if we are restarting a resync, round
         * un_rs_resync_done down to the previous resync region boundary. This
         * ensures that we send a RESYNC_NEXT message before resyncing any
         * blocks
         */
        if (MD_MNSET_SETNO(setno)) {
                blks = ((MD_DEF_RESYNC_BLK_SZ + blk_size + skip - 1)/
                    (blk_size + skip));
                un->un_rs_resync_done = (un->un_rs_resync_done/blks) * blks;
        }
        /*
         * un_rs_resync_done is the number of ('size' + 'skip') increments
         * already resynced from the base 'block'
         * un_rs_resync_2_do is the number of iterations in
         * this component resync.
         */
        ASSERT(count >= un->un_rs_resync_done);
        un->un_rs_resync_2_do = (diskaddr_t)count;

        un->c.un_status |= MD_UN_WAR;
        sm->sm_flags |= MD_SM_RESYNC_TARGET;
        md_unit_writerexit(ui);

        /* For MN sets, resync NOTIFY is done when processing resync messages */
        if (!MD_MNSET_SETNO(setno)) {
                SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_START,
                    SVM_TAG_METADEVICE, setno, MD_SID(un));
        }
        un = (mm_unit_t *)md_unit_readerlock(ui);

        /* check to see if we've been asked to terminate */
        if (resync_kill_pending(un, MDI_UNIT(MD_SID(un)), MD_READER_HELD)) {
                if (un->c.un_status & MD_UN_RESYNC_CANCEL)
                        broke_out = RESYNC_ERR;
        }
        /*
         * Check that we are still performing the same component
         * resync. If not, another node must have completed it
         * so we have no more work to do.
         */
        if (un->un_rs_type != old_rs_type) {
                md_unit_readerexit(ui);
                (void) md_unit_writerlock(ui);
                return;
        }
        /*
         * Adjust resync_done, resync_2_do, start of resync area and count to
         * skip already resync'd data. We need to recalculate resync_done as
         * we have dropped the unit lock above and may have lost ownership to
         * another node, with a different resync buffer size and it may have
         * sent us new values of resync_done and resync_2_do based on its
         * resync buffer size
         */
        recalc_resync_done(un, resync_2_do, initblock, blk_size, skip);
        un->un_rs_resync_2_do = resync_2_do;
        count -= un->un_rs_resync_done;
        block = initblock + ((blk_size + skip) * (int)un->un_rs_resync_done);

        un->un_rs_dropped_lock = 1;
        while ((count > 0) && (broke_out != RESYNC_ERR)) {
                old_rs_done = un->un_rs_resync_done;
                /*
                 * For MN mirrors send a message to the other nodes. This
                 * message includes the size of the region that must be blocked
                 * for all writes
                 */
                if (MD_MNSET_SETNO(setno)) {
                        if ((un->un_rs_resync_done%blks == 0)) {
                                un->un_resync_startbl = block;
                                send_mn_resync_next_message(un, block,
                                    (blk_size+skip)*blks, flags1);
                                flags1 = 0;
                                /*
                                 * check to see if we've been asked to
                                 * terminate
                                 */
                                if (resync_kill_pending(un,
                                    MDI_UNIT(MD_SID(un)), MD_READER_HELD)) {
                                        if (un->c.un_status &
                                            MD_UN_RESYNC_CANCEL) {
                                                broke_out = RESYNC_ERR;
                                                break;
                                        }
                                }

                                /*
                                 * Check that we are still performing the same
                                 * component resync. If not, another node must
                                 * have completed it so we have no more work to
                                 * do. Also reset count to remaining resync as
                                 * we may have lost ownership in in
                                 * send_mn_resync_next_message while another
                                 * node continued with the resync and
                                 * incremented resync_done.
                                 */
                                if (un->un_rs_type != old_rs_type) {
                                        md_unit_readerexit(ui);
                                        (void) md_unit_writerlock(ui);
                                        return;
                                }
                                /*
                                 * recalculate resync_done, resync_2_do
                                 * We need to recalculate resync_done as
                                 * we have dropped the unit lock in
                                 * send_mn_resync_next_message above and may
                                 * have lost ownership to another node, with a
                                 * different resync buffer size and it may have
                                 * sent us new values of resync_done and
                                 * resync_2_do based on its resync buffer size
                                 */
                                recalc_resync_done(un, resync_2_do, initblock,
                                    blk_size, skip);
                                un->un_rs_resync_2_do = resync_2_do;
                                count = un->un_rs_resync_2_do -
                                    un->un_rs_resync_done;
                                /*
                                 * Adjust start of resync area to skip already
                                 * resync'd data
                                 */
                                block = initblock + ((blk_size + skip) *
                                    (int)un->un_rs_resync_done);
                                old_rs_done = un->un_rs_resync_done;
                        }
                }
                err = resync_read_blk_range(un, block, block + size,
                    MD_READER_HELD, MD_RESYNC_FLAG_ERR);

                /* resync_read_blk_range releases/grabs a new lock */
                un = (mm_unit_t *)MD_UNIT(mnum);

                if (err) {
                        broke_out = RESYNC_ERR;
                        break;
                }
                /*
                 * If we are no longer resyncing this component, return as
                 * another node has progressed the resync.
                 */
                if (un->un_rs_type != old_rs_type) {
                        md_unit_readerexit(ui);
                        (void) md_unit_writerlock(ui);
                        return;
                }

                /*
                 * recalculate resync_done, resync_2_do. We need to recalculate
                 * resync_done as we have dropped the unit lock in
                 * resync_read_blk_range above and may have lost ownership to
                 * another node, with a different resync buffer size and it may
                 * have sent us new values of resync_done and resync_2_do based
                 * on its resync buffer size
                 */
                recalc_resync_done(un, resync_2_do, initblock, blk_size, skip);
                un->un_rs_resync_2_do = resync_2_do;

                /*
                 * Reset count to remaining resync as we may have blocked in
                 * resync_read_blk_range while another node continued
                 * with the resync and incremented resync_done. Also adjust
                 * start of resync area to skip already resync'd data.
                 */
                count = un->un_rs_resync_2_do - un->un_rs_resync_done;
                block = initblock +((blk_size + skip) *
                    (int)un->un_rs_resync_done);

                /*
                 * If we are picking up from another node, we retry the last
                 * block otherwise step on to the next block
                 */
                if (old_rs_done == un->un_rs_resync_done) {
                        block += blk_size + skip;
                        un->un_rs_resync_done++;
                        count--;
                }

                if ((count == 1) && frag)
                        size = frag;
                if (shared->ms_state == CS_ERRED) {
                        err = 1;
                        broke_out = RESYNC_ERR;
                        break;
                }

                /* Check to see if we've completed the resync cleanly */
                if (un->un_rs_thread_flags & MD_RI_SHUTDOWN)
                        break;
        }

        md_unit_readerexit(ui);
        un = (mm_unit_t *)md_unit_writerlock(ui);

        /*
         * If MN set send message to all nodes to indicate resync
         * phase is complete. The processing of the message will update the
         * mirror state
         */
        if (MD_MNSET_SETNO(setno)) {
                send_mn_resync_done_message(un, broke_out);
        } else {
                un->c.un_status &= ~MD_UN_WAR;
                sm->sm_flags &= ~MD_SM_RESYNC_TARGET;

                if (err)
                        shared->ms_flags |= MDM_S_RS_TRIED;
                else
                        /*
                         * As we don't transmit the changes,
                         * no need to drop the lock.
                         */
                        set_sm_comp_state(un, smi, ci, CS_OKAY, 0,
                            MD_STATE_NO_XMIT, (IOLOCK *)NULL);
        }

        /* For MN sets, resync NOTIFY is done when processing resync messages */
        if (!MD_MNSET_SETNO(setno)) {
                if (broke_out) {
                        SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_FAILED,
                            SVM_TAG_METADEVICE, setno, MD_SID(un));
                } else {
                        SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_DONE,
                            SVM_TAG_METADEVICE, setno, MD_SID(un));
                }
                SET_RS_TYPE_NONE(un->un_rs_type);
        }
}

static void
submirror_resync(mm_unit_t *un)
{
        mdi_unit_t              *ui;
        minor_t                 mnum;
        mm_submirror_t          *sm;
        mm_submirror_ic_t       *smic;
        int                     smi;
        diskaddr_t              chunk;
        diskaddr_t              curblk;
        int                     err;
        int                     cnt;
        set_t                   setno;
        int                     broke_out = 0;
        int                     i;
        int                     flags1 = MD_FIRST_RESYNC_NEXT;
        int                     compcnt;

        mnum = MD_SID(un);
        ui = MDI_UNIT(mnum);
        setno = MD_UN2SET(un);

        /*
         * If the submirror_index is non-zero, we are continuing a resync
         * so restart resync from last submirror marked as being resynced.
         */
        if (RS_SMI(un->un_rs_type) != 0) {
                smi = RS_SMI(un->un_rs_type);
                sm = &un->un_sm[smi];
                smic = &un->un_smic[smi];
                if (!SMS_IS(sm, SMS_ATTACHED_RESYNC)) {
                        for (smi = 0; smi < NMIRROR; smi++) {
                                sm = &un->un_sm[smi];
                                smic = &un->un_smic[smi];
                                if (SMS_IS(sm, SMS_ATTACHED_RESYNC))
                                        break;
                        }
                }
        } else {
                for (smi = 0; smi < NMIRROR; smi++) {
                        sm = &un->un_sm[smi];
                        smic = &un->un_smic[smi];
                        if (SMS_IS(sm, SMS_ATTACHED_RESYNC))
                                break;
                }
        }
        if (smi == NMIRROR) {
                SET_RS_TYPE_NONE(un->un_rs_type);
                return;
        }

        /*
         * If we've only got one component we can fail on a resync write
         * if an error is encountered. This stops an unnecessary read of the
         * whole mirror on a target write error.
         */
        compcnt = (*(smic->sm_get_component_count))(sm->sm_dev, sm);
        if (compcnt == 1)
                flags1 |= MD_RESYNC_FLAG_ERR;

        un->c.un_status |= MD_UN_WAR;
        sm->sm_flags |= MD_SM_RESYNC_TARGET;
        SET_RS_SMI(un->un_rs_type, smi);
        md_unit_writerexit(ui);

        /* For MN sets, resync NOTIFY is done when processing resync messages */
        if (!MD_MNSET_SETNO(setno)) {
                SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_START,
                    SVM_TAG_METADEVICE, setno, MD_SID(un));
        }
        un = (mm_unit_t *)md_unit_readerlock(ui);

        un->un_rs_dropped_lock = 1;

        /* check to see if we've been asked to terminate */
        if (resync_kill_pending(un, MDI_UNIT(MD_SID(un)), MD_READER_HELD)) {
                if (un->c.un_status & MD_UN_RESYNC_CANCEL)
                        broke_out = RESYNC_ERR;
        }
        /*
         * Check that we are still performing the same submirror
         * resync. If not, another node must have completed it
         * so we have no more work to do.
         */
        if (RS_TYPE(un->un_rs_type) != MD_RS_SUBMIRROR) {
                md_unit_readerexit(ui);
                (void) md_unit_writerlock(ui);
                return;
        }

        /* if > 1TB mirror, increase percent done granularity */
        if (un->c.un_total_blocks > MD_MAX_BLKS_FOR_SMALL_DEVS)
                chunk = un->c.un_total_blocks / 1000;
        else
                chunk = un->c.un_total_blocks / 100;
        if (chunk == 0)
                chunk = un->c.un_total_blocks;
        /*
         * If a MN set, round the chunk size up to a multiple of
         * MD_DEF_RESYNC_BLK_SZ
         */
        if (MD_MNSET_SETNO(setno)) {
                chunk = ((chunk + MD_DEF_RESYNC_BLK_SZ)/MD_DEF_RESYNC_BLK_SZ)
                    * MD_DEF_RESYNC_BLK_SZ;
                if (chunk > un->c.un_total_blocks)
                        chunk = un->c.un_total_blocks;
        }
        /*
         * Handle restartable resyncs that continue from where the previous
         * resync left off. The new resync range is from un_rs_resync_done ..
         * un_rs_resync_2_do
         */
        curblk = 0;
        if (un->un_rs_resync_done == 0) {
                un->un_rs_resync_2_do = un->c.un_total_blocks;
        } else {
                curblk = un->un_rs_resync_done;
        }
        while ((curblk != un->c.un_total_blocks) && (broke_out != RESYNC_ERR)) {
                diskaddr_t      rs_done;

                rs_done = un->un_rs_resync_done;
                err = resync_read_blk_range(un, curblk, curblk + chunk,
                    MD_READER_HELD, MD_SEND_MESS_XMIT | flags1);
                flags1 = (compcnt == 1 ? MD_RESYNC_FLAG_ERR : 0);

                /* resync_read_blk_range releases/grabs a new lock */
                un = (mm_unit_t *)MD_UNIT(mnum);

                if (err) {
                        broke_out = RESYNC_ERR;
                        break;
                }

                /*
                 * If we are no longer executing a submirror resync, return
                 * as another node has completed the submirror resync.
                 */
                if (RS_TYPE(un->un_rs_type) != MD_RS_SUBMIRROR) {
                        md_unit_readerexit(ui);
                        (void) md_unit_writerlock(ui);
                        return;
                }
                /*
                 * If resync_done has changed, we must have blocked
                 * in resync_read_blk_range while another node
                 * continued with the resync so restart from resync_done.
                 */
                if (rs_done != un->un_rs_resync_done) {
                        curblk = un->un_rs_resync_done;
                } else {
                        curblk += chunk;
                        un->un_rs_resync_done = curblk;
                }

                if ((curblk + chunk) > un->c.un_total_blocks)
                        chunk = un->c.un_total_blocks - curblk;
                for (i = 0, cnt = 0; i < NMIRROR; i++)
                        if (SUBMIRROR_IS_WRITEABLE(un, i) &&
                            !SMS_BY_INDEX_IS(un, i, SMS_ALL_ERRED) &&
                            (un->un_sm[i].sm_flags & MD_SM_RESYNC_TARGET))
                                cnt++;
                if (cnt == 0) {
                        broke_out = RESYNC_ERR;
                        break;
                }

                /* Check to see if we've completed the resync cleanly */
                if (un->un_rs_thread_flags & MD_RI_SHUTDOWN)
                        break;
        }
        md_unit_readerexit(ui);
        un = (mm_unit_t *)md_unit_writerlock(ui);

        /*
         * If MN set send message to all nodes to indicate resync
         * phase is complete. The processing of the message will update the
         * mirror state
         */
        if (MD_MNSET_SETNO(setno)) {
                send_mn_resync_done_message(un, broke_out);
        } else {
                sm->sm_flags &= ~MD_SM_RESYNC_TARGET;
                if (err) {
                        mirror_set_sm_state(sm, smic, SMS_ATTACHED, 1);
                } else {
                        mirror_set_sm_state(sm, smic, SMS_RUNNING, 0);
                }
                un->c.un_status &= ~MD_UN_WAR;
                mirror_commit(un, SMI2BIT(smi), 0);
        }

        /* For MN sets, resync NOTIFY is done when processing resync messages */
        if (!MD_MNSET_SETNO(setno)) {
                if (broke_out) {
                        SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_FAILED,
                            SVM_TAG_METADEVICE, setno, MD_SID(un));
                } else {
                        SE_NOTIFY(EC_SVM_STATE, ESC_SVM_RESYNC_DONE,
                            SVM_TAG_METADEVICE, setno, MD_SID(un));
                }
        }
}

static void
component_resync(mm_unit_t *un)
{
        mm_submirror_t          *sm;
        mm_submirror_ic_t       *smic;
        int                     ci;
        int                     i;
        int                     compcnt;

        /*
         * Handle the case where we are picking up a partially complete
         * component resync. In this case un_rs_type contains the submirror
         * and component index of where we should restart the resync.
         */
        while (un->un_rs_type != MD_RS_COMPONENT) {
                i = RS_SMI(un->un_rs_type);
                ci = RS_CI(un->un_rs_type);
                check_comp_4_resync(un, i, ci);
                if (resync_kill_pending(un, MDI_UNIT(MD_SID(un)),
                    MD_WRITER_HELD))
                        return;
                /*
                 * If we have no current resync, contine to scan submirror and
                 * components. If the resync has moved on to another component,
                 * restart it and if the resync is no longer a component
                 * resync, just exit
                 */
                if (RS_TYPE(un->un_rs_type) == MD_RS_NONE)
                        break;
                if (RS_TYPE(un->un_rs_type) != MD_RS_COMPONENT)
                        return;
        }
        /* Now continue scanning _all_ submirrors and components */
        for (i = 0; i < NMIRROR; i++) {
                sm = &un->un_sm[i];
                smic = &un->un_smic[i];
                if (!SMS_IS(sm, SMS_RUNNING | SMS_LIMPING))
                        continue;
                compcnt = (*(smic->sm_get_component_count))(sm->sm_dev, sm);
                for (ci = 0; ci < compcnt; ci++) {
                        SET_RS_SMI(un->un_rs_type, i);
                        SET_RS_CI(un->un_rs_type, ci);
                        SET_RS_TYPE(un->un_rs_type, MD_RS_COMPONENT);
                        check_comp_4_resync(un, i, ci);
                        /* Bail out if we've been asked to abort/shutdown */
                        if (resync_kill_pending(un, MDI_UNIT(MD_SID(un)),
                            MD_WRITER_HELD))
                                return;
                        /*
                         * Now check if another node has continued with the
                         * resync, if we are no longer in component resync,
                         * exit, otherwise update to the current component - 1
                         * so that the next call of check_comp_4 resync() will
                         * resync the current component.
                         */
                        if ((RS_TYPE(un->un_rs_type) != MD_RS_NONE) &&
                            (RS_TYPE(un->un_rs_type) != MD_RS_COMPONENT))
                                return;
                        else {
                                if (RS_SMI(un->un_rs_type) != i) {
                                        i = RS_SMI(un->un_rs_type);
                                        ci = RS_CI(un->un_rs_type) - 1;
                                } else if (RS_CI(un->un_rs_type) != ci)
                                        ci = RS_CI(un->un_rs_type) - 1;
                        }
                }
        }
}

static void
reset_comp_flags(mm_unit_t *un)
{
        mm_submirror_t          *sm;
        mm_submirror_ic_t       *smic;
        md_m_shared_t           *shared;
        int                     ci;
        int                     i;
        int                     compcnt;

        for (i = 0; i < NMIRROR; i++) {
                sm = &un->un_sm[i];
                smic = &un->un_smic[i];
                if (!SMS_IS(sm, SMS_INUSE))
                        continue;
                compcnt = (*(smic->sm_get_component_count))(sm->sm_dev, sm);
                for (ci = 0; ci < compcnt; ci++) {
                        shared = (md_m_shared_t *)(*(smic->sm_shared_by_indx))
                            (sm->sm_dev, sm, ci);
                        shared->ms_flags &= ~MDM_S_RS_TRIED;
                }
        }
}

/*
 * resync_progress_thread:
 * ----------------------
 * Thread started on first resync of a unit which simply blocks until woken up
 * by a cv_signal, and then updates the mddb for the mirror unit record. This
 * saves the resync progress information (un_rs_resync_done, un_rs_resync_2_do)
 * so that an aborted resync can be continued after an intervening reboot.
 */
static void
resync_progress_thread(minor_t mnum)
{
        mm_unit_t       *un = MD_UNIT(mnum);
        mdi_unit_t      *ui = MDI_UNIT(mnum);
        set_t           setno = MD_MIN2SET(mnum);

        while (un->c.un_status & MD_UN_RESYNC_ACTIVE) {
                mutex_enter(&un->un_rs_progress_mx);
                cv_wait(&un->un_rs_progress_cv, &un->un_rs_progress_mx);
                mutex_exit(&un->un_rs_progress_mx);
                if (un->un_rs_progress_flags & MD_RI_KILL)
                        break;

                /*
                 * Commit mirror unit if we're the Master node in a multi-node
                 * environment
                 */
                if (MD_MNSET_SETNO(setno) && md_set[setno].s_am_i_master) {
                        (void) md_unit_readerlock(ui);
                        mirror_commit(un, NO_SUBMIRRORS, 0);
                        md_unit_readerexit(ui);
                }
        }
        thread_exit();
}

/*
 * resync_progress:
 * ---------------
 * Timeout handler for updating the progress of the resync thread.
 * Simply wake up the resync progress daemon which will then mirror_commit() the
 * unit structure to the mddb. This snapshots the current progress of the resync
 */
static void
resync_progress(void *arg)
{
        mm_unit_t       *un = (mm_unit_t *)arg;
        mdi_unit_t      *ui = MDI_UNIT(MD_SID(un));
        uint_t          active;

        mutex_enter(&un->un_rs_progress_mx);
        cv_signal(&un->un_rs_progress_cv);
        mutex_exit(&un->un_rs_progress_mx);

        /* schedule the next timeout if the resync is still marked active */
        (void) md_unit_readerlock(ui);
        active = un->c.un_status & MD_UN_RESYNC_ACTIVE ? 1 : 0;
        md_unit_readerexit(ui);
        if (active) {
                un->un_rs_resync_to_id = timeout(resync_progress, un,
                    (clock_t)(drv_usectohz(60000000) *
                    md_mirror_resync_update_intvl));
        }
}

/*
 * resync_unit:
 * -----------
 * Resync thread which drives all forms of resync (optimized, component,
 * submirror). Must handle thread suspension and kill to allow multi-node
 * resync to run without undue ownership changes.
 *
 * For a MN set, the reync mechanism is as follows:
 *
 * When a resync is started, either via metattach, metaonline, metareplace,
 * metasync or by a hotspare kicking in, a message is sent to all nodes, which
 * calls mirror_resync_thread. If there is currently no mirror owner, the
 * master node sends a CHOOSE_OWNER message to the handler on the master. This
 * chooses a mirror owner and sends a CHANGE_OWNER message requesting the
 * selected node to become the owner.
 * If this node is not the owner it sets itself to block in resync_kill_pending
 * and if there is no owner all nodes will block until the chosen owner is
 * selected, in which case it will unblock itself. So, on entry to this
 * function only one node will continue past resync_kill_pending().
 * Once the resync thread is started, it basically cycles through the optimized,
 * component and submirrors resyncs until there is no more work to do.
 *
 * For an ABR mirror, once a mirror owner is chosen it will complete the resync
 * unless the nodes dies in which case a new owner will be chosen and it will
 * have to complete the resync from the point at which the previous owner died.
 * To do this we broadcast a RESYNC_NEXT message before each region to be
 * resynced and this message contains the address and length of the region
 * being resynced and the current progress through the resync. The size of
 * this region is MD_DEF_RESYNC_BLK_SZ blocks. It is larger than the resync
 * block size to limit the amount of inter node traffic. The RESYNC_NEXT
 * message also indicates to all other nodes that all writes to this block
 * must be blocked until the next RESYNC_NEXT message is received. This ensures
 * that no node can write to a block that is being resynced. For all MN
 * mirrors we also block the whole resync region on the resync owner node so
 * that all writes to the resync region are blocked on all nodes. There is a
 * difference here between a MN set and a regular set in that for a MN set
 * we protect the mirror from writes to the current resync block by blocking
 * a larger region. For a regular set we just block writes to the current
 * resync block.
 *
 * For a non-ABR mirror the same RESYNC_NEXT message is sent with an
 * additional purpose. In this case, there is only one mirror owner at a time
 * and rather than continually switching ownership between the chosen mirror
 * owner and the node that is writing to the mirror, we move the resync to the
 * mirror owner. When we swich ownership, we block the old owner and unblock
 * the resync thread on the new owner. To enable the new owner to continue the
 * resync, all nodes need to have the latest resync status, Then, following each
 * resync write, we check to see if the resync state has changed and if it
 * has this must be because we have lost ownership to another node(s) for a
 * period and then have become owner again later in the resync process. If we
 * are still dealing with the same resync, we just adjust addresses and counts
 * and then continue. If the resync has moved on to a different type, for
 * example from an optimized to a submirror resync, we move on to process the
 * resync described by rs_type and continue from the position described by
 * resync_done and resync_startbl.
 *
 * Note that for non-ABR mirrors it is possible for a write to be made on a
 * non resync-owner node without a change of ownership. This is the case when
 * the mirror has a soft part created on it and a write in ABR mode is made
 * to that soft part. Therefore we still need to block writes to the resync
 * region on all nodes.
 *
 * Sending the latest resync state to all nodes also enables them to continue
 * a resync in the event that the mirror owner dies. If a mirror owner for
 * a non-ABR mirror has died, there will be dirty resync regions. Therefore,
 * regardless of whether another type of resync was in progress, we must first
 * do an optimized resync to clean up the dirty regions before continuing
 * with the interrupted resync.
 *
 * The resync status is held in the unit structure
 * On disk
 * un_rs_resync_done    The number of contiguous resyc blocks done so far
 * un_rs_resync_2_do    The total number of contiguous resync blocks
 * un_rs_type           The resync type (inc submirror and component numbers)
 * In core
 * un_resync_startbl    The address of the current resync block being processed
 *
 * In the event that the whole cluster fails we need to just use
 * un_rs_resync_done to restart the resync and to ensure that this is
 * periodically written to disk, we have a thread which writes the record
 * to disk every 5 minutes. As the granularity of un_rs_resync_done is
 * usually coarse ( for an optimized resync 1001 is the max value) there is
 * little point in writing this more frequently.
 */
static void
resync_unit(minor_t mnum)
{
        mdi_unit_t      *ui;
        mm_unit_t       *un;
        md_error_t      mde = mdnullerror;
        int             mn_resync = 0;
        int             resync_finish = 0;
        set_t           setno = MD_MIN2SET(mnum);
        uint_t          old_rs_type = MD_RS_NONE;
        uint_t          old_rs_done = 0, old_rs_2_do = 0;
        uint_t          old_rs_startbl = 0;
        int             block_resync = 1;
        char            cpr_name[23];   /* Unique CPR name */
        int             rs_copysize;
        char            *rs_buffer;
        int             nretries = 0;

resync_restart:
#ifdef DEBUG
        if (mirror_debug_flag)
                printf("Resync started (mnum = %x)\n", mnum);
#endif
        /*
         * increment the mirror resync count
         */
        mutex_enter(&md_cpr_resync.md_resync_mutex);
        md_cpr_resync.md_mirror_resync++;
        mutex_exit(&md_cpr_resync.md_resync_mutex);

        ui = MDI_UNIT(mnum);
        un = MD_UNIT(mnum);

        rs_copysize = un->un_rs_copysize;
        if (rs_copysize == 0) {
                /*
                 * Don't allow buffer size to fall outside the
                 * range 0 < bufsize <= md_max_xfer_bufsz.
                 */
                if (md_resync_bufsz <= 0)
                        md_resync_bufsz = MD_DEF_RESYNC_BUF_SIZE;
                rs_copysize = MIN(md_resync_bufsz, md_max_xfer_bufsz);
        }
        rs_buffer = kmem_zalloc(dbtob(rs_copysize), KM_SLEEP);
        un = md_unit_writerlock(ui);
        un->un_rs_copysize = rs_copysize;
        un->un_rs_buffer = rs_buffer;

        if (MD_MNSET_SETNO(setno)) {
                /*
                 * Register this resync thread with the CPR mechanism. This
                 * allows us to detect when the system is suspended and so
                 * keep track of the RPC failure condition.
                 */
                (void) snprintf(cpr_name, sizeof (cpr_name),
                    "mirror_resync%x", mnum);
                CALLB_CPR_INIT(&un->un_rs_cprinfo, &un->un_rs_cpr_mx,
                    callb_md_mrs_cpr, cpr_name);

                if (ui->ui_tstate & MD_RESYNC_NOT_DONE) {
                        /*
                         * If this is the first resync following the initial
                         * snarf (MD_RESYNC_NOT_DONE still set) and we've
                         * been started outside a reconfig step (e.g. by being
                         * added to an existing set) we need to query the
                         * existing submirror state for this mirror.
                         * The set_status flags will have MD_MN_SET_MIR_STATE_RC
                         * set if we've been through a step4 reconfig, so only
                         * query the master if this isn't (yet) set. In this
                         * case we must continue the resync thread as there is
                         * not guaranteed to be a currently running resync on
                         * any of the other nodes. Worst case is that we will
                         * initiate an ownership change to this node and then
                         * find that there is no resync to perform. However, we
                         * will then have correct status across the cluster.
                         */
                        if (!md_set[setno].s_am_i_master) {
                                if (!(md_get_setstatus(setno) &
                                    MD_SET_MN_MIR_STATE_RC)) {
                                        mirror_get_status(un, NULL);
                                        block_resync = 0;
#ifdef DEBUG
                                        if (mirror_debug_flag) {
                                                mm_submirror_t *sm;
                                                int i;
                                                for (i = 0; i < NMIRROR; i++) {
                                                        sm = &un->un_sm[i];
                                                        printf(
                                                            "sm[%d] state=%4x"
                                                            " flags=%4x\n", i,
                                                            sm->sm_state,
                                                            sm->sm_flags);
                                                }
                                        }
#endif
                                }
                        }
                        ui->ui_tstate &= ~MD_RESYNC_NOT_DONE;
                }
                /*
                 * For MN set, if we have an owner, then start the resync on it.
                 * If there is no owner the master must send a message to
                 * choose the owner. This message will contain the current
                 * resync count and it will only be sent to the master, where
                 * the resync count will be used to choose the next node to
                 * perform a resync, by cycling through the nodes in the set.
                 * The message handler will then send a CHANGE_OWNER message to
                 * all nodes, and on receipt of that message, the chosen owner
                 * will issue a SET_OWNER ioctl to become the owner. This ioctl
                 * will be requested to spawn a thread to issue the
                 * REQUEST_OWNER message to become the owner which avoids the
                 * need for concurrent ioctl requests.
                 * After sending the message, we will block waiting for one
                 * of the nodes to become the owner and start the resync
                 */
                if (MD_MN_NO_MIRROR_OWNER(un)) {
                        /*
                         * There is no owner, block and then the master will
                         * choose the owner. Only perform this if 'block_resync'
                         * is set.
                         */
                        if (block_resync) {
                                mutex_enter(&un->un_rs_thread_mx);
                                un->un_rs_thread_flags |= MD_RI_BLOCK_OWNER;
                                mutex_exit(&un->un_rs_thread_mx);
                        }
                        if (md_set[setno].s_am_i_master) {
                                md_unit_writerexit(ui);
                                (void) mirror_choose_owner(un, NULL);
                                (void) md_unit_writerlock(ui);
                        }
                } else {
                        /* There is an owner, block if we are not it */
                        if (!MD_MN_MIRROR_OWNER(un)) {
                                mutex_enter(&un->un_rs_thread_mx);
                                un->un_rs_thread_flags |= MD_RI_BLOCK_OWNER;
                                mutex_exit(&un->un_rs_thread_mx);
                        }
                }
        }
        /*
         * Start a timeout chain to update the resync progress to the mddb.
         * This will run every md_mirror_resync_update_intvl minutes and allows
         * a resync to be continued over a reboot.
         */
        ASSERT(un->un_rs_resync_to_id == 0);
        un->un_rs_resync_to_id = timeout(resync_progress, un,
            (clock_t)(drv_usectohz(60000000) * md_mirror_resync_update_intvl));

        /*
         * Handle resync restart from the last logged position. The contents
         * of un_rs_resync_2_do and un_rs_resync_done are dependent on the
         * type of resync that was in progress.
         */
        if (MD_MNSET_SETNO(setno)) {
                switch ((uint_t)RS_TYPE(un->un_rs_type)) {
                case MD_RS_NONE:
                case MD_RS_OPTIMIZED:
                case MD_RS_COMPONENT:
                case MD_RS_SUBMIRROR:
                case MD_RS_ABR:
                        break;
                default:
                        un->un_rs_type = MD_RS_NONE;
                }
                /* Allocate a resync message, if required */
                if (un->un_rs_msg == NULL) {
                        un->un_rs_msg = (md_mn_msg_resync_t *)kmem_zalloc(
                            sizeof (md_mn_msg_resync_t), KM_SLEEP);
                }
                mn_resync = 1;
        }

        /* Check to see if we've been requested to block/kill */
        if (resync_kill_pending(un, ui, MD_WRITER_HELD)) {
                goto bail_out;
        }

        do {
                un->un_rs_dropped_lock = 0;
                /*
                 * Always perform an optimized resync first as this will bring
                 * the mirror into an available state in the shortest time.
                 * If we are resuming an interrupted resync, other than an
                 * optimized resync, we save the type and amount done so that
                 * we can resume the appropriate resync after the optimized
                 * resync has completed.
                 */
                if ((RS_TYPE(un->un_rs_type) != MD_RS_NONE) &&
                    (RS_TYPE(un->un_rs_type) != MD_RS_OPTIMIZED)) {
                        old_rs_type = un->un_rs_type;
                        old_rs_done = un->un_rs_resync_done;
                        old_rs_2_do = un->un_rs_resync_2_do;
                        old_rs_startbl = un->un_resync_startbl;
                }
                SET_RS_TYPE(un->un_rs_type, MD_RS_OPTIMIZED);
                /*
                 * If we are continuing a resync that is not an
                 * OPTIMIZED one, then we start from the beginning when
                 * doing this optimized resync
                 */
                if (RS_TYPE(old_rs_type) != MD_RS_OPTIMIZED) {
                        un->un_rs_resync_done = 0;
                        un->un_rs_resync_2_do = 0;
                        un->un_resync_startbl = 0;
                }
                optimized_resync(un);
                /* Check to see if we've been requested to block/kill */
                if (resync_kill_pending(un, ui, MD_WRITER_HELD)) {
                        goto bail_out;
                }
                un = (mm_unit_t *)MD_UNIT(mnum);
                /*
                 * If another node has moved the resync on, we must
                 * restart the correct resync
                 */
                if (mn_resync &&
                    (RS_TYPE(un->un_rs_type) != MD_RS_NONE)) {
                        old_rs_type = un->un_rs_type;
                        old_rs_done = un->un_rs_resync_done;
                        old_rs_2_do = un->un_rs_resync_2_do;
                        old_rs_startbl = un->un_resync_startbl;
                }

                /*
                 * Restore previous resync progress or move onto a
                 * component resync.
                 */
                if (RS_TYPE(old_rs_type) != MD_RS_NONE) {
                        un->un_rs_type = old_rs_type;
                        un->un_rs_resync_done = old_rs_done;
                        un->un_rs_resync_2_do = old_rs_2_do;
                        un->un_resync_startbl = old_rs_startbl;
                } else {
                        un->un_rs_type = MD_RS_COMPONENT;
                        un->un_rs_resync_done = 0;
                        un->un_rs_resync_2_do = 0;
                        un->un_resync_startbl = 0;
                }

                if (RS_TYPE(un->un_rs_type) == MD_RS_COMPONENT) {
                        component_resync(un);
                        /* Check to see if we've been requested to block/kill */
                        if (resync_kill_pending(un, ui, MD_WRITER_HELD)) {
                                goto bail_out;
                        }
                        un = (mm_unit_t *)MD_UNIT(mnum);
                        /*
                         * If we have moved on from a component resync, another
                         * node must have completed it and started a submirror
                         * resync, so leave the resync state alone. For non
                         * multi-node sets we move onto the submirror resync.
                         */
                        if (mn_resync) {
                                if (RS_TYPE(un->un_rs_type) == MD_RS_NONE) {
                                        un->un_rs_type = MD_RS_SUBMIRROR;
                                        un->un_rs_resync_done =
                                            un->un_rs_resync_2_do = 0;
                                        un->un_resync_startbl = 0;
                                }
                        } else {
                                un->un_rs_type = MD_RS_SUBMIRROR;
                                un->un_rs_resync_done = 0;
                                un->un_rs_resync_2_do = 0;
                                un->un_resync_startbl = 0;
                        }
                }
                if (RS_TYPE(un->un_rs_type) == MD_RS_SUBMIRROR) {
                        submirror_resync(un);
                        /* Check to see if we've been requested to block/kill */
                        if (resync_kill_pending(un, ui, MD_WRITER_HELD)) {
                                goto bail_out;
                        }
                        un = (mm_unit_t *)MD_UNIT(mnum);
                        /*
                         * If we have moved on from a submirror resync, another
                         * node must have completed it and started a different
                         * resync, so leave the resync state alone
                         */
                        if (mn_resync) {
                                if (RS_TYPE(un->un_rs_type) == MD_RS_NONE) {
                                        un->un_rs_resync_done =
                                            un->un_rs_resync_2_do = 0;
                                        un->un_resync_startbl = 0;
                                }
                        } else {
                                /* If non-MN mirror, reinitialize state */
                                un->un_rs_type = MD_RS_NONE;
                                un->un_rs_resync_done = 0;
                                un->un_rs_resync_2_do = 0;
                                un->un_resync_startbl = 0;
                        }
                }
        } while (un->un_rs_dropped_lock);
        mutex_enter(&un->un_rs_thread_mx);
        un->un_rs_thread_flags |= MD_RI_SHUTDOWN;
        mutex_exit(&un->un_rs_thread_mx);

        resync_finish = 1;
bail_out:
#ifdef DEBUG
        if (mirror_debug_flag)
                printf("Resync stopped (mnum = %x), resync_finish = %d\n",
                    mnum, resync_finish);
#endif
        kmem_free(un->un_rs_buffer, dbtob(un->un_rs_copysize));

        mutex_enter(&un->un_rs_progress_mx);
        un->un_rs_progress_flags |= MD_RI_KILL;
        cv_signal(&un->un_rs_progress_cv);
        mutex_exit(&un->un_rs_progress_mx);

        /*
         * For MN Set, send a RESYNC_FINISH if this node completed the resync.
         * There is no need to grow unit here, it will be done in the
         * handler for the RESYNC_FINISH message together with resetting
         * MD_UN_RESYNC_ACTIVE.
         */
        if (mn_resync) {
                if (resync_finish) {
                        /*
                         * Normal resync completion. Issue a RESYNC_FINISH
                         * message if we're part of a multi-node set.
                         */
                        md_mn_kresult_t *kres;
                        md_mn_msg_resync_t *rmsg;
                        int             rval;

                        rmsg = (md_mn_msg_resync_t *)un->un_rs_msg;
                        md_unit_writerexit(ui);

                        rmsg->msg_resync_mnum = mnum;
                        rmsg->msg_resync_type = 0;
                        rmsg->msg_resync_done = 0;
                        rmsg->msg_resync_2_do = 0;
                        rmsg->msg_originator = md_mn_mynode_id;

                        kres = kmem_alloc(sizeof (md_mn_kresult_t), KM_SLEEP);

smrf_msg:
                        mutex_enter(&un->un_rs_cpr_mx);
                        CALLB_CPR_SAFE_BEGIN(&un->un_rs_cprinfo);

                        rval = mdmn_ksend_message(setno,
                            MD_MN_MSG_RESYNC_FINISH, MD_MSGF_NO_LOG, 0,
                            (char *)rmsg, sizeof (md_mn_msg_resync_t), kres);

                        CALLB_CPR_SAFE_END(&un->un_rs_cprinfo,
                            &un->un_rs_cpr_mx);
                        mutex_exit(&un->un_rs_cpr_mx);

                        if (!MDMN_KSEND_MSG_OK(rval, kres)) {
                                mdmn_ksend_show_error(rval, kres,
                                    "RESYNC_FINISH");
                                /* If we're shutting down, pause things here. */
                                if (kres->kmmr_comm_state == MDMNE_RPC_FAIL) {
                                        while (!md_mn_is_commd_present()) {
                                                delay(md_hz);
                                        }
                                        /*
                                         * commd is now available again. Retry
                                         * the message once. If this fails we
                                         * panic as the system is in an
                                         * unexpected state.
                                         */
                                        if (nretries++ == 0)
                                                goto smrf_msg;
                                }
                                cmn_err(CE_PANIC,
                                    "ksend_message failure: RESYNC_FINISH");
                        }
                        kmem_free(kres, sizeof (md_mn_kresult_t));
                        (void) md_unit_writerlock(ui);
                }
                /*
                 * If the resync has been cancelled, clear flags, reset owner
                 * for ABR mirror and release the resync region parent
                 * structure.
                 */
                if (un->c.un_status & MD_UN_RESYNC_CANCEL) {
                        md_mps_t        *ps;

                        if (ui->ui_tstate & MD_ABR_CAP) {
                                /* Resync finished, if ABR set owner to NULL */
                                mutex_enter(&un->un_owner_mx);
                                un->un_mirror_owner = 0;
                                mutex_exit(&un->un_owner_mx);
                        }

                        un->c.un_status &= ~(MD_UN_RESYNC_CANCEL |
                            MD_UN_RESYNC_ACTIVE);
                        ps = un->un_rs_prev_overlap;
                        if (ps != NULL) {
                                /* Remove previous overlap resync region */
                                if (ps->ps_flags & MD_MPS_ON_OVERLAP)
                                mirror_overlap_tree_remove(ps);
                                /*
                                 * Release the overlap range reference
                                 */
                                un->un_rs_prev_overlap = NULL;
                                kmem_cache_free(mirror_parent_cache,
                                    ps);
                        }
                }

                /*
                 * Release resync message buffer. This will be reallocated on
                 * the next invocation of the resync_unit thread.
                 */
                if (un->un_rs_msg) {
                        kmem_free(un->un_rs_msg, sizeof (md_mn_msg_resync_t));
                        un->un_rs_msg = NULL;
                }
        } else {
                /* For non-MN sets deal with any pending grows */
                un->c.un_status &= ~MD_UN_RESYNC_ACTIVE;
                if (un->c.un_status & MD_UN_GROW_PENDING) {
                        if ((mirror_grow_unit(un, &mde) != 0) ||
                            (! mdismderror(&mde, MDE_GROW_DELAYED))) {
                                un->c.un_status &= ~MD_UN_GROW_PENDING;
                        }
                }
        }

        reset_comp_flags(un);
        un->un_resync_completed = 0;
        mirror_commit(un, NO_SUBMIRRORS, 0);
        md_unit_writerexit(ui);

        /*
         * Stop the resync progress thread.
         */
        if (un->un_rs_resync_to_id != 0) {
                (void) untimeout(un->un_rs_resync_to_id);
                un->un_rs_resync_to_id = 0;
        }

        /*
         * Calling mirror_internal_close() makes further reference to un / ui
         * dangerous. If we are the only consumer of the mirror it is possible
         * for a metaclear to be processed after completion of the m_i_c()
         * routine. As we need to handle the case where another resync has been
         * scheduled for the mirror, we raise the open count on the device
         * which protects against the close / metaclear / lock => panic scenario
         */
        (void) md_unit_incopen(MD_SID(un), FREAD|FWRITE, OTYP_LYR);
        (void) mirror_internal_close(MD_SID(un), OTYP_LYR, 0, (IOLOCK *)NULL);

        /*
         * deccrement the mirror resync count
         */
        mutex_enter(&md_cpr_resync.md_resync_mutex);
        md_cpr_resync.md_mirror_resync--;
        mutex_exit(&md_cpr_resync.md_resync_mutex);

        /*
         * Remove the thread reference as we're about to exit. This allows a
         * subsequent mirror_resync_unit() to start a new thread.
         * If RESYNC_ACTIVE is set, mirror_resync_unit() must have been
         * called to start a new resync, so reopen the mirror and go back to
         * the start.
         */
        (void) md_unit_writerlock(ui);
        mutex_enter(&un->un_rs_thread_mx);
        un->un_rs_thread_flags &= ~(MD_RI_KILL|MD_RI_SHUTDOWN);
        mutex_exit(&un->un_rs_thread_mx);
        if (un->c.un_status & MD_UN_RESYNC_ACTIVE) {
                md_unit_writerexit(ui);
                if (mirror_internal_open(MD_SID(un), (FREAD|FWRITE),
                    OTYP_LYR, 0, (IOLOCK *)NULL) == 0) {
                        /* Release the reference grabbed above */
                        (void) mirror_internal_close(MD_SID(un), OTYP_LYR, 0,
                            (IOLOCK *)NULL);
                        goto resync_restart;
                }
                (void) md_unit_writerlock(ui);
                cmn_err(CE_NOTE,
                    "Could not open metadevice (%x) for resync\n",
                    MD_SID(un));
        }
        un->un_rs_thread = NULL;
        md_unit_writerexit(ui);

        /*
         * Check for hotspares once we've cleared the resync thread reference.
         * If there are any errored units a poke_hotspares() will result in
         * a call to mirror_resync_unit() which we need to allow to start.
         */
        (void) poke_hotspares();

        /*
         * Remove this thread from the CPR callback table.
         */
        if (mn_resync) {
                mutex_enter(&un->un_rs_cpr_mx);
                CALLB_CPR_EXIT(&un->un_rs_cprinfo);
        }

        /*
         * Remove the extra reference to the unit we generated above. After
         * this call it is *unsafe* to reference either ui or un as they may
         * no longer be allocated.
         */
        (void) mirror_internal_close(MD_SID(un), OTYP_LYR, 0, (IOLOCK *)NULL);

        thread_exit();
}

/*
 * mirror_resync_unit:
 * ------------------
 * Start a resync for the given mirror metadevice. Save the resync thread ID in
 * un->un_rs_thread for later manipulation.
 *
 * Returns:
 *      0       Success
 *      !=0     Error
 */
/*ARGSUSED*/
int
mirror_resync_unit(
        minor_t                 mnum,
        md_resync_ioctl_t       *ri,
        md_error_t              *ep,
        IOLOCK                  *lockp
)
{
        mdi_unit_t              *ui;
        mm_unit_t               *un;
        set_t                   setno = MD_MIN2SET(mnum);

        ui = MDI_UNIT(mnum);

        if (md_get_setstatus(setno) & MD_SET_STALE)
                return (mdmddberror(ep, MDE_DB_STALE, mnum, setno));

        if (mirror_internal_open(mnum, (FREAD|FWRITE), OTYP_LYR, 0, lockp)) {
                return (mdmderror(ep, MDE_MIRROR_OPEN_FAILURE, mnum));
        }
        if (lockp) {
                un = (mm_unit_t *)md_ioctl_writerlock(lockp, ui);
        } else {
                un = (mm_unit_t *)md_unit_writerlock(ui);
        }

        /*
         * Check to see if we're attempting to start a resync while one is
         * already running.
         */
        if (un->c.un_status & MD_UN_RESYNC_ACTIVE ||
            un->un_rs_thread != NULL) {
                /*
                 * Ensure RESYNC_ACTIVE set, it may not be if the resync thread
                 * is in the process of terminating, setting the flag will
                 * cause the resync thread to return to the beginning
                 */
                un->c.un_status |= MD_UN_RESYNC_ACTIVE;
                if (lockp) {
                        md_ioctl_writerexit(lockp);
                } else {
                        md_unit_writerexit(ui);
                }
                (void) mirror_internal_close(mnum, OTYP_LYR, 0, lockp);
                return (0);
        }
        un->c.un_status |= MD_UN_RESYNC_ACTIVE;
        un->c.un_status &= ~MD_UN_RESYNC_CANCEL;
        if ((ri) && (ri->ri_copysize > 0) &&
            (ri->ri_copysize <= md_max_xfer_bufsz))
                un->un_rs_copysize = ri->ri_copysize;
        else
                un->un_rs_copysize = 0;

        /* Start the resync progress thread off */
        un->un_rs_progress_flags = 0;
        (void) thread_create(NULL, 0, resync_progress_thread,
            (caddr_t)(uintptr_t)mnum, 0, &p0, TS_RUN, minclsyspri);

        /*
         * We have to store the thread ID in the unit structure so do not
         * drop writerlock until the thread is active. This means resync_unit
         * may spin on its first md_unit_readerlock(), but deadlock won't occur.
         */
        mutex_enter(&un->un_rs_thread_mx);
        un->un_rs_thread_flags &= ~(MD_RI_KILL|MD_RI_SHUTDOWN);
        mutex_exit(&un->un_rs_thread_mx);
        un->un_rs_thread = thread_create(NULL, 0, resync_unit,
            (caddr_t)(uintptr_t)mnum, 0, &p0, TS_RUN, 60);
        if (un->un_rs_thread == (kthread_id_t)NULL) {
                un->c.un_status &= ~MD_UN_RESYNC_ACTIVE;
                if (lockp) {
                        md_ioctl_writerexit(lockp);
                } else {
                        md_unit_writerexit(ui);
                }
                (void) mirror_internal_close(mnum, OTYP_LYR, 0, lockp);
                return (mdmderror(ep, MDE_MIRROR_THREAD_FAILURE, mnum));
        } else {
                if (lockp) {
                        md_ioctl_writerexit(lockp);
                } else {
                        md_unit_writerexit(ui);
                }
        }

        return (0);
}

/*
 * mirror_ioctl_resync:
 * -------------------
 * Called as a result of an MD_IOCSETSYNC ioctl. Either start, block, unblock
 * or kill the resync thread associated with the specified unit.
 * Can return with locks held since mdioctl will free any locks
 * that are marked in lock->l_flags.
 *
 * Returns:
 *      0       Success
 *      !=0     Error Code
 */
int
mirror_ioctl_resync(
        md_resync_ioctl_t       *ri,
        IOLOCK                  *lock
)
{
        minor_t                 mnum = ri->ri_mnum;
        mm_unit_t               *un;
        uint_t                  bits;
        mm_submirror_t          *sm;
        mm_submirror_ic_t       *smic;
        int                     smi;
        kt_did_t                tid;
        set_t                   setno = MD_MIN2SET(mnum);

        mdclrerror(&ri->mde);

        if ((setno >= md_nsets) ||
            (MD_MIN2UNIT(mnum) >= md_nunits)) {
                return (mdmderror(&ri->mde, MDE_INVAL_UNIT, mnum));
        }

        /* RD_LOCK flag grabs the md_ioctl_readerlock */
        un = mirror_getun(mnum, &ri->mde, RD_LOCK, lock);

        if (un == NULL) {
                return (mdmderror(&ri->mde, MDE_UNIT_NOT_SETUP, mnum));
        }
        if (un->c.un_type != MD_METAMIRROR) {
                return (mdmderror(&ri->mde, MDE_NOT_MM, mnum));
        }
        if (un->un_nsm < 2) {
                return (0);
        }

        /*
         * Determine the action to take based on the ri_flags field:
         *      MD_RI_BLOCK:    Block current resync thread
         *      MD_RI_UNBLOCK:  Unblock resync thread
         *      MD_RI_KILL:     Abort resync thread
         *      MD_RI_RESYNC_FORCE_MNSTART: Directly start resync thread
         *              without using rpc.mdcommd messages.
         *      any other:      Start resync thread
         */
        switch (ri->ri_flags & (MD_RI_BLOCK|MD_RI_UNBLOCK|MD_RI_KILL)) {

        case MD_RI_BLOCK:
                /* Halt resync thread by setting flag in un_rs_flags */
                if (!(un->c.un_status & MD_UN_RESYNC_ACTIVE)) {
                        return (0);
                }
                mutex_enter(&un->un_rs_thread_mx);
                un->un_rs_thread_flags |= MD_RI_BLOCK;
                mutex_exit(&un->un_rs_thread_mx);
                return (0);

        case MD_RI_UNBLOCK:
                /*
                 * Restart resync thread by clearing flag in un_rs_flags and
                 * cv_signal'ing the blocked thread.
                 */
                if (!(un->c.un_status & MD_UN_RESYNC_ACTIVE)) {
                        return (0);
                }
                mutex_enter(&un->un_rs_thread_mx);
                un->un_rs_thread_flags &= ~MD_RI_BLOCK;
                cv_signal(&un->un_rs_thread_cv);
                mutex_exit(&un->un_rs_thread_mx);
                return (0);

        case MD_RI_KILL:
                /* Abort resync thread. */
                if (!(un->c.un_status & MD_UN_RESYNC_ACTIVE)) {
                        return (0);
                }
                mutex_enter(&un->un_rs_thread_mx);
                tid = un->un_rs_thread ? (un->un_rs_thread)->t_did : 0;
                un->un_rs_thread_flags &= ~(MD_RI_BLOCK|MD_RI_BLOCK_OWNER);
                un->un_rs_thread_flags |= MD_RI_KILL;
                cv_signal(&un->un_rs_thread_cv);
                mutex_exit(&un->un_rs_thread_mx);
                if (tid != 0) {
                        if (!(ri->ri_flags & MD_RI_NO_WAIT)) {
                                md_ioctl_readerexit(lock);
                                thread_join(tid);
                                un->un_rs_thread_flags &= ~MD_RI_KILL;
                                un->un_rs_thread = NULL;
                                cmn_err(CE_WARN, "md: %s: Resync cancelled\n",
                                    md_shortname(MD_SID(un)));
                        }
                }
                return (0);
        }

        md_ioctl_readerexit(lock);

        bits = 0;
        for (smi = 0; smi < NMIRROR; smi++) {
                sm = &un->un_sm[smi];
                smic = &un->un_smic[smi];
                if (!SMS_IS(sm, SMS_ATTACHED))
                        continue;
                mirror_set_sm_state(sm, smic, SMS_ATTACHED_RESYNC, 1);
                bits |= SMI2BIT(smi);
        }
        if (bits != 0)
                mirror_commit(un, bits, 0);

        /*
         * If we are resyncing a mirror in a MN set and the rpc.mdcommd
         * can be used, we do not start the resync at this point.
         * Instead, the metasync command that issued the ioctl
         * will send a RESYNC_STARTING message to start the resync thread. The
         * reason we do it this way is to ensure that the metasync ioctl is
         * executed on all nodes before the resync thread is started.
         *
         * If a MN set and the MD_RI_RESYNC_FORCE_MNSTART flag is set, then
         * don't use rpc.mdcommd, but just start the resync thread.  This
         * flag is set on a node when it is being added to a diskset
         * so that the resync threads are started on the newly added node.
         */
        if ((!(MD_MNSET_SETNO(setno))) ||
            (ri->ri_flags & MD_RI_RESYNC_FORCE_MNSTART)) {
                return (mirror_resync_unit(mnum, ri, &ri->mde, lock));
        } else {
                return (0);
        }
}

int
mirror_mark_resync_region_non_owner(struct mm_unit *un,
        diskaddr_t startblk, diskaddr_t endblk, md_mn_nodeid_t source_node)
{
        int                     no_change;
        size_t                  start_rr;
        size_t                  current_rr;
        size_t                  end_rr;
        md_mn_msg_rr_dirty_t    *rr;
        md_mn_kresult_t         *kres;
        set_t                   setno = MD_UN2SET(un);
        int                     rval;
        md_mn_nodeid_t          node_idx = source_node - 1;
        mdi_unit_t              *ui = MDI_UNIT(MD_SID(un));
        md_mn_nodeid_t          owner_node;
        minor_t                 mnum = MD_SID(un);

        if (un->un_nsm < 2)
                return (0);

        /*
         * Check to see if we have a un_pernode_dirty_bm[] entry allocated. If
         * not, allocate it and then fill the [start..end] entries.
         * Update un_pernode_dirty_sum if we've gone 0->1.
         * Update un_dirty_bm if the corresponding entries are clear.
         */
        rw_enter(&un->un_pernode_dirty_mx[node_idx], RW_WRITER);
        if (un->un_pernode_dirty_bm[node_idx] == NULL) {
                un->un_pernode_dirty_bm[node_idx] =
                    (uchar_t *)kmem_zalloc(
                    (uint_t)howmany(un->un_rrd_num, NBBY), KM_SLEEP);
        }
        rw_exit(&un->un_pernode_dirty_mx[node_idx]);

        BLK_TO_RR(end_rr, endblk, un);
        BLK_TO_RR(start_rr, startblk, un);

        no_change = 1;

        mutex_enter(&un->un_resync_mx);
        rw_enter(&un->un_pernode_dirty_mx[node_idx], RW_READER);
        for (current_rr = start_rr; current_rr <= end_rr; current_rr++) {
                un->un_outstanding_writes[current_rr]++;
                if (!IS_PERNODE_DIRTY(source_node, current_rr, un)) {
                        un->un_pernode_dirty_sum[current_rr]++;
                        SET_PERNODE_DIRTY(source_node, current_rr, un);
                }
                CLR_GOING_CLEAN(current_rr, un);
                if (!IS_REGION_DIRTY(current_rr, un)) {
                        no_change = 0;
                        SET_REGION_DIRTY(current_rr, un);
                        SET_GOING_DIRTY(current_rr, un);
                } else if (IS_GOING_DIRTY(current_rr, un))
                        no_change = 0;
        }
        rw_exit(&un->un_pernode_dirty_mx[node_idx]);
        mutex_exit(&un->un_resync_mx);

        if (no_change) {
                return (0);
        }

        /*
         * If we have dirty regions to commit, send a
         * message to the owning node so that the
         * in-core bitmap gets updated appropriately.
         * TODO: make this a kmem_cache pool to improve
         * alloc/free performance ???
         */
        kres = (md_mn_kresult_t *)kmem_alloc(sizeof (md_mn_kresult_t),
            KM_SLEEP);
        rr = (md_mn_msg_rr_dirty_t *)kmem_alloc(sizeof (md_mn_msg_rr_dirty_t),
            KM_SLEEP);

resend_mmrr:
        owner_node = un->un_mirror_owner;

        rr->rr_mnum = mnum;
        rr->rr_nodeid = md_mn_mynode_id;
        rr->rr_range = (ushort_t)start_rr << 16;
        rr->rr_range |= (ushort_t)end_rr & 0xFFFF;

        /* release readerlock before sending message */
        md_unit_readerexit(ui);

        rval = mdmn_ksend_message(setno, MD_MN_MSG_RR_DIRTY,
            MD_MSGF_NO_LOG|MD_MSGF_BLK_SIGNAL|MD_MSGF_DIRECTED,
            un->un_mirror_owner, (char *)rr,
            sizeof (md_mn_msg_rr_dirty_t), kres);

        /* reaquire readerlock on message completion */
        (void) md_unit_readerlock(ui);

        /* if the message send failed, note it, and pass an error back up */
        if (!MDMN_KSEND_MSG_OK(rval, kres)) {
                /* if commd is gone, no point in printing a message */
                if (md_mn_is_commd_present())
                        mdmn_ksend_show_error(rval, kres, "RR_DIRTY");
                kmem_free(kres, sizeof (md_mn_kresult_t));
                kmem_free(rr, sizeof (md_mn_msg_rr_dirty_t));
                return (1);
        }

        /*
         * if the owner changed while we were sending the message, and it's
         * not us, the new mirror owner won't yet have done the right thing
         * with our data.  Let him know.  If we became the owner, we'll
         * deal with that differently below.  Note that receiving a message
         * about another node twice won't hurt anything.
         */
        if (un->un_mirror_owner != owner_node && !MD_MN_MIRROR_OWNER(un))
                goto resend_mmrr;

        kmem_free(kres, sizeof (md_mn_kresult_t));
        kmem_free(rr, sizeof (md_mn_msg_rr_dirty_t));

        mutex_enter(&un->un_resync_mx);

        /*
         * If we became the owner changed while we were sending the message,
         * we have dirty bits in the un_pernode_bm that aren't yet reflected
         * in the un_dirty_bm, as it was re-read from disk, and our bits
         * are also not reflected in the on-disk DRL.  Fix that now.
         */
        if (MD_MN_MIRROR_OWNER(un)) {
                rw_enter(&un->un_pernode_dirty_mx[node_idx], RW_WRITER);
                mirror_copy_rr(howmany(un->un_rrd_num, NBBY),
                    un->un_pernode_dirty_bm[node_idx], un->un_dirty_bm);
                rw_exit(&un->un_pernode_dirty_mx[node_idx]);

                un->un_resync_flg |= MM_RF_COMMITING | MM_RF_GATECLOSED;

                mutex_exit(&un->un_resync_mx);
                mddb_commitrec_wrapper(un->un_rr_dirty_recid);
                mutex_enter(&un->un_resync_mx);

                un->un_resync_flg &= ~(MM_RF_COMMITING | MM_RF_GATECLOSED);
                cv_broadcast(&un->un_resync_cv);
        }

        for (current_rr = start_rr; current_rr <= end_rr; current_rr++)
                CLR_GOING_DIRTY(current_rr, un);

        mutex_exit(&un->un_resync_mx);

        return (0);
}

int
mirror_mark_resync_region_owner(struct mm_unit *un,
        diskaddr_t startblk, diskaddr_t endblk, md_mn_nodeid_t source_node)
{
        int                     no_change;
        size_t                  start_rr;
        size_t                  current_rr;
        size_t                  end_rr;
        int                     mnset = MD_MNSET_SETNO(MD_UN2SET(un));
        md_mn_nodeid_t          node_idx = source_node - 1;

        if (un->un_nsm < 2)
                return (0);

        /*
         * Check to see if we have a un_pernode_dirty_bm[] entry allocated. If
         * not, allocate it and then fill the [start..end] entries.
         * Update un_pernode_dirty_sum if we've gone 0->1.
         * Update un_dirty_bm if the corresponding entries are clear.
         */
        if (mnset) {
                rw_enter(&un->un_pernode_dirty_mx[node_idx], RW_WRITER);
                if (un->un_pernode_dirty_bm[node_idx] == NULL) {
                        un->un_pernode_dirty_bm[node_idx] =
                            (uchar_t *)kmem_zalloc(
                            (uint_t)howmany(un->un_rrd_num, NBBY), KM_SLEEP);
                }
                rw_exit(&un->un_pernode_dirty_mx[node_idx]);
        }

        mutex_enter(&un->un_resync_mx);

        if (mnset)
                rw_enter(&un->un_pernode_dirty_mx[node_idx], RW_READER);

        no_change = 1;
        BLK_TO_RR(end_rr, endblk, un);
        BLK_TO_RR(start_rr, startblk, un);
        for (current_rr = start_rr; current_rr <= end_rr; current_rr++) {
                if (!mnset || source_node == md_mn_mynode_id)
                        un->un_outstanding_writes[current_rr]++;
                if (mnset) {
                        if (!IS_PERNODE_DIRTY(source_node, current_rr, un))
                                un->un_pernode_dirty_sum[current_rr]++;
                        SET_PERNODE_DIRTY(source_node, current_rr, un);
                }
                CLR_GOING_CLEAN(current_rr, un);
                if (!IS_REGION_DIRTY(current_rr, un))
                        no_change = 0;
                if (IS_GOING_DIRTY(current_rr, un))
                        no_change = 0;
        }

        if (mnset)
                rw_exit(&un->un_pernode_dirty_mx[node_idx]);

        if (no_change) {
                mutex_exit(&un->un_resync_mx);
                return (0);
        }
        un->un_waiting_to_mark++;
        while (un->un_resync_flg & MM_RF_GATECLOSED) {
                if (panicstr)
                        return (1);
                cv_wait(&un->un_resync_cv, &un->un_resync_mx);
        }
        un->un_waiting_to_mark--;

        no_change = 1;
        for (current_rr = start_rr; current_rr <= end_rr; current_rr++) {
                if (!IS_REGION_DIRTY(current_rr, un)) {
                        SET_REGION_DIRTY(current_rr, un);
                        SET_GOING_DIRTY(current_rr, un);
                        no_change = 0;
                } else {
                        if (IS_GOING_DIRTY(current_rr, un))
                                no_change = 0;
                }
        }
        if (no_change) {
                if (un->un_waiting_to_mark == 0 || un->un_waiting_to_clear != 0)
                        cv_broadcast(&un->un_resync_cv);
                mutex_exit(&un->un_resync_mx);
                return (0);
        }

        un->un_resync_flg |= MM_RF_COMMIT_NEEDED;
        un->un_waiting_to_commit++;
        while (un->un_waiting_to_mark != 0 &&
            !(un->un_resync_flg & MM_RF_GATECLOSED)) {
                if (panicstr)
                        return (1);
                cv_wait(&un->un_resync_cv, &un->un_resync_mx);
        }

        if (un->un_resync_flg & MM_RF_COMMIT_NEEDED) {
                un->un_resync_flg |= MM_RF_COMMITING | MM_RF_GATECLOSED;
                un->un_resync_flg &= ~MM_RF_COMMIT_NEEDED;

                mutex_exit(&un->un_resync_mx);
                mddb_commitrec_wrapper(un->un_rr_dirty_recid);
                mutex_enter(&un->un_resync_mx);

                un->un_resync_flg &= ~MM_RF_COMMITING;
                cv_broadcast(&un->un_resync_cv);
        }
        while (un->un_resync_flg & MM_RF_COMMITING) {
                if (panicstr)
                        return (1);
                cv_wait(&un->un_resync_cv, &un->un_resync_mx);
        }

        for (current_rr = start_rr; current_rr <= end_rr; current_rr++)
                CLR_GOING_DIRTY(current_rr, un);

        if (--un->un_waiting_to_commit == 0) {
                un->un_resync_flg &= ~MM_RF_GATECLOSED;
                cv_broadcast(&un->un_resync_cv);
        }
        mutex_exit(&un->un_resync_mx);

        return (0);
}

int
mirror_mark_resync_region(struct mm_unit *un,
        diskaddr_t startblk, diskaddr_t endblk, md_mn_nodeid_t source_node)
{
        int     mnset = MD_MNSET_SETNO(MD_UN2SET(un));

        if (mnset && !MD_MN_MIRROR_OWNER(un)) {
                return (mirror_mark_resync_region_non_owner(un, startblk,
                    endblk, source_node));
        } else {
                return (mirror_mark_resync_region_owner(un, startblk, endblk,
                    source_node));
        }
}

int
mirror_resize_resync_regions(mm_unit_t *un, diskaddr_t new_tb)
{
        short           *owp;
        optim_resync_t  *orp;
        uint_t          rr_mult = 1;
        uint_t          old_nregions, new_nregions;
        int             old_bm_size, new_bm_size;
        size_t          size;
        mddb_recid_t    recid, old_recid;
        uchar_t         *old_dirty_bm;
        int             i, j;
        mddb_type_t     typ1;
        set_t           setno = MD_UN2SET(un);
        uchar_t         *old_pns;

        old_nregions = un->un_rrd_num;
        new_nregions = (uint_t)((new_tb/un->un_rrd_blksize) + 1);

        while (new_nregions > MD_MAX_NUM_RR) {
                new_nregions >>= 1;
                rr_mult <<= 1;
        }

        new_bm_size = howmany(new_nregions, NBBY);
        old_bm_size = howmany(old_nregions, NBBY);

        size = new_bm_size + sizeof (*orp) - sizeof (orp->or_rr);

        typ1 = (mddb_type_t)md_getshared_key(setno,
            mirror_md_ops.md_driver.md_drivername);
        recid = mddb_createrec(size, typ1, RESYNC_REC,
            MD_CRO_OPTIMIZE|MD_CRO_32BIT, setno);
        if (recid < 0)
                return (-1);

        orp = (struct optim_resync *)mddb_getrecaddr(recid);
        ASSERT(orp != NULL);

        orp->or_magic = OR_MAGIC;               /* Magic # */
        orp->or_blksize = un->un_rrd_blksize;   /* Same block size */
        orp->or_num = new_nregions;             /* New number of regions */

        old_dirty_bm = un->un_dirty_bm;
        un->un_dirty_bm = orp->or_rr;

        kmem_free((caddr_t)un->un_goingdirty_bm, old_bm_size);
        un->un_goingdirty_bm = (uchar_t *)kmem_zalloc(new_bm_size, KM_SLEEP);

        kmem_free((caddr_t)un->un_goingclean_bm, old_bm_size);
        un->un_goingclean_bm = (uchar_t *)kmem_zalloc(new_bm_size, KM_SLEEP);

        kmem_free((caddr_t)un->un_resync_bm, old_bm_size);
        un->un_resync_bm = (uchar_t *)kmem_zalloc(new_bm_size, KM_SLEEP);

        owp = un->un_outstanding_writes;
        un->un_outstanding_writes = (short *)kmem_zalloc(
            new_nregions * sizeof (short), KM_SLEEP);

        old_pns = un->un_pernode_dirty_sum;
        if (old_pns)
                un->un_pernode_dirty_sum = (uchar_t *)kmem_zalloc(new_nregions,
                    KM_SLEEP);

        /*
         * Now translate the old records into the new
         * records
         */
        for (i = 0; i < old_nregions; i++) {
                /*
                 * only bring forward the
                 * outstanding write counters and the dirty bits and also
                 * the pernode_summary counts
                 */
                if (!isset(old_dirty_bm, i))
                        continue;

                setbit(un->un_dirty_bm, (i / rr_mult));
                un->un_outstanding_writes[(i / rr_mult)] += owp[i];
                if (old_pns)
                        un->un_pernode_dirty_sum[(i / rr_mult)] += old_pns[i];
        }
        kmem_free((caddr_t)owp, old_nregions * sizeof (short));
        if (old_pns)
                kmem_free((caddr_t)old_pns, old_nregions);

        /*
         * Copy all non-zero un_pernode_dirty_bm[] arrays to new versions
         */
        for (j = 0; j < MD_MNMAXSIDES; j++) {
                rw_enter(&un->un_pernode_dirty_mx[j], RW_WRITER);
                old_dirty_bm = un->un_pernode_dirty_bm[j];
                if (old_dirty_bm) {
                        un->un_pernode_dirty_bm[j] = (uchar_t *)kmem_zalloc(
                            new_bm_size, KM_SLEEP);
                        for (i = 0; i < old_nregions; i++) {
                                if (!isset(old_dirty_bm, i))
                                        continue;

                                setbit(un->un_pernode_dirty_bm[j],
                                    (i / rr_mult));
                        }
                        kmem_free((caddr_t)old_dirty_bm, old_bm_size);
                }
                rw_exit(&un->un_pernode_dirty_mx[j]);
        }

        /* Save the old record id */
        old_recid = un->un_rr_dirty_recid;

        /* Update the mirror unit struct */
        un->un_rr_dirty_recid = recid;
        un->un_rrd_num = new_nregions;
        un->un_rrd_blksize = un->un_rrd_blksize * rr_mult;

        orp->or_blksize = un->un_rrd_blksize;

        /*
         * NOTE: The reason there are distinct calls to mddb_commitrec_wrapper
         * instead of using mddb_commitrecs_wrapper, is that you cannot
         * atomically commit optimized records.
         */
        mddb_commitrec_wrapper(recid);
        mddb_commitrec_wrapper(un->c.un_record_id);
        mddb_deleterec_wrapper(old_recid);
        return (0);
}

/* lockp can be NULL for !MN diksets */
int
mirror_add_resync_regions(mm_unit_t *un, diskaddr_t new_tb)
{
        uchar_t         *old;
        short           *owp;
        optim_resync_t  *orp;
        uint_t          old_nregions, new_nregions;
        int             old_bm_size, new_bm_size;
        size_t          size;
        mddb_recid_t    recid, old_recid;
        mddb_type_t     typ1;
        set_t           setno = MD_UN2SET(un);
        int             i;

        old_nregions = un->un_rrd_num;
        new_nregions = (uint_t)((new_tb/un->un_rrd_blksize) + 1);

        new_bm_size = howmany(new_nregions, NBBY);
        old_bm_size = howmany(old_nregions, NBBY);

        size = new_bm_size + sizeof (*orp) - sizeof (orp->or_rr);

        typ1 = (mddb_type_t)md_getshared_key(setno,
            mirror_md_ops.md_driver.md_drivername);

        recid = mddb_createrec(size, typ1, RESYNC_REC,
            MD_CRO_OPTIMIZE|MD_CRO_32BIT, setno);
        if (recid < 0)
                return (-1);

        orp = (struct optim_resync *)mddb_getrecaddr(recid);
        ASSERT(orp != NULL);

        orp->or_magic = OR_MAGIC;               /* Magic # */
        orp->or_blksize = un->un_rrd_blksize;   /* Same block size */
        orp->or_num = new_nregions;             /* New number of regions */

        /* Copy the old bm over the new bm */
        bcopy((caddr_t)un->un_dirty_bm, (caddr_t)orp->or_rr, old_bm_size);

        /*
         * Create new bigger incore arrays, copy, and free old ones:
         *              un_goingdirty_bm
         *              un_goingclean_bm
         *              un_resync_bm
         *              un_outstanding_writes
         *              un_pernode_dirty_sum
         *              un_pernode_dirty_bm[]
         */
        old = un->un_goingdirty_bm;
        un->un_goingdirty_bm = (uchar_t *)kmem_zalloc(new_bm_size, KM_SLEEP);
        bcopy((caddr_t)old, (caddr_t)un->un_goingdirty_bm, old_bm_size);
        kmem_free((caddr_t)old, old_bm_size);

        old = un->un_goingclean_bm;
        un->un_goingclean_bm = (uchar_t *)kmem_zalloc(new_bm_size, KM_SLEEP);
        bcopy((caddr_t)old, (caddr_t)un->un_goingclean_bm, old_bm_size);
        kmem_free((caddr_t)old, old_bm_size);

        old = un->un_resync_bm;
        un->un_resync_bm = (uchar_t *)kmem_zalloc(new_bm_size, KM_SLEEP);
        bcopy((caddr_t)old, (caddr_t)un->un_resync_bm, old_bm_size);
        kmem_free((caddr_t)old, old_bm_size);

        owp = un->un_outstanding_writes;
        un->un_outstanding_writes = (short *)kmem_zalloc(
            (uint_t)new_nregions * sizeof (short), KM_SLEEP);
        bcopy((caddr_t)owp, (caddr_t)un->un_outstanding_writes,
            old_nregions * sizeof (short));
        kmem_free((caddr_t)owp, (old_nregions * sizeof (short)));

        old = un->un_pernode_dirty_sum;
        if (old) {
                un->un_pernode_dirty_sum = (uchar_t *)kmem_zalloc(
                    new_nregions, KM_SLEEP);
                bcopy((caddr_t)old, (caddr_t)un->un_pernode_dirty_sum,
                    old_nregions);
                kmem_free((caddr_t)old, old_nregions);
        }

        for (i = 0; i < MD_MNMAXSIDES; i++) {
                rw_enter(&un->un_pernode_dirty_mx[i], RW_WRITER);
                old = un->un_pernode_dirty_bm[i];
                if (old) {
                        un->un_pernode_dirty_bm[i] = (uchar_t *)kmem_zalloc(
                            new_bm_size, KM_SLEEP);
                        bcopy((caddr_t)old, (caddr_t)un->un_pernode_dirty_bm[i],
                            old_bm_size);
                        kmem_free((caddr_t)old, old_bm_size);
                }
                rw_exit(&un->un_pernode_dirty_mx[i]);
        }

        /* Save the old record id */
        old_recid = un->un_rr_dirty_recid;

        /* Update the mirror unit struct */
        un->un_rr_dirty_recid = recid;
        un->un_rrd_num = new_nregions;
        un->un_dirty_bm = orp->or_rr;

        /*
         * NOTE: The reason there are distinct calls to mddb_commitrec_wrapper
         * instead of using mddb_commitrecs_wrapper, is that you cannot
         * atomically commit optimized records.
         */
        mddb_commitrec_wrapper(recid);
        mddb_commitrec_wrapper(un->c.un_record_id);
        mddb_deleterec_wrapper(old_recid);
        return (0);
}

/*
 * mirror_copy_rr:
 * --------------
 * Combine the dirty record bitmap with the in-core resync bitmap. This allows
 * us to carry a resync over an ownership change.
 */
void
mirror_copy_rr(int sz, uchar_t *src, uchar_t *dest)
{
        int     i;

        for (i = 0; i < sz; i++)
                *dest++ |= *src++;
}

/*
 * mirror_set_dirty_rr:
 * -------------------
 * Set the pernode_dirty_bm[node] entries and un_dirty_bm[] if appropriate.
 * For the owning node (DRL/mirror owner) update the on-disk RR if needed.
 * Called on every clean->dirty transition for the originating writer node.
 * Note: only the non-owning nodes will initiate this message and it is only
 * the owning node that has to process it.
 */
int
mirror_set_dirty_rr(md_mn_rr_dirty_params_t *iocp)
{

        minor_t                 mnum = iocp->rr_mnum;
        mm_unit_t               *un;
        int                     start = (int)iocp->rr_start;
        int                     end = (int)iocp->rr_end;
        set_t                   setno = MD_MIN2SET(mnum);
        md_mn_nodeid_t          orignode = iocp->rr_nodeid;     /* 1-based */
        diskaddr_t              startblk, endblk;

        mdclrerror(&iocp->mde);

        if ((setno >= md_nsets) ||
            (MD_MIN2UNIT(mnum) >= md_nunits)) {
                return (mdmderror(&iocp->mde, MDE_INVAL_UNIT, mnum));
        }

        /* Must have _NO_ ioctl lock set if we update the RR on-disk */
        un = mirror_getun(mnum, &iocp->mde, NO_LOCK, NULL);

        if (un == NULL) {
                return (mdmderror(&iocp->mde, MDE_UNIT_NOT_SETUP, mnum));
        }
        if (un->c.un_type != MD_METAMIRROR) {
                return (mdmderror(&iocp->mde, MDE_NOT_MM, mnum));
        }
        if (orignode < 1 || orignode >= MD_MNMAXSIDES) {
                return (mdmderror(&iocp->mde, MDE_INVAL_UNIT, mnum));
        }
        if (un->un_nsm < 2) {
                return (0);
        }

        /*
         * Only process this message if we're the owner of the mirror.
         */
        if (!MD_MN_MIRROR_OWNER(un)) {
                return (0);
        }

        RR_TO_BLK(startblk, start, un);
        RR_TO_BLK(endblk, end, un);
        return (mirror_mark_resync_region_owner(un, startblk, endblk,
            orignode));
}

/*
 * mirror_clean_rr_bits:
 * --------------------
 * Clear the pernode_dirty_bm[node] entries which are passed in the bitmap
 * Once _all_ references are removed (pernode_dirty_count[x] == 0) this region
 * is 'cleanable' and will get flushed out by clearing un_dirty_bm[] on all
 * nodes. Callable from ioctl / interrupt / whatever context.
 * un_resync_mx is held on entry.
 */
static void
mirror_clean_rr_bits(
        md_mn_rr_clean_params_t *iocp)
{
        minor_t                 mnum = iocp->rr_mnum;
        mm_unit_t               *un;
        uint_t                  cleared_bits;
        md_mn_nodeid_t          node = iocp->rr_nodeid - 1;
        md_mn_nodeid_t          orignode = iocp->rr_nodeid;
        int                     i, start, end;

        un = mirror_getun(mnum, &iocp->mde, NO_LOCK, NULL);

        cleared_bits = 0;
        start = MDMN_RR_CLEAN_PARAMS_START_BIT(iocp);
        end = start + MDMN_RR_CLEAN_PARAMS_DATA_BYTES(iocp) * NBBY;
        rw_enter(&un->un_pernode_dirty_mx[node], RW_READER);
        for (i = start; i < end; i++) {
                if (isset(MDMN_RR_CLEAN_PARAMS_DATA(iocp), i - start)) {
                        if (IS_PERNODE_DIRTY(orignode, i, un)) {
                                un->un_pernode_dirty_sum[i]--;
                                CLR_PERNODE_DIRTY(orignode, i, un);
                        }
                        if (un->un_pernode_dirty_sum[i] == 0) {
                                cleared_bits++;
                                CLR_REGION_DIRTY(i, un);
                                CLR_GOING_CLEAN(i, un);
                        }
                }
        }
        rw_exit(&un->un_pernode_dirty_mx[node]);
        if (cleared_bits) {
                /*
                 * We can only be called iff we are the mirror owner, however
                 * as this is a (potentially) decoupled routine the ownership
                 * may have moved from us by the time we get to execute the
                 * bit clearing. Hence we still need to check for being the
                 * owner before flushing the DRL to the replica.
                 */
                if (MD_MN_MIRROR_OWNER(un)) {
                        mutex_exit(&un->un_resync_mx);
                        mddb_commitrec_wrapper(un->un_rr_dirty_recid);
                        mutex_enter(&un->un_resync_mx);
                }
        }
}

/*
 * mirror_drl_task:
 * ---------------
 * Service routine for clearing the DRL bits on a deferred MD_MN_RR_CLEAN call
 * We need to obtain exclusive access to the un_resync_cv and then clear the
 * necessary bits.
 * On completion, we must also free the passed in argument as it is allocated
 * at the end of the ioctl handler and won't be freed on completion.
 */
static void
mirror_drl_task(void *arg)
{
        md_mn_rr_clean_params_t *iocp = (md_mn_rr_clean_params_t *)arg;
        minor_t                 mnum = iocp->rr_mnum;
        mm_unit_t               *un;

        un = mirror_getun(mnum, &iocp->mde, NO_LOCK, NULL);

        mutex_enter(&un->un_rrp_inflight_mx);
        mutex_enter(&un->un_resync_mx);
        un->un_waiting_to_clear++;
        while (un->un_resync_flg & MM_RF_STALL_CLEAN)
                cv_wait(&un->un_resync_cv, &un->un_resync_mx);
        un->un_waiting_to_clear--;

        un->un_resync_flg |= MM_RF_GATECLOSED;
        mirror_clean_rr_bits(iocp);
        un->un_resync_flg &= ~MM_RF_GATECLOSED;
        if (un->un_waiting_to_mark != 0 || un->un_waiting_to_clear != 0) {
                cv_broadcast(&un->un_resync_cv);
        }
        mutex_exit(&un->un_resync_mx);
        mutex_exit(&un->un_rrp_inflight_mx);

        kmem_free((caddr_t)iocp, MDMN_RR_CLEAN_PARAMS_SIZE(iocp));
}

/*
 * mirror_set_clean_rr:
 * -------------------
 * Clear the pernode_dirty_bm[node] entries which are passed in the bitmap
 * Once _all_ references are removed (pernode_dirty_count[x] == 0) this region
 * is 'cleanable' and will get flushed out by clearing un_dirty_bm[] on all
 * nodes.
 *
 * Only the mirror-owner need process this message as it is the only RR updater.
 * Non-owner nodes issue this request, but as we have no point-to-point message
 * support we will receive the message on all nodes.
 */
int
mirror_set_clean_rr(md_mn_rr_clean_params_t *iocp)
{

        minor_t                 mnum = iocp->rr_mnum;
        mm_unit_t               *un;
        set_t                   setno = MD_MIN2SET(mnum);
        md_mn_nodeid_t          node = iocp->rr_nodeid - 1;
        int                     can_clear = 0;
        md_mn_rr_clean_params_t *newiocp;
        int                     rval = 0;

        mdclrerror(&iocp->mde);

        if ((setno >= md_nsets) ||
            (MD_MIN2UNIT(mnum) >= md_nunits)) {
                return (mdmderror(&iocp->mde, MDE_INVAL_UNIT, mnum));
        }

        /* Must have _NO_ ioctl lock set if we update the RR on-disk */
        un = mirror_getun(mnum, &iocp->mde, NO_LOCK, NULL);

        if (un == NULL) {
                return (mdmderror(&iocp->mde, MDE_UNIT_NOT_SETUP, mnum));
        }
        if (un->c.un_type != MD_METAMIRROR) {
                return (mdmderror(&iocp->mde, MDE_NOT_MM, mnum));
        }
        if (un->un_nsm < 2) {
                return (0);
        }

        /*
         * Check to see if we're the mirror owner. If not, there's nothing
         * for us to to.
         */
        if (!MD_MN_MIRROR_OWNER(un)) {
                return (0);
        }

        /*
         * Process the to-be-cleaned bitmap. We need to update the pernode_dirty
         * bits and pernode_dirty_sum[n], and if, and only if, the sum goes 0
         * we can then mark the un_dirty_bm entry as GOINGCLEAN. Alternatively
         * we can just defer this cleaning until the next process_resync_regions
         * timeout.
         */
        rw_enter(&un->un_pernode_dirty_mx[node], RW_WRITER);
        if (un->un_pernode_dirty_bm[node] == NULL) {
                un->un_pernode_dirty_bm[node] = (uchar_t *)kmem_zalloc(
                    howmany(un->un_rrd_num, NBBY), KM_SLEEP);
        }
        rw_exit(&un->un_pernode_dirty_mx[node]);

        /*
         * See if we can simply clear the un_dirty_bm[] entries. If we're not
         * the issuing node _and_ we aren't in the process of marking/clearing
         * the RR bitmaps, we can simply update the bits as needed.
         * If we're the owning node and _not_ the issuing node, we should also
         * sync the RR if we clear any bits in it.
         */
        mutex_enter(&un->un_resync_mx);
        can_clear = (un->un_resync_flg & MM_RF_STALL_CLEAN) ? 0 : 1;
        if (can_clear) {
                un->un_resync_flg |= MM_RF_GATECLOSED;
                mirror_clean_rr_bits(iocp);
                un->un_resync_flg &= ~MM_RF_GATECLOSED;
                if (un->un_waiting_to_mark != 0 ||
                    un->un_waiting_to_clear != 0) {
                        cv_broadcast(&un->un_resync_cv);
                }
        }
        mutex_exit(&un->un_resync_mx);

        /*
         * If we couldn't clear the bits, due to DRL update from m_m_r_r / p_r_r
         * we must schedule a blocking call to update the DRL on this node.
         * As we're invoked from an ioctl we are going to have the original data
         * disappear (kmem_free) once we return. So, copy the data into a new
         * structure and let the taskq routine release it on completion.
         */
        if (!can_clear) {
                size_t  sz = MDMN_RR_CLEAN_PARAMS_SIZE(iocp);

                newiocp = (md_mn_rr_clean_params_t *)kmem_alloc(sz, KM_SLEEP);

                bcopy(iocp, newiocp, sz);

                if (ddi_taskq_dispatch(un->un_drl_task, mirror_drl_task,
                    newiocp, DDI_NOSLEEP) != DDI_SUCCESS) {
                        kmem_free(newiocp, sz);
                        rval = ENOMEM;  /* probably starvation */
                }
        }

        return (rval);
}