util.c

Go to the documentation of this file.

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

#include "lib/trace.h"        /* m0_trace_set_mmapped_buffer */
#include "rpc/rpclib.h"       /* M0_RPCLIB_MAX_RETRIES */
#include "motr/client_internal.h" /* m0_client */
#include "layout/plan.h"      /* m0_layout_io_plop */

#include "iscservice/isc.h"
#include "util.h"

#ifndef DEBUG
#define DEBUG 0
#endif

/* static variables */
static struct m0_container      container;
static struct m0_idx_dix_config dix_conf = {};

/* global variables */
struct m0_realm     uber_realm;
int                 trace_level = 0;
bool                m0trace_on = false;
struct m0_semaphore isc_sem;
struct m0_list      isc_reqs;

uint64_t isc_m0gs(struct m0_obj *obj, struct m0_client *cinst)
{
        unsigned long           usz; /* unit size */
        struct m0_pool_version *pver;

        pver = m0_pool_version_find(&cinst->m0c_pools_common,
                                    &obj->ob_attr.oa_pver);
        if (pver == NULL) {
                ERR("invalid object pool version: "FID_F"\n",
                    FID_P(&obj->ob_attr.oa_pver));
                return 0;
        }

        usz = m0_obj_layout_id_to_unit_size(obj->ob_attr.oa_layout_id);

        return usz * pver->pv_attr.pa_N;
}

void free_segs(struct m0_bufvec *data, struct m0_indexvec *ext,
               struct m0_bufvec *attr)
{
        m0_indexvec_free(ext);
        m0_bufvec_free(data);
        m0_bufvec_free(attr);
}

int alloc_segs(struct m0_bufvec *data, struct m0_indexvec *ext,
               struct m0_bufvec *attr, uint64_t bsz, uint32_t cnt)
{
        int i, rc;

        rc = m0_bufvec_alloc(data, cnt, bsz) ?:
             m0_bufvec_alloc(attr, cnt, 1) ?:
             m0_indexvec_alloc(ext, cnt);
        if (rc != 0)
                goto err;

        for (i = 0; i < cnt; i++)
                attr->ov_vec.v_count[i] = 0; /* no attrs */

        return 0;
 err:
        free_segs(data, ext, attr);
        return rc;
}

uint64_t set_exts(struct m0_indexvec *ext, uint64_t off, uint64_t bsz)
{
        uint32_t i;

        for (i = 0; i < ext->iv_vec.v_nr; i++) {
                ext->iv_index[i] = off;
                ext->iv_vec.v_count[i] = bsz;
                off += bsz;
        }

        return i * bsz;
}

int isc_req_prepare(struct isc_req *req, struct m0_buf *args,
                    const struct m0_fid *comp_fid,
                    struct m0_layout_io_plop *iop, uint32_t reply_len)
{
        int                    rc;
        struct m0_rpc_session *sess = iop->iop_session;
        struct m0_fop_isc     *fop_isc = &req->cir_isc_fop;
        struct m0_fop         *arg_fop = &req->cir_fop;

        req->cir_plop = &iop->iop_base;
        fop_isc->fi_comp_id = *comp_fid;
        m0_rpc_at_init(&fop_isc->fi_args);
        rc = m0_rpc_at_add(&fop_isc->fi_args, args, sess->s_conn);
        if (rc != 0) {
                m0_rpc_at_fini(&fop_isc->fi_args);
                ERR("m0_rpc_at_add() failed with %d\n", rc);
                return rc;
        }

        /* Initialise the reply RPC AT buffer to be received.*/
        m0_rpc_at_init(&fop_isc->fi_ret);
        rc = m0_rpc_at_recv(&fop_isc->fi_ret, sess->s_conn, reply_len, false);
        if (rc != 0) {
                m0_rpc_at_fini(&fop_isc->fi_args);
                m0_rpc_at_fini(&fop_isc->fi_ret);
                ERR("m0_rpc_at_recv() failed with %d\n", rc);
                return rc;
        }

        m0_fop_init(arg_fop, &m0_fop_isc_fopt, fop_isc, m0_fop_release);
        req->cir_rpc_sess = sess;

        return rc;
}

void isc_req_replied(struct m0_rpc_item *item)
{
        int                    rc;
        struct m0_fop         *fop = M0_AMB(fop, item, f_item);
        struct m0_fop         *reply_fop;
        struct m0_fop_isc_rep *isc_reply;
        struct isc_req        *req = M0_AMB(req, fop, cir_fop);
        const char *addr = m0_rpc_conn_addr(req->cir_rpc_sess->s_conn);

        if (item->ri_error != 0) {
                req->cir_rc = item->ri_error;
                ERR("No reply from %s: rc=%d.\n", addr, item->ri_error);
                goto err;
        }

        reply_fop = m0_rpc_item_to_fop(req->cir_fop.f_item.ri_reply);
        isc_reply = (struct m0_fop_isc_rep *)m0_fop_data(reply_fop);
        rc = req->cir_rc = isc_reply->fir_rc;
        if (rc != 0) {
                ERR("Got error in reply from %s: rc=%d.\n", addr, rc);
                if (rc == -ENOENT)
                        ERR("Was isc .so library is loaded?\n");
                goto err;
        }

        rc = m0_rpc_at_rep_get(&req->cir_isc_fop.fi_ret, &isc_reply->fir_ret,
                               &req->cir_result);
        if (rc != 0)
                ERR("rpc_at_rep_get() from %s failed: rc=%d\n", addr, rc);
 err:
        m0_semaphore_up(&isc_sem);
}

static const struct m0_rpc_item_ops isc_item_ops = {
        .rio_replied = isc_req_replied,
};

static void ireqs_list_add_in_order(struct isc_req *req)
{
        struct isc_req *r;
        struct m0_layout_io_plop *pl1;
        struct m0_layout_io_plop *pl2 = M0_AMB(pl2, req->cir_plop, iop_base);

        /*
         * No protection of the list is needed here as isc_req_send()
         * is called from the same thread.
         */
        m0_list_for_each_entry(&isc_reqs, r, struct isc_req, cir_link) {
                pl1 = M0_AMB(pl1, r->cir_plop, iop_base);
                if (pl1->iop_goff > pl2->iop_goff)
                        break;
        }
        m0_list_add_before(&r->cir_link, &req->cir_link);
}

int isc_req_send(struct isc_req *req)
{
        int                    rc;
        struct m0_rpc_item    *item;

        item              = &req->cir_fop.f_item;
        item->ri_session  = req->cir_rpc_sess;
        item->ri_prio     = M0_RPC_ITEM_PRIO_MID;
        item->ri_deadline = M0_TIME_IMMEDIATELY;
        item->ri_nr_sent_max = M0_RPCLIB_MAX_RETRIES;
        item->ri_ops      = &isc_item_ops;

        rc = m0_rpc_post(item);
        if (rc != 0)
                ERR("Failed to send request to %s: rc=%d\n",
                    m0_rpc_conn_addr(req->cir_rpc_sess->s_conn), rc);

        ireqs_list_add_in_order(req);

        return rc;
}

static void fop_fini_lock(struct m0_fop *fop)
{
        struct m0_rpc_machine *rmach = m0_fop_rpc_machine(fop);

        m0_rpc_machine_lock(rmach);
        m0_fop_fini(fop);
        m0_rpc_machine_unlock(rmach);
}

void isc_req_fini(struct isc_req *req)
{
        struct m0_fop *reply_fop = NULL;

        if (req->cir_fop.f_item.ri_reply != NULL)
                reply_fop = m0_rpc_item_to_fop(req->cir_fop.f_item.ri_reply);
        if (reply_fop != NULL)
                m0_fop_put_lock(reply_fop);
        req->cir_fop.f_item.ri_reply = NULL;
        m0_rpc_at_fini(&req->cir_isc_fop.fi_args);
        m0_rpc_at_fini(&req->cir_isc_fop.fi_ret);
        req->cir_fop.f_data.fd_data = NULL;
        fop_fini_lock(&req->cir_fop);
}

/*
 * init client resources.
 */
int isc_init(struct m0_config *conf, struct m0_client **cinst)
{
        int   rc;

        conf->mc_is_oostore            = true;
        conf->mc_is_read_verify        = false;
        conf->mc_tm_recv_queue_min_len = M0_NET_TM_RECV_QUEUE_DEF_LEN;
        conf->mc_max_rpc_msg_size      = M0_RPC_DEF_MAX_RPC_MSG_SIZE;
        conf->mc_layout_id             = 0;
        conf->mc_idx_service_id        = M0_IDX_DIX;
        conf->mc_idx_service_conf      = &dix_conf;

        if (!m0trace_on)
                m0_trace_set_mmapped_buffer(false);

        rc = m0_client_init(cinst, conf, true);
        if (rc != 0) {
                ERR("failed to initilise the Client API\n");
                return rc;
        }

        m0_container_init(&container, NULL, &M0_UBER_REALM, *cinst);
        rc = container.co_realm.re_entity.en_sm.sm_rc;
        if (rc != 0) {
                ERR("failed to open uber realm\n");
                return rc;
        }
        uber_realm = container.co_realm;

        m0_list_init(&isc_reqs);

        return 0;
}

void isc_fini(struct m0_client *cinst)
{
        m0_client_fini(cinst, true);
}

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