<commit_before>/* -*- mode: C; c-basic-offset: 4 -*- */
#ident "$Id$"
#ident "Copyright (c) 2007-2010 Tokutek Inc.  All rights reserved."
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."

#include "includes.h"

static void note_txn_closing (TOKUTXN txn);

void
toku_poll_txn_progress_function(TOKUTXN txn, uint8_t is_commit, uint8_t stall_for_checkpoint) {
    if (txn->progress_poll_fun) {
        TOKU_TXN_PROGRESS_S progress = {
            .entries_total     = txn->num_rollentries,
            .entries_processed = txn->num_rollentries_processed,
            .is_commit = is_commit,
            .stalled_on_checkpoint = stall_for_checkpoint};
        txn->progress_poll_fun(&progress, txn->progress_poll_fun_extra);
    }
}

int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
    int r=0;
    rolltype_dispatch_assign(item, toku_commit_, r, txn, yield, yieldv, lsn);
    txn->num_rollentries_processed++;
    if (txn->num_rollentries_processed % 1024 == 0)
        toku_poll_txn_progress_function(txn, TRUE, FALSE);
    return r;
}

int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
    int r=0;
    rolltype_dispatch_assign(item, toku_rollback_, r, txn, yield, yieldv, lsn);
    txn->num_rollentries_processed++;
    if (txn->num_rollentries_processed % 1024 == 0)
        toku_poll_txn_progress_function(txn, FALSE, FALSE);
    return r;
}

static inline int
txn_has_inprogress_rollback_log(TOKUTXN txn) {
    return txn->current_rollback.b != ROLLBACK_NONE.b;
}

static inline int
txn_has_spilled_rollback_logs(TOKUTXN txn) {
    return txn->spilled_rollback_tail.b != ROLLBACK_NONE.b;
}

static void rollback_unpin_remove_callback(CACHEKEY* cachekey, BOOL for_checkpoint, void* extra) {
    struct brt_header* h = extra;
    toku_free_blocknum(
        h->blocktable, 
        cachekey,
        h,
        for_checkpoint
        );
}


int
toku_delete_rollback_log(TOKUTXN txn, ROLLBACK_LOG_NODE log) {
    int r;
    CACHEFILE cf = txn->logger->rollback_cachefile;
    struct brt_header *h = toku_cachefile_get_userdata(cf);
    if (txn->pinned_inprogress_rollback_log == log) {
        txn->pinned_inprogress_rollback_log = NULL;
    }
    r = toku_cachetable_unpin_and_remove (cf, log->thislogname, rollback_unpin_remove_callback, h);
    assert(r==0);
    return r;
}

static int
toku_apply_txn (TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn,
                apply_rollback_item func) {
    int r = 0;
    // do the commit/abort calls and free everything
    // we do the commit/abort calls in reverse order too.
    struct roll_entry *item;
    //printf("%s:%d abort\n", __FILE__, __LINE__);
    int count=0;

    BLOCKNUM next_log      = ROLLBACK_NONE;
    uint32_t next_log_hash = 0;

    BOOL is_current = FALSE;
    if (txn_has_inprogress_rollback_log(txn)) {
        next_log      = txn->current_rollback;
        next_log_hash = txn->current_rollback_hash;
        is_current = TRUE;
    }
    else if (txn_has_spilled_rollback_logs(txn)) {
        next_log      = txn->spilled_rollback_tail;
        next_log_hash = txn->spilled_rollback_tail_hash;
    }

    uint64_t last_sequence = txn->num_rollback_nodes;
    BOOL found_head = FALSE;
    while (next_log.b != ROLLBACK_NONE.b) {
        ROLLBACK_LOG_NODE log;
        //pin log
        r = toku_get_and_pin_rollback_log(txn, txn->txnid64, last_sequence-1, next_log, next_log_hash, &log);
        assert(r==0);

        r = toku_maybe_prefetch_older_rollback_log(txn, log);
        assert(r==0);
        
        last_sequence = log->sequence;
        if (func) {
            while ((item=log->newest_logentry)) {
                log->newest_logentry = item->prev;
                r = func(txn, item, yield, yieldv, lsn);
                if (r!=0) return r;
                count++;
                if (count%2 == 0) yield(NULL, NULL, yieldv);
            }
        }
        if (next_log.b == txn->spilled_rollback_head.b) {
            assert(!found_head);
            found_head = TRUE;
            assert(log->sequence == 0);
        }<commit_msg>[t:4468] Transaction yiled frequency changes merged to main.<commit_after>/* -*- mode: C; c-basic-offset: 4 -*- */
#ident "$Id$"
#ident "Copyright (c) 2007-2010 Tokutek Inc.  All rights reserved."
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."

#include "includes.h"

static void note_txn_closing (TOKUTXN txn);

void
toku_poll_txn_progress_function(TOKUTXN txn, uint8_t is_commit, uint8_t stall_for_checkpoint) {
    if (txn->progress_poll_fun) {
        TOKU_TXN_PROGRESS_S progress = {
            .entries_total     = txn->num_rollentries,
            .entries_processed = txn->num_rollentries_processed,
            .is_commit = is_commit,
            .stalled_on_checkpoint = stall_for_checkpoint};
        txn->progress_poll_fun(&progress, txn->progress_poll_fun_extra);
    }
}

int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
    int r=0;
    rolltype_dispatch_assign(item, toku_commit_, r, txn, yield, yieldv, lsn);
    txn->num_rollentries_processed++;
    if (txn->num_rollentries_processed % 1024 == 0)
        toku_poll_txn_progress_function(txn, TRUE, FALSE);
    return r;
}

int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
    int r=0;
    rolltype_dispatch_assign(item, toku_rollback_, r, txn, yield, yieldv, lsn);
    txn->num_rollentries_processed++;
    if (txn->num_rollentries_processed % 1024 == 0)
        toku_poll_txn_progress_function(txn, FALSE, FALSE);
    return r;
}

static inline int
txn_has_inprogress_rollback_log(TOKUTXN txn) {
    return txn->current_rollback.b != ROLLBACK_NONE.b;
}

static inline int
txn_has_spilled_rollback_logs(TOKUTXN txn) {
    return txn->spilled_rollback_tail.b != ROLLBACK_NONE.b;
}

static void rollback_unpin_remove_callback(CACHEKEY* cachekey, BOOL for_checkpoint, void* extra) {
    struct brt_header* h = extra;
    toku_free_blocknum(
        h->blocktable, 
        cachekey,
        h,
        for_checkpoint
        );
}


int
toku_delete_rollback_log(TOKUTXN txn, ROLLBACK_LOG_NODE log) {
    int r;
    CACHEFILE cf = txn->logger->rollback_cachefile;
    struct brt_header *h = toku_cachefile_get_userdata(cf);
    if (txn->pinned_inprogress_rollback_log == log) {
        txn->pinned_inprogress_rollback_log = NULL;
    }
    r = toku_cachetable_unpin_and_remove (cf, log->thislogname, rollback_unpin_remove_callback, h);
    assert(r==0);
    return r;
}

static int
toku_apply_txn (TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn,
                apply_rollback_item func) {
    int r = 0;
    // do the commit/abort calls and free everything
    // we do the commit/abort calls in reverse order too.
    struct roll_entry *item;
    //printf("%s:%d abort\n", __FILE__, __LINE__);
    int count=0;

    BLOCKNUM next_log      = ROLLBACK_NONE;
    uint32_t next_log_hash = 0;

    BOOL is_current = FALSE;
    if (txn_has_inprogress_rollback_log(txn)) {
        next_log      = txn->current_rollback;
        next_log_hash = txn->current_rollback_hash;
        is_current = TRUE;
    }
    else if (txn_has_spilled_rollback_logs(txn)) {
        next_log      = txn->spilled_rollback_tail;
        next_log_hash = txn->spilled_rollback_tail_hash;
    }

    uint64_t last_sequence = txn->num_rollback_nodes;
    BOOL found_head = FALSE;
    while (next_log.b != ROLLBACK_NONE.b) {
        ROLLBACK_LOG_NODE log;
        //pin log
        r = toku_get_and_pin_rollback_log(txn, txn->txnid64, last_sequence-1, next_log, next_log_hash, &log);
        assert(r==0);

        r = toku_maybe_prefetch_older_rollback_log(txn, log);
        assert(r==0);
        
        last_sequence = log->sequence;
        if (func) {
            while ((item=log->newest_logentry)) {
                log->newest_logentry = item->prev;
                r = func(txn, item, yield, yieldv, lsn);
                if (r!=0) return r;
                count++;
                // We occassionally yield here to prevent transactions
                // from hogging the log.  This yield will allow other
                // threads to grab the ydb lock.  However, we don't
                // want any transaction doing more than one log
                // operation to always yield the ydb lock, as it must
                // wait for the ydb lock to be released to proceed.
                if (count % 8 == 0) {
                    yield(NULL, NULL, yieldv);
                }
            }
        }
        if (next_log.b == txn->spilled_rollback_head.b) {
            assert(!found_head);
            found_head = TRUE;
            assert(log->sequence == 0);
        }
{'fraction_non_alphanumeric': 0.09626128061882251, 'fraction_numerical': 0.010743446497636441, 'mean_word_length': 2.851468762929251, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4394157', 'n_tokens_mistral': 3329, 'n_tokens_neox': 3185, 'n_words': 735}