/*-
 * Copyright (c) 2014-2019 MongoDB, Inc.
 * Copyright (c) 2008-2014 WiredTiger, Inc.
 *	All rights reserved.
 *
 * See the file LICENSE for redistribution information.
 */

#include "wt_internal.h"

/*
 * __nsnap_destroy --
 *     Destroy a named snapshot structure.
 */
static void
__nsnap_destroy(WT_SESSION_IMPL *session, WT_NAMED_SNAPSHOT *nsnap)
{
    __wt_free(session, nsnap->name);
    __wt_free(session, nsnap->snapshot);
    __wt_free(session, nsnap);
}

/*
 * __nsnap_drop_one --
 *     Drop a single named snapshot. The named snapshot lock must be held write locked.
 */
static int
__nsnap_drop_one(WT_SESSION_IMPL *session, WT_CONFIG_ITEM *name)
{
    WT_NAMED_SNAPSHOT *found;
    WT_TXN_GLOBAL *txn_global;

    txn_global = &S2C(session)->txn_global;

    TAILQ_FOREACH (found, &txn_global->nsnaph, q)
        if (WT_STRING_MATCH(found->name, name->str, name->len))
            break;

    if (found == NULL)
        return (WT_NOTFOUND);

    /* Bump the global ID if we are removing the first entry */
    if (found == TAILQ_FIRST(&txn_global->nsnaph)) {
        WT_ASSERT(session, !__wt_txn_visible_all(session, txn_global->nsnap_oldest_id, WT_TS_NONE));
        txn_global->nsnap_oldest_id =
          (TAILQ_NEXT(found, q) != NULL) ? TAILQ_NEXT(found, q)->pinned_id : WT_TXN_NONE;
        WT_DIAGNOSTIC_YIELD;
        WT_ASSERT(session, txn_global->nsnap_oldest_id == WT_TXN_NONE ||
            !__wt_txn_visible_all(session, txn_global->nsnap_oldest_id, WT_TS_NONE));
    }
    TAILQ_REMOVE(&txn_global->nsnaph, found, q);
    __nsnap_destroy(session, found);
    WT_STAT_CONN_INCR(session, txn_snapshots_dropped);

    return (0);
}

/*
 * __nsnap_drop_to --
 *     Drop named snapshots, if the name is NULL all snapshots will be dropped. The named snapshot
 *     lock must be held write locked.
 */
static int
__nsnap_drop_to(WT_SESSION_IMPL *session, WT_CONFIG_ITEM *name, bool inclusive)
{
    WT_NAMED_SNAPSHOT *last, *nsnap, *prev;
    WT_TXN_GLOBAL *txn_global;
    uint64_t new_nsnap_oldest;

    last = nsnap = prev = NULL;
    txn_global = &S2C(session)->txn_global;

    if (TAILQ_EMPTY(&txn_global->nsnaph)) {
        if (name == NULL)
            return (0);
        /*
         * Dropping specific snapshots when there aren't any it's an error.
         */
        WT_RET_MSG(
          session, EINVAL, "Named snapshot '%.*s' for drop not found", (int)name->len, name->str);
    }

    /*
     * The new ID will be none if we are removing all named snapshots which is the default behavior
     * of this loop.
     */
    new_nsnap_oldest = WT_TXN_NONE;
    if (name != NULL) {
        TAILQ_FOREACH (last, &txn_global->nsnaph, q) {
            if (WT_STRING_MATCH(last->name, name->str, name->len))
                break;
            prev = last;
        }
        if (last == NULL)
            WT_RET_MSG(session, EINVAL, "Named snapshot '%.*s' for drop not found", (int)name->len,
              name->str);

        if (!inclusive) {
            /* We are done if a drop before points to the head */
            if (prev == 0)
                return (0);
            last = prev;
        }

        if (TAILQ_NEXT(last, q) != NULL)
            new_nsnap_oldest = TAILQ_NEXT(last, q)->pinned_id;
    }

    do {
        nsnap = TAILQ_FIRST(&txn_global->nsnaph);
        WT_ASSERT(session, nsnap != NULL);
        TAILQ_REMOVE(&txn_global->nsnaph, nsnap, q);
        __nsnap_destroy(session, nsnap);
        WT_STAT_CONN_INCR(session, txn_snapshots_dropped);
        /* Last will be NULL in the all case so it will never match */
    } while (nsnap != last && !TAILQ_EMPTY(&txn_global->nsnaph));

    /* Now that the queue of named snapshots is updated, update the ID */
    WT_ASSERT(session, !__wt_txn_visible_all(session, txn_global->nsnap_oldest_id, WT_TS_NONE) &&
        (new_nsnap_oldest == WT_TXN_NONE ||
                WT_TXNID_LE(txn_global->nsnap_oldest_id, new_nsnap_oldest)));
    txn_global->nsnap_oldest_id = new_nsnap_oldest;
    WT_DIAGNOSTIC_YIELD;
    WT_ASSERT(session, new_nsnap_oldest == WT_TXN_NONE ||
        !__wt_txn_visible_all(session, new_nsnap_oldest, WT_TS_NONE));

    return (0);
}

/*
 * __wt_txn_named_snapshot_begin --
 *     Begin an named in-memory snapshot.
 */
int
__wt_txn_named_snapshot_begin(WT_SESSION_IMPL *session, const char *cfg[])
{
    WT_CONFIG_ITEM cval;
    WT_DECL_RET;
    WT_NAMED_SNAPSHOT *nsnap, *nsnap_new;
    WT_TXN *txn;
    WT_TXN_GLOBAL *txn_global;
    const char *txn_cfg[] = {
      WT_CONFIG_BASE(session, WT_SESSION_begin_transaction), "isolation=snapshot", NULL};
    bool include_updates, started_txn;

    started_txn = false;
    nsnap_new = NULL;
    txn_global = &S2C(session)->txn_global;
    txn = &session->txn;

    WT_RET(__wt_config_gets_def(session, cfg, "include_updates", 0, &cval));
    include_updates = cval.val != 0;

    WT_RET(__wt_config_gets_def(session, cfg, "name", 0, &cval));
    WT_ASSERT(session, cval.len != 0);

    if (!F_ISSET(txn, WT_TXN_RUNNING)) {
        if (include_updates)
            WT_RET_MSG(session, EINVAL,
              "A transaction must be "
              "running to include updates in a named snapshot");

        WT_RET(__wt_txn_begin(session, txn_cfg));
        started_txn = true;
    }
    if (!include_updates)
        F_SET(txn, WT_TXN_READONLY);

    /* Save a copy of the transaction's snapshot. */
    WT_ERR(__wt_calloc_one(session, &nsnap_new));
    nsnap = nsnap_new;
    WT_ERR(__wt_strndup(session, cval.str, cval.len, &nsnap->name));

    /*
     * To include updates from a writing transaction, make sure a transaction ID has been allocated.
     */
    if (include_updates) {
        WT_ERR(__wt_txn_id_check(session));
        WT_ASSERT(session, txn->id != WT_TXN_NONE);
        nsnap->id = txn->id;
    } else
        nsnap->id = WT_TXN_NONE;
    nsnap->pinned_id = WT_SESSION_TXN_STATE(session)->pinned_id;
    nsnap->snap_min = txn->snap_min;
    nsnap->snap_max = txn->snap_max;
    if (txn->snapshot_count > 0) {
        WT_ERR(__wt_calloc_def(session, txn->snapshot_count, &nsnap->snapshot));
        memcpy(nsnap->snapshot, txn->snapshot, txn->snapshot_count * sizeof(*nsnap->snapshot));
    }
    nsnap->snapshot_count = txn->snapshot_count;

    /* Update the list. */

    /*
     * The semantic is that a new snapshot with the same name as an existing snapshot will replace
     * the old one.
     */
    WT_ERR_NOTFOUND_OK(__nsnap_drop_one(session, &cval));

    if (TAILQ_EMPTY(&txn_global->nsnaph)) {
        WT_ASSERT(session, txn_global->nsnap_oldest_id == WT_TXN_NONE &&
            !__wt_txn_visible_all(session, nsnap_new->pinned_id, WT_TS_NONE));
        __wt_readlock(session, &txn_global->rwlock);
        txn_global->nsnap_oldest_id = nsnap_new->pinned_id;
        __wt_readunlock(session, &txn_global->rwlock);
    }
    TAILQ_INSERT_TAIL(&txn_global->nsnaph, nsnap_new, q);
    WT_STAT_CONN_INCR(session, txn_snapshots_created);
    nsnap_new = NULL;

err:
    if (started_txn) {
#ifdef HAVE_DIAGNOSTIC
        uint64_t pinned_id = WT_SESSION_TXN_STATE(session)->pinned_id;
#endif
        WT_TRET(__wt_txn_rollback(session, NULL));
        WT_DIAGNOSTIC_YIELD;
        WT_ASSERT(session, !__wt_txn_visible_all(session, pinned_id, WT_TS_NONE));
    }

    if (nsnap_new != NULL)
        __nsnap_destroy(session, nsnap_new);

    return (ret);
}

/*
 * __wt_txn_named_snapshot_drop --
 *     Drop named snapshots
 */
int
__wt_txn_named_snapshot_drop(WT_SESSION_IMPL *session, const char *cfg[])
{
    WT_CONFIG objectconf;
    WT_CONFIG_ITEM all_config, before_config, k, names_config, to_config, v;
    WT_DECL_RET;

    WT_RET(__wt_config_gets_def(session, cfg, "drop.all", 0, &all_config));
    WT_RET(__wt_config_gets_def(session, cfg, "drop.names", 0, &names_config));
    WT_RET(__wt_config_gets_def(session, cfg, "drop.to", 0, &to_config));
    WT_RET(__wt_config_gets_def(session, cfg, "drop.before", 0, &before_config));

    if (all_config.val != 0)
        WT_RET(__nsnap_drop_to(session, NULL, true));
    else if (before_config.len != 0)
        WT_RET(__nsnap_drop_to(session, &before_config, false));
    else if (to_config.len != 0)
        WT_RET(__nsnap_drop_to(session, &to_config, true));

    /* We are done if there are no named drops */

    if (names_config.len != 0) {
        __wt_config_subinit(session, &objectconf, &names_config);
        while ((ret = __wt_config_next(&objectconf, &k, &v)) == 0) {
            ret = __nsnap_drop_one(session, &k);
            if (ret != 0)
                WT_RET_MSG(
                  session, EINVAL, "Named snapshot '%.*s' for drop not found", (int)k.len, k.str);
        }
        if (ret == WT_NOTFOUND)
            ret = 0;
    }

    return (ret);
}

/*
 * __wt_txn_named_snapshot_get --
 *     Lookup a named snapshot for a transaction.
 */
int
__wt_txn_named_snapshot_get(WT_SESSION_IMPL *session, WT_CONFIG_ITEM *nameval)
{
    WT_NAMED_SNAPSHOT *nsnap;
    WT_TXN *txn;
    WT_TXN_GLOBAL *txn_global;
    WT_TXN_STATE *txn_state;

    txn = &session->txn;
    txn_global = &S2C(session)->txn_global;
    txn_state = WT_SESSION_TXN_STATE(session);

    txn->isolation = WT_ISO_SNAPSHOT;
    if (session->ncursors > 0)
        WT_RET(__wt_session_copy_values(session));

    __wt_readlock(session, &txn_global->nsnap_rwlock);
    TAILQ_FOREACH (nsnap, &txn_global->nsnaph, q)
        if (WT_STRING_MATCH(nsnap->name, nameval->str, nameval->len)) {
            /*
             * Acquire the scan lock so the oldest ID can't move forward without seeing our pinned
             * ID.
             */
            __wt_readlock(session, &txn_global->rwlock);
            txn_state->pinned_id = nsnap->pinned_id;
            __wt_readunlock(session, &txn_global->rwlock);

            WT_ASSERT(session, !__wt_txn_visible_all(session, txn_state->pinned_id, WT_TS_NONE) &&
                txn_global->nsnap_oldest_id != WT_TXN_NONE &&
                WT_TXNID_LE(txn_global->nsnap_oldest_id, txn_state->pinned_id));
            txn->snap_min = nsnap->snap_min;
            txn->snap_max = nsnap->snap_max;
            if ((txn->snapshot_count = nsnap->snapshot_count) != 0)
                memcpy(
                  txn->snapshot, nsnap->snapshot, nsnap->snapshot_count * sizeof(*nsnap->snapshot));
            if (nsnap->id != WT_TXN_NONE) {
                WT_ASSERT(session, txn->id == WT_TXN_NONE);
                txn->id = nsnap->id;
                F_SET(txn, WT_TXN_READONLY);
            }
            F_SET(txn, WT_TXN_HAS_SNAPSHOT);
            break;
        }
    __wt_readunlock(session, &txn_global->nsnap_rwlock);

    if (nsnap == NULL)
        WT_RET_MSG(
          session, EINVAL, "Named snapshot '%.*s' not found", (int)nameval->len, nameval->str);

    /* Flag that this transaction is opened on a named snapshot */
    F_SET(txn, WT_TXN_NAMED_SNAPSHOT);

    return (0);
}

/*
 * __wt_txn_named_snapshot_config --
 *     Check the configuration for a named snapshot
 */
int
__wt_txn_named_snapshot_config(
  WT_SESSION_IMPL *session, const char *cfg[], bool *has_create, bool *has_drops)
{
    WT_CONFIG_ITEM all_config, before_config, names_config, to_config;
    WT_CONFIG_ITEM cval;
    WT_TXN *txn;

    txn = &session->txn;
    *has_create = *has_drops = false;

    /* Verify that the name is legal. */
    WT_RET(__wt_config_gets_def(session, cfg, "name", 0, &cval));
    if (cval.len != 0) {
        if (WT_STRING_MATCH("all", cval.str, cval.len))
            WT_RET_MSG(session, EINVAL, "Can't create snapshot with reserved \"all\" name");

        WT_RET(__wt_name_check(session, cval.str, cval.len));

        if (F_ISSET(txn, WT_TXN_RUNNING) && txn->isolation != WT_ISO_SNAPSHOT)
            WT_RET_MSG(session, EINVAL,
              "Can't create a named snapshot from a running "
              "transaction that isn't snapshot isolation");
        else if (F_ISSET(txn, WT_TXN_RUNNING) && txn->mod_count != 0)
            WT_RET_MSG(session, EINVAL,
              "Can't create a named snapshot from a running "
              "transaction that has made updates");
        *has_create = true;
    }

    /* Verify that the drop configuration is sane. */
    WT_RET(__wt_config_gets_def(session, cfg, "drop.all", 0, &all_config));
    WT_RET(__wt_config_gets_def(session, cfg, "drop.names", 0, &names_config));
    WT_RET(__wt_config_gets_def(session, cfg, "drop.to", 0, &to_config));
    WT_RET(__wt_config_gets_def(session, cfg, "drop.before", 0, &before_config));

    /* Avoid more work if no drops are configured. */
    if (all_config.val != 0 || names_config.len != 0 || before_config.len != 0 ||
      to_config.len != 0) {
        if (before_config.len != 0 && to_config.len != 0)
            WT_RET_MSG(session, EINVAL,
              "Illegal configuration; named snapshot drop can't "
              "specify both before and to options");
        if (all_config.val != 0 &&
          (names_config.len != 0 || to_config.len != 0 || before_config.len != 0))
            WT_RET_MSG(session, EINVAL,
              "Illegal configuration; named snapshot drop can't "
              "specify all and any other options");
        *has_drops = true;
    }

    if (!*has_create && !*has_drops)
        WT_RET_MSG(session, EINVAL,
          "WT_SESSION::snapshot API called without any drop or "
          "name option");

    return (0);
}

/*
 * __wt_txn_named_snapshot_destroy --
 *     Destroy all named snapshots on connection close
 */
void
__wt_txn_named_snapshot_destroy(WT_SESSION_IMPL *session)
{
    WT_NAMED_SNAPSHOT *nsnap;
    WT_TXN_GLOBAL *txn_global;

    txn_global = &S2C(session)->txn_global;
    txn_global->nsnap_oldest_id = WT_TXN_NONE;

    while ((nsnap = TAILQ_FIRST(&txn_global->nsnaph)) != NULL) {
        TAILQ_REMOVE(&txn_global->nsnaph, nsnap, q);
        __nsnap_destroy(session, nsnap);
    }
}