layout_db.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 2012-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.
 *
 */


#include "lib/errno.h"
#include "lib/memory.h"
#include "lib/misc.h"  /* memset() */
#include "lib/vec.h"   /* M0_BUFVEC_INIT_BUF() */
#include "lib/finject.h"

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

#include "layout/layout_internal.h"
#include "layout/layout_db.h"

M0_INTERNAL void m0_layout_pair_set(struct m0_db_pair *pair, uint64_t *lid,
                                    void *area, m0_bcount_t num_bytes)
{
        pair->dp_key.db_buf.b_addr = lid;
        pair->dp_key.db_buf.b_nob  = sizeof *lid;
        pair->dp_rec.db_buf.b_addr = area;
        pair->dp_rec.db_buf.b_nob  = num_bytes;
}

static int pair_init(struct m0_db_pair *pair,
                     struct m0_layout *l,
                     struct m0_db_tx *tx,
                     enum m0_layout_xcode_op op,
                     m0_bcount_t recsize)
{
        void                    *key_buf = pair->dp_key.db_buf.b_addr;
        void                    *rec_buf = pair->dp_rec.db_buf.b_addr;
        struct m0_bufvec         bv;
        struct m0_bufvec_cursor  rec_cur;
        int                      rc;

        M0_PRE(key_buf != NULL);
        M0_PRE(rec_buf != NULL);
        M0_PRE(pair->dp_key.db_buf.b_nob == sizeof l->l_id);
        M0_PRE(pair->dp_rec.db_buf.b_nob >= recsize);
        M0_PRE(M0_IN(op, (M0_LXO_DB_LOOKUP, M0_LXO_DB_ADD,
                          M0_LXO_DB_UPDATE, M0_LXO_DB_DELETE)));
        M0_PRE(recsize >= sizeof(struct m0_layout_rec));

        *(uint64_t *)key_buf = l->l_id;
        memset(pair->dp_rec.db_buf.b_addr, 0, pair->dp_rec.db_buf.b_nob);
        m0_db_pair_setup(pair, &l->l_dom->ld_layouts,
                         key_buf, sizeof l->l_id,
                         rec_buf, recsize);
        if (op == M0_LXO_DB_LOOKUP)
                rc = 0;
        else {
                bv = (struct m0_bufvec)M0_BUFVEC_INIT_BUF(&rec_buf,
                                                  &pair->dp_rec.db_buf.b_nob);
                m0_bufvec_cursor_init(&rec_cur, &bv);

                rc = m0_layout_encode(l, op, tx, &rec_cur);
                if (rc != 0) {
                        m0_layout__log("pair_init", "m0_layout_encode() failed",
                                       l->l_id, rc);
                        m0_db_pair_fini(pair);
                }
        }
        return M0_RC(rc);
}

M0_INTERNAL int m0_layout_lookup(struct m0_layout_domain *dom,
                                 uint64_t lid,
                                 struct m0_layout_type *lt,
                                 struct m0_db_tx *tx,
                                 struct m0_db_pair *pair,
                                 struct m0_layout **out)
{
        int                      rc;
        struct m0_bufvec         bv;
        struct m0_bufvec_cursor  cur;
        m0_bcount_t              max_recsize;
        m0_bcount_t              recsize;
        struct m0_layout        *l;
        struct m0_layout        *ghost;

        M0_PRE(m0_layout__domain_invariant(dom));
        M0_PRE(lid > 0);
        M0_PRE(lt != NULL);
        M0_PRE(tx != NULL);
        M0_PRE(pair != NULL);
        M0_PRE(out != NULL);

        M0_ENTRY("lid %llu", (unsigned long long)lid);
        if (dom->ld_type[lt->lt_id] != lt) {
                m0_layout__log("m0_layout_lookup", "Unregistered layout type",
                               lid, -EPROTO);
                return M0_ERR(-EPROTO);
        }

        m0_mutex_lock(&dom->ld_lock);
        l = m0_layout__list_lookup(dom, lid, true);
        m0_mutex_unlock(&dom->ld_lock);
        if (l != NULL) {
                /*
                 * Layout object exists in memory and m0_layout__list_lookup()
                 * has now acquired a reference on it.
                 */
                *out = l;
                M0_POST(m0_layout__invariant(*out));
                M0_LEAVE("lid %llu, rc %d", (unsigned long long)lid, 0);
                return 0;
        }

        /* Allocate outside of the domain lock to improve concurrency. */
        rc = lt->lt_ops->lto_allocate(dom, lid, &l);
        if (rc != 0) {
                m0_layout__log("m0_layout_lookup", "lto_allocate() failed",
                               lid, rc);
                return M0_RC(rc);
        }
        /* Here, lto_allocate() has locked l->l_lock. */

        if (M0_FI_ENABLED("ghost_creation")) {}

        /* Re-check for possible concurrent layout creation. */
        m0_mutex_lock(&dom->ld_lock);
        ghost = m0_layout__list_lookup(dom, lid, true);
        if (ghost != NULL) {
                /*
                 * Another instance of the layout with the same layout id
                 * "ghost" was created while the domain lock was released.
                 * Use it. m0_layout__list_lookup() has now acquired a
                 * reference on "ghost".
                 */
                m0_mutex_unlock(&dom->ld_lock);
                l->l_ops->lo_delete(l);

                /* Wait for possible decoding completion. */
                m0_mutex_lock(&ghost->l_lock);
                m0_mutex_unlock(&ghost->l_lock);

                *out = ghost;
                M0_POST(m0_layout__invariant(*out));
                M0_POST(m0_ref_read(&(*out)->l_ref) > 1);
                M0_LEAVE("lid %llu, ghost found, rc %d",
                         (unsigned long long)lid, 0);
                return 0;
        }
        m0_mutex_unlock(&dom->ld_lock);

        max_recsize = m0_layout_max_recsize(dom);
        recsize = pair->dp_rec.db_buf.b_nob <= max_recsize ?
                  pair->dp_rec.db_buf.b_nob : max_recsize;
        rc = pair_init(pair, l, tx, M0_LXO_DB_LOOKUP, recsize);
        M0_ASSERT(rc == 0);
        rc = m0_table_lookup(tx, pair);
        if (rc != 0) {
                /* Error covered in UT. */
                l->l_ops->lo_delete(l);
                m0_layout__log("m0_layout_lookup", "m0_table_lookup() failed",
                               lid, rc);
                goto out;
        }

        bv = (struct m0_bufvec)M0_BUFVEC_INIT_BUF(&pair->dp_rec.db_buf.b_addr,
                                                  &recsize);
        m0_bufvec_cursor_init(&cur, &bv);
        rc = m0_layout_decode(l, &cur, M0_LXO_DB_LOOKUP, tx);
        if (rc != 0) {
                /* Error covered in UT. */
                l->l_ops->lo_delete(l);
                m0_layout__log("m0_layout_lookup", "m0_layout_decode() failed",
                               lid, rc);
                goto out;
        }
        *out = l;
        M0_POST(m0_layout__invariant(*out) && m0_ref_read(&l->l_ref) > 0);
        m0_mutex_unlock(&l->l_lock);
out:
        m0_db_pair_fini(pair);
        M0_LEAVE("lid %llu, rc %d", (unsigned long long)lid, rc);
        return M0_RC(rc);
}

M0_INTERNAL int m0_layout_add(struct m0_layout *l,
                              struct m0_db_tx *tx, struct m0_db_pair *pair)
{
        m0_bcount_t recsize;
        int         rc;

        M0_PRE(m0_layout__invariant(l));
        M0_PRE(tx != NULL);
        M0_PRE(pair != NULL);

        M0_ENTRY("lid %llu", (unsigned long long)l->l_id);
        m0_mutex_lock(&l->l_lock);
        recsize = l->l_ops->lo_recsize(l);
        rc = pair_init(pair, l, tx, M0_LXO_DB_ADD, recsize);
        if (rc == 0) {
                rc = m0_table_insert(tx, pair);
                if (rc != 0)
                        m0_layout__log("m0_layout_add",
                                       "m0_table_insert() failed", l->l_id, rc);
                m0_db_pair_fini(pair);
        } else
                m0_layout__log("m0_layout_add", "pair_init() failed",
                               l->l_id, rc);
        m0_mutex_unlock(&l->l_lock);
        M0_LEAVE("lid %llu, rc %d", (unsigned long long)l->l_id, rc);
        return M0_RC(rc);
}

M0_INTERNAL int m0_layout_update(struct m0_layout *l,
                                 struct m0_db_tx *tx, struct m0_db_pair *pair)
{
        m0_bcount_t recsize;
        int         rc;

        M0_PRE(m0_layout__invariant(l));
        M0_PRE(tx != NULL);
        M0_PRE(pair != NULL);

        M0_ENTRY("lid %llu", (unsigned long long)l->l_id);
        m0_mutex_lock(&l->l_lock);
        recsize = l->l_ops->lo_recsize(l);
        rc = pair_init(pair, l, tx, M0_LXO_DB_UPDATE, recsize);
        if (rc == 0) {
                if (M0_FI_ENABLED("table_update_err"))
                        { rc = L_TABLE_UPDATE_ERR; goto err1_injected; }
                rc = m0_table_update(tx, pair);
err1_injected:
                if (rc != 0)
                        m0_layout__log("m0_layout_update",
                                       "m0_table_update() failed", l->l_id, rc);
                m0_db_pair_fini(pair);
        } else
                m0_layout__log("m0_layout_update",
                               "pair_init() failed", l->l_id, rc);
        m0_mutex_unlock(&l->l_lock);
        M0_LEAVE("lid %llu, rc %d", (unsigned long long)l->l_id, rc);
        return M0_RC(rc);
}

M0_INTERNAL int m0_layout_delete(struct m0_layout *l,
                                 struct m0_db_tx *tx, struct m0_db_pair *pair)
{
        m0_bcount_t recsize;
        int         rc;

        M0_PRE(m0_layout__invariant(l));
        M0_PRE(tx != NULL);
        M0_PRE(pair != NULL);

        M0_ENTRY("lid %llu", (unsigned long long)l->l_id);
        m0_mutex_lock(&l->l_lock);
        if (l->l_user_count > 0) {
                M0_LOG(M0_ERROR, "lid %llu, user_count %lu, Invalid "
                       "user_count, rc %d", (unsigned long long)l->l_id,
                       (unsigned long)l->l_user_count, -EPROTO);
                m0_mutex_unlock(&l->l_lock);
                return M0_ERR(-EPROTO);
        }

        recsize = l->l_ops->lo_recsize(l);
        rc = pair_init(pair, l, tx, M0_LXO_DB_DELETE, recsize);
        if (rc == 0) {
                rc = m0_table_delete(tx, pair);
                if (rc != 0)
                        m0_layout__log("m0_layout_delete",
                                       "m0_table_delete() failed", l->l_id, rc);
                m0_db_pair_fini(pair);
        } else
                m0_layout__log("m0_layout_delete",
                               "pair_init() failed", l->l_id, rc);
        m0_mutex_unlock(&l->l_lock);
        M0_LEAVE("lid %llu, rc %d", (unsigned long long)l->l_id, rc);
        return M0_RC(rc);
}

#undef M0_TRACE_SUBSYSTEM

/*
 *  Local variables:
 *  c-indentation-style: "K&R"
 *  c-basic-offset: 8
 *  tab-width: 8
 *  fill-column: 79
 *  scroll-step: 1
 *  End:
 */