recovery.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 2014-2020 Seagate Technology LLC and/or its Affiliates
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * For any questions about this software or licensing,
 * please email opensource@seagate.com or cortx-questions@seagate.com.
 *
 */


#define M0_TRACE_SUBSYSTEM M0_TRACE_SUBSYS_BE
#include "lib/trace.h"

#include "be/recovery.h"

#include "lib/arith.h"          /* max_check */
#include "lib/errno.h"          /* -ENOSYS */
#include "lib/memory.h"
#include "be/fmt.h"
#include "be/log.h"
#include "motr/magic.h"         /* M0_BE_RECOVERY_MAGIC */

M0_TL_DESCR_DEFINE(log_record_iter, "m0_be_log_record_iter list in recovery",
                   static, struct m0_be_log_record_iter, lri_linkage, lri_magic,
                   M0_BE_RECOVERY_MAGIC, M0_BE_RECOVERY_HEAD_MAGIC);
M0_TL_DEFINE(log_record_iter, static, struct m0_be_log_record_iter);

M0_INTERNAL void m0_be_recovery_init(struct m0_be_recovery     *rvr,
                                     struct m0_be_recovery_cfg *cfg)
{
        rvr->brec_cfg = *cfg;
        m0_mutex_init(&rvr->brec_lock);
        log_record_iter_tlist_init(&rvr->brec_iters);
}

M0_INTERNAL void m0_be_recovery_fini(struct m0_be_recovery *rvr)
{
        m0_mutex_fini(&rvr->brec_lock);
        log_record_iter_tlist_fini(&rvr->brec_iters);
}

static int be_recovery_log_record_iter_new(struct m0_be_log_record_iter **iter)
{
        int rc;

        M0_ALLOC_PTR(*iter);
        rc = *iter == NULL ? -ENOMEM : 0;
        rc = rc ?: m0_be_log_record_iter_init(*iter);

        return rc;
}

static void
be_recovery_log_record_iter_destroy(struct m0_be_log_record_iter *iter)
{
        if (iter != NULL) {
                m0_be_log_record_iter_fini(iter);
                m0_free(iter);
        }
}

M0_INTERNAL int m0_be_recovery_run(struct m0_be_recovery *rvr)
{
        struct m0_be_fmt_log_header   log_hdr = {};
        struct m0_be_log_record_iter *prev    = NULL;
        struct m0_be_log_record_iter *iter;
        struct m0_be_log             *log = rvr->brec_cfg.brc_log;
        m0_bindex_t                   last_discarded;
        m0_bindex_t                   log_discarded;
        m0_bindex_t                   next_pos = M0_BINDEX_MAX;
        int                           rc;

        M0_ENTRY("rvr = %p, log = %p", rvr, log);

        /* TODO avoid reading of header from disk, log reads it during init */
        rc = m0_be_fmt_log_header_init(&log_hdr, NULL);
        M0_ASSERT(rc == 0);
        rc = m0_be_log_header_read(log, &log_hdr);
        M0_ASSERT(rc == 0);
        log_discarded = log_hdr.flh_discarded;
        M0_LOG(M0_DEBUG, "log_discarded=%"PRIu64, log_discarded);

        rc = be_recovery_log_record_iter_new(&iter);
        rc = rc ?: m0_be_log_record_initial(log, iter);
        while (rc == 0) {
                log_record_iter_tlink_init_at_tail(iter, &rvr->brec_iters);
                prev = iter;
                M0_LOG(M0_DEBUG, "Forward movement : prev header "
                                  BFLRH_F, BFLRH_P(&prev->lri_header));
                rc = be_recovery_log_record_iter_new(&iter);
                rc = rc ?: m0_be_log_record_next(log, prev, iter);
        }
        be_recovery_log_record_iter_destroy(iter);
        if (!M0_IN(rc, (0, -ENOENT))){
                M0_LOG(M0_ERROR, "Error: rc=%d", rc);
                goto err;
        }
        if (rc == -ENOENT && prev == NULL){
                M0_LOG(M0_DEBUG, "Empty");
                goto empty;
        }

        last_discarded = max_check(prev->lri_header.lrh_discarded,
                                   log_discarded);
        prev = log_record_iter_tlist_head(&rvr->brec_iters);
        while (prev != NULL && prev->lri_header.lrh_pos < last_discarded) {
                next_pos = prev->lri_header.lrh_pos + prev->lri_header.lrh_size;
                M0_LOG(M0_DEBUG, "next_pos=%"PRIu64, next_pos);
                log_record_iter_tlink_del_fini(prev);
                be_recovery_log_record_iter_destroy(prev);
                prev = log_record_iter_tlist_head(&rvr->brec_iters);
        }
        M0_ASSERT(ergo(prev == NULL,
                       last_discarded == log_discarded &&
                       last_discarded == next_pos));
        if (prev == NULL){
                M0_LOG(M0_DEBUG, "Empty");
                goto empty;
        }

        rc = 0;
        while (rc == 0 && prev->lri_header.lrh_pos > last_discarded) {
                rc = be_recovery_log_record_iter_new(&iter);
                rc = rc ?: m0_be_log_record_prev(log, prev, iter);
                if (rc == 0) {
                        log_record_iter_tlink_init_at(iter, &rvr->brec_iters);
                        prev = iter;
                        M0_LOG(M0_DEBUG, "Backward movement : prev header "
                                          BFLRH_F, BFLRH_P(&(prev->lri_header)));
                } else
                        be_recovery_log_record_iter_destroy(iter);
        }
        M0_ASSERT(ergo(rc == 0, prev->lri_header.lrh_pos == last_discarded));

        if (rc != 0){
                M0_LOG(M0_ERROR, "Error: rc=%d", rc);
                goto err;
        }

        iter = log_record_iter_tlist_tail(&rvr->brec_iters);

        rvr->brec_last_record_pos  = iter->lri_header.lrh_pos;
        rvr->brec_last_record_size = iter->lri_header.lrh_size;
        rvr->brec_current          = rvr->brec_last_record_pos +
                                     rvr->brec_last_record_size;
        rvr->brec_discarded        = prev->lri_header.lrh_pos;
        M0_LOG(M0_DEBUG, "Recovery scan complete : last_record_pos=%"PRIu64
                         " last_record_size=%"PRIu64
                         " current position=%"PRIu64
                         " discarded position=%"PRIu64,
                         rvr->brec_last_record_pos,
                         rvr->brec_last_record_size,
                         rvr->brec_current,
                         rvr->brec_discarded);

        goto out;

err:
        M0_LOG(M0_ERROR, "Error during recovery scan. rc = %d", rc);
        m0_tl_for(log_record_iter, &rvr->brec_iters, iter) {
                log_record_iter_tlink_del_fini(iter);
                be_recovery_log_record_iter_destroy(iter);
        } m0_tl_endfor;
        goto out;

empty:
        rc = 0;
        rvr->brec_last_record_pos  = log_hdr.flh_group_lsn;
        rvr->brec_last_record_size = log_hdr.flh_group_size;
        rvr->brec_current          = log_discarded;
        rvr->brec_discarded        = log_discarded;
        M0_LOG(M0_DEBUG, "Empty Logs : last_record_pos=%"PRIu64
                         " last_record_size=%"PRIu64
                         " current position=%"PRIu64
                         " discarded position=%"PRIu64,
                         rvr->brec_last_record_pos,
                         rvr->brec_last_record_size,
                         rvr->brec_current,
                         rvr->brec_discarded);
        M0_POST(log_record_iter_tlist_is_empty(&rvr->brec_iters));
out:
        m0_be_fmt_log_header_fini(&log_hdr);
        M0_LEAVE();
        return rc;
}

M0_INTERNAL bool
m0_be_recovery_log_record_available(struct m0_be_recovery *rvr)
{
        bool result;

        m0_mutex_lock(&rvr->brec_lock);
        result = !log_record_iter_tlist_is_empty(&rvr->brec_iters);
        m0_mutex_unlock(&rvr->brec_lock);

        return result;
}

M0_INTERNAL void
m0_be_recovery_log_record_get(struct m0_be_recovery        *rvr,
                              struct m0_be_log_record_iter *iter)
{
        struct m0_be_log_record_iter *next;

        m0_mutex_lock(&rvr->brec_lock);
        next = log_record_iter_tlist_pop(&rvr->brec_iters);
        m0_mutex_unlock(&rvr->brec_lock);
        M0_ASSERT(next != NULL);
        m0_be_log_record_iter_copy(iter, next);
        log_record_iter_tlink_fini(next);
        be_recovery_log_record_iter_destroy(next);
        M0_LOG(M0_DEBUG, "iter pos=%" PRId64 " size=%" PRId64 " discarded=%"PRId64,
               iter->lri_header.lrh_pos,
               iter->lri_header.lrh_size,
               iter->lri_header.lrh_discarded);
}

/*
 *  Local variables:
 *  c-indentation-style: "K&R"
 *  c-basic-offset: 8
 *  tab-width: 8
 *  fill-column: 80
 *  scroll-step: 1
 *  End:
 */
/*
 * vim: tabstop=8 shiftwidth=8 noexpandtab textwidth=80 nowrap
 */