<commit_before>#include "base_txn_btree.h"

rc_t
base_txn_btree::do_search(transaction &t, const varstr &k, value_reader &vr)
{
    t.ensure_active();

    key_writer key_writer(&k);
    const varstr * const key_str =
    key_writer.fully_materialize(true, t.string_allocator());

    // search the underlying btree to map k=>(btree_node|tuple)
    dbtuple * tuple{};
    OID oid;
    concurrent_btree::versioned_node_t sinfo;
    const bool found = this->underlying_btree.search(varkey(key_str), oid, tuple, t.xc, &sinfo);
    if (found)
        return t.do_tuple_read(tuple, vr);
#ifdef PHANTOM_PROT_NODE_SET
    else {
        rc_t rc = t.do_node_read(sinfo.first, sinfo.second);
        if (rc_is_abort(rc))
            return rc;
    }
#endif
    return rc_t{RC_FALSE};
}

std::map<std::string, uint64_t>
base_txn_btree::unsafe_purge(bool dump_stats)
{
    ALWAYS_ASSERT(!been_destructed);
    been_destructed = true;
    purge_tree_walker w;
    underlying_btree.tree_walk(w);
    underlying_btree.clear();
    return std::map<std::string, uint64_t>();
}

void
base_txn_btree::purge_tree_walker::on_node_begin(const typename concurrent_btree::node_opaque_t *n)
{
    INVARIANT(spec_values.empty());
    spec_values = concurrent_btree::ExtractValues(n);
}

void
base_txn_btree::purge_tree_walker::on_node_success()
{
    spec_values.clear();
}

void
base_txn_btree::purge_tree_walker::on_node_failure()
{
    spec_values.clear();
}

rc_t base_txn_btree::do_tree_put(
    transaction &t,
    const varstr *k,
    const varstr *v,
    bool expect_new)
{
    INVARIANT(k);
    INVARIANT(!expect_new || v); // makes little sense to remove() a key you expect
                                 // to not be present, so we assert this doesn't happen
                                 // for now [since this would indicate a suboptimality]
    t.ensure_active();

    // FIXME: tzwang: try_insert_new_tuple only tries once (no waiting, just one cas),
    // it fails if somebody else acted faster to insert new, we then
    // (fall back to) with the normal update procedure.
    // try_insert_new_tuple should add tuple to write-set too, if succeeded.
    if (expect_new and t.try_insert_new_tuple(&this->underlying_btree, k, v, this->fid))
        return rc_t{RC_TRUE};

    // do regular search
    dbtuple * bv = 0;
    OID oid = 0;
    if (!this->underlying_btree.search(varkey(k), oid, bv, t.xc))
        return rc_t{RC_ABORT_INTERNAL};
#ifdef PHANTOM_PROT_TABLE_LOCK
    // for delete
    bool instant_lock = false;
    table_lock_t *l = NULL;
    if (not v) {
        l = this->underlying_btree.tuple_vec()->lock_ptr();
        transaction::table_lock_set_t::iterator it =
            std::find(t.table_locks.begin(), t.table_locks.end(), l);
        if (it == t.table_locks.end()) {
            if (not object_vector::lock(l, TABLE_LOCK_X))
                return rc_t{RC_ABORT_PHANTOM};
            instant_lock = true;
        }
        else {
            if (not object_vector::upgrade_lock(l))
                return rc_t{RC_ABORT_PHANTOM};
        }
        ASSERT((volatile_read(*l) & TABLE_LOCK_MODE_MASK) == TABLE_LOCK_X or
               (volatile_read(*l) & TABLE_LOCK_MODE_MASK) == TABLE_LOCK_SIX);
    }
#endif

    dbtuple *tuple = NULL;
    // first *updater* wins
    dbtuple *prev = oidmgr->oid_put_update(this->underlying_btree.tuple_vec(), oid, v, t.xc, tuple);

    if (prev) { // succeeded
        ASSERT(tuple);
        ASSERT(t.xc);
#ifdef USE_PARALLEL_SSI
        ASSERT(prev->sstamp == NULL_PTR);
        // Therotically if there's a committed T3 and concurrent readers,
        // as the updater (T2) I can abort early here to avoid closing a cycle,
        // as it's easier to restart the T2. In practice, when retry-aborted-tx
        // is turned on, and if we don't do pre_commit if the tx is aborted
        // before entering precommit (ie, do log->discard directly), this causes
        // extremely slow performance with more than 20% cycles spent on
        // register/deregister_reader_tx. Actually if we don't abort here and
        // still do log->precommit at abort, we also get higher performance.
        // So we defer this possible abort to commit time (meanwhile the reader
        // might not really commit after all...)
        /*
        if (t.xc->ct3 and serial_get_tuple_readers(prev, true)) {
            // unlink the version here (note abort_impl won't be able to catch
            // it because it's not yet in the write set), same as in SSN impl.
            oidmgr->oid_unlink(this->fid, oid, tuple);
#ifdef PHANTOM_PROT_TABLE_LOCK
            if (instant_lock)
                object_vector::unlock(l);
#endif
            return rc_t{RC_ABORT_SERIAL};
        }
        */
#endif
#ifdef USE_PARALLEL_SSN
        // update hi watermark
        // Overwriting a version could trigger outbound anti-dep,
        // i.e., I'll depend on some tx who has read the version that's
        // being overwritten by me. So I'll need to see the version's
        // access stamp to tell if the read happened.<commit_msg>More details about SSI pivot abort at update.<commit_after>#include "base_txn_btree.h"

rc_t
base_txn_btree::do_search(transaction &t, const varstr &k, value_reader &vr)
{
    t.ensure_active();

    key_writer key_writer(&k);
    const varstr * const key_str =
    key_writer.fully_materialize(true, t.string_allocator());

    // search the underlying btree to map k=>(btree_node|tuple)
    dbtuple * tuple{};
    OID oid;
    concurrent_btree::versioned_node_t sinfo;
    const bool found = this->underlying_btree.search(varkey(key_str), oid, tuple, t.xc, &sinfo);
    if (found)
        return t.do_tuple_read(tuple, vr);
#ifdef PHANTOM_PROT_NODE_SET
    else {
        rc_t rc = t.do_node_read(sinfo.first, sinfo.second);
        if (rc_is_abort(rc))
            return rc;
    }
#endif
    return rc_t{RC_FALSE};
}

std::map<std::string, uint64_t>
base_txn_btree::unsafe_purge(bool dump_stats)
{
    ALWAYS_ASSERT(!been_destructed);
    been_destructed = true;
    purge_tree_walker w;
    underlying_btree.tree_walk(w);
    underlying_btree.clear();
    return std::map<std::string, uint64_t>();
}

void
base_txn_btree::purge_tree_walker::on_node_begin(const typename concurrent_btree::node_opaque_t *n)
{
    INVARIANT(spec_values.empty());
    spec_values = concurrent_btree::ExtractValues(n);
}

void
base_txn_btree::purge_tree_walker::on_node_success()
{
    spec_values.clear();
}

void
base_txn_btree::purge_tree_walker::on_node_failure()
{
    spec_values.clear();
}

rc_t base_txn_btree::do_tree_put(
    transaction &t,
    const varstr *k,
    const varstr *v,
    bool expect_new)
{
    INVARIANT(k);
    INVARIANT(!expect_new || v); // makes little sense to remove() a key you expect
                                 // to not be present, so we assert this doesn't happen
                                 // for now [since this would indicate a suboptimality]
    t.ensure_active();

    // FIXME: tzwang: try_insert_new_tuple only tries once (no waiting, just one cas),
    // it fails if somebody else acted faster to insert new, we then
    // (fall back to) with the normal update procedure.
    // try_insert_new_tuple should add tuple to write-set too, if succeeded.
    if (expect_new and t.try_insert_new_tuple(&this->underlying_btree, k, v, this->fid))
        return rc_t{RC_TRUE};

    // do regular search
    dbtuple * bv = 0;
    OID oid = 0;
    if (!this->underlying_btree.search(varkey(k), oid, bv, t.xc))
        return rc_t{RC_ABORT_INTERNAL};
#ifdef PHANTOM_PROT_TABLE_LOCK
    // for delete
    bool instant_lock = false;
    table_lock_t *l = NULL;
    if (not v) {
        l = this->underlying_btree.tuple_vec()->lock_ptr();
        transaction::table_lock_set_t::iterator it =
            std::find(t.table_locks.begin(), t.table_locks.end(), l);
        if (it == t.table_locks.end()) {
            if (not object_vector::lock(l, TABLE_LOCK_X))
                return rc_t{RC_ABORT_PHANTOM};
            instant_lock = true;
        }
        else {
            if (not object_vector::upgrade_lock(l))
                return rc_t{RC_ABORT_PHANTOM};
        }
        ASSERT((volatile_read(*l) & TABLE_LOCK_MODE_MASK) == TABLE_LOCK_X or
               (volatile_read(*l) & TABLE_LOCK_MODE_MASK) == TABLE_LOCK_SIX);
    }
#endif

    dbtuple *tuple = NULL;
    // first *updater* wins
    dbtuple *prev = oidmgr->oid_put_update(this->underlying_btree.tuple_vec(), oid, v, t.xc, tuple);

    if (prev) { // succeeded
        ASSERT(tuple);
        ASSERT(t.xc);
#ifdef USE_PARALLEL_SSI
        /*
         * TPC-C profile:
         * 1. Abort when xstamp > ct3 with concurrent readers: favors new order;
         * 2. Abort when xstamp > ct3 with concurrent readers begin ts > ct3: favors payment
         * But either way kills a lot of writers, favors readers in general.
         * 1 is simple, settle with that for now.
         */
        ASSERT(prev->sstamp == NULL_PTR);
        if (t.xc->ct3 and (volatile_read(prev->xstamp) > t.xc->ct3 or serial_get_tuple_readers(prev, true))) {
            oidmgr->oid_unlink(this->underlying_btree.tuple_vec(), oid, tuple);
            return {RC_ABORT_SERIAL};
        }

        /*
        if (t.xc->ct3) {
            readers_list::bitmap_t readers = serial_get_tuple_readers(prev, true);
            while (readers) {
                int i = __builtin_ctzll(readers);
                ASSERT(i >= 0 and i < readers_list::XIDS_PER_READER_KEY);
                readers &= (readers-1);
                XID rxid = volatile_read(rlist.xids[i]);
                ASSERT(rxid != xc->owner);
                if (!rxid._val)    // reader is gone, check xstamp in the end
                    continue;
                XID reader_owner = INVALID_XID;
                uint64_t reader_begin = 0;
                xid_context *reader_xc = NULL;
                reader_xc = xid_get_context(rxid);
                if (not reader_xc)  // context change, consult xstamp later
                    continue;

                // copy everything before doing anything
                reader_begin = volatile_read(reader_xc->begin).offset();
                reader_owner = volatile_read(reader_xc->owner);
                if (reader_owner != rxid)  // consult xstamp later
                    continue;

                if (reader_begin > t.xc->ct3) {
                    oidmgr->oid_unlink(this->underlying_btree.tuple_vec(), oid, tuple);
                    return {RC_ABORT_SERIAL};
                }
            }
            if (volatile_read(prev->xstamp) > t.xc->ct3) {
                oidmgr->oid_unlink(this->underlying_btree.tuple_vec(), oid, tuple);
                return {RC_ABORT_SERIAL};
            }
        }*/
#endif
#ifdef USE_PARALLEL_SSN
        // update hi watermark
        // Overwriting a version could trigger outbound anti-dep,
        // i.e., I'll depend on some tx who has read the version that's
        // being overwritten by me. So I'll need to see the version's
        // access stamp to tell if the read happened.
{'fraction_non_alphanumeric': 0.1067917380716154, 'fraction_numerical': 0.003066654640570037, 'mean_word_length': 2.4704225352112674, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 55, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3465799', 'n_tokens_mistral': 3721, 'n_tokens_neox': 3598, 'n_words': 907}