obj.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 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 "motr/client.h"
#include "motr/client_internal.h"
#include "motr/layout.h"
#include "motr/idx.h"
#include "motr/sync.h"

#include "lib/errno.h"
#include "fid/fid.h"               /* m0_fid */
#include "lib/locality.h"          /* m0_locality_here() */
#include "lib/misc.h"              /* m0_strtou64() */
#include "ioservice/fid_convert.h" /* m0_fid_convert_ */
#include "layout/layout.h"         /* m0_lid_to_unit_map */
#include "conf/helpers.h"          /* m0_confc_root_open */

#define M0_TRACE_SUBSYSTEM M0_TRACE_SUBSYS_CLIENT
#include "lib/trace.h"
#include "lib/finject.h"


M0_INTERNAL struct m0_client*
m0__oo_instance(struct m0_op_obj *oo)
{
        M0_PRE(oo != NULL);

        return m0__entity_instance(oo->oo_oc.oc_op.op_entity);
}

M0_INTERNAL struct m0_obj*
m0__obj_entity(struct m0_entity *entity)
{
        struct m0_obj *obj;

        M0_PRE(entity != NULL);

        return M0_AMB(obj, entity, ob_entity);
}

static bool m0_op_common_invariant(struct m0_op_common *oc)
{
        return M0_RC(oc != NULL &&
                     m0_op_common_bob_check(oc));
}

M0_INTERNAL bool m0_op_obj_ast_rc_invariant(struct m0_ast_rc *ar)
{
        return M0_RC(ar != NULL &&
                     m0_ast_rc_bob_check(ar));
}

M0_INTERNAL bool m0_op_obj_invariant(struct m0_op_obj *oo)
{
        return M0_RC(oo != NULL &&
                     m0_op_obj_bob_check(oo) &&
                     oo->oo_oc.oc_op.op_size >= sizeof *oo &&
                     m0_op_obj_ast_rc_invariant(&oo->oo_ar) &&
                     m0_op_common_invariant(&oo->oo_oc));
}

static bool obj_op_obj_invariant(struct m0_op_obj *oo)
{
        bool rc;

        /* Don't use a m0__xxx_instance, as they assert */
        if(oo == NULL)
                rc = false;
        else if(oo->oo_oc.oc_op.op_entity == NULL)
                rc = false;
        else if(oo->oo_oc.oc_op.op_entity->en_realm == NULL)
                rc = false;
        else if(oo->oo_oc.oc_op.op_entity->en_realm->re_instance == NULL)
                rc = false;
        else
                rc = true;

        return M0_RC(rc);
}

static bool obj_layout_id_invariant(uint64_t layout_id)
{

        int      ltype = M0_OBJ_LAYOUT_TYPE(layout_id);
        uint64_t lid = M0_OBJ_LAYOUT_ID(layout_id);

        return M0_RC(M0_IN(ltype, (M0_LT_PDCLUST,
                                   M0_LT_COMPOSITE,
                                   M0_LT_CAPTURE)) &&
                     lid > 0 && lid < m0_lid_to_unit_map_nr);
}

M0_INTERNAL uint64_t m0__obj_lid(struct m0_obj *obj)
{
        uint64_t lid;

        M0_ENTRY();
        M0_PRE(obj != NULL);
        M0_PRE(obj_layout_id_invariant(obj->ob_attr.oa_layout_id));
        lid = M0_OBJ_LAYOUT_ID(obj->ob_attr.oa_layout_id);
        M0_LEAVE();
        return lid;
}

M0_INTERNAL enum m0_client_layout_type
m0__obj_layout_type(struct m0_obj *obj)
{
        int type;

        M0_ENTRY();
        M0_PRE(obj != NULL);
        M0_PRE(obj_layout_id_invariant(obj->ob_attr.oa_layout_id));
        type = M0_OBJ_LAYOUT_TYPE(obj->ob_attr.oa_layout_id);
        M0_LEAVE();
        return type;
}

M0_INTERNAL void m0__obj_attr_set(struct m0_obj *obj,
                                  struct m0_fid pver,
                                  uint64_t layout_id)
{
        M0_ENTRY();
        M0_PRE(obj != NULL);
        M0_PRE(obj_layout_id_invariant(layout_id));
        obj->ob_attr.oa_layout_id = layout_id;
        obj->ob_attr.oa_pver = pver;
        M0_LEAVE();
}

M0_INTERNAL struct m0_fid m0__obj_pver(struct m0_obj *obj)
{
        return obj->ob_attr.oa_pver;
}

M0_INTERNAL bool
m0__obj_pool_version_is_valid(const struct m0_obj *obj)
{
        struct m0_client *cinst;

        M0_PRE(obj != NULL);
        cinst = m0__entity_instance(&obj->ob_entity);
        return m0_conf_fid_is_valid(&obj->ob_attr.oa_pver) && (
               m0_pool_version_find(&cinst->m0c_pools_common,
                                    &obj->ob_attr.oa_pver) != NULL);
}

static void obj_namei_cb_cancel(struct m0_op_common *oc)
{
        int               rc;
        struct m0_op_obj *oo;
        struct m0_op     *op;

        M0_ENTRY();

        M0_PRE(oc != NULL);
        op = &oc->oc_op;
        M0_PRE(op->op_entity != NULL);
        M0_PRE(m0_sm_group_is_locked(&op->op_sm_group));
        M0_PRE(M0_IN(op->op_code, (M0_EO_CREATE,
                                   M0_EO_DELETE,
                                   M0_EO_OPEN)));

        oo = bob_of(oc, struct m0_op_obj, oo_oc, &oo_bobtype);
        M0_PRE(obj_op_obj_invariant(oo));

        rc = m0__obj_namei_cancel(op);
        op->op_rc = rc;
        M0_LEAVE();
}

static void obj_namei_cb_launch(struct m0_op_common *oc)
{
        int               rc;
        struct m0_op_obj *oo;
        struct m0_op     *op;

        M0_ENTRY();

        M0_PRE(oc != NULL);
        op = &oc->oc_op;
        M0_PRE(op->op_entity != NULL);
        M0_PRE(m0_uint128_cmp(&M0_ID_APP,
                              &op->op_entity->en_id) < 0);
        M0_PRE(M0_IN(op->op_code, (M0_EO_CREATE,
                                   M0_EO_DELETE,
                                   M0_EO_OPEN)));
        M0_PRE(M0_IN(op->op_entity->en_sm.sm_state, (M0_ES_INIT,
                                                     M0_ES_OPEN)));
        M0_PRE(m0_sm_group_is_locked(&op->op_sm_group));
        M0_ASSERT(entity_invariant_full(op->op_entity));

        oo = bob_of(oc, struct m0_op_obj, oo_oc, &oo_bobtype);
        M0_PRE(obj_op_obj_invariant(oo));

        /* Move to a different state and call the control function. */
        m0_sm_group_lock(&op->op_entity->en_sm_group);
        switch (op->op_code) {
                case M0_EO_CREATE:
                        m0_sm_move(&op->op_entity->en_sm, 0,
                                   M0_ES_CREATING);
                        break;
                case M0_EO_DELETE:
                        m0_sm_move(&op->op_entity->en_sm, 0,
                                   M0_ES_DELETING);
                        break;
                case M0_EO_OPEN:
                        m0_sm_move(&op->op_entity->en_sm, 0,
                                   M0_ES_OPENING);
                        break;
                default:
                        M0_IMPOSSIBLE("Management operation not implemented");
        }
        m0_sm_group_unlock(&op->op_entity->en_sm_group);

        rc = m0__obj_namei_send(oo);
        if (rc == 0) {
                m0_sm_move(&op->op_sm, 0, M0_OS_LAUNCHED);
        } else if (rc == MOTR_MDCOB_LOOKUP_SKIP) {
                /* This means meta-data lookup is skipped and op state is
                 * already moved to LAUNCHED --> EXECUTED --> STABLE, so
                 * skipped m0_sm_move() and resetting rc state to 0
                 * */
                (void)rc;
        }

        M0_LEAVE();
}

static void obj_namei_cb_free(struct m0_op_common *oc)
{
        struct m0_op_obj *oo;

        M0_ENTRY();

        M0_PRE(oc != NULL);
        M0_PRE((oc->oc_op.op_size >= sizeof *oo));

        /* By now, fini() has been called and bob_of cannot be used */
        oo = M0_AMB(oo, oc, oo_oc);
        M0_PRE(obj_op_obj_invariant(oo));

        m0_free(oo);

        M0_LEAVE();
}

static void obj_namei_cb_fini(struct m0_op_common *oc)
{
        struct m0_op_obj *oo;
        struct m0_client *cinst;

        M0_ENTRY();
        M0_PRE(oc != NULL);
        M0_PRE(oc->oc_op.op_size >= sizeof *oo);

        oo = bob_of(oc, struct m0_op_obj, oo_oc, &oo_bobtype);
        M0_PRE(m0_op_obj_invariant(oo));

        cinst = m0__oo_instance(oo);
        M0_ASSERT(cinst != NULL);

        if (oo->oo_layout_instance != NULL) {
                m0_layout_instance_fini(oo->oo_layout_instance);
                oo->oo_layout_instance = NULL;
        }

        m0_op_common_bob_fini(&oo->oo_oc);
        m0_ast_rc_bob_fini(&oo->oo_ar);
        m0_op_obj_bob_fini(oo);

        M0_SET0(&oo->oo_fid);

#ifdef CLIENT_FOR_M0T1FS
        M0_SET0(&oo->oo_pfid);
        m0_buf_free(&oo->oo_name);
#endif

        M0_LEAVE();
}

M0_INTERNAL int
m0__obj_pool_version_get(struct m0_obj *obj,
                         struct m0_pool_version **pv)
{
        int               rc;
        struct m0_client *cinst;
        struct m0_fid    *pool;

        M0_ENTRY();

        cinst = m0__obj_instance(obj);

        if (pv == NULL)
                return M0_ERR(-ENOENT);

        if (M0_FI_ENABLED("fake_pool_version")) {
                *pv = cinst->m0c_pools_common.pc_cur_pver;
                return 0;
        }

        if (m0__obj_pool_version_is_valid(obj)) {
                *pv = m0_pool_version_find(&cinst->m0c_pools_common,
                                           &obj->ob_attr.oa_pver);
                rc = (*pv != NULL)? 0 : -ENOENT;
        } else {
                pool = m0_fid_is_set(&obj->ob_attr.oa_pool)?
                       &obj->ob_attr.oa_pool : NULL;
                rc = m0_pool_version_get(&cinst->m0c_pools_common, pool, pv);
                if (rc != 0)
                        return M0_ERR(rc);
                M0_ASSERT(*pv != NULL);
                obj->ob_attr.oa_pver = (*pv)->pv_id;
        }

        return M0_RC(rc);
}

M0_INTERNAL uint64_t
m0__obj_layout_id_get(struct m0_op_obj *oo)
{
        uint64_t       lid;
        struct m0_obj *obj;

        M0_ENTRY();

        if (M0_FI_ENABLED("fake_obj_layout_id"))
                return 1;

        obj = M0_AMB(obj, oo->oo_oc.oc_op.op_entity, ob_entity);
        M0_ASSERT(obj_layout_id_invariant(obj->ob_attr.oa_layout_id));
        lid = M0_OBJ_LAYOUT_ID(obj->ob_attr.oa_layout_id);

        M0_LEAVE();
        return lid;
}

M0_INTERNAL int
m0__obj_layout_instance_build(struct m0_client *cinst,
                              const uint64_t layout_id,
                              const struct m0_fid *fid,
                              struct m0_layout_instance **linst)
{
        int               rc = -EINVAL;
        struct m0_layout *layout;

        M0_PRE(cinst != NULL);
        M0_PRE(linst != NULL);
        M0_PRE(fid != NULL);

        /*
         * All the layouts should already be generated on startup and added
         * to the list unless wrong layout_id is used.
         */
        layout = m0_layout_find(&cinst->m0c_reqh.rh_ldom, layout_id);
        if (layout != NULL) {
                *linst = NULL;
                rc = m0_layout_instance_build(layout, fid, linst);
                m0_layout_put(layout);
        }

        return M0_RC(rc);
}

#ifdef CLIENT_FOR_M0T1FS

int obj_fid_make_name(char *name, size_t name_len,
                      const struct m0_fid *fid)
{
        int rc;

        M0_PRE(name != NULL);
        M0_PRE(name_len > 0);
        M0_PRE(fid != NULL);

        rc = snprintf(name, name_len, "%" PRIx64 ":%"PRIx64, FID_P(fid));
        M0_ASSERT(rc >= 0 && rc < name_len);
        return M0_RC(0);
}
#endif

static int obj_namei_op_init(struct m0_entity *entity,
                             struct m0_op *op)
{
        int                         rc;
        char                       *obj_name;
        uint64_t                    obj_key;
        uint64_t                    obj_container;
        uint64_t                    lid;
        struct m0_op_obj           *oo;
        struct m0_op_common        *oc;
        struct m0_layout_instance  *linst;
        struct m0_client           *cinst;

        M0_ENTRY();

        M0_PRE(entity != NULL);
        M0_PRE(op != NULL);

        if (M0_FI_ENABLED("fake_msg_size")) {
                rc = M0_ERR(-EMSGSIZE);
                goto error;
        }

        if (op->op_size < sizeof *oo) {
                rc = M0_ERR(-EMSGSIZE);
                goto error;
        }

        oc = bob_of(op, struct m0_op_common, oc_op, &oc_bobtype);
        M0_PRE(m0_op_common_invariant(oc));
        oo = bob_of(oc, struct m0_op_obj, oo_oc, &oo_bobtype);
        M0_PRE(m0_op_obj_invariant(oo));

        cinst = m0__oo_instance(oo);
        M0_ASSERT(cinst != NULL);

        /* Set the op_common's callbacks. */
        oc->oc_cb_launch = obj_namei_cb_launch;
        oc->oc_cb_cancel = obj_namei_cb_cancel;
        oc->oc_cb_fini = obj_namei_cb_fini;
        oc->oc_cb_free = obj_namei_cb_free;

        /*
         * Set the object's fid.
         *
         * Something about the fid at current motr implementation (fid_convert.h)
         * (1) fid is 128 bits long,  global fid and cob fid both use the highest
         * 8 bits to represent object type and the lowest 96 bits to store object
         * id. The interpretion of these 96 bits depends on the users. For
         * example, as the name of fid::f_container suggests, the 32 bits (or
         * any number of bits) in f_container can be viewed as 'application
         * container' id, so supporting multiple application containers is
         * possible in current Motr implementation.
         *
         * (2) The difference of global fid and cob fid is in the 24 bits in
         * fid::f_container. cob fid uses these 24 bits to store device id in a
         * layout (md is 0, and devices in ioservices ranges from 1 to P).
         *
         * (3) Does Client need to check if an object's container id matches
         * the container id inside its fid?
         */
        obj_container = entity->en_id.u_hi;
        obj_key = entity->en_id.u_lo;
        m0_fid_gob_make(&oo->oo_fid, obj_container, obj_key);

        /* Get a layout instance for the object. */
        lid = m0_pool_version2layout_id(&oo->oo_pver,
                                        m0__obj_layout_id_get(oo));
        rc = m0__obj_layout_instance_build(cinst, lid,
                                           &oo->oo_fid, &linst);
        if (rc != 0) {
                M0_ERR(rc);
                goto error;
        }
        oo->oo_layout_instance = linst;

#ifdef CLIENT_FOR_M0T1FS
        /* Set the object's parent's fid. */
        if (!m0_fid_is_set(&cinst->m0c_root_fid) ||
            !m0_fid_is_valid(&cinst->m0c_root_fid)) {
                rc = M0_ERR(-EINVAL);
                goto error;
        }
        oo->oo_pfid = cinst->m0c_root_fid;

        /* Generate a valid oo_name. */
        obj_name = m0_alloc(M0_OBJ_NAME_MAX_LEN);
        rc = obj_fid_make_name(obj_name, M0_OBJ_NAME_MAX_LEN, &oo->oo_fid);
        if (rc != 0) {
                M0_ERR(rc);
                goto error;
        }
        m0_buf_init(&oo->oo_name, obj_name, strlen(obj_name));
#endif
        M0_ASSERT(rc == 0);
        return M0_RC(rc);

error:
        M0_ASSERT(rc != 0);
        return M0_ERR(rc);
}

static void obj_optimal_lid_set(struct m0_obj *obj,
                                struct m0_layout_domain *ldom)
{
        uint64_t *lid;

        lid = &obj->ob_attr.oa_layout_id;

        /* Find optimal layout id when pver id is set and layout id is not */
        if (*lid == 0 && m0_fid_is_set(&obj->ob_attr.oa_pver)) {
                *lid = m0_layout_find_by_buffsize(ldom, &obj->ob_attr.oa_pver,
                                                  obj->ob_attr.oa_buf_size);
        }
        /* Set default layout id when both layout id and pver id is unset */
        else if (*lid == 0 && !m0_fid_is_set(&obj->ob_attr.oa_pver)) {
                *lid = M0_DEFAULT_LAYOUT_ID;
        }
}

static int obj_op_obj_init(struct m0_op_obj *oo)
{
        int                      rc;
        struct m0_locality      *locality;
        struct m0_pool_version  *pv;
        struct m0_obj           *obj;
        struct m0_client        *cinst;
        struct m0_layout_domain *ldom;

        M0_ENTRY();
        M0_PRE(oo != NULL);
        M0_PRE(M0_IN(OP_OBJ2CODE(oo), (M0_EO_CREATE,
                                       M0_EO_DELETE,
                                       M0_EO_OPEN)));

        obj = m0__obj_entity(oo->oo_oc.oc_op.op_entity);
        cinst = m0__obj_instance(obj);
        if (OP_OBJ2CODE(oo) == M0_EO_CREATE) {
                rc = m0__obj_pool_version_get(obj, &pv);
                if (rc != 0)
                        return M0_ERR(rc);
                oo->oo_pver = pv->pv_id;

                ldom = &cinst->m0c_reqh.rh_ldom;
                obj_optimal_lid_set(obj, ldom);
        } else if (OP_OBJ2CODE(oo) == M0_EO_OPEN) {
                /*
                 * XXX:Not required to assign pool version for operation other
                 *     than OBJECT CREATE.
                 * OBJECT OPEN operation will fetch pool version from meta-data
                 * and cache it to m0_obj::ob_layout::oa_pver
                 * MOTR-2871 will fix and verify this issue separately.
                 */
                        pv = m0_pool_version_md_get(&cinst->m0c_pools_common);
                        M0_ASSERT(pv != NULL);
                        oo->oo_pver = pv->pv_id;
        } else {
                oo->oo_pver = m0__obj_pver(obj);
        }
        locality = m0__locality_pick(m0__oo_instance(oo));
        M0_ASSERT(locality != NULL);
        oo->oo_sm_grp = locality->lo_grp;
        M0_SET0(&oo->oo_ar);

        oo->oo_layout_instance = NULL;
        M0_SET0(&oo->oo_fid);

#ifdef CLIENT_FOR_M0T1FS
        M0_SET0(&oo->oo_pfid);
        M0_SET0(&oo->oo_name);
#endif

        m0_op_obj_bob_init(oo);
        m0_ast_rc_bob_init(&oo->oo_ar);

        M0_POST(m0_op_obj_invariant(oo));
        return M0_RC(0);
}

static int obj_op_prepare(struct m0_entity *entity,
                          struct m0_op **op,
                          enum m0_entity_opcode opcode)
{
        int    rc;
        bool   alloced = false;
        struct m0_op_common *oc;
        struct m0_op_obj    *oo;

        M0_ENTRY();

        M0_PRE(entity != NULL);
        M0_ASSERT(entity_invariant_full(entity));
        M0_PRE(M0_IN(entity->en_sm.sm_state, (M0_ES_INIT,
                                              M0_ES_OPEN)));
        M0_PRE(op != NULL);
        /* We may want to gain this lock, check we don't already hold it. */
        M0_PRE(*op == NULL ||
               !m0_sm_group_is_locked(&(*op)->op_sm_group));

        /* Allocate the op if necessary. */
        if (*op == NULL) {
                rc = m0_op_alloc(op, sizeof *oo);
                if (rc != 0)
                        return M0_ERR(rc);
                alloced = true;
        } else {
                size_t cached_size = (*op)->op_size;

                if ((*op)->op_size < sizeof *oo)
                        return M0_ERR(-EMSGSIZE);

                /* 0 the pre-allocated operation. */
                memset(*op, 0, cached_size);
                (*op)->op_size = cached_size;
        }
        m0_mutex_init(&(*op)->op_pending_tx_lock);
        spti_tlist_init(&(*op)->op_pending_tx);

        /* Initialise the operation's generic part. */
        (*op)->op_code = opcode;
        rc = m0_op_init(*op, &m0_op_conf, entity);
        if (rc != 0)
                goto op_free;

        /* No bob_init()'s have been called yet: we use M0_AMB(). */
        oc = M0_AMB(oc, *op, oc_op);
        m0_op_common_bob_init(oc);

        /* Init the m0_op_obj part. */
        oo = M0_AMB(oo, oc, oo_oc);
        rc = obj_op_obj_init(oo);
        if (rc != 0)
                goto op_fini;

        return M0_RC(0);

op_fini:
        m0_op_fini(*op);
op_free:
        if (alloced) {
                m0_op_free(*op);
                *op = NULL;
        }

        return M0_RC(rc);
}

static int entity_namei_op(struct m0_entity *entity,
                           struct m0_op **op,
                           enum m0_entity_opcode opcode)
{
        int  rc;
        bool pre_allocated = false;

        M0_ENTRY();

        M0_PRE(entity != NULL);
        M0_ASSERT(entity_invariant_full(entity));
        M0_PRE(M0_IN(entity->en_sm.sm_state, (M0_ES_INIT,
                                              M0_ES_OPEN)));
        M0_PRE(op != NULL);
        M0_PRE(M0_IN(opcode, (M0_EO_CREATE,
                              M0_EO_DELETE,
                              M0_EO_OPEN)));

        switch (entity->en_type) {
        case M0_ET_OBJ:
                pre_allocated = (*op != NULL);
                /* Allocate an op on an object and initialise common stuff. */
                rc = obj_op_prepare(entity, op, opcode);
                if (rc != 0)
                        goto error;

                /* Initialise the stuff specific to a obj namespace operation. */
                rc = obj_namei_op_init(entity, *op);
                if (rc != 0)
                        goto op_fini;
                break;
        case M0_ET_IDX:
                rc = m0_idx_op_namei(entity, op, opcode);
                if (rc != 0)
                        goto error;
                break;
        default:
                M0_IMPOSSIBLE("Entity type not yet implemented.");
        }

        M0_POST(rc == 0);
        M0_POST(*op != NULL);
        m0_sm_group_lock(&(*op)->op_sm_group);
        M0_POST((*op)->op_sm.sm_rc == 0);
        m0_sm_group_unlock(&(*op)->op_sm_group);

        return M0_RC(0);
op_fini:
        m0_op_fini(*op);
        if (!pre_allocated) {
                m0_op_free(*op);
                *op = NULL;
        }
error:
        M0_ASSERT(rc != 0);
        return M0_ERR(rc);
}

int m0_entity_create(struct m0_fid *pool,
                     struct m0_entity *entity,
                     struct m0_op **op)
{
        struct m0_obj *obj;

        M0_ENTRY();
        if (entity->en_flags & M0_ENF_META)
                M0_LOG(M0_DEBUG, "M0_ENF_META FLAG is set from application");

        M0_PRE(entity != NULL);
        M0_PRE(op != NULL);

        /* Currently, pool selection is only for objects. */
        if (entity->en_type == M0_ET_OBJ && pool != NULL) {
                obj = M0_AMB(obj, entity, ob_entity);
                obj->ob_attr.oa_pool = *pool;
        }

        return M0_RC(entity_namei_op(entity, op, M0_EO_CREATE));
}
M0_EXPORTED(m0_entity_create);

int m0_entity_delete(struct m0_entity *entity,
                     struct m0_op **op)
{
        M0_ENTRY();

        M0_PRE(entity != NULL);
        M0_PRE(op != NULL);

        return M0_RC(entity_namei_op(entity, op, M0_EO_DELETE));
}
M0_EXPORTED(m0_entity_delete);

uint64_t m0_obj_unit_size_to_layout_id(int unit_size)
{
        uint64_t i;

        for (i = 0; i < m0_lid_to_unit_map_nr; i++)
                if (m0_lid_to_unit_map[i] == unit_size)
                        break;

        if (i == m0_lid_to_unit_map_nr)
                return 0;
        else
                return i;
}
M0_EXPORTED(m0_obj_unit_size_to_layout_id);

int m0_obj_layout_id_to_unit_size(uint64_t layout_id)
{
        M0_ASSERT(layout_id > 0 && layout_id < m0_lid_to_unit_map_nr);

        return m0_lid_to_unit_map[layout_id];
}
M0_EXPORTED(m0_obj_layout_id_to_unit_size);

uint64_t m0_client_layout_id(const struct m0_client *instance)
{
        uint64_t lid = M0_DEFAULT_LAYOUT_ID;

        M0_ENTRY();
        M0_PRE(instance != NULL);

        /*
         * TODO:This layout selection is a temporary solution for s3 team
         * requirement. In future this has to be replaced by more sophisticated
         * solution.
         */
        if (instance->m0c_config->mc_layout_id != 0)
                lid = instance->m0c_config->mc_layout_id;

        M0_LEAVE("lid=%"PRIu64, lid);
        return lid;
}
M0_EXPORTED(m0_client_layout_id);

enum m0_client_layout_type m0_obj_layout_type(struct m0_obj *obj)
{
        return M0_OBJ_LAYOUT_TYPE(obj->ob_attr.oa_layout_id);
}
M0_EXPORTED(m0_obj_layout_type);

int m0_entity_open(struct m0_entity *entity,
                   struct m0_op **op)
{
        M0_ENTRY();

        M0_PRE(entity != NULL);
        M0_PRE(op != NULL);

        if (entity->en_type == M0_ET_IDX) {
                /* Since pool version for index entity is got by index query,
                 * move the entity state into OPEN and return success. */
                m0_sm_group_lock(&entity->en_sm_group);
                m0_sm_move(&entity->en_sm, 0, M0_ES_OPENING);
                m0_sm_move(&entity->en_sm, 0, M0_ES_OPEN);
                m0_sm_group_unlock(&entity->en_sm_group);
                return M0_RC(0);
        } else
                return M0_RC(entity_namei_op(entity, op,
                                             M0_EO_OPEN));
}
M0_EXPORTED(m0_entity_open);

#undef M0_TRACE_SUBSYSTEM

/*
 *  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
 */