/*
 * 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 2007 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

/*
 * Copyright (c) 2015 by Delphix. All rights reserved.
 */

#include <sys/systm.h>
#include <sys/types.h>
#include <sys/vnode.h>
#include <sys/errno.h>
#include <sys/sysmacros.h>
#include <sys/debug.h>
#include <sys/kmem.h>
#include <sys/conf.h>
#include <sys/proc.h>
#include <sys/taskq.h>
#include <sys/cmn_err.h>
#include <sys/fs/ufs_inode.h>
#include <sys/fs/ufs_filio.h>
#include <sys/fs/ufs_log.h>
#include <sys/fs/ufs_bio.h>

/*
 * FILE SYSTEM INTERFACE TO TRANSACTION OPERATIONS (TOP; like VOP)
 */

uint_t topkey; /* tsd transaction key */

/*
 * declare a delta
 */
void
top_delta(
        ufsvfs_t *ufsvfsp,
        offset_t mof,
        off_t nb,
        delta_t dtyp,
        int (*func)(),
        ulong_t arg)
{
        ml_unit_t               *ul     = ufsvfsp->vfs_log;
        threadtrans_t           *tp     = tsd_get(topkey);

        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);
        ASSERT(nb);
        ASSERT(((ul->un_debug & (MT_TRANSACT|MT_MATAMAP)) == 0) ||
            top_delta_debug(ul, mof, nb, dtyp));

        deltamap_add(ul->un_deltamap, mof, nb, dtyp, func, arg, tp);

        ul->un_logmap->mtm_ref = 1; /* for roll thread's heuristic */
        if (tp) {
                tp->any_deltas = 1;
        }
}

/*
 * cancel a delta
 */
void
top_cancel(ufsvfs_t *ufsvfsp, offset_t mof, off_t nb, int flags)
{
        ml_unit_t       *ul     = ufsvfsp->vfs_log;
        int             metadata = flags & (I_DIR|I_IBLK|I_SHAD|I_QUOTA);

        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);
        ASSERT(nb);
        ASSERT(((ul->un_debug & (MT_TRANSACT|MT_MATAMAP)) == 0) ||
            (!(flags & metadata) ||
            top_delta_debug(ul, mof, nb, DT_CANCEL)));

        if (metadata)
                deltamap_del(ul->un_deltamap, mof, nb);

        logmap_cancel(ul, mof, nb, metadata);

        /*
         * needed for the roll thread's heuristic
         */
        ul->un_logmap->mtm_ref = 1;
}

/*
 * check if this delta has been canceled (metadata -> userdata)
 */
int
top_iscancel(ufsvfs_t *ufsvfsp, offset_t mof, off_t nb)
{
        ml_unit_t       *ul     = ufsvfsp->vfs_log;

        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);
        ASSERT(nb);
        if (logmap_iscancel(ul->un_logmap, mof, nb))
                return (1);
        if (ul->un_flags & LDL_ERROR)
                return (1);
        return (0);
}

/*
 * put device into error state
 */
void
top_seterror(ufsvfs_t *ufsvfsp)
{
        ml_unit_t       *ul     = ufsvfsp->vfs_log;

        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);
        ldl_seterror(ul, "ufs is forcing a ufs log error");
}

/*
 * issue a empty sync op to help empty the delta/log map or the log
 */
static void
top_issue_sync(ufsvfs_t *ufsvfsp)
{
        int error = 0;

        if ((curthread->t_flag & T_DONTBLOCK) == 0)
                curthread->t_flag |= T_DONTBLOCK;
        top_begin_sync(ufsvfsp, TOP_COMMIT_ASYNC, 0, &error);
        if (!error) {
                top_end_sync(ufsvfsp, &error, TOP_COMMIT_ASYNC, 0);
        }
}

static void
top_issue_from_taskq(void *arg)
{
        ufsvfs_t *ufsvfsp = arg;
        ml_unit_t *ul = ufsvfsp->vfs_log;
        mt_map_t *mtm = ul->un_logmap;

        top_issue_sync(ufsvfsp);

        /*
         * We were called from the taskq_dispatch() in top_begin_async(), so
         * decrement mtm_taskq_sync_count and wake up the thread waiting
         * on the mtm_cv if the mtm_taskq_sync_count hits zero.
         */
        ASSERT(taskq_member(system_taskq, curthread));

        mutex_enter(&mtm->mtm_lock);
        mtm->mtm_taskq_sync_count--;
        if (mtm->mtm_taskq_sync_count == 0) {
                cv_signal(&mtm->mtm_cv);
        }
        mutex_exit(&mtm->mtm_lock);
}

/*
 * MOBY TRANSACTION ROUTINES
 * begin a moby transaction
 *      sync ops enter until first sync op finishes
 *      async ops enter until last sync op finishes
 * end a moby transaction
 *              outstanding deltas are pushed thru log
 *              log buffer is committed (incore only)
 *              next trans is open to async ops
 *              log buffer is committed on the log
 *              next trans is open to sync ops
 */

/*ARGSUSED*/
void
top_begin_sync(ufsvfs_t *ufsvfsp, top_t topid, ulong_t size, int *error)
{
        ml_unit_t       *ul     = ufsvfsp->vfs_log;
        mt_map_t        *mtm = ul->un_logmap;
        threadtrans_t   *tp;
        ushort_t        seq;

        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);
        ASSERT(error != NULL);
        ASSERT(*error == 0);

        mutex_enter(&mtm->mtm_lock);
        if (topid == TOP_FSYNC) {
                /*
                 * Error the fsync immediately if this is an nfs thread
                 * and its last transaction has already been committed.
                 * The only transactions outstanding are those
                 * where no commit has even started
                 * (last_async_tid == mtm->mtm_tid)
                 * or those where a commit is in progress
                 * (last_async_tid == mtm->mtm_committid)
                 */
                if (curthread->t_flag & T_DONTPEND) {
                        tp = tsd_get(topkey);
                        if (tp && (tp->last_async_tid != mtm->mtm_tid) &&
                            (tp->last_async_tid != mtm->mtm_committid)) {
                                mutex_exit(&mtm->mtm_lock);
                                *error = 1;
                                return;
                        }
                }

                /*
                 * If there's already other synchronous transactions
                 * and we haven't allowed async ones to start yet
                 * then just wait for the commit to complete.
                 */
                if (((mtm->mtm_closed & (TOP_SYNC | TOP_ASYNC)) ==
                    (TOP_SYNC | TOP_ASYNC)) || mtm->mtm_activesync) {
                        seq = mtm->mtm_seq;
                        do {
                                cv_wait(&mtm->mtm_cv_commit, &mtm->mtm_lock);
                        } while (seq == mtm->mtm_seq);
                        mutex_exit(&mtm->mtm_lock);
                        *error = 1;
                        return;
                }
                if (mtm->mtm_closed & TOP_SYNC) {
                        /*
                         * We know we're in the window where a thread is
                         * committing a transaction in top_end_sync() and
                         * has allowed async threads to start but hasn't
                         * got the completion on the commit write to
                         * allow sync threads to start.
                         * So wait for that commit completion then retest
                         * for the quick nfs check and if that fails
                         * go on to start a transaction
                         */
                        seq = mtm->mtm_seq;
                        do {
                                cv_wait(&mtm->mtm_cv_commit, &mtm->mtm_lock);
                        } while (seq == mtm->mtm_seq);

                        /* tp is set above if T_DONTPEND */
                        if ((curthread->t_flag & T_DONTPEND) && tp &&
                            (tp->last_async_tid != mtm->mtm_tid) &&
                            (tp->last_async_tid != mtm->mtm_committid)) {
                                mutex_exit(&mtm->mtm_lock);
                                *error = 1;
                                return;
                        }
                }
        }
retry:
        mtm->mtm_ref = 1;
        /*
         * current transaction closed to sync ops; try for next transaction
         */
        if ((mtm->mtm_closed & TOP_SYNC) && !panicstr) {
                ulong_t         resv;

                /*
                 * We know a commit is in progress, if we are trying to
                 * commit and we haven't allowed async ones to start yet,
                 * then just wait for the commit completion
                 */
                if ((size == TOP_COMMIT_SIZE) &&
                    (((mtm->mtm_closed & (TOP_SYNC | TOP_ASYNC)) ==
                    (TOP_SYNC | TOP_ASYNC)) || (mtm->mtm_activesync))) {
                        seq = mtm->mtm_seq;
                        do {
                                cv_wait(&mtm->mtm_cv_commit, &mtm->mtm_lock);
                        } while (seq == mtm->mtm_seq);
                        mutex_exit(&mtm->mtm_lock);
                        *error = 1;
                        return;
                }

                /*
                 * next transaction is full; try for next transaction
                 */
                resv = size + ul->un_resv_wantin + ul->un_resv;
                if (resv > ul->un_maxresv) {
                        cv_wait(&mtm->mtm_cv_commit, &mtm->mtm_lock);
                        goto retry;
                }
                /*
                 * we are in the next transaction; wait for it to start
                 */
                mtm->mtm_wantin++;
                ul->un_resv_wantin += size;
                /*
                 * The corresponding cv_broadcast wakes up
                 * all threads that have been validated to go into
                 * the next transaction. However, because spurious
                 * cv_wait wakeups are possible we use a sequence
                 * number to check that the commit and cv_broadcast
                 * has really occurred. We couldn't use mtm_tid
                 * because on error that doesn't get incremented.
                 */
                seq = mtm->mtm_seq;
                do {
                        cv_wait(&mtm->mtm_cv_commit, &mtm->mtm_lock);
                } while (seq == mtm->mtm_seq);
        } else {
                /*
                 * if the current transaction is full; try the next one
                 */
                if (size && (ul->un_resv && ((size + ul->un_resv) >
                    ul->un_maxresv)) && !panicstr) {
                        /*
                         * log is over reserved and no one will unresv the space
                         *      so generate empty sync op to unresv the space
                         */
                        if (mtm->mtm_activesync == 0) {
                                mutex_exit(&mtm->mtm_lock);
                                top_issue_sync(ufsvfsp);
                                mutex_enter(&mtm->mtm_lock);
                                goto retry;
                        }
                        cv_wait(&mtm->mtm_cv_commit, &mtm->mtm_lock);
                        goto retry;
                }
                /*
                 * we are in the current transaction
                 */
                mtm->mtm_active++;
                mtm->mtm_activesync++;
                ul->un_resv += size;
        }

        ASSERT(mtm->mtm_active > 0);
        ASSERT(mtm->mtm_activesync > 0);
        mutex_exit(&mtm->mtm_lock);

        ASSERT(((ul->un_debug & MT_TRANSACT) == 0) ||
            top_begin_debug(ul, topid, size));
}

int tryfail_cnt;

int
top_begin_async(ufsvfs_t *ufsvfsp, top_t topid, ulong_t size, int tryasync)
{
        ml_unit_t       *ul     = ufsvfsp->vfs_log;
        mt_map_t        *mtm    = ul->un_logmap;
        threadtrans_t   *tp;

        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);

        tp = tsd_get(topkey);
        if (tp == NULL) {
                tp = kmem_zalloc(sizeof (threadtrans_t), KM_SLEEP);
                (void) tsd_set(topkey, tp);
        }
        tp->deltas_size = 0;
        tp->any_deltas = 0;

        mutex_enter(&mtm->mtm_lock);
retry:
        mtm->mtm_ref = 1;
        /*
         * current transaction closed to async ops; try for next transaction
         */
        if ((mtm->mtm_closed & TOP_ASYNC) && !panicstr) {
                if (tryasync) {
                        mutex_exit(&mtm->mtm_lock);
                        tryfail_cnt++;
                        return (EWOULDBLOCK);
                }
                cv_wait(&mtm->mtm_cv_next, &mtm->mtm_lock);
                goto retry;
        }

        /*
         * if the current transaction is full; try the next one
         */
        if (((size + ul->un_resv + ul->un_resv_wantin) > ul->un_maxresv) &&
            !panicstr) {
                /*
                 * log is overreserved and no one will unresv the space
                 *      so generate empty sync op to unresv the space
                 * We need TOP_SYNC_FORCED because we want to know when
                 * a top_end_sync is completed.
                 * mtm_taskq_sync_count is needed because we want to keep track
                 * of the pending top_issue_sync dispatches so that during
                 * forced umount we can wait for these to complete.
                 * mtm_taskq_sync_count is decremented in top_issue_sync and
                 * can remain set even after top_end_sync completes.
                 * We have a window between the clearing of TOP_SYNC_FORCED
                 * flag and the decrementing of mtm_taskq_sync_count.
                 * If in this window new async transactions start consuming
                 * log space, the log can get overreserved.
                 * Subsequently a new async transaction would fail to generate
                 * an empty sync transaction via the taskq, since it finds
                 * the mtm_taskq_sync_count set. This can cause a hang.
                 * Hence we do not test for mtm_taskq_sync_count being zero.
                 * Instead, the TOP_SYNC_FORCED flag is tested here.
                 */
                if ((mtm->mtm_activesync == 0) &&
                    (!(mtm->mtm_closed & TOP_SYNC_FORCED))) {
                        /*
                         * Set flag to stop multiple forced empty
                         * sync transactions. Increment mtm_taskq_sync_count.
                         */
                        mtm->mtm_closed |= TOP_SYNC_FORCED;
                        mtm->mtm_taskq_sync_count++;
                        mutex_exit(&mtm->mtm_lock);
                        (void) taskq_dispatch(system_taskq,
                            top_issue_from_taskq, ufsvfsp, TQ_SLEEP);
                        if (tryasync) {
                                tryfail_cnt++;
                                return (EWOULDBLOCK);
                        }
                        mutex_enter(&mtm->mtm_lock);
                        goto retry;
                }
                if (tryasync) {
                        mutex_exit(&mtm->mtm_lock);
                        tryfail_cnt++;
                        return (EWOULDBLOCK);
                }
                cv_wait(&mtm->mtm_cv_next, &mtm->mtm_lock);
                goto retry;
        }
        /*
         * we are in the current transaction
         */
        mtm->mtm_active++;
        ul->un_resv += size;

        ASSERT(mtm->mtm_active > 0);
        mutex_exit(&mtm->mtm_lock);

        ASSERT(((ul->un_debug & MT_TRANSACT) == 0) ||
            top_begin_debug(ul, topid, size));
        return (0);
}

/*ARGSUSED*/
void
top_end_sync(ufsvfs_t *ufsvfsp, int *ep, top_t topid, ulong_t size)
{
        ml_unit_t       *ul     = ufsvfsp->vfs_log;
        mt_map_t        *mtm    = ul->un_logmap;
        mapentry_t      *cancellist;
        uint32_t        tid;

        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);
        ASSERT(((ul->un_debug & MT_TRANSACT) == 0) ||
            top_end_debug(ul, mtm, topid, size));

        mutex_enter(&mtm->mtm_lock);
        tid = mtm->mtm_tid;

        mtm->mtm_activesync--;
        mtm->mtm_active--;

        mtm->mtm_ref = 1;

        /*
         * wait for last syncop to complete
         */
        if (mtm->mtm_activesync || panicstr) {
                ushort_t seq = mtm->mtm_seq;

                mtm->mtm_closed = TOP_SYNC;

                do {
                        cv_wait(&mtm->mtm_cv_commit, &mtm->mtm_lock);
                } while (seq == mtm->mtm_seq);
                mutex_exit(&mtm->mtm_lock);
                goto out;
        }
        /*
         * last syncop; close current transaction to all ops
         */
        mtm->mtm_closed = TOP_SYNC|TOP_ASYNC;

        /*
         * wait for last asyncop to finish
         */
        while (mtm->mtm_active) {
                cv_wait(&mtm->mtm_cv_eot, &mtm->mtm_lock);
        }

        /*
         * push dirty metadata thru the log
         */
        deltamap_push(ul);

        ASSERT(((ul->un_debug & MT_FORCEROLL) == 0) ||
            top_roll_debug(ul));

        mtm->mtm_tid = tid + 1; /* can overflow to 0 */

        /*
         * Empty the cancellist, but save it for logmap_free_cancel
         */
        mutex_enter(&mtm->mtm_mutex);
        cancellist = mtm->mtm_cancel;
        mtm->mtm_cancel = NULL;
        mutex_exit(&mtm->mtm_mutex);

        /*
         * allow async ops
         */
        ASSERT(mtm->mtm_active == 0);
        ul->un_resv = 0; /* unreserve the log space */
        mtm->mtm_closed = TOP_SYNC;
        /*
         * Hold the un_log_mutex here until we are done writing
         * the commit record to prevent any more deltas to be written
         * to the log after we allow async operations.
         */
        mutex_enter(&ul->un_log_mutex);
        mutex_exit(&mtm->mtm_lock);
        cv_broadcast(&mtm->mtm_cv_next);

        /*
         * asynchronously write the commit record,
         */
        logmap_commit(ul, tid);

        /*
         * wait for outstanding log writes (e.g., commits) to finish
         */
        ldl_waito(ul);

        /*
         * Now that we are sure the commit has been written to the log
         * we can free any canceled deltas.  If we free them before
         * guaranteeing that the commit was written, we could panic before
         * the commit, but after an async thread has allocated and written
         * to canceled freed block.
         */

        logmap_free_cancel(mtm, &cancellist);
        mutex_exit(&ul->un_log_mutex);

        /*
         * now, allow all ops
         */
        mutex_enter(&mtm->mtm_lock);
        mtm->mtm_active += mtm->mtm_wantin;
        ul->un_resv += ul->un_resv_wantin;
        mtm->mtm_activesync = mtm->mtm_wantin;
        mtm->mtm_wantin = 0;
        mtm->mtm_closed = 0;
        ul->un_resv_wantin = 0;
        mtm->mtm_committid = mtm->mtm_tid;
        mtm->mtm_seq++;
        mutex_exit(&mtm->mtm_lock);

        /*
         * Finish any other synchronous transactions and
         * start any waiting new synchronous transactions
         */
        cv_broadcast(&mtm->mtm_cv_commit);

        /*
         * if the logmap is getting full; roll something
         */
        if (logmap_need_roll_sync(mtm)) {
                logmap_forceroll_nowait(mtm);
        }

out:
        if (ul->un_flags & LDL_ERROR)
                *ep = EIO;
}

/*ARGSUSED*/
void
top_end_async(ufsvfs_t *ufsvfsp, top_t topid, ulong_t size)
{
        ml_unit_t       *ul     = ufsvfsp->vfs_log;
        mt_map_t        *mtm    = ul->un_logmap;
        threadtrans_t   *tp     = tsd_get(topkey);
        int             wakeup_needed = 0;

        ASSERT(tp);
        ASSERT(ufsvfsp->vfs_dev == ul->un_dev);
        ASSERT(((ul->un_debug & MT_TRANSACT) == 0) ||
            top_end_debug(ul, mtm, topid, size));

        mutex_enter(&mtm->mtm_lock);

        if (size > tp->deltas_size) {
                ul->un_resv -= (size - tp->deltas_size);
        }
        if (tp->any_deltas) {
                tp->last_async_tid = mtm->mtm_tid;
        }
        mtm->mtm_ref = 1;

        mtm->mtm_active--;
        if ((mtm->mtm_active == 0) &&
            (mtm->mtm_closed == (TOP_SYNC|TOP_ASYNC))) {
                wakeup_needed = 1;
        }
        mutex_exit(&mtm->mtm_lock);
        if (wakeup_needed)
                cv_signal(&mtm->mtm_cv_eot);

        /*
         * Generate a sync op if the log, logmap, or deltamap are heavily used.
         * Unless we are possibly holding any VM locks, since if we are holding
         * any VM locks and we issue a top_end_sync(), we could deadlock.
         */
        if ((mtm->mtm_activesync == 0) &&
            !(mtm->mtm_closed & TOP_SYNC) &&
            (deltamap_need_commit(ul->un_deltamap) ||
            logmap_need_commit(mtm) ||
            ldl_need_commit(ul)) &&
            (topid != TOP_GETPAGE)) {
                top_issue_sync(ufsvfsp);
        }
        /*
         * roll something from the log if the logmap is too full
         */
        if (logmap_need_roll_async(mtm))
                logmap_forceroll_nowait(mtm);
}

/*
 * Called from roll thread;
 *      buffer set for reading master
 * Returns
 *      0 - success, can continue with next buffer
 *      1 - failure due to logmap deltas being in use
 */
int
top_read_roll(rollbuf_t *rbp, ml_unit_t *ul)
{
        buf_t           *bp     = &rbp->rb_bh;
        offset_t        mof     = ldbtob(bp->b_blkno);

        /*
         * get a list of deltas
         */
        if (logmap_list_get_roll(ul->un_logmap, mof, rbp)) {
                /* logmap deltas are in use */
                return (1);
        }

        /*
         * no deltas were found, nothing to roll
         */
        if (rbp->rb_age == NULL) {
                bp->b_flags |= B_INVAL;
                return (0);
        }

        /*
         * If there is one cached roll buffer that cover all the deltas then
         * we can use that instead of copying to a separate roll buffer.
         */
        if (rbp->rb_crb) {
                rbp->rb_bh.b_blkno = lbtodb(rbp->rb_crb->c_mof);
                return (0);
        }

        /*
         * Set up the read.
         * If no read is needed logmap_setup_read() returns 0.
         */
        if (logmap_setup_read(rbp->rb_age, rbp)) {
                /*
                 * async read the data from master
                 */
                logstats.ls_rreads.value.ui64++;
                bp->b_bcount = MAPBLOCKSIZE;
                (void) bdev_strategy(bp);
                lwp_stat_update(LWP_STAT_INBLK, 1);
        } else {
                sema_v(&bp->b_io); /* mark read as complete */
        }
        return (0);
}

int ufs_crb_enable = 1;

/*
 * move deltas from deltamap into the log
 */
void
top_log(ufsvfs_t *ufsvfsp, char *va, offset_t vamof, off_t nb,
    caddr_t buf, uint32_t bufsz)
{
        ml_unit_t       *ul = ufsvfsp->vfs_log;
        mapentry_t      *me;
        offset_t        hmof;
        uint32_t        hnb, nb1;

        /*
         * needed for the roll thread's heuristic
         */
        ul->un_logmap->mtm_ref = 1;

        if (buf && ufs_crb_enable) {
                ASSERT((bufsz & DEV_BMASK) == 0);
                /*
                 * Move any deltas to the logmap. Split requests that
                 * straddle MAPBLOCKSIZE hash boundaries (i.e. summary info).
                 */
                for (hmof = vamof - (va - buf), nb1 = nb; bufsz;
                    bufsz -= hnb, hmof += hnb, buf += hnb, nb1 -= hnb) {
                        hnb = MAPBLOCKSIZE - (hmof & MAPBLOCKOFF);
                        if (hnb > bufsz)
                                hnb = bufsz;
                        me = deltamap_remove(ul->un_deltamap,
                            MAX(hmof, vamof), MIN(hnb, nb1));
                        if (me) {
                                logmap_add_buf(ul, va, hmof, me, buf, hnb);
                        }
                }
        } else {
                /*
                 * if there are deltas
                 */
                me = deltamap_remove(ul->un_deltamap, vamof, nb);
                if (me) {
                        /*
                         * move to logmap
                         */
                        logmap_add(ul, va, vamof, me);
                }
        }

        ASSERT((ul->un_matamap == NULL) ||
            matamap_within(ul->un_matamap, vamof, nb));
}


static void
top_threadtrans_destroy(void *tp)
{
        kmem_free(tp, sizeof (threadtrans_t));
}

void
_init_top(void)
{
        ASSERT(top_init_debug());

        /*
         * set up the delta layer
         */
        _init_map();

        /*
         * Initialise the thread specific data transaction key
         */
        tsd_create(&topkey, top_threadtrans_destroy);
}