rm_rwlock.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 2015-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_RM
#include "lib/trace.h"

#include "lib/memory.h"    /* M0_ALLOC_PTR */
#include "fid/fid_xc.h"
#include "xcode/xcode.h"
#include "rm/rm_rwlock.h"
#include "addb2/addb2.h"

/* Forward Declarations */
static bool rwlockable_equal(const struct m0_rm_resource *resource0,
                             const struct m0_rm_resource *resource1);
static m0_bcount_t rwlockable_len(const struct m0_rm_resource *resource);
static int rwlockable_encode(struct m0_bufvec_cursor     *cur,
                             const struct m0_rm_resource *resource);
static int rwlockable_decode(struct m0_bufvec_cursor *cur,
                             struct m0_rm_resource  **resource);
static void rwlockable_credit_init(struct m0_rm_resource *resource,
                                   struct m0_rm_credit   *credit);
static void rwlockable_resource_free(struct m0_rm_resource *resource);

static bool rwlock_cr_intersects(const struct m0_rm_credit *self,
                                 const struct m0_rm_credit *c1);
static m0_bcount_t rwlock_cr_len(const struct m0_rm_credit *c0);

static int rwlock_cr_join(struct m0_rm_credit       *self,
                          const struct m0_rm_credit *c1);
static int rwlock_cr_disjoin(struct m0_rm_credit       *self,
                             const struct m0_rm_credit *c1,
                             struct m0_rm_credit       *intersection);
static int rwlock_cr_copy(struct m0_rm_credit       *dest,
                          const struct m0_rm_credit *self);
static int rwlock_cr_diff(struct m0_rm_credit       *self,
                          const struct m0_rm_credit *c1);
static bool rwlock_cr_conflicts(const struct m0_rm_credit *self,
                                const struct m0_rm_credit *c1);
static bool rwlock_cr_is_subset(const struct m0_rm_credit *self,
                                const struct m0_rm_credit *c1);
static int rwlock_cr_encode(struct m0_rm_credit     *self,
                            struct m0_bufvec_cursor *cur);
static int rwlock_cr_decode(struct m0_rm_credit     *self,
                            struct m0_bufvec_cursor *cur);
static void rwlock_cr_free(struct m0_rm_credit *self);
static void rwlock_cr_initial_capital(struct m0_rm_credit *self);

const struct m0_rm_resource_type_ops rwlockable_type_ops = {
        .rto_eq     = rwlockable_equal,
        .rto_len    = rwlockable_len,
        .rto_decode = rwlockable_decode,
        .rto_encode = rwlockable_encode,
};

const struct m0_rm_resource_ops rwlockable_ops = {
        .rop_credit_init   = rwlockable_credit_init,
        .rop_resource_free = rwlockable_resource_free,
};

const struct m0_rm_credit_ops rwlock_credit_ops = {
        .cro_intersects      = rwlock_cr_intersects,
        .cro_join            = rwlock_cr_join,
        .cro_copy            = rwlock_cr_copy,
        .cro_diff            = rwlock_cr_diff,
        .cro_free            = rwlock_cr_free,
        .cro_encode          = rwlock_cr_encode,
        .cro_decode          = rwlock_cr_decode,
        .cro_len             = rwlock_cr_len,
        .cro_is_subset       = rwlock_cr_is_subset,
        .cro_disjoin         = rwlock_cr_disjoin,
        .cro_conflicts       = rwlock_cr_conflicts,
        .cro_initial_capital = rwlock_cr_initial_capital,
};

#define R_RW(res) container_of(res, struct m0_rw_lockable, rwl_resource)
#define RW_XO(rw) (&M0_XCODE_OBJ(m0_fid_xc, (void *)(rw)->rwl_fid))
#define CR_XO(cr) (&M0_XCODE_OBJ(&M0_XT_U64, (void *)&(cr)->cr_datum))

M0_INTERNAL int
m0_rwlockable_domain_type_init(struct m0_rm_domain        *rwl_dom,
                               struct m0_rm_resource_type *rwl_rt)
{
        M0_SET0(rwl_dom);
        M0_SET0(rwl_rt);
        m0_rm_domain_init(rwl_dom);
        m0_rw_lockable_type_register(rwl_dom, rwl_rt);
        return 0;
}

M0_INTERNAL void
m0_rwlockable_domain_type_fini(struct m0_rm_domain        *rwl_dom,
                               struct m0_rm_resource_type *rwl_rt)
{
        m0_rw_lockable_type_deregister(rwl_rt);
        m0_rm_domain_fini(rwl_dom);
}

static bool rwlockable_equal(const struct m0_rm_resource *resource0,
                             const struct m0_rm_resource *resource1)
{
        struct m0_rw_lockable *lockable0;
        struct m0_rw_lockable *lockable1;

        M0_PRE(resource0 != NULL && resource1 != NULL);

        lockable0 = R_RW(resource0);
        lockable1 = R_RW(resource1);

        return m0_fid_eq(lockable0->rwl_fid, lockable1->rwl_fid);
}

static m0_bcount_t rwlockable_len(const struct m0_rm_resource *resource)
{
        struct m0_rw_lockable *lockable;
        struct m0_xcode_obj    fidobj;
        struct m0_xcode_ctx    ctx;
        static m0_bcount_t     len;

        M0_PRE(resource != NULL);

        /*
         * len depends only on size of fid, which is constant,
         * therefore len can be calculated once and then reused.
         */
        if (len == 0) {
                lockable = R_RW(resource);
                fidobj.xo_type = m0_fid_xc;
                fidobj.xo_ptr  = (void *)lockable->rwl_fid;
                m0_xcode_ctx_init(&ctx, &fidobj);
                len = m0_xcode_length(&ctx);
                M0_ASSERT(len > 0);
        }

        return len;
}

static int rwlockable_encode(struct m0_bufvec_cursor     *cur,
                             const struct m0_rm_resource *resource)
{
        M0_ENTRY();
        M0_PRE(resource != NULL);

        return M0_RC(m0_xcode_encdec(RW_XO(R_RW(resource)),
                                     cur, M0_XCODE_ENCODE));
}

static int rwlockable_decode(struct m0_bufvec_cursor *cur,
                             struct m0_rm_resource  **resource)
{
        struct m0_rw_lockable *lockable;
        struct m0_xcode_obj   *xo;
        int                    rc;

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

        M0_ALLOC_PTR(lockable);
        if (lockable == NULL)
                return M0_ERR(-ENOMEM);

        xo = RW_XO(lockable);
        rc = m0_xcode_encdec(xo, cur, M0_XCODE_DECODE);
        if (rc == 0) {
                lockable->rwl_fid = xo->xo_ptr;
                lockable->rwl_resource.r_ops = &rwlockable_ops;
                *resource = &lockable->rwl_resource;
        } else {
                m0_free(lockable);
        }

        return M0_RC(rc);
}

static void rwlockable_credit_init(struct m0_rm_resource *resource,
                                   struct m0_rm_credit   *credit)
{
        M0_PRE(credit != NULL);
        credit->cr_datum = 0;
        credit->cr_ops = &rwlock_credit_ops;
}

static void rwlockable_resource_free(struct m0_rm_resource *resource)
{
        struct m0_rw_lockable *lockable;

        lockable = R_RW(resource);
        m0_xcode_free_obj(&M0_XCODE_OBJ(m0_fid_xc, (void *)lockable->rwl_fid));
        m0_free(lockable);
}

static bool rwlock_credit_invariant(const struct m0_rm_credit *rwlock_cr)
{
        return rwlock_cr->cr_datum <= RM_RW_WRITE_LOCK;
}

static bool rwlock_cr_intersects(const struct m0_rm_credit *self,
                                 const struct m0_rm_credit *c1)
{
        M0_PRE(c1 != NULL);
        M0_PRE(rwlock_credit_invariant(self) &&
               rwlock_credit_invariant(c1));

        return  self->cr_datum != 0 && c1->cr_datum != 0;
}

static m0_bcount_t rwlock_cr_len(const struct m0_rm_credit *c0)
{
        struct m0_xcode_obj datumobj;
        struct m0_xcode_ctx ctx;

        M0_PRE(c0 != NULL);
        M0_PRE(rwlock_credit_invariant(c0));

        datumobj.xo_type = &M0_XT_U64;
        datumobj.xo_ptr  = (void *)&c0->cr_datum;
        m0_xcode_ctx_init(&ctx, &datumobj);
        return m0_xcode_length(&ctx);
}

static int rwlock_cr_join(struct m0_rm_credit       *self,
                          const struct m0_rm_credit *c1)
{
        M0_PRE(rwlock_credit_invariant(self) &&
               rwlock_credit_invariant(c1));
        /* Max credit value is write lock, also check for overflow */
        M0_PRE(self->cr_datum + c1->cr_datum >= self->cr_datum &&
               self->cr_datum + c1->cr_datum <= RM_RW_WRITE_LOCK);

        self->cr_datum += c1->cr_datum;

        M0_POST(rwlock_credit_invariant(self));
        return 0;
}

static int rwlock_cr_disjoin(struct m0_rm_credit       *self,
                             const struct m0_rm_credit *c1,
                             struct m0_rm_credit       *intersection)
{
        M0_PRE(rwlock_credit_invariant(self) &&
               rwlock_credit_invariant(c1));

        if (self->cr_datum > c1->cr_datum) {
                self->cr_datum -= c1->cr_datum;
                intersection->cr_datum = c1->cr_datum;
        } else {
                self->cr_datum = 0;
                intersection->cr_datum = self->cr_datum;
        }

        M0_POST(rwlock_credit_invariant(self) &&
                rwlock_credit_invariant(intersection));
        return 0;
}

static int rwlock_cr_copy(struct m0_rm_credit       *dest,
                          const struct m0_rm_credit *self)
{
        M0_PRE(dest != NULL);
        M0_PRE(rwlock_credit_invariant(self));

        dest->cr_datum = self->cr_datum;
        dest->cr_owner = self->cr_owner;
        dest->cr_ops = self->cr_ops;
        M0_POST(rwlock_credit_invariant(dest));
        return 0;
}

static int rwlock_cr_diff(struct m0_rm_credit       *self,
                          const struct m0_rm_credit *c1)
{
        M0_PRE(c1 != NULL);
        M0_PRE(rwlock_credit_invariant(self) &&
               rwlock_credit_invariant(c1));

        self->cr_datum = (self->cr_datum > c1->cr_datum) ?
                                self->cr_datum-c1->cr_datum : 0;
        M0_POST(rwlock_credit_invariant(self));
        return 0;
}

static bool rwlock_cr_conflicts(const struct m0_rm_credit *self,
                                const struct m0_rm_credit *c1)
{
        M0_PRE(c1 != NULL);
        M0_PRE(rwlock_credit_invariant(self) &&
               rwlock_credit_invariant(c1));

        return (self->cr_datum != 0 && c1->cr_datum == RM_RW_WRITE_LOCK) ||
               (self->cr_datum == RM_RW_WRITE_LOCK && c1->cr_datum != 0);
}

static bool rwlock_cr_is_subset(const struct m0_rm_credit *self,
                                const struct m0_rm_credit *c1)
{
        M0_PRE(rwlock_credit_invariant(self) &&
               rwlock_credit_invariant(c1));
        return self->cr_datum <= c1->cr_datum;
}

static int rwlock_cr_encode(struct m0_rm_credit     *self,
                            struct m0_bufvec_cursor *cur)
{
        M0_PRE(rwlock_credit_invariant(self));
        return M0_RC(m0_xcode_encdec(CR_XO(self), cur, M0_XCODE_ENCODE));
}

static int rwlock_cr_decode(struct m0_rm_credit     *self,
                            struct m0_bufvec_cursor *cur)
{
        return M0_RC(m0_xcode_encdec(CR_XO(self), cur, M0_XCODE_DECODE));
}

static void rwlock_cr_free(struct m0_rm_credit *self)
{
        self->cr_datum = 0;
}

static void rwlock_cr_initial_capital(struct m0_rm_credit *self)
{
         self->cr_datum = RM_RW_WRITE_LOCK;
}
 /* end internal RWLockInternal */

M0_INTERNAL void m0_rw_lockable_init(struct m0_rw_lockable *lockable,
                                     const struct m0_fid   *fid,
                                     struct m0_rm_domain   *dom)
{
        M0_PRE(lockable != NULL);
        M0_PRE(fid != NULL);

        lockable->rwl_resource.r_ops = &rwlockable_ops;
        lockable->rwl_fid = fid;
        if (dom != NULL)
                m0_rm_resource_add(dom->rd_types[M0_RM_RWLOCKABLE_RT],
                                   &lockable->rwl_resource);
}
M0_EXPORTED(m0_rw_lockable_init);

M0_INTERNAL void m0_rw_lockable_fini(struct m0_rw_lockable *lockable)
{
        /* Delete embedded resource from domain */
        m0_rm_resource_del(&lockable->rwl_resource);
        lockable->rwl_resource.r_ops = NULL;
        lockable->rwl_fid = NULL;
}
M0_EXPORTED(m0_rw_lockable_fini);

M0_INTERNAL void m0_rm_rwlock_owner_init(struct m0_rm_owner    *owner,
                                         struct m0_fid         *fid,
                                         struct m0_rw_lockable *lockable,
                                         struct m0_rm_remote   *creditor)
{
        m0_rm_owner_init(owner, fid, &m0_rm_no_group,
                &lockable->rwl_resource, creditor);
}
M0_EXPORTED(m0_rm_rwlock_owner_init);

M0_INTERNAL void m0_rm_rwlock_owner_fini(struct m0_rm_owner *owner)
{
        m0_rm_owner_fini(owner);
}
M0_EXPORTED(m0_rm_rwlock_owner_fini);

M0_INTERNAL void m0_rm_rwlock_req_init(struct m0_rm_incoming           *req,
                                       struct m0_rm_owner              *owner,
                                       const struct m0_rm_incoming_ops *ops,
                                       enum m0_rm_incoming_flags        flags,
                                       enum m0_rm_rwlock_req_type       type)
{
        M0_ENTRY();
        M0_PRE(M0_IN(type, (RM_RWLOCK_WRITE, RM_RWLOCK_READ)));

        m0_rm_incoming_init(req, owner, M0_RIT_LOCAL, RIP_NONE, flags);

        req->rin_want.cr_datum = (type == RM_RWLOCK_WRITE) ?
                                  RM_RW_WRITE_LOCK : RM_RW_READ_LOCK;
        req->rin_ops = ops;

        M0_LEAVE();
}
M0_EXPORTED(m0_rm_rwlock_req_init);

M0_INTERNAL void m0_rm_rwlock_req_fini(struct m0_rm_incoming *req)
{
        M0_ENTRY();
        m0_rm_incoming_fini(req);
        M0_LEAVE();
}
M0_EXPORTED(m0_rm_rwlock_req_fini);

M0_INTERNAL int m0_rw_lockable_type_register(struct m0_rm_domain        *dom,
                                             struct m0_rm_resource_type *rtype)
{
        M0_ENTRY();

        rtype->rt_id = M0_RM_RWLOCKABLE_RT;
        rtype->rt_name = "rw-lockable";
        rtype->rt_ops = &rwlockable_type_ops;
        return M0_RC(m0_rm_type_register(dom, rtype));
}
M0_EXPORTED(m0_rw_lockable_type_register);

M0_INTERNAL
void m0_rw_lockable_type_deregister(struct m0_rm_resource_type *rtype)
{
        M0_ENTRY();
        m0_rm_type_deregister(rtype);
        M0_LEAVE();
}
M0_EXPORTED(m0_rw_lockable_type_deregister);

const struct m0_fid M0_RWLOCK_FID = M0_FID_TINIT('R', 'W', 'L');

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