10633947029

Committed 30 Aug 2024 01:50PM UTC coverage: 87.197% (+0.001%) from 87.196%

Build # 10633947029

Build Type

push

github

Committed by

locker

Commit Message

applier: fix memory leak when replica is removed from cluster by master

When replica is deleted from cluster it receives the deletion statement
and then connection to master is closed by master. Since the commit
25aac111ce10 ("box: apply deletion of replica from `_cluster` on the
deleted replica") we kill applier to master on the deletion. Thus the
exit message produced due to EOF on connection to master is never
executed. This message in next in queue after the deletion statement.
As a result the exit message diag with SocketError from applier thread
is not freed.

Closes #10490
Part-of #10211

NO_TEST=covered by existing tests
NO_DOC=bugfix

(cherry picked from commit 6f0e19404)

Run Details

68327 of 121603 branches covered (56.19%)

1 of 1 new or added line in 1 file covered. (100.0%)

77 existing lines in 18 files now uncovered.

100994 of 115823 relevant lines covered (87.2%)

2448265.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.93

/src/box/vy_read_iterator.c

/*
 * Copyright 2010-2017, Tarantool AUTHORS, please see AUTHORS file.
 *
 * 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.
 *
 * THIS SOFTWARE IS PROVIDED BY AUTHORS ``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
 * AUTHORS 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.
 */
#include "vy_read_iterator.h"
#include "vy_run.h"
#include "vy_mem.h"
#include "vy_cache.h"
#include "vy_tx.h"
#include "fiber.h"
#include "vy_history.h"
#include "vy_lsm.h"
#include "vy_stat.h"

/**
 * Merge source, support structure for vy_read_iterator.
 * Contains source iterator and merge state.
 */
struct vy_read_src {
        /** Source iterator. */
        union {
                struct vy_run_iterator run_iterator;
                struct vy_mem_iterator mem_iterator;
                struct vy_txw_iterator txw_iterator;
                struct vy_cache_iterator cache_iterator;
        };
        /** Set if the iterator was started. */
        bool is_started;
        /**
         * Set if this is the last (deepest) source that may store tuples
         * matching the search criteria.
         */
        bool is_last;
        /** See vy_read_iterator->front_id. */
        uint32_t front_id;
        /** History of the key the iterator is positioned at. */
        struct vy_history history;
};

/**
 * Extend internal source array capacity to fit capacity sources.
 * Not necessary to call is but calling it allows to optimize internal memory
 * allocation
 */
static NODISCARD int
vy_read_iterator_reserve(struct vy_read_iterator *itr, uint32_t capacity)
{
        if (itr->src_capacity >= capacity)
                return 0;
        struct vy_read_src *new_src = calloc(capacity, sizeof(*new_src));
        if (new_src == NULL) {
                diag_set(OutOfMemory, capacity * sizeof(*new_src),
                         "calloc", "new_src");
                return -1;
        }
        memcpy(new_src, itr->src, itr->src_count * sizeof(*new_src));
        for (uint32_t i = 0; i < itr->src_count; i++) {
                vy_history_create(&new_src[i].history,
                                  &itr->lsm->env->history_node_pool);
                vy_history_splice(&new_src[i].history, &itr->src[i].history);
        }
        free(itr->src);
        itr->src = new_src;
        itr->src_capacity = capacity;
        return 0;
}

/**
 * Add another source to read iterator. Must be called before actual
 * iteration start and must not be called after.
 */
static struct vy_read_src *
vy_read_iterator_add_src(struct vy_read_iterator *itr)
{
        if (itr->src_count == itr->src_capacity) {
                if (vy_read_iterator_reserve(itr, itr->src_count + 1) != 0)
                        return NULL;
        }
        struct vy_read_src *src = &itr->src[itr->src_count++];
        memset(src, 0, sizeof(*src));
        vy_history_create(&src->history, &itr->lsm->env->history_node_pool);
        return src;
}

/**
 * Pin all slices open by the read iterator.
 * Used to make sure no run slice is invalidated by
 * compaction while we are fetching data from disk.
 */
static void
vy_read_iterator_pin_slices(struct vy_read_iterator *itr)
{
        for (uint32_t i = itr->disk_src; i < itr->src_count; i++) {
                struct vy_read_src *src = &itr->src[i];
                vy_slice_pin(src->run_iterator.slice);
        }
}

/**
 * Unpin all slices open by the read iterator.
 * See also: vy_read_iterator_pin_slices().
 */
static void
vy_read_iterator_unpin_slices(struct vy_read_iterator *itr)
{
        for (uint32_t i = itr->disk_src; i < itr->src_count; i++) {
                struct vy_read_src *src = &itr->src[i];
                vy_slice_unpin(src->run_iterator.slice);
        }
}

/**
 * Return true if the current candidate for the next key is outside
 * the current range and hence we should move to the next range.
 *
 * If we are looking for a match (EQ, REQ) and the search key
 * doesn't intersect with the current range's boundary, the next
 * range can't contain statements matching the search criteria
 * and hence there's no point in iterating to it.
 */
static bool
vy_read_iterator_range_is_done(struct vy_read_iterator *itr,
                               struct vy_entry next)
{
        struct vy_range *range = itr->curr_range;
        struct key_def *cmp_def = itr->lsm->cmp_def;
        int dir = iterator_direction(itr->iterator_type);

        if (dir > 0 && range->end.stmt != NULL &&
            (next.stmt == NULL || vy_entry_compare(next, range->end,
                                                   cmp_def) >= 0) &&
            (itr->iterator_type != ITER_EQ ||
             vy_entry_compare(itr->key, range->end, cmp_def) >= 0))
                return true;

        if (dir < 0 && range->begin.stmt != NULL &&
            (next.stmt == NULL || vy_entry_compare(next, range->begin,
                                                   cmp_def) < 0) &&
            (itr->iterator_type != ITER_REQ ||
             vy_entry_compare(itr->key, range->begin, cmp_def) <= 0))
                return true;

        return false;
}

/**
 * Compare two tuples from the read iterator perspective.
 *
 * Returns:
 *  < 0 if statement @a precedes statement @b in the iterator output
 * == 0 if statements @a and @b are at the same position
 *  > 0 if statement @a supersedes statement @b
 *
 * NULL denotes the statement following the last one.
 */
static inline int
vy_read_iterator_cmp_stmt(struct vy_read_iterator *itr,
                          struct vy_entry a, struct vy_entry b)
{
        if (a.stmt == NULL && b.stmt != NULL)
                return 1;
        if (a.stmt != NULL && b.stmt == NULL)
                return -1;
        if (a.stmt == NULL && b.stmt == NULL)
                return 0;
        return iterator_direction(itr->iterator_type) *
                vy_entry_compare(a, b, itr->lsm->cmp_def);
}

/**
 * Check if the statement at which the given read source
 * is positioned precedes the current candidate for the
 * next key ('next') and update the latter if so.
 * The 'stop' flag is set if the next key is found and
 * older sources don't need to be evaluated.
 */
static void
vy_read_iterator_evaluate_src(struct vy_read_iterator *itr,
                              struct vy_read_src *src,
                              struct vy_entry *next, bool *stop)
{
        uint32_t src_id = src - itr->src;
        struct vy_entry entry = vy_history_last_stmt(&src->history);
        int cmp = vy_read_iterator_cmp_stmt(itr, entry, *next);
        if (cmp < 0) {
                assert(entry.stmt != NULL);
                *next = entry;
                itr->front_id++;
        }
        if (cmp <= 0)
                src->front_id = itr->front_id;

        if (src->is_last)
                goto stop;

        if (itr->check_exact_match &&
            cmp < 0 && vy_history_is_terminal(&src->history)) {
                /*
                 * So this is a terminal statement that might be the first one
                 * in the output and the iterator may return at most one tuple
                 * equal to the search key. Let's check if this statement
                 * equals the search key. If it is, there cannot be a better
                 * candidate in deeper sources so we may skip them.
                 *
                 * No need to check for equality if it's EQ iterator because
                 * it must have been already checked by the source iterator.
                 * Sic: for REQ the check is still required (see need_check_eq).
                 */
                if (itr->iterator_type == ITER_EQ ||
                    vy_entry_compare(entry, itr->key, itr->lsm->cmp_def) == 0) {
                        /*
                         * If we get an exact match for EQ/REQ search, we don't
                         * need to check deeper sources on next iterations so
                         * mark this source last. Note that we might still need
                         * to scan this source again though - if we encounter
                         * a DELETE statement - because in this case there may
                         * be a newer REPLACE statement for the same key in it.
                         */
                        if (itr->iterator_type == ITER_EQ ||
                            itr->iterator_type == ITER_REQ)
                                src->is_last = true;
                        goto stop;
                }
        }

        itr->skipped_src = MAX(itr->skipped_src, src_id + 1);
        return;
stop:
        itr->skipped_src = src_id + 1;
        *stop = true;
}

/**
 * Reevaluate scanned (not skipped) read sources and position 'next' to
 * the statement that is minimal from this read iterator's perspective.
 * This function assumes that all scanned read sources are up-to-date.
 * See also vy_read_iterator_evaluate_src().
 */
static void
vy_read_iterator_reevaluate_srcs(struct vy_read_iterator *itr,
                                 struct vy_entry *next)
{
        *next = vy_entry_none();
        for (uint32_t i = 0; i < itr->src_count; i++) {
                if (i >= itr->skipped_src)
                        break;
                struct vy_read_src *src = &itr->src[i];
                struct vy_entry entry = vy_history_last_stmt(&src->history);
                int cmp = vy_read_iterator_cmp_stmt(itr, entry, *next);
                if (cmp < 0) {
                        *next = entry;
                        itr->front_id++;
                }
                if (cmp <= 0)
                        src->front_id = itr->front_id;
        }
}

/*
 * Each of the functions from the vy_read_iterator_scan_* family
 * is used by vy_read_iterator_advance() to:
 *
 * 1. Update the position of a read source, which implies:
 *
 *    - Starting iteration over the source if it has not been done
 *      yet or restoring the iterator position in case the source
 *      has been modified since the last iteration.
 *
 *    - Advancing the iterator position to the first statement
 *      following the one returned on the previous iteration.
 *      To avoid an extra tuple comparison, we maintain front_id
 *      for each source: all sources with front_id equal to the
 *      front_id of the read iterator were used on the previous
 *      iteration and hence need to be advanced.
 *
 * 2. Update the candidate for the next key ('next') if the
 *    statement at which the source is positioned precedes it.
 *    The 'stop' flag is set if older sources do not need to be
 *    scanned (e.g. because a chain was found in the cache).
 *    See also vy_read_iterator_evaluate_src().
 */

static NODISCARD int
vy_read_iterator_scan_txw(struct vy_read_iterator *itr,
                          struct vy_entry *next, bool *stop)
{
        struct vy_read_src *src = &itr->src[itr->txw_src];
        struct vy_txw_iterator *src_itr = &src->txw_iterator;

        if (itr->tx == NULL)
                return 0;

        assert(itr->txw_src < itr->skipped_src);

        int rc = vy_txw_iterator_restore(src_itr, itr->last, &src->history);
        if (rc == 0) {
                if (!src->is_started) {
                        rc = vy_txw_iterator_skip(src_itr, itr->last,
                                                  &src->history);
                } else if (src->front_id == itr->prev_front_id) {
                        rc = vy_txw_iterator_next(src_itr, &src->history);
                }
                src->is_started = true;
        }
        if (rc < 0)
                return -1;

        vy_read_iterator_evaluate_src(itr, src, next, stop);
        return 0;
}

static NODISCARD int
vy_read_iterator_scan_cache(struct vy_read_iterator *itr,
                            struct vy_entry *next, bool *stop)
{
        bool is_interval = false;
        struct vy_read_src *src = &itr->src[itr->cache_src];
        struct vy_cache_iterator *src_itr = &src->cache_iterator;

        int rc = vy_cache_iterator_restore(src_itr, itr->last,
                                           &src->history, &is_interval);
        if (rc == 0) {
                if (!src->is_started || itr->cache_src >= itr->skipped_src) {
                        rc = vy_cache_iterator_skip(src_itr, itr->last,
                                                &src->history, &is_interval);
                } else if (src->front_id == itr->prev_front_id) {
                        rc = vy_cache_iterator_next(src_itr, &src->history,
                                                    &is_interval);
                }
                src->is_started = true;
        }
        if (rc < 0)
                return -1;

        vy_read_iterator_evaluate_src(itr, src, next, stop);
        if (is_interval) {
                itr->skipped_src = itr->cache_src + 1;
                *stop = true;
        }
        return 0;
}

static NODISCARD int
vy_read_iterator_scan_mem(struct vy_read_iterator *itr, uint32_t mem_src,
                          struct vy_entry *next, bool *stop,
                          int64_t *min_skipped_plsn)
{
        int rc;
        struct vy_read_src *src = &itr->src[mem_src];
        struct vy_mem_iterator *src_itr = &src->mem_iterator;

        assert(mem_src >= itr->mem_src && mem_src < itr->disk_src);

        rc = vy_mem_iterator_restore(src_itr, itr->last, &src->history);
        if (rc == 0) {
                if (!src->is_started || mem_src >= itr->skipped_src) {
                        rc = vy_mem_iterator_skip(src_itr, itr->last,
                                                  &src->history);
                } else if (src->front_id == itr->prev_front_id) {
                        rc = vy_mem_iterator_next(src_itr, &src->history);
                }
                src->is_started = true;
        }
        if (rc < 0)
                return -1;
        vy_read_iterator_evaluate_src(itr, src, next, stop);
        *min_skipped_plsn = MIN(*min_skipped_plsn, src_itr->min_skipped_plsn);
        return 0;
}

static NODISCARD int
vy_read_iterator_scan_disk(struct vy_read_iterator *itr, uint32_t disk_src,
                           struct vy_entry *next, bool *stop)
{
        int rc = 0;
        struct vy_read_src *src = &itr->src[disk_src];
        struct vy_run_iterator *src_itr = &src->run_iterator;

        assert(disk_src >= itr->disk_src && disk_src < itr->src_count);

        if (!src->is_started || disk_src >= itr->skipped_src)
                rc = vy_run_iterator_skip(src_itr, itr->last,
                                          &src->history);
        else if (src->front_id == itr->prev_front_id)
                rc = vy_run_iterator_next(src_itr, &src->history);
        src->is_started = true;

        if (rc < 0)
                return -1;

        vy_read_iterator_evaluate_src(itr, src, next, stop);
        return 0;
}

/**
 * Restore the position of the active in-memory tree iterator
 * after a yield caused by a disk read and update 'next'
 * if necessary.
 */
static NODISCARD int
vy_read_iterator_restore_mem(struct vy_read_iterator *itr,
                             struct vy_entry *next)
{
        int rc;
        int cmp;
        struct vy_read_src *src = &itr->src[itr->mem_src];

        /*
         * 'next' may refer to a statement in the memory source history,
         * which may be cleaned up by vy_mem_iterator_restore(), so we need
         * to take a reference to it.
         */
        struct tuple *next_stmt_ref = next->stmt;
        if (next_stmt_ref != NULL)
                tuple_ref(next_stmt_ref);

        rc = vy_mem_iterator_restore(&src->mem_iterator,
                                     itr->last, &src->history);
        if (rc < 0)
                goto out; /* memory allocation error */
        if (rc == 0)
                goto out; /* nothing changed */

        /* The memory source was updated. Reevaluate it for 'next'. */
        rc = 0;
        struct vy_entry entry = vy_history_last_stmt(&src->history);
        cmp = vy_read_iterator_cmp_stmt(itr, entry, *next);
        if (cmp > 0) {
                /*
                 * Normally, memory trees are append-only so if the source is
                 * not on top of the heap after restoration, it was not before.
                 * There's one exception to this rule though: a statement may
                 * be deleted from a memory tree on rollback after a WAL write
                 * failure. If the deleted statement was on top of the heap,
                 * we need to reevaluate all read sources to reposition the
                 * iterator to the minimal statement.
                 */
                if (src->front_id == itr->front_id)
                        vy_read_iterator_reevaluate_srcs(itr, next);
                goto out;
        }
        /* The new statement is a better candidate for 'next'. */
        *next = entry;
        if (cmp < 0) {
                /*
                 * The new statement precedes the current
                 * candidate for the next key.
                 */
                itr->front_id++;
        } else {
                /*
                 * The new statement updates the next key.
                 * Make sure we don't read the old value
                 * from the cache while applying UPSERTs.
                 */
                struct vy_read_src *cache_src = &itr->src[itr->cache_src];
                if (cache_src->front_id == itr->front_id)
                        vy_history_cleanup(&cache_src->history);
        }
        src->front_id = itr->front_id;
out:
        if (next_stmt_ref != NULL)
                tuple_unref(next_stmt_ref);
        return rc;
}

static void
vy_read_iterator_restore(struct vy_read_iterator *itr);

static void
vy_read_iterator_next_range(struct vy_read_iterator *itr);

/**
 * Advance the iterator to the next key.
 * Returns 0 on success, -1 on error.
 */
static NODISCARD int
vy_read_iterator_advance(struct vy_read_iterator *itr)
{
        /*
         * Restore the iterator position if the LSM tree has changed
         * since the last iteration or this is the first iteration.
         */
        if (!itr->is_started ||
            itr->mem_list_version != itr->lsm->mem_list_version ||
            itr->range_tree_version != itr->lsm->range_tree_version ||
            itr->range_version != itr->curr_range->version) {
                vy_read_iterator_restore(itr);
        }
        itr->is_started = true;
restart:
        itr->prev_front_id = itr->front_id;
        itr->front_id++;

        /*
         * Look up the next key in read sources starting
         * from the one that stores newest data.
         */
        bool stop = false;
        struct vy_entry next = vy_entry_none();
        if (vy_read_iterator_scan_txw(itr, &next, &stop) != 0)
                return -1;
        if (stop)
                goto done;
        if (vy_read_iterator_scan_cache(itr, &next, &stop) != 0)
                return -1;
        if (stop)
                goto done;

        int64_t min_skipped_plsn = INT64_MAX;
        for (uint32_t i = itr->mem_src; i < itr->disk_src && !stop; i++) {
                if (vy_read_iterator_scan_mem(itr, i, &next, &stop,
                                              &min_skipped_plsn) != 0)
                        return -1;
        }
        if (itr->tx != NULL && min_skipped_plsn != INT64_MAX) {
                if (vy_tx_send_to_read_view(itr->tx, min_skipped_plsn) != 0)
                        return -1;
                if (itr->tx->state == VINYL_TX_ABORT) {
                        diag_set(ClientError, ER_TRANSACTION_CONFLICT);
                        return -1;
                }
        }
        if (stop)
                goto done;
rescan_disk:
        /* The following code may yield as it needs to access disk. */
        vy_read_iterator_pin_slices(itr);
        for (uint32_t i = itr->disk_src; i < itr->src_count; i++) {
                if (vy_read_iterator_scan_disk(itr, i, &next, &stop) != 0) {
                        vy_read_iterator_unpin_slices(itr);
                        return -1;
                }
                if (stop)
                        break;
        }
        vy_read_iterator_unpin_slices(itr);
        /*
         * The transaction could have been aborted while we were
         * reading disk. We must stop now and return an error as
         * this function could be called by a DML request that
         * was aborted by a DDL operation: failing will prevent
         * it from dereferencing a destroyed space.
         */
        if (itr->tx != NULL && itr->tx->state == VINYL_TX_ABORT) {
                diag_set(ClientError, ER_TRANSACTION_CONFLICT);
                return -1;
        }
        /*
         * The list of in-memory indexes and/or the range tree could
         * have been modified by dump/compaction while we were fetching
         * data from disk. Restart the iterator if this is the case.
         * Note, we don't need to check the current range's version,
         * because all slices were pinned and hence could not be
         * removed.
         */
        if (itr->mem_list_version != itr->lsm->mem_list_version ||
            itr->range_tree_version != itr->lsm->range_tree_version) {
                vy_read_iterator_restore(itr);
                goto restart;
        }
        /*
         * The transaction write set couldn't change during the yield
         * as it is owned exclusively by the current fiber so the only
         * source to check is the active in-memory tree.
         */
        if (vy_read_iterator_restore_mem(itr, &next) != 0)
                return -1;
        /*
         * Scan the next range in case we transgressed the current
         * range's boundaries.
         */
        if (vy_read_iterator_range_is_done(itr, next)) {
                vy_read_iterator_next_range(itr);
                goto rescan_disk;
        }
done:
#ifndef NDEBUG
        /* Check that the statement meets search criteria. */
        if (next.stmt != NULL) {
                int cmp = vy_entry_compare(next, itr->key, itr->lsm->cmp_def);
                cmp *= iterator_direction(itr->iterator_type);
                if (itr->iterator_type == ITER_GT ||
                    itr->iterator_type == ITER_LT)
                        assert(cmp > 0);
                else
                        assert(cmp >= 0);
        }
        /*
         * Ensure the read iterator does not return duplicates
         * and respects statement order.
         */
        if (itr->last.stmt != NULL && next.stmt != NULL) {
               assert(vy_read_iterator_cmp_stmt(itr, next, itr->last) > 0);
        }
#endif
        if (itr->need_check_eq && next.stmt != NULL &&
            vy_entry_compare(next, itr->key, itr->lsm->cmp_def) != 0)
                itr->front_id++;
        return 0;
}

/** Add the transaction source to the read iterator. */
static void
vy_read_iterator_add_tx(struct vy_read_iterator *itr)
{
        assert(itr->tx != NULL);
        enum iterator_type iterator_type = (itr->iterator_type != ITER_REQ ?
                                            itr->iterator_type : ITER_LE);
        struct vy_txw_iterator_stat *stat = &itr->lsm->stat.txw.iterator;
        struct vy_read_src *sub_src = vy_read_iterator_add_src(itr);
        vy_txw_iterator_open(&sub_src->txw_iterator, stat, itr->tx, itr->lsm,
                             iterator_type, itr->key);
}

/** Add the cache source to the read iterator. */
static void
vy_read_iterator_add_cache(struct vy_read_iterator *itr, bool is_prepared_ok)
{
        enum iterator_type iterator_type = (itr->iterator_type != ITER_REQ ?
                                            itr->iterator_type : ITER_LE);
        struct vy_read_src *sub_src = vy_read_iterator_add_src(itr);
        vy_cache_iterator_open(&sub_src->cache_iterator, &itr->lsm->cache,
                               iterator_type, itr->key, itr->read_view,
                               is_prepared_ok);
}

/** Add the memory level source to the read iterator. */
static void
vy_read_iterator_add_mem(struct vy_read_iterator *itr, bool is_prepared_ok)
{
        enum iterator_type iterator_type = (itr->iterator_type != ITER_REQ ?
                                            itr->iterator_type : ITER_LE);
        struct vy_lsm *lsm = itr->lsm;
        struct vy_read_src *sub_src;

        /* Add the active in-memory index. */
        assert(lsm->mem != NULL);
        sub_src = vy_read_iterator_add_src(itr);
        vy_mem_iterator_open(&sub_src->mem_iterator, &lsm->stat.memory.iterator,
                             lsm->mem, iterator_type, itr->key, itr->read_view,
                             is_prepared_ok);
        /* Add sealed in-memory indexes. */
        struct vy_mem *mem;
        rlist_foreach_entry(mem, &lsm->sealed, in_sealed) {
                sub_src = vy_read_iterator_add_src(itr);
                vy_mem_iterator_open(&sub_src->mem_iterator,
                                     &lsm->stat.memory.iterator,
                                     mem, iterator_type, itr->key,
                                     itr->read_view, is_prepared_ok);
        }
}

/** Add the disk level source to the read iterator. */
static void
vy_read_iterator_add_disk(struct vy_read_iterator *itr)
{
        assert(itr->curr_range != NULL);
        enum iterator_type iterator_type = (itr->iterator_type != ITER_REQ ?
                                            itr->iterator_type : ITER_LE);
        struct vy_lsm *lsm = itr->lsm;
        struct vy_slice *slice;
        /*
         * The format of the statement must be exactly the space
         * format with the same identifier to fully match the
         * format in vy_mem.
         */
        rlist_foreach_entry(slice, &itr->curr_range->slices, in_range) {
                struct vy_read_src *sub_src = vy_read_iterator_add_src(itr);
                vy_run_iterator_open(&sub_src->run_iterator,
                                     &lsm->stat.disk.iterator, slice,
                                     iterator_type, itr->key,
                                     itr->read_view, lsm->cmp_def,
                                     lsm->key_def, lsm->disk_format);
        }
}

/**
 * Close all open sources and reset the merge state.
 */
static void
vy_read_iterator_cleanup(struct vy_read_iterator *itr)
{
        uint32_t i;
        struct vy_read_src *src;

        if (itr->txw_src < itr->src_count) {
                src = &itr->src[itr->txw_src];
                vy_history_cleanup(&src->history);
                vy_txw_iterator_close(&src->txw_iterator);
        }
        if (itr->cache_src < itr->src_count) {
                src = &itr->src[itr->cache_src];
                vy_history_cleanup(&src->history);
                vy_cache_iterator_close(&src->cache_iterator);
        }
        for (i = itr->mem_src; i < itr->disk_src; i++) {
                src = &itr->src[i];
                vy_history_cleanup(&src->history);
                vy_mem_iterator_close(&src->mem_iterator);
        }
        for (i = itr->disk_src; i < itr->src_count; i++) {
                src = &itr->src[i];
                vy_history_cleanup(&src->history);
                vy_run_iterator_close(&src->run_iterator);
        }

        itr->txw_src = UINT32_MAX;
        itr->cache_src = UINT32_MAX;
        itr->mem_src = UINT32_MAX;
        itr->disk_src = UINT32_MAX;
        itr->skipped_src = UINT32_MAX;
        itr->src_count = 0;
}

void
vy_read_iterator_open_after(struct vy_read_iterator *itr, struct vy_lsm *lsm,
                            struct vy_tx *tx, enum iterator_type iterator_type,
                            struct vy_entry key, struct vy_entry last,
                            const struct vy_read_view **rv)
{
        memset(itr, 0, sizeof(*itr));

        itr->lsm = lsm;
        itr->tx = tx;
        itr->iterator_type = iterator_type;
        itr->key = key;
        itr->read_view = rv;
        itr->last = last;
        itr->last_cached = vy_entry_none();
        itr->is_first_cached = (itr->last.stmt == NULL);

        if (vy_stmt_is_empty_key(key.stmt)) {
                /*
                 * Strictly speaking, a GT/LT iterator should return
                 * nothing if the key is empty, because every key is
                 * equal to the empty key, but historically we return
                 * all keys instead. So use GE/LE instead of GT/LT
                 * in this case.
                 */
                itr->iterator_type = iterator_direction(iterator_type) > 0 ?
                                     ITER_GE : ITER_LE;
        }

        if (iterator_type == ITER_ALL)
                itr->iterator_type = ITER_GE;

        if (iterator_type == ITER_REQ) {
                /*
                 * Source iterators cannot handle ITER_REQ and
                 * use ITER_LE instead, so we need to enable EQ
                 * check in this case.
                 *
                 * See vy_read_iterator_add_{tx,cache,mem,run}.
                 */
                itr->need_check_eq = true;
        }

        itr->check_exact_match =
                (iterator_type == ITER_EQ || iterator_type == ITER_REQ ||
                 iterator_type == ITER_GE || iterator_type == ITER_LE) &&
                vy_stmt_is_exact_key(key.stmt, lsm->cmp_def, lsm->key_def,
                                     lsm->opts.is_unique);
}

/**
 * Restart the read iterator from the position following
 * the last statement returned to the user. Called when
 * the current range or the whole range tree is changed.
 * Also used for preparing the iterator for the first
 * iteration.
 */
static void
vy_read_iterator_restore(struct vy_read_iterator *itr)
{
        vy_read_iterator_cleanup(itr);

        itr->mem_list_version = itr->lsm->mem_list_version;
        itr->range_tree_version = itr->lsm->range_tree_version;
        itr->curr_range = vy_range_tree_find_by_key(&itr->lsm->range_tree,
                                                    itr->iterator_type,
                                                    itr->last.stmt != NULL ?
                                                    itr->last : itr->key);
        itr->range_version = itr->curr_range->version;

        bool is_prepared_ok = true;
        if (itr->tx != NULL) {
                is_prepared_ok = vy_tx_is_prepared_ok(itr->tx);
                itr->txw_src = itr->src_count;
                vy_read_iterator_add_tx(itr);
        }

        itr->cache_src = itr->src_count;
        vy_read_iterator_add_cache(itr, is_prepared_ok);

        itr->mem_src = itr->src_count;
        vy_read_iterator_add_mem(itr, is_prepared_ok);

        itr->disk_src = itr->src_count;
        vy_read_iterator_add_disk(itr);
}

/**
 * Iterate to the next range.
 */
static void
vy_read_iterator_next_range(struct vy_read_iterator *itr)
{
        struct vy_range *range = itr->curr_range;
        struct key_def *cmp_def = itr->lsm->cmp_def;
        int dir = iterator_direction(itr->iterator_type);

        assert(range != NULL);
        while (true) {
                range = dir > 0 ?
                        vy_range_tree_next(&itr->lsm->range_tree, range) :
                        vy_range_tree_prev(&itr->lsm->range_tree, range);
                assert(range != NULL);

                if (itr->last.stmt == NULL)
                        break;
                /*
                 * We could skip an entire range due to the cache.
                 * Make sure the next statement falls in the range.
                 */
                if (dir > 0 && (range->end.stmt == NULL ||
                                vy_entry_compare(itr->last, range->end,
                                                 cmp_def) < 0))
                        break;
                if (dir < 0 && (range->begin.stmt == NULL ||
                                vy_entry_compare(itr->last, range->begin,
                                                 cmp_def) > 0))
                        break;
        }
        itr->curr_range = range;
        itr->range_version = range->version;

        for (uint32_t i = itr->disk_src; i < itr->src_count; i++) {
                struct vy_read_src *src = &itr->src[i];
                vy_run_iterator_close(&src->run_iterator);
        }
        itr->src_count = itr->disk_src;

        vy_read_iterator_add_disk(itr);
}

/**
 * Get a resultant statement for the current key.
 * Returns 0 on success, -1 on error.
 */
static NODISCARD int
vy_read_iterator_apply_history(struct vy_read_iterator *itr,
                               struct vy_entry *ret)
{
        struct vy_lsm *lsm = itr->lsm;
        struct vy_history history;
        vy_history_create(&history, &lsm->env->history_node_pool);

        for (uint32_t i = 0; i < itr->src_count; i++) {
                struct vy_read_src *src = &itr->src[i];
                if (src->front_id == itr->front_id) {
                        vy_history_splice(&history, &src->history);
                        if (vy_history_is_terminal(&history))
                                break;
                }
        }

        int upserts_applied = 0;
        int rc = vy_history_apply(&history, lsm->cmp_def,
                                  true, &upserts_applied, ret);

        lsm->stat.upsert.applied += upserts_applied;
        vy_history_cleanup(&history);
        return rc;
}

/**
 * Track a read in the conflict manager.
 */
static int
vy_read_iterator_track_read(struct vy_read_iterator *itr, struct vy_entry entry)
{
        if (itr->tx == NULL)
                return 0;

        if (entry.stmt == NULL) {
                entry = (itr->iterator_type == ITER_EQ ||
                         itr->iterator_type == ITER_REQ ?
                         itr->key : itr->lsm->env->empty_key);
        }

        int rc;
        if (iterator_direction(itr->iterator_type) >= 0) {
                rc = vy_tx_track(itr->tx, itr->lsm, itr->key,
                                 itr->iterator_type != ITER_GT,
                                 entry, true);
        } else {
                rc = vy_tx_track(itr->tx, itr->lsm, entry, true,
                                 itr->key, itr->iterator_type != ITER_LT);
        }
        return rc;
}

NODISCARD int
vy_read_iterator_next(struct vy_read_iterator *itr, struct vy_entry *result)
{
        assert(itr->tx == NULL || itr->tx->state == VINYL_TX_READY);

        struct vy_entry entry;
next_key:
        if (vy_read_iterator_advance(itr) != 0)
                return -1;
        if (vy_read_iterator_apply_history(itr, &entry) != 0)
                return -1;
        if (vy_read_iterator_track_read(itr, entry) != 0)
                return -1;

        if (itr->last.stmt != NULL)
                tuple_unref(itr->last.stmt);
        itr->last = entry;

        if (entry.stmt != NULL && vy_stmt_type(entry.stmt) == IPROTO_DELETE) {
                /*
                 * We don't return DELETEs so skip to the next key.
                 * If the DELETE was read from TX write set, there
                 * is a good chance that the space actually has
                 * the deleted key and hence we must not consider
                 * previous + current tuple as an unbroken chain.
                 */
                if (vy_stmt_lsn(entry.stmt) == INT64_MAX) {
                        if (itr->last_cached.stmt != NULL)
                                tuple_unref(itr->last_cached.stmt);
                        itr->last_cached = vy_entry_none();
                }
                goto next_key;
        }
        assert(entry.stmt == NULL ||
               vy_stmt_type(entry.stmt) == IPROTO_INSERT ||
               vy_stmt_type(entry.stmt) == IPROTO_REPLACE);

        itr->check_exact_match = false;
        *result = entry;
        return 0;
}

void
vy_read_iterator_cache_add(struct vy_read_iterator *itr, struct vy_entry entry)
{
        if ((**itr->read_view).vlsn != INT64_MAX) {
                if (itr->last_cached.stmt != NULL)
                        tuple_unref(itr->last_cached.stmt);
                itr->last_cached = vy_entry_none();
                return;
        }
        vy_cache_add(&itr->lsm->cache, entry, itr->last_cached,
                     itr->is_first_cached, itr->key, itr->iterator_type);
        if (entry.stmt != NULL)
                tuple_ref(entry.stmt);
        if (itr->last_cached.stmt != NULL)
                tuple_unref(itr->last_cached.stmt);
        itr->last_cached = entry;
        itr->is_first_cached = false;
}

/**
 * Close the iterator and free resources
 */
void
vy_read_iterator_close(struct vy_read_iterator *itr)
{
        if (itr->last.stmt != NULL)
                tuple_unref(itr->last.stmt);
        if (itr->last_cached.stmt != NULL)
                tuple_unref(itr->last_cached.stmt);
        vy_read_iterator_cleanup(itr);
        free(itr->src);
        TRASH(itr);
}

tarantool / tarantool / 10633947029

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous