/*
 * Copyright (c) 2015-2018 The DragonFly Project.  All rights reserved.
 *
 * This code is derived from software contributed to The DragonFly Project
 * by Matthew Dillon <dillon@dragonflybsd.org>
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 * 3. Neither the name of The DragonFly Project nor the names of its
 *    contributors may be used to endorse or promote products derived
 *    from this software without specific, prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
 * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
 * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
 * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
/*
 * This module implements the cluster synchronizer.  Basically the way
 * it works is that a thread is created for each cluster node in a PFS.
 * This thread is responsible for synchronizing the current node using
 * data from other nodes.
 *
 * Any out of sync master or slave can get back into synchronization as
 * long as a quorum of masters agree on the update_tid.  If a quorum is
 * not available it may still be possible to synchronize to the highest
 * available update_tid as a way of trying to catch up as much as possible
 * until a quorum is available.
 *
 * If no quorum is possible (which can happen even if all masters are
 * available, if the update_tid does not match), then manual intervention
 * may be required to resolve discrepancies.
 */
#include "hammer2.h"

typedef struct hammer2_deferred_ip {
        struct hammer2_deferred_ip *next;
        hammer2_inode_t *ip;
} hammer2_deferred_ip_t;

typedef struct hammer2_deferred_list {
        hammer2_deferred_ip_t   *base;
        int                     count;
} hammer2_deferred_list_t;


#define HAMMER2_SYNCHRO_DEBUG 1

static int hammer2_sync_slaves(hammer2_thread_t *thr, hammer2_inode_t *ip,
                                hammer2_deferred_list_t *list, int isroot);
#if 0
static void hammer2_update_pfs_status(hammer2_thread_t *thr, uint32_t flags);
                                nerror = hammer2_sync_insert(
                                                thr, &parent, &chain,
                                                focus->bref.modify_tid,
                                                idx, focus);
#endif
static int hammer2_sync_insert(hammer2_thread_t *thr,
                        hammer2_chain_t **parentp, hammer2_chain_t **chainp,
                        hammer2_tid_t modify_tid, int idx,
                        hammer2_xop_head_t *xop, hammer2_chain_t *focus);
static int hammer2_sync_destroy(hammer2_thread_t *thr,
                        hammer2_chain_t **parentp, hammer2_chain_t **chainp,
                        hammer2_tid_t mtid, int idx);
static int hammer2_sync_replace(hammer2_thread_t *thr,
                        hammer2_chain_t *parent, hammer2_chain_t *chain,
                        hammer2_tid_t mtid, int idx,
                        hammer2_xop_head_t *xop, hammer2_chain_t *focus,
                        int isroot);

/****************************************************************************
 *                          HAMMER2 SYNC THREADS                            *
 ****************************************************************************/
/*
 * Primary management thread for an element of a node.  A thread will exist
 * for each element requiring management.
 *
 * No management threads are needed for the SPMP or for any PMP with only
 * a single MASTER.
 *
 * On the SPMP - handles bulkfree and dedup operations
 * On a PFS    - handles remastering and synchronization
 */
void
hammer2_primary_sync_thread(void *arg)
{
        hammer2_thread_t *thr = arg;
        hammer2_pfs_t *pmp;
        hammer2_deferred_list_t list;
        hammer2_deferred_ip_t *defer;
        int error;
        uint32_t flags;
        uint32_t nflags;

        pmp = thr->pmp;
        bzero(&list, sizeof(list));

        for (;;) {
                flags = thr->flags;
                cpu_ccfence();

                /*
                 * Handle stop request
                 */
                if (flags & HAMMER2_THREAD_STOP)
                        break;

                /*
                 * Handle freeze request
                 */
                if (flags & HAMMER2_THREAD_FREEZE) {
                        nflags = (flags & ~(HAMMER2_THREAD_FREEZE |
                                            HAMMER2_THREAD_WAITING)) |
                                 HAMMER2_THREAD_FROZEN;
                        if (!atomic_cmpset_int(&thr->flags, flags, nflags))
                                continue;
                        if (flags & HAMMER2_THREAD_WAITING)
                                wakeup(&thr->flags);
                        continue;
                }

                if (flags & HAMMER2_THREAD_UNFREEZE) {
                        nflags = flags & ~(HAMMER2_THREAD_UNFREEZE |
                                           HAMMER2_THREAD_FROZEN |
                                           HAMMER2_THREAD_WAITING);
                        if (!atomic_cmpset_int(&thr->flags, flags, nflags))
                                continue;
                        if (flags & HAMMER2_THREAD_WAITING)
                                wakeup(&thr->flags);
                        continue;
                }

                /*
                 * Force idle if frozen until unfrozen or stopped.
                 */
                if (flags & HAMMER2_THREAD_FROZEN) {
                        nflags = flags | HAMMER2_THREAD_WAITING;

                        tsleep_interlock(&thr->flags, 0);
                        if (atomic_cmpset_int(&thr->flags, flags, nflags))
                                tsleep(&thr->flags, PINTERLOCKED, "frozen", 0);
                        continue;
                }

                /*
                 * Reset state on REMASTER request
                 */
                if (thr->flags & HAMMER2_THREAD_REMASTER) {
                        nflags = flags & ~HAMMER2_THREAD_REMASTER;
                        if (atomic_cmpset_int(&thr->flags, flags, nflags)) {
                                /* reset state here */
                        }
                        continue;
                }

                /*
                 * Synchronization scan.
                 */
                if (hammer2_debug & 0x8000)
                        kprintf("sync_slaves pfs %s clindex %d\n",
                                pmp->pfs_names[thr->clindex], thr->clindex);
                hammer2_trans_init(pmp, 0);

                hammer2_inode_ref(pmp->iroot);

                for (;;) {
                        int didbreak = 0;
                        /* XXX lock synchronize pmp->modify_tid */
                        error = hammer2_sync_slaves(thr, pmp->iroot, &list, 1);
                        if (hammer2_debug & 0x8000) {
                                kprintf("sync_slaves error %d defer %p\n",
                                        error, list.base);
                        }
                        if (error != HAMMER2_ERROR_EAGAIN)
                                break;
                        while ((defer = list.base) != NULL) {
                                hammer2_inode_t *nip;

                                nip = defer->ip;
                                error = hammer2_sync_slaves(thr, nip, &list,
                                                        (nip == pmp->iroot));
                                if (error &&
                                    error != HAMMER2_ERROR_EAGAIN &&
                                    error != HAMMER2_ERROR_ENOENT) {
                                        break;
                                }
                                if (hammer2_thr_break(thr)) {
                                        didbreak = 1;
                                        break;
                                }

                                /*
                                 * If no additional defers occurred we can
                                 * remove this one, otherwise keep it on
                                 * the list and retry once the additional
                                 * defers have completed.
                                 */
                                if (defer == list.base) {
                                        --list.count;
                                        list.base = defer->next;
                                        kfree(defer, M_HAMMER2);
                                        defer = NULL;   /* safety */
                                        hammer2_inode_drop(nip);
                                }
                        }

                        /*
                         * If the thread is being remastered, frozen, or
                         * stopped, clean up any left-over deferals.
                         */
                        if (didbreak ||
                            (error && error != HAMMER2_ERROR_EAGAIN)) {
                                kprintf("didbreak\n");
                                while ((defer = list.base) != NULL) {
                                        --list.count;
                                        hammer2_inode_drop(defer->ip);
                                        list.base = defer->next;
                                        kfree(defer, M_HAMMER2);
                                }
                                if (error == 0 || error == HAMMER2_ERROR_EAGAIN)
                                        error = HAMMER2_ERROR_EINPROGRESS;
                                break;
                        }
                }

                hammer2_inode_drop(pmp->iroot);
                hammer2_trans_done(pmp, 0);

                if (error && error != HAMMER2_ERROR_EINPROGRESS)
                        kprintf("hammer2_sync_slaves: error %d\n", error);

                /*
                 * Wait for event, or 5-second poll.
                 */
                nflags = flags | HAMMER2_THREAD_WAITING;
                tsleep_interlock(&thr->flags, 0);
                if (atomic_cmpset_int(&thr->flags, flags, nflags)) {
                        tsleep(&thr->flags, 0, "h2idle", hz * 5);
                }
        }
        thr->td = NULL;
        hammer2_thr_signal(thr, HAMMER2_THREAD_STOPPED);
        /* thr structure can go invalid after this point */
}

#if 0
/*
 * Given a locked cluster created from pmp->iroot, update the PFS's
 * reporting status.
 */
static
void
hammer2_update_pfs_status(hammer2_thread_t *thr, uint32_t flags)
{
        hammer2_pfs_t *pmp = thr->pmp;

        flags &= HAMMER2_CLUSTER_ZFLAGS;
        if (pmp->cluster_flags == flags)
                return;
        pmp->cluster_flags = flags;

        kprintf("pfs %p", pmp);
        if (flags & HAMMER2_CLUSTER_MSYNCED)
                kprintf(" masters-all-good");
        if (flags & HAMMER2_CLUSTER_SSYNCED)
                kprintf(" slaves-all-good");

        if (flags & HAMMER2_CLUSTER_WRHARD)
                kprintf(" quorum/rw");
        else if (flags & HAMMER2_CLUSTER_RDHARD)
                kprintf(" quorum/ro");

        if (flags & HAMMER2_CLUSTER_UNHARD)
                kprintf(" out-of-sync-masters");
        else if (flags & HAMMER2_CLUSTER_NOHARD)
                kprintf(" no-masters-visible");

        if (flags & HAMMER2_CLUSTER_WRSOFT)
                kprintf(" soft/rw");
        else if (flags & HAMMER2_CLUSTER_RDSOFT)
                kprintf(" soft/ro");

        if (flags & HAMMER2_CLUSTER_UNSOFT)
                kprintf(" out-of-sync-slaves");
        else if (flags & HAMMER2_CLUSTER_NOSOFT)
                kprintf(" no-slaves-visible");
        kprintf("\n");
}
#endif

#if 0
static
void
dumpcluster(const char *label,
            hammer2_cluster_t *cparent, hammer2_cluster_t *cluster)
{
        hammer2_chain_t *chain;
        int i;

        if ((hammer2_debug & 1) == 0)
                return;

        kprintf("%s\t", label);
        KKASSERT(cparent->nchains == cluster->nchains);
        for (i = 0; i < cparent->nchains; ++i) {
                if (i)
                        kprintf("\t");
                kprintf("%d ", i);
                if ((chain = cparent->array[i].chain) != NULL) {
                        kprintf("%016jx%s ",
                                chain->bref.key,
                                ((cparent->array[i].flags &
                                  HAMMER2_CITEM_INVALID) ? "(I)" : "   ")
                        );
                } else {
                        kprintf("      NULL      %s ", "   ");
                }
                if ((chain = cluster->array[i].chain) != NULL) {
                        kprintf("%016jx%s ",
                                chain->bref.key,
                                ((cluster->array[i].flags &
                                  HAMMER2_CITEM_INVALID) ? "(I)" : "   ")
                        );
                } else {
                        kprintf("      NULL      %s ", "   ");
                }
                kprintf("\n");
        }
}
#endif

/*
 * Each out of sync node sync-thread must issue an all-nodes XOP scan of
 * the inode.  This creates a multiplication effect since the XOP scan itself
 * issues to all nodes.  However, this is the only way we can safely
 * synchronize nodes which might have disparate I/O bandwidths and the only
 * way we can safely deal with stalled nodes.
 *
 * XXX serror / merror rollup and handling.
 */
static
int
hammer2_sync_slaves(hammer2_thread_t *thr, hammer2_inode_t *ip,
                    hammer2_deferred_list_t *list, int isroot)
{
        hammer2_xop_scanall_t *xop;
        hammer2_chain_t *parent;
        hammer2_chain_t *chain;
        hammer2_pfs_t *pmp;
        hammer2_key_t key_next;
        hammer2_tid_t sync_tid;
        int needrescan;
        int want_update;
        int serror;             /* slave error */
        int merror;             /* master error (from xop_collect) */
        int nerror;             /* temporary error */
        int idx;
        int n;

        pmp = ip->pmp;
        idx = thr->clindex;     /* cluster node we are responsible for */
        needrescan = 0;
        want_update = 0;
        sync_tid = 0;
        chain = NULL;
        parent = NULL;

#if 0
        /*
         * Nothing to do if all slaves are synchronized.
         * Nothing to do if cluster not authoritatively readable.
         */
        if (pmp->cluster_flags & HAMMER2_CLUSTER_SSYNCED)
                return(0);
        if ((pmp->cluster_flags & HAMMER2_CLUSTER_RDHARD) == 0)
                return(HAMMER2_ERROR_INCOMPLETE);
#endif

        merror = 0;

        /*
         * Resolve the root inode of the PFS and determine if synchronization
         * is needed by checking modify_tid.
         *
         * Retain the synchronization TID from the focus inode and use it
         * later to synchronize the focus inode if/when the recursion
         * succeeds.
         */
        {
                hammer2_xop_ipcluster_t *xop2;
                hammer2_chain_t *focus;

                hammer2_inode_lock(ip, HAMMER2_RESOLVE_SHARED);
                xop2 = hammer2_xop_alloc(ip, HAMMER2_XOP_MODIFYING);
                hammer2_xop_start_except(&xop2->head, &hammer2_ipcluster_desc,
                                         idx);
                hammer2_inode_unlock(ip);
                merror = hammer2_xop_collect(&xop2->head, 0);
                if (merror == 0 && (focus = xop2->head.cluster.focus) != NULL) {
                        sync_tid = focus->bref.modify_tid;
                        chain = hammer2_inode_chain_and_parent(ip, idx,
                                                    &parent,
                                                    HAMMER2_RESOLVE_ALWAYS |
                                                    HAMMER2_RESOLVE_SHARED);
                        want_update = (chain->bref.modify_tid != sync_tid);
                        if (chain) {
                                hammer2_chain_unlock(chain);
                                hammer2_chain_drop(chain);
                                chain = NULL;
                        }
                        if (parent) {
                                hammer2_chain_unlock(parent);
                                hammer2_chain_drop(parent);
                                parent = NULL;
                        }
                }
                hammer2_xop_retire(&xop2->head, HAMMER2_XOPMASK_VOP);
        }

        if (want_update == 0)
                return(0);

        /*
         * The inode is left unlocked during the scan.  Issue a XOP
         * that does *not* include our cluster index to iterate
         * properly synchronized elements and resolve our cluster index
         * against it.
         */
        hammer2_inode_lock(ip, HAMMER2_RESOLVE_SHARED);
        xop = hammer2_xop_alloc(ip, HAMMER2_XOP_MODIFYING);
        xop->key_beg = HAMMER2_KEY_MIN;
        xop->key_end = HAMMER2_KEY_MAX;
        xop->resolve_flags = HAMMER2_RESOLVE_SHARED |
                             HAMMER2_RESOLVE_ALWAYS;
        xop->lookup_flags = HAMMER2_LOOKUP_SHARED |
                            HAMMER2_LOOKUP_NODIRECT |
                            HAMMER2_LOOKUP_ALWAYS;
        hammer2_xop_start_except(&xop->head, &hammer2_scanall_desc, idx);
        parent = hammer2_inode_chain(ip, idx,
                                     HAMMER2_RESOLVE_ALWAYS |
                                     HAMMER2_RESOLVE_SHARED);
        hammer2_inode_unlock(ip);

        chain = hammer2_chain_lookup(&parent, &key_next,
                                     HAMMER2_KEY_MIN, HAMMER2_KEY_MAX,
                                     &serror,
                                     HAMMER2_LOOKUP_SHARED |
                                     HAMMER2_LOOKUP_NODIRECT |
                                     HAMMER2_LOOKUP_NODATA);
        merror = hammer2_xop_collect(&xop->head, 0);
        if (hammer2_debug & 0x8000) {
                kprintf("START_SCAN IP=%016jx chain=%p (%016jx)\n",
                        ip->meta.name_key, chain,
                        (chain ? chain->bref.key : -1));
        }

        for (;;) {
                /*
                 * We are done if our scan is done and the XOP scan is done.
                 * We are done if the XOP scan failed (that is, we don't
                 * have authoritative data to synchronize with).
                 */
                int advance_local = 0;
                int advance_xop = 0;
                int dodefer = 0;
                hammer2_chain_t *focus;

                if (chain == NULL && merror == HAMMER2_ERROR_ENOENT)
                        break;
                if (merror && merror != HAMMER2_ERROR_ENOENT)
                        break;

                /*
                 * Compare
                 */
                if (chain && merror == HAMMER2_ERROR_ENOENT) {
                        /*
                         * If we have local chains but the XOP scan is done,
                         * the chains need to be deleted.
                         */
                        n = -1;
                        focus = NULL;
                } else if (chain == NULL) {
                        /*
                         * If our local scan is done but the XOP scan is not,
                         * we need to create the missing chain(s).
                         */
                        n = 1;
                        focus = xop->head.cluster.focus;
                } else {
                        /*
                         * Otherwise compare to determine the action
                         * needed.
                         */
                        focus = xop->head.cluster.focus;
                        n = hammer2_chain_cmp(chain, focus);
                }

                /*
                 * Take action based on comparison results.
                 */
                if (n < 0) {
                        /*
                         * Delete extranious local data.  This will
                         * automatically advance the chain.
                         */
                        nerror = hammer2_sync_destroy(thr, &parent, &chain,
                                                      0, idx);
                } else if (n == 0 && chain->bref.modify_tid !=
                                     focus->bref.modify_tid) {
                        /*
                         * Matching key but local data or meta-data requires
                         * updating.  If we will recurse, we still need to
                         * update to compatible content first but we do not
                         * synchronize modify_tid until the entire recursion
                         * has completed successfully.
                         */
                        if (focus->bref.type == HAMMER2_BREF_TYPE_INODE) {
                                nerror = hammer2_sync_replace(
                                                thr, parent, chain,
                                                0,
                                                idx, &xop->head, focus, 0);
                                dodefer = 1;
                        } else {
                                nerror = hammer2_sync_replace(
                                                thr, parent, chain,
                                                focus->bref.modify_tid,
                                                idx, &xop->head, focus, 0);
                        }
                        advance_local = 1;
                        advance_xop = 1;
                } else if (n == 0) {
                        /*
                         * 100% match, advance both
                         */
                        advance_local = 1;
                        advance_xop = 1;
                        nerror = 0;
                } else if (n > 0) {
                        /*
                         * Insert missing local data.
                         *
                         * If we will recurse, we still need to update to
                         * compatible content first but we do not synchronize
                         * modify_tid until the entire recursion has
                         * completed successfully.
                         */
                        if (focus->bref.type == HAMMER2_BREF_TYPE_INODE) {
                                nerror = hammer2_sync_insert(
                                                thr, &parent, &chain,
                                                0,
                                                idx, &xop->head, focus);
                                dodefer = 2;
                        } else {
                                nerror = hammer2_sync_insert(
                                                thr, &parent, &chain,
                                                focus->bref.modify_tid,
                                                idx, &xop->head, focus);
                        }
                        advance_local = 1;
                        advance_xop = 1;
                }

                /*
                 * We cannot recurse depth-first because the XOP is still
                 * running in node threads for this scan.  Create a placemarker
                 * by obtaining and record the hammer2_inode.
                 *
                 * We excluded our node from the XOP so we must temporarily
                 * add it to xop->head.cluster so it is properly incorporated
                 * into the inode.
                 *
                 * The deferral is pushed onto a LIFO list for bottom-up
                 * synchronization.
                 */
                if (merror == 0 && dodefer) {
                        hammer2_inode_t *nip;
                        hammer2_deferred_ip_t *defer;

                        KKASSERT(focus->bref.type == HAMMER2_BREF_TYPE_INODE);

                        defer = kmalloc(sizeof(*defer), M_HAMMER2,
                                        M_WAITOK | M_ZERO);
                        KKASSERT(xop->head.cluster.array[idx].chain == NULL);
                        xop->head.cluster.array[idx].flags =
                                                        HAMMER2_CITEM_INVALID;
                        xop->head.cluster.array[idx].chain = chain;
                        nip = hammer2_inode_get(pmp, &xop->head, -1, idx);
                        xop->head.cluster.array[idx].chain = NULL;

                        hammer2_inode_ref(nip);
                        hammer2_inode_unlock(nip);

                        defer->next = list->base;
                        defer->ip = nip;
                        list->base = defer;
                        ++list->count;
                        needrescan = 1;
                }

                /*
                 * If at least one deferral was added and the deferral
                 * list has grown too large, stop adding more.  This
                 * will trigger an HAMMER2_ERROR_EAGAIN return.
                 */
                if (needrescan && list->count > 1000)
                        break;

                /*
                 * Advancements for iteration.
                 */
                if (advance_xop) {
                        merror = hammer2_xop_collect(&xop->head, 0);
                }
                if (advance_local) {
                        chain = hammer2_chain_next(&parent, chain, &key_next,
                                                   key_next, HAMMER2_KEY_MAX,
                                                   &serror,
                                                   HAMMER2_LOOKUP_SHARED |
                                                   HAMMER2_LOOKUP_NODIRECT |
                                                   HAMMER2_LOOKUP_NODATA);
                }
        }
        hammer2_xop_retire(&xop->head, HAMMER2_XOPMASK_VOP);
        if (chain) {
                hammer2_chain_unlock(chain);
                hammer2_chain_drop(chain);
        }
        if (parent) {
                hammer2_chain_unlock(parent);
                hammer2_chain_drop(parent);
        }

        /*
         * If we added deferrals we want the caller to synchronize them
         * and then call us again.
         *
         * NOTE: In this situation we do not yet want to synchronize our
         *       inode, setting the error code also has that effect.
         */
        if ((merror == 0 || merror == HAMMER2_ERROR_ENOENT) && needrescan)
                merror = HAMMER2_ERROR_EAGAIN;

        /*
         * If no error occurred we can synchronize the inode meta-data
         * and modify_tid.  Only limited changes are made to PFSROOTs.
         *
         * XXX inode lock was lost
         */
        if (merror == 0 || merror == HAMMER2_ERROR_ENOENT) {
                hammer2_xop_ipcluster_t *xop2;
                hammer2_chain_t *focus;

                hammer2_inode_lock(ip, HAMMER2_RESOLVE_SHARED);
                xop2 = hammer2_xop_alloc(ip, HAMMER2_XOP_MODIFYING);
                hammer2_xop_start_except(&xop2->head, &hammer2_ipcluster_desc,
                                         idx);
                hammer2_inode_unlock(ip);
                merror = hammer2_xop_collect(&xop2->head, 0);
                if (merror == 0) {
                        focus = xop2->head.cluster.focus;
                        if ((hammer2_debug & 0x8000) && focus) {
                                const char *filename;

                                filename = hammer2_xop_gdata(&xop2->head)->
                                                ipdata.filename;
                                kprintf("syncthr: update inode %p (%s)\n",
                                        focus, filename);
                                hammer2_xop_pdata(&xop2->head);
                        }
                        chain = hammer2_inode_chain_and_parent(ip, idx,
                                                    &parent,
                                                    HAMMER2_RESOLVE_ALWAYS |
                                                    HAMMER2_RESOLVE_SHARED);

                        KKASSERT(parent != NULL);
                        nerror = hammer2_sync_replace(
                                        thr, parent, chain,
                                        sync_tid,
                                        idx, &xop2->head, focus, isroot);
                        hammer2_chain_unlock(chain);
                        hammer2_chain_drop(chain);
                        hammer2_chain_unlock(parent);
                        hammer2_chain_drop(parent);
                        /* XXX */
                }
                hammer2_xop_retire(&xop2->head, HAMMER2_XOPMASK_VOP);
        }

        return merror;
}

/*
 * Create a missing chain by copying the focus from another device.
 *
 * On entry *parentp and focus are both locked shared.  The chain will be
 * created and returned in *chainp also locked shared.
 */
static
int
hammer2_sync_insert(hammer2_thread_t *thr,
                    hammer2_chain_t **parentp, hammer2_chain_t **chainp,
                    hammer2_tid_t mtid, int idx, hammer2_xop_head_t *xop,
                    hammer2_chain_t *focus)
{
        hammer2_chain_t *chain;
        hammer2_key_t dummy;
        int error;

#if HAMMER2_SYNCHRO_DEBUG
        if (hammer2_debug & 1)
        kprintf("insert rec par=%p/%d.%016jx slave %d %d.%016jx mod=%016jx\n",
                *parentp,
                (*parentp)->bref.type,
                (*parentp)->bref.key,
                idx,
                focus->bref.type, focus->bref.key, mtid);
#endif

        /*
         * Parent requires an exclusive lock for the insertion.
         * We must unlock the child to avoid deadlocks while
         * relocking the parent.
         */
        if (*chainp) {
                hammer2_chain_unlock(*chainp);
                hammer2_chain_drop(*chainp);
                *chainp = NULL;
        }
        hammer2_chain_unlock(*parentp);
        hammer2_chain_lock(*parentp, HAMMER2_RESOLVE_ALWAYS);

        /*
         * We must reissue the lookup to properly position (*parentp)
         * for the insertion.
         */
        chain = hammer2_chain_lookup(parentp, &dummy,
                                     focus->bref.key, focus->bref.key,
                                     &error,
                                     HAMMER2_LOOKUP_NODIRECT |
                                     HAMMER2_LOOKUP_ALWAYS);
        KKASSERT(chain == NULL);

        chain = NULL;
        error = hammer2_chain_create(parentp, &chain, NULL, thr->pmp,
                                     focus->bref.methods,
                                     focus->bref.key, focus->bref.keybits,
                                     focus->bref.type, focus->bytes,
                                     mtid, 0, 0);
        if (error == 0) {
                const hammer2_media_data_t *data;

                error = hammer2_chain_modify(chain, mtid, 0, 0);
                if (error)
                        goto failed;

                /*
                 * Copy focus to new chain
                 */

                /* type already set */
                chain->bref.methods = focus->bref.methods;
                /* keybits already set */
                chain->bref.vradix = focus->bref.vradix;
                /* mirror_tid set by flush */
                KKASSERT(chain->bref.modify_tid == mtid);
                chain->bref.flags = focus->bref.flags;
                /* key already present */
                /* check code will be recalculated */

                /*
                 * Copy data body.
                 */
                switch(chain->bref.type) {
                case HAMMER2_BREF_TYPE_INODE:
                        data = hammer2_xop_gdata(xop);

                        if ((data->ipdata.meta.op_flags &
                             HAMMER2_OPFLAG_DIRECTDATA) == 0) {
                                /* do not copy block table */
                                bcopy(data, chain->data,
                                      offsetof(hammer2_inode_data_t, u));
                                hammer2_xop_pdata(xop);
                                break;
                        }
                        hammer2_xop_pdata(xop);
                        /* fall through copy whole thing */
                case HAMMER2_BREF_TYPE_DATA:
                        data = hammer2_xop_gdata(xop);
                        bcopy(data, chain->data, chain->bytes);
                        hammer2_chain_setcheck(chain, chain->data);
                        hammer2_xop_pdata(xop);
                        break;
                case HAMMER2_BREF_TYPE_DIRENT:
                        /*
                         * Directory entries embed data in the blockref.
                         */
                        if (chain->bytes) {
                                data = hammer2_xop_gdata(xop);
                                bcopy(data, chain->data, chain->bytes);
                                hammer2_chain_setcheck(chain, chain->data);
                                hammer2_xop_pdata(xop);
                        } else {
                                chain->bref.check = focus->bref.check;
                        }
                        chain->bref.embed = focus->bref.embed;
                        break;
                default:
                        KKASSERT(0);
                        break;
                }
        }

failed:
        if (chain)
                hammer2_chain_unlock(chain);    /* unlock, leave ref */
        *chainp = chain;                        /* will be returned locked */

        /*
         * Avoid an ordering deadlock when relocking shared.
         */
        hammer2_chain_unlock(*parentp);
        hammer2_chain_lock(*parentp, HAMMER2_RESOLVE_SHARED |
                                     HAMMER2_RESOLVE_ALWAYS);
        if (chain) {
                hammer2_chain_lock(chain, HAMMER2_RESOLVE_SHARED |
                                          HAMMER2_RESOLVE_ALWAYS);
                error = chain->error;
        }

        return error;
}

/*
 * Destroy an extranious chain.
 *
 * Both *parentp and *chainp are locked shared.
 *
 * On return, *chainp will be adjusted to point to the next element in the
 * iteration and locked shared.
 */
static
int
hammer2_sync_destroy(hammer2_thread_t *thr,
                     hammer2_chain_t **parentp, hammer2_chain_t **chainp,
                     hammer2_tid_t mtid, int idx)
{
        hammer2_chain_t *chain;
        hammer2_key_t key_next;
        hammer2_key_t save_key;
        int error;

        chain = *chainp;

#if HAMMER2_SYNCHRO_DEBUG
        if (hammer2_debug & 1)
        kprintf("destroy rec %p/%p slave %d %d.%016jx\n",
                *parentp, chain,
                idx, chain->bref.type, chain->bref.key);
#endif

        save_key = chain->bref.key;
        if (save_key != HAMMER2_KEY_MAX)
                ++save_key;

        /*
         * Try to avoid unnecessary I/O.
         *
         * XXX accounting not propagated up properly.  We might have to do
         *     a RESOLVE_MAYBE here and pass 0 for the flags.
         */
        hammer2_chain_unlock(chain);    /* relock exclusive */
        hammer2_chain_unlock(*parentp);
        hammer2_chain_lock(*parentp, HAMMER2_RESOLVE_ALWAYS);
        hammer2_chain_lock(chain, HAMMER2_RESOLVE_NEVER);

        hammer2_chain_delete(*parentp, chain, mtid, HAMMER2_DELETE_PERMANENT);
        hammer2_chain_unlock(chain);
        hammer2_chain_drop(chain);
        chain = NULL;                   /* safety */

        hammer2_chain_unlock(*parentp); /* relock shared */
        hammer2_chain_lock(*parentp, HAMMER2_RESOLVE_SHARED |
                                     HAMMER2_RESOLVE_ALWAYS);
        *chainp = hammer2_chain_lookup(parentp, &key_next,
                                     save_key, HAMMER2_KEY_MAX,
                                     &error,
                                     HAMMER2_LOOKUP_SHARED |
                                     HAMMER2_LOOKUP_NODIRECT |
                                     HAMMER2_LOOKUP_NODATA);
        return error;
}

/*
 * cparent is locked exclusively, with an extra ref, cluster is not locked.
 * Replace element [i] in the cluster.
 */
static
int
hammer2_sync_replace(hammer2_thread_t *thr,
                     hammer2_chain_t *parent, hammer2_chain_t *chain,
                     hammer2_tid_t mtid, int idx,
                     hammer2_xop_head_t *xop, hammer2_chain_t *focus,
                     int isroot)
{
        uint8_t otype;
        int nradix;
        int error;

#if HAMMER2_SYNCHRO_DEBUG
        if (hammer2_debug & 1)
        kprintf("replace rec %p slave %d %d.%016jx mod=%016jx\n",
                chain,
                idx,
                focus->bref.type, focus->bref.key, mtid);
#endif
        hammer2_chain_unlock(chain);
        hammer2_chain_lock(chain, HAMMER2_RESOLVE_ALWAYS);
        error = chain->error;
        if (error == 0) {
                const hammer2_media_data_t *data;

                if (chain->bytes != focus->bytes) {
                        /* XXX what if compressed? */
                        nradix = hammer2_getradix(chain->bytes);
                        error = hammer2_chain_resize(chain, mtid, 0, nradix, 0);
                        if (error)
                                goto failed;
                }
                error = hammer2_chain_modify(chain, mtid, 0, 0);
                if (error)
                        goto failed;
                otype = chain->bref.type;
                data = hammer2_xop_gdata(xop);
                chain->bref.type = focus->bref.type;
                chain->bref.methods = focus->bref.methods;
                chain->bref.keybits = focus->bref.keybits;
                chain->bref.vradix = focus->bref.vradix;
                /* mirror_tid updated by flush */
                KKASSERT(mtid == 0 || chain->bref.modify_tid == mtid);
                chain->bref.flags = focus->bref.flags;
                /* key already present */
                /* check code will be recalculated */

                /*
                 * Copy data body.
                 */
                switch(chain->bref.type) {
                case HAMMER2_BREF_TYPE_INODE:
                        /*
                         * Special case PFSROOTs, only limited changes can
                         * be made since the meta-data contains miscellanious
                         * distinguishing fields.
                         */
                        if (isroot) {
                                chain->data->ipdata.meta.uflags =
                                        data->ipdata.meta.uflags;
                                chain->data->ipdata.meta.rmajor =
                                        data->ipdata.meta.rmajor;
                                chain->data->ipdata.meta.rminor =
                                        data->ipdata.meta.rminor;
                                chain->data->ipdata.meta.ctime =
                                        data->ipdata.meta.ctime;
                                chain->data->ipdata.meta.mtime =
                                        data->ipdata.meta.mtime;
                                chain->data->ipdata.meta.atime =
                                        data->ipdata.meta.atime;
                                /* not btime */
                                chain->data->ipdata.meta.uid =
                                        data->ipdata.meta.uid;
                                chain->data->ipdata.meta.gid =
                                        data->ipdata.meta.gid;
                                chain->data->ipdata.meta.mode =
                                        data->ipdata.meta.mode;
                                chain->data->ipdata.meta.ncopies =
                                        data->ipdata.meta.ncopies;
                                chain->data->ipdata.meta.comp_algo =
                                        data->ipdata.meta.comp_algo;
                                chain->data->ipdata.meta.check_algo =
                                        data->ipdata.meta.check_algo;
                                chain->data->ipdata.meta.data_quota =
                                        data->ipdata.meta.data_quota;
                                chain->data->ipdata.meta.inode_quota =
                                        data->ipdata.meta.inode_quota;

                                /*
                                 * last snapshot tid controls overwrite
                                 */
                                if (chain->data->ipdata.meta.pfs_lsnap_tid <
                                    data->ipdata.meta.pfs_lsnap_tid) {
                                        chain->data->ipdata.meta.pfs_lsnap_tid =
                                        data->ipdata.meta.pfs_lsnap_tid;
                                }

                                hammer2_chain_setcheck(chain, chain->data);
                                break;
                        }

                        /*
                         * Normal replacement.
                         */
                        if ((data->ipdata.meta.op_flags &
                             HAMMER2_OPFLAG_DIRECTDATA) == 0) {
                                /*
                                 * If DIRECTDATA is transitioning to 0 or the
                                 * old chain is not an inode we have to
                                 * initialize the block table.
                                 */
                                if (otype != HAMMER2_BREF_TYPE_INODE ||
                                    (chain->data->ipdata.meta.op_flags &
                                     HAMMER2_OPFLAG_DIRECTDATA)) {
                                        kprintf("chain inode trans "
                                                "away from dd\n");
                                        bzero(&chain->data->ipdata.u,
                                              sizeof(chain->data->ipdata.u));
                                }
                                bcopy(data, chain->data,
                                      offsetof(hammer2_inode_data_t, u));
                                /* XXX setcheck on inode should not be needed */
                                hammer2_chain_setcheck(chain, chain->data);
                                break;
                        }
                        /* fall through */
                case HAMMER2_BREF_TYPE_DATA:
                        bcopy(data, chain->data, chain->bytes);
                        hammer2_chain_setcheck(chain, chain->data);
                        break;
                case HAMMER2_BREF_TYPE_DIRENT:
                        /*
                         * Directory entries embed data in the blockref.
                         */
                        if (chain->bytes) {
                                bcopy(data, chain->data, chain->bytes);
                                hammer2_chain_setcheck(chain, chain->data);
                        } else {
                                chain->bref.check = focus->bref.check;
                        }
                        chain->bref.embed = focus->bref.embed;
                        break;
                default:
                        KKASSERT(0);
                        break;
                }
                hammer2_xop_pdata(xop);
        }

failed:
        hammer2_chain_unlock(chain);
        hammer2_chain_lock(chain, HAMMER2_RESOLVE_SHARED |
                                  HAMMER2_RESOLVE_MAYBE);

        return error;
}