sync.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/addb.h"
#include "motr/client.h"
#include "motr/client_internal.h"
#include "motr/sync.h"               /* sync_interactions */

#define M0_TRACE_SUBSYSTEM M0_TRACE_SUBSYS_CLIENT
#include "lib/trace.h"
#include "lib/finject.h"
#include "mdservice/fsync_fops.h"       /* m0_fop_fsync_mds_fopt */
#include "fop/fom_generic.h"            /* m0_rpc_item_is_generic_reply_fop */
#include "lib/memory.h"                 /* m0_alloc, m0_free */
#include "lib/tlist.h"
#include "lib/hash.h"                   /* m0_htable */
#include "file/file.h"
#include "motr/magic.h"                 /* M0_T1FS_FFW_TLIST_MAGIC? */
#include "pool/pool.h"                  /* pools_common_svc_ctx_tl */

static const struct m0_bob_type os_bobtype;
M0_BOB_DEFINE(static, &os_bobtype,  m0_op_sync);
static const struct m0_bob_type os_bobtype = {
        .bt_name         = "os_bobtype",
        .bt_magix_offset = offsetof(struct m0_op_sync, os_magic),
        .bt_magix        = M0_OS_MAGIC,
        .bt_check        = NULL,
};

M0_TL_DESCR_DEFINE(sync_target,
                   "Targets to synced for a client SYNC request",
                   static, struct sync_target,
                   srt_tlink, srt_tlink_magic,
                   M0_SYNC_TGT_TL_MAGIC, M0_SYNC_TGT_TL_MAGIC);

M0_TL_DEFINE(sync_target, static, struct sync_target);

/* TODO: use Client-defined magic values */
M0_TL_DESCR_DEFINE(spf, "sync_fop_wrappers pending fsync-fops",
                   static, struct sync_fop_wrapper, sfw_tlink,
                   sfw_tlink_magic, M0_T1FS_FFW_TLIST_MAGIC1,
                   M0_T1FS_FFW_TLIST_MAGIC2);

M0_TL_DEFINE(spf, static, struct sync_fop_wrapper);

/* SPTI -> Service to Pending Transaction Id list */
M0_TL_DESCR_DEFINE(spti, "m0_reqh_service_txid pending list", M0_INTERNAL,
                        struct m0_reqh_service_txid,
                        stx_tlink, stx_link_magic,
                        M0_INSTANCE_PTI_MAGIC,
                        M0_INSTANCE_PTI_MAGIC);

M0_TL_DEFINE(spti, M0_INTERNAL, struct m0_reqh_service_txid);

static struct sync_interactions si = {
        .si_post_rpc       = &m0_rpc_post,
        .si_wait_for_reply = &m0_rpc_item_wait_for_reply,
        /* fini is for requests, allocated in a bigger structure */
        .si_fop_fini       = &m0_fop_fini,
        /* put is for replies, allocated by a lower layer */
        .si_fop_put        = &m0_fop_put_lock,
};

static void sync_fop_cleanup(struct m0_ref *ref)
{
        struct m0_fop                  *fop;
        struct sync_fop_wrapper *sfw;

        M0_ENTRY();
        M0_PRE(si.si_fop_fini != NULL);

        fop = M0_AMB(fop, ref, f_ref);
        si.si_fop_fini(fop);

        sfw = M0_AMB(sfw, fop, sfw_fop);
        m0_free(sfw);

        M0_LEAVE("sync_fop_cleanup");
}

static void sync_request_done_locked(struct sync_request *sreq)
{
        struct m0_op      *op;
        struct m0_op_sync *os;

        os = sreq->sr_op_sync;
        op = &os->os_oc.oc_op;

        /* Updates SYNC op's state. */
        m0_sm_move(&op->op_sm, 0, M0_OS_EXECUTED);
        m0_op_executed(op);

        /*
         * Currently, an SYNC request is considered success only if all services
         * involved persist requested txid successfully.
         * TODO: in some cases an SYNC request can be considered success even
         * some FSYNC fops fail. For example, when an application writs
         * to an N+K object, it is not necessary to wait for all N+K services.
         * N services returnning successfully should be considered success for
         * an SYNC request.
         */
        /* A server reply will cause the op to complete stably,
         * with return code being reported in op->op_rc */
        op->op_rc = sreq->sr_rc;
        m0_sm_move(&op->op_sm, 0, M0_OS_STABLE);
        m0_op_stable(op);
}

static void sync_request_done(struct sync_request *sreq)
{
        struct m0_sm_group *op_grp;

        op_grp = &sreq->sr_op_sync->os_oc.oc_op.op_sm_group;
        m0_sm_group_lock(op_grp);
        sync_request_done_locked(sreq);
        m0_sm_group_unlock(op_grp);
}

/*
 * AST callback for completion of an SYNC request (all fops have been replied).
 * The AST is in the sm group of SYNC op.
 */
static void sync_request_ast(struct m0_sm_group *grp,
                                    struct m0_sm_ast *ast)
{
        struct sync_request *sreq;

        M0_ENTRY();

        M0_PRE(ast != NULL);
        M0_PRE(grp != NULL);
        M0_PRE(m0_sm_group_is_locked(grp));

        sreq = M0_AMB(sreq, ast, sr_ast);
        sync_request_done(sreq);

        M0_LEAVE();
}

static void sync_fop_done(struct sync_fop_wrapper *sfw, int rc)
{
        struct m0_op_sync       *os;
        struct sync_request     *sreq;
        struct m0_fop           *fop;

        M0_ENTRY();

        sreq = sfw->sfw_req;
        os = sreq->sr_op_sync;

        m0_mutex_lock(&sreq->sr_fops_lock);
        sreq->sr_nr_fops--;
        spf_tlist_del(sfw);
        fop = &sfw->sfw_fop;
        m0_fop_put_lock(fop);

        if (sreq->sr_nr_fops == 0) {
                /* All fops for this SYNC request have been replied. */
                sreq->sr_rc = sreq->sr_rc?:rc;
                m0_sm_ast_post(os->os_sm_grp, &sreq->sr_ast);
        }
        m0_mutex_unlock(&sreq->sr_fops_lock);

        M0_LEAVE();
}

static void sync_fop_stx_update(struct m0_reqh_service_txid *stx,
                                       uint64_t txid)
{
        M0_PRE(stx != NULL);

        if (stx->stx_tri.tri_txid <= txid) {
                /*
                 * Our transaction got committed, update the record to be
                 * ignored in the future.
                 */
                M0_SET0(&stx->stx_tri);
        }
        /*
         * Else the stx_maximum_txid got increased while we were waiting, it
         * is acceptable for fsync to return as long as the
         * correct-at-time-of-sending txn was committed (which the caller
         * should assert).
         */
}

static int sync_reply_process(struct sync_fop_wrapper *sfw)
{
        int                          rc;
        uint64_t                     reply_txid;
        struct m0_fop               *fop;
        struct m0_fop_fsync         *ffd;
        struct m0_fop_fsync_rep     *ffr;
        struct m0_rpc_item          *item;
        struct m0_reqh_service_ctx  *service;
        struct m0_reqh_service_txid *stx;
        struct m0_tl                *pending_tx_tl;
        struct m0_mutex             *pending_tx_lock;
        struct m0_entity            *ent;
        struct m0_op                *op;
        struct sync_request         *sreq;
        struct sync_target          *tgt;

        M0_ENTRY();

        M0_PRE(sfw != NULL);

        fop  = &sfw->sfw_fop;
        sreq = sfw->sfw_req;
        item = &fop->f_item;
        rc = m0_rpc_item_error(item);
        if (rc != 0) {
                M0_LOG(M0_ERROR, "rpc item error = %d", rc);
                goto out;
        }

        /* Gets the {fop,reply} data */
        ffd = m0_fop_data(fop);
        M0_ASSERT(ffd != NULL);
        ffr = m0_fop_data(m0_rpc_item_to_fop(item->ri_reply));
        M0_ASSERT(ffr != NULL);

        rc = ffr->ffr_rc;
        if (rc != 0) {
                M0_LOG(M0_ERROR, "reply rc=%d", rc);
                goto out;
        }

        /* Is this a valid reply to our request */
        reply_txid = ffr->ffr_be_remid.tri_txid;
        if (reply_txid < ffd->ff_be_remid.tri_txid) {
                /* Error (most likely) caused by an ioservice. */
                rc = M0_ERR(-EIO);
                M0_LOG(M0_ERROR, "Committed transaction is smaller "
                                            "than that requested.");
                goto out;
        }

        /* Update the stx stored in sfw to avoid sending repeated fops. */
        sync_fop_stx_update(sfw->sfw_stx, reply_txid);

        /* Update matched stx in all targets of an SYNC request. */
        service = sfw->sfw_stx->stx_service_ctx;
        m0_tl_for(sync_target, &sreq->sr_targets, tgt) {
                if (tgt->srt_type == SYNC_ENTITY) {
                        ent = tgt->u.srt_ent;
                        pending_tx_tl = &ent->en_pending_tx;
                        pending_tx_lock = &ent->en_pending_tx_lock;
                } else if (tgt->srt_type == SYNC_OP) {
                        op = tgt->u.srt_op;
                        pending_tx_tl = &op->op_pending_tx;
                        pending_tx_lock = &op->op_pending_tx_lock;
                } else
                        M0_IMPOSSIBLE("SYNC type not supported yet.");

                m0_mutex_lock(pending_tx_lock);
                stx = m0_tl_find(spti, stx, pending_tx_tl,
                                 stx->stx_service_ctx == service);
                if (stx != NULL)
                        sync_fop_stx_update(stx, reply_txid);
                m0_mutex_unlock(pending_tx_lock);
        } m0_tl_endfor;

        /* Updates Client instance wide stx. */
        m0_mutex_lock(&service->sc_max_pending_tx_lock);
        sync_fop_stx_update(&service->sc_max_pending_tx, reply_txid);
        m0_mutex_unlock(&service->sc_max_pending_tx_lock);

out:
        return M0_RC(rc);
}

static void sync_fop_ast(struct m0_sm_group *grp, struct m0_sm_ast *ast)
{
        int                             rc;
        struct sync_fop_wrapper *sfw;

        M0_ENTRY();

        M0_PRE(grp != NULL);
        M0_PRE(m0_sm_group_is_locked(grp));
        M0_PRE(ast != NULL);

        sfw = M0_AMB(sfw, ast, sfw_ast);
        rc = sync_reply_process(sfw);
        sync_fop_done(sfw, rc);

        M0_LEAVE();
}

static void sync_rio_replied(struct m0_rpc_item *item)
{
        struct m0_fop           *fop;
        struct m0_op_sync       *os;
        struct sync_fop_wrapper *sfw;

        M0_ENTRY();

        fop = m0_rpc_item_to_fop(item);
        sfw = M0_AMB(sfw, fop, sfw_fop);
        os = sfw->sfw_req->sr_op_sync;

        /* Trigger AST for post processing FSYNC fop. */
        sfw->sfw_ast.sa_cb = &sync_fop_ast;
        m0_sm_ast_post(os->os_sm_grp, &sfw->sfw_ast);

        M0_LEAVE();
        return;
}

static const struct m0_rpc_item_ops sync_ri_ops = {
        .rio_replied = sync_rio_replied,
};

static
int sync_request_fop_send(struct sync_request         *sreq,
                          struct m0_reqh_service_txid *stx,
                          enum m0_fsync_mode           mode,
                          bool                         set_ri_ops,
                          struct sync_fop_wrapper    **sfw_out)
{
        int                             rc;
        struct m0_fop                  *fop;
        struct m0_rpc_item             *item;
        struct m0_fop_fsync            *ffd;
        struct m0_fop_type             *fopt;
        struct sync_fop_wrapper        *sfw;

        M0_ENTRY();

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

        /* Stores the pending txid reference with the fop */
        sfw->sfw_stx = stx;
        sfw->sfw_req = sreq;

        rc = m0_rpc_session_validate(&stx->stx_service_ctx->sc_rlink.rlk_sess);
        if (rc != 0) {
                m0_free(sfw);
                return M0_ERR_INFO(rc, "Service tx session invalid");
        }

        if (stx->stx_service_ctx->sc_type == M0_CST_MDS)
                fopt = &m0_fop_fsync_mds_fopt;
        else if (stx->stx_service_ctx->sc_type == M0_CST_IOS)
                fopt = &m0_fop_fsync_ios_fopt;
        else if (stx->stx_service_ctx->sc_type == M0_CST_CAS)
                fopt = &m0_fop_fsync_cas_fopt;
        else
                M0_IMPOSSIBLE("invalid service type:%d",
                              stx->stx_service_ctx->sc_type);

        fop = &sfw->sfw_fop;
        m0_fop_init(fop, fopt, NULL, &sync_fop_cleanup);
        rc = m0_fop_data_alloc(fop);
        if (rc != 0) {
                m0_free(sfw);
                return M0_ERR_INFO(rc, "Allocating sync fop data failed.");
        }

        ffd = m0_fop_data(fop);
        ffd->ff_be_remid = stx->stx_tri;
        ffd->ff_fsync_mode = mode;

        /*
         *  Posts the rpc_item directly so that this is asyncronous.
         *  Prepare the fop as an rpc item
         */
        item = &fop->f_item;
        item->ri_ops = (set_ri_ops == true)?&sync_ri_ops:NULL;
        item->ri_session = &stx->stx_service_ctx->sc_rlink.rlk_sess;
        item->ri_prio = M0_RPC_ITEM_PRIO_MID;
        item->ri_deadline = 0;
        item->ri_nr_sent_max = M0_RPC_MAX_RETRIES;
        item->ri_resend_interval = M0_RPC_RESEND_INTERVAL;

        rc = si.si_post_rpc(item);
        if (rc != 0) {
                si.si_fop_fini(fop);
                return M0_ERR_INFO(rc, "Calling m0_rpc_post() failed.");
        }

        *sfw_out = sfw;
        return M0_RC(0);
}

/*
 * If wait_after_launch == true, caller of this function will wait for
 * the FSYNC reply fops.
 */
static int sync_request_launch(struct sync_request *sreq,
                               enum m0_fsync_mode mode,
                               bool wait_after_launch)
{
        int                             rc;
        int                             saved_error = 0;
        struct m0_reqh_service_txid    *iter;
        struct sync_fop_wrapper        *sfw = NULL;
        struct m0_tl                   *stx_tl;

        M0_ENTRY();

        if (M0_FI_ENABLED("launch_failed"))
                return M0_ERR(-EAGAIN);

        /*
         * Finds the services with pending transactions for each entry,
         * send an fsync fop. This is the fop sending loop.
         */
        m0_mutex_lock(&sreq->sr_fops_lock);
        stx_tl = &sreq->sr_stxs;
        m0_tl_for(spti, stx_tl, iter) {
                /*
                 * Sends an fsync fop for
                 * iter->stx_maximum_txid to iter->stx_service_ctx
                 */

                /* Checks if this service has any pending transactions. */
                if (iter->stx_tri.tri_txid == 0)
                        continue;

                /* Creates and sends an FSYNC fop. */
                rc = sync_request_fop_send(sreq, iter, mode,
                                                  !wait_after_launch, &sfw);
                if (rc != 0) {
                        saved_error = rc;
                        break;
                } else {
                        /* Add to list of pending fops  */
                        M0_CNT_INC(sreq->sr_nr_fops);
                        spf_tlink_init_at(sfw, &sreq->sr_fops);
                }
        } m0_tl_endfor;

        /*
         * How to handle those fops which have been sent?
         * (1) Cancel them by calling m0_rpc_item_cancel,
         * (2) Simply treat them as 'normal' fsync fops and let rpc item
         *     callback to handle the replies.
         * As m0_rpc_item_cancel() only cancels rpc item locally, fops are still
         * executed in service side, client will choose the option 2 to allow
         * client to update its records on FSYNC.
         *
         * Returns saved_error directly only if no fops are sent, otherwise the
         * error is stored in sync_request::sr_rc.
         */
        if (sreq->sr_nr_fops == 0) {
                if (saved_error == 0)
                        /*
                         * It may happen when there are no pending txid. For
                         * example, sync op is launched even before
                         * a write request gets replies and sets txid.
                         */
                        rc = -EAGAIN;
                else
                        rc = M0_ERR(saved_error);
        } else {
                sreq->sr_rc = saved_error != 0 ? M0_ERR(saved_error) :
                                saved_error;
                rc = 0;
        }
        m0_mutex_unlock(&sreq->sr_fops_lock);
        return M0_RC(rc);
}

static int sync_reply_wait(struct sync_fop_wrapper *sfw)
{
        int                 rc;
        struct m0_rpc_item *item;
        struct m0_fop      *fop;

        M0_ENTRY();

        fop = &sfw->sfw_fop;
        item = &fop->f_item;

        rc = si.si_wait_for_reply(item, m0_time_from_now(M0_RPC_TIMEOUT, 0));
        if (rc != 0)
                goto out;

        rc = sync_reply_process(sfw);

out:
        si.si_fop_put(fop);

        return M0_RC(rc);
}

static int sync_request_launch_and_wait(struct sync_request *sreq,
                                               enum m0_fsync_mode mode)
{
        int                             rc;
        int                             saved_error;
        struct sync_fop_wrapper        *sfw;

        M0_ENTRY();

        M0_PRE(sreq != NULL);

        /*
         * After calling sync_core() we may have sent some fops,
         * but stopped when one failed - collect all the replies before
         * returning.
         */
        rc = sync_request_launch(sreq, mode, true);
        if (rc != 0)
                return M0_ERR(rc);

        /* This is the fop-reply receiving loop. */
        saved_error = sreq->sr_rc;
        m0_tl_teardown(spf, &sreq->sr_fops, sfw) {
                /* Get and process the reply. */
                rc = sync_reply_wait(sfw);
                saved_error = saved_error ? : rc;
        }

        return M0_RC(saved_error);
}

static int sync_request_target_add(struct sync_request *sreq,
                                          int type, void *target)
{
        struct m0_entity    *ent;
        struct m0_op        *op;
        struct sync_target  *stgt;

        M0_ENTRY();

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

        switch(type) {
        case SYNC_ENTITY:
                ent = (struct m0_entity *)target;
                M0_ASSERT(M0_IN(ent->en_type,
                                (M0_ET_OBJ, M0_ET_IDX)));
                stgt->srt_type = SYNC_ENTITY;
                stgt->u.srt_ent = ent;
                break;
        case SYNC_OP:
                /*
                 * Only those ops which have been executed and received txid
                 * can be sync-ed in current version.
                 */
                op = (struct m0_op *)target;
                stgt->srt_type = SYNC_OP;
                stgt->u.srt_op = op;
                break;
        case SYNC_INSTANCE:
                stgt->srt_type = SYNC_INSTANCE;
                break;
        default:
                M0_IMPOSSIBLE("Unknow type for SYNC request.");
                break;
        }
        sync_target_tlink_init_at(stgt, &sreq->sr_targets);

        return M0_RC(0);
}

static int sync_request_stx_add(struct sync_request *sreq,
                                       struct m0_tl *pending_tx_tl,
                                       struct m0_mutex *pending_tx_lock)
{
        int                          i = 0;
        int                          rc = 0;
        int                          nr_saved_stxs = 0;
        struct m0_reqh_service_txid *saved_stxs;
        struct m0_reqh_service_txid *stx = NULL;
        struct m0_reqh_service_txid *iter;

        m0_mutex_lock(pending_tx_lock);

        M0_ALLOC_ARR(saved_stxs, spti_tlist_length(pending_tx_tl));
        if (saved_stxs == NULL) {
                rc = M0_ERR(-ENOMEM);
                goto error;
        }

        m0_tl_for(spti, pending_tx_tl, iter) {
                /* Find the record for this service */
                stx = m0_tl_find(spti, stx, &sreq->sr_stxs,
                                 stx->stx_service_ctx == iter->stx_service_ctx);
                if (stx != NULL) {
                        saved_stxs[nr_saved_stxs].stx_service_ctx =
                                                        stx->stx_service_ctx;
                        saved_stxs[nr_saved_stxs].stx_tri = stx->stx_tri;
                        if (iter->stx_tri.tri_txid > stx->stx_tri.tri_txid)
                                stx->stx_tri = iter->stx_tri;
                } else {
                        /* Not found - add a new stx. */
                        M0_ALLOC_PTR(stx);
                        if (stx == NULL) {
                                rc = M0_ERR(-ENOMEM);
                                goto undo;
                        }
                        stx->stx_service_ctx = iter->stx_service_ctx;
                        stx->stx_tri = iter->stx_tri;
                        spti_tlink_init_at(stx, &sreq->sr_stxs);
                }
                nr_saved_stxs++;
        } m0_tl_endfor;

        m0_free(saved_stxs);
        m0_mutex_unlock(pending_tx_lock);
        return M0_RC(0);

undo:
        m0_tl_for(spti, pending_tx_tl, iter) {
                stx = m0_tl_find(spti, stx, &sreq->sr_stxs,
                                 stx->stx_service_ctx == iter->stx_service_ctx);
                M0_ASSERT(stx != NULL);

                if (saved_stxs[i].stx_service_ctx == NULL)
                        /* It is a newly created stx, so simply remove it. */
                        spti_tlist_del(stx);
                else if (stx->stx_tri.tri_txid >
                        saved_stxs[i].stx_tri.tri_txid) {
                        /* Retores the old value. */
                        M0_ASSERT(stx->stx_service_ctx ==
                                  saved_stxs[i].stx_service_ctx);
                        stx->stx_tri = saved_stxs[i].stx_tri;
                }

                i++;
                if ( i == nr_saved_stxs)
                        break;
        } m0_tl_endfor;

error:
        m0_free(saved_stxs);
        m0_mutex_unlock(pending_tx_lock);
        return M0_RC(rc);
}

static void sync_pending_stx_update(struct m0_reqh_service_ctx *service,
                                           struct m0_mutex *pending_tx_lock,
                                           struct m0_tl *pending_tx,
                                           struct m0_be_tx_remid *btr)
{
        struct m0_reqh_service_txid *stx;

        /*
          * TODO: replace this O(N) search with something better.
          * Embbed the struct m0_reqh_service_txid in a list of
          * 'services for this inode'? See RB1667
          */
        /* Find the record for this service */
        m0_mutex_lock(pending_tx_lock);
        stx = m0_tl_find(spti, stx, pending_tx,
                         stx->stx_service_ctx == service);

        if (stx != NULL) {
                if (btr->tri_txid > stx->stx_tri.tri_txid)
                        stx->stx_tri = *btr;
        } else {
                /*
                 * not found - add a new record
                 */
                M0_ALLOC_PTR(stx);
                if (stx != NULL) {
                        stx->stx_service_ctx = service;
                        stx->stx_tri = *btr;

                        spti_tlink_init_at(stx, pending_tx);
                }
        }
        m0_mutex_unlock(pending_tx_lock);
}

void sync_record_update(struct m0_reqh_service_ctx *service,
                        struct m0_entity           *ent,
                        struct m0_op               *op,
                        struct m0_be_tx_remid      *btr)
{
        struct m0_reqh_service_txid *stx = NULL;

        M0_ENTRY();

        M0_PRE(service != NULL);

        /* Updates pending transaction number in the entity. */
        if (ent != NULL)
                sync_pending_stx_update(service, &ent->en_pending_tx_lock,
                                        &ent->en_pending_tx, btr);

        /* Updates pending transaction number in the op. */
        if (op != NULL)
                sync_pending_stx_update(service, &op->op_pending_tx_lock,
                                        &op->op_pending_tx, btr);

        /* update pending transaction number in the Client instance */
        m0_mutex_lock(&service->sc_max_pending_tx_lock);
        stx = &service->sc_max_pending_tx;
        /* update the value from the reply_fop */
        if (btr->tri_txid > stx->stx_tri.tri_txid) {
                stx->stx_service_ctx = service;
                stx->stx_tri = *btr;
        }
        m0_mutex_unlock(&service->sc_max_pending_tx_lock);

        M0_LEAVE("Client sync record updated.");
}

static bool sync_op_invariant(struct m0_op_sync *os)
{
        return M0_RC(os != NULL &&
                     m0_op_sync_bob_check(os) &&
                     os->os_oc.oc_op.op_size >= sizeof *os &&
                     m0_ast_rc_bob_check(&os->os_ar) &&
                     m0_op_common_bob_check(&os->os_oc));
}

static void sync_op_cb_fini(struct m0_op_common *oc)
{
        struct m0_op_sync *os;

        M0_ENTRY();

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

        os = bob_of(oc, struct m0_op_sync, os_oc, &os_bobtype);
        M0_PRE(sync_op_invariant(os));
        m0_op_common_bob_fini(&os->os_oc);
        m0_ast_rc_bob_fini(&os->os_ar);
        m0_op_sync_bob_fini(os);

        M0_LEAVE();
}

static void sync_op_cb_free(struct m0_op_common *oc)
{
        struct m0_op_sync           *os;
        struct sync_target          *tgt;
        struct m0_reqh_service_txid *stx = NULL;

        M0_ENTRY();

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

        /* By now, fini() has been called and bob_of cannot be used */
        os = M0_AMB(os, oc, os_oc);
        m0_tl_teardown(spti, &os->os_req->sr_stxs, stx)
                m0_free(stx);
        m0_tl_teardown(sync_target, &os->os_req->sr_targets, tgt)
                m0_free(tgt);
        m0_free(os->os_req);
        m0_free(os);

        M0_LEAVE();
}

static void sync_op_cb_launch(struct m0_op_common *oc)
{
        int                         rc = 0;
        struct m0_op        *op;
        struct m0_op_sync   *os;
        struct sync_request *sreq;

        M0_ENTRY();

        M0_PRE(oc != NULL);
        os = bob_of(oc, struct m0_op_sync, os_oc, &os_bobtype);
        op = &oc->oc_op;

        sreq = os->os_req;
        rc = sync_request_launch(sreq, os->os_mode, false);
        m0_sm_move(&op->op_sm, 0, M0_OS_LAUNCHED);

        if (rc != 0) {
                /*
                 * If the SYNC request is not sent to services, update the op's
                 * state here. Note: m0_op_launch_one() has held the
                 * group lock.
                 */
                sreq->sr_rc = sreq->sr_rc?:rc;
                sync_request_done_locked(sreq);
        }
}

static void sync_request_init(struct sync_request *sreq)
{
        M0_SET0(sreq);
        spf_tlist_init(&sreq->sr_fops);
        spti_tlist_init(&sreq->sr_stxs);
        sync_target_tlist_init(&sreq->sr_targets);
        m0_mutex_init(&sreq->sr_fops_lock);
}

static int sync_op_init(struct m0_op *op)
{
        int                         rc;
        struct m0_op_common *oc;
        struct m0_op_sync   *os;
        struct m0_locality  *locality;
        struct sync_request *sreq;

        M0_ENTRY();

        op->op_code = M0_EO_SYNC;
        rc = m0_op_init(op, &m0_op_conf, NULL);
        if (rc != 0)
                return M0_RC(rc);
        /*
         * Initialise m0_op_common part.
         * bob_init()'s haven't been called yet: we use M0_AMB().
         */
        oc = M0_AMB(oc, op, oc_op);
        os = M0_AMB(os, oc, os_oc);
        os->os_mode = M0_FSYNC_MODE_ACTIVE;

        m0_op_common_bob_init(oc);
        oc->oc_cb_launch = sync_op_cb_launch;
        oc->oc_cb_fini   = sync_op_cb_fini;
        oc->oc_cb_free   = sync_op_cb_free;

        /* Allocates and initialises a sync request. */
        M0_ALLOC_PTR(sreq);
        if (sreq == NULL)
                return M0_ERR(-ENOMEM);
        sync_request_init(sreq);
        sreq->sr_op_sync = os;
        sreq->sr_ast.sa_cb = sync_request_ast;
        os->os_req = sreq;

        /* Picks a locality thread for this op. */
        locality = m0__locality_pick(NULL);
        M0_ASSERT(locality != NULL);
        os->os_sm_grp = locality->lo_grp;
        M0_SET0(&os->os_ar);

        m0_op_sync_bob_init(os);
        m0_ast_rc_bob_init(&os->os_ar);

        return M0_RC(0);
}

int m0_sync_op_init(struct m0_op **sop)
{
        int rc = 0;

        M0_ENTRY();

        rc = m0_op_alloc(sop, sizeof(struct m0_op_sync))?:
             sync_op_init(*sop);

        return M0_RC(rc);
}
M0_EXPORTED(m0_sync_op_init);

int m0_sync_entity_add(struct m0_op *sop,
                              struct m0_entity *ent)
{
        int                         rc;
        struct m0_op_common *oc;
        struct m0_op_sync   *os;
        struct sync_request *sreq;

        M0_ENTRY();

        /*
         * New elements can only be added to the SYNC op
         * before the op is launched.
         */
        M0_PRE(sop != NULL);
        M0_PRE(sop->op_sm.sm_state == M0_OS_INITIALISED);

        oc = bob_of(sop, struct m0_op_common, oc_op, &oc_bobtype);
        os = bob_of(oc, struct m0_op_sync, os_oc, &os_bobtype);

        /* Stores the target. */
        sreq = os->os_req;
        M0_ASSERT(sreq != NULL);
        rc = sync_request_target_add(sreq, SYNC_ENTITY, ent);
        if (rc != 0)
                return M0_ERR(rc);

        /* Adds service txid (if stx exists in the list, then merge.).*/
        rc = sync_request_stx_add(sreq, &ent->en_pending_tx,
                                         &ent->en_pending_tx_lock);

        return M0_RC(rc);
}
M0_EXPORTED(m0_sync_entity_add);

int m0_sync_op_add(struct m0_op *sop,
                          struct m0_op *op)
{
        int                         rc;
        struct m0_op_common *oc;
        struct m0_op_sync   *os;
        struct sync_request *sreq;

        M0_ENTRY();

        /*
         * New elements can only be added to the SYNC op
         * before the op is launched.
         */
        M0_PRE(sop != NULL);
        M0_PRE(sop->op_sm.sm_state == M0_OS_INITIALISED);
        M0_PRE(op != NULL);
        M0_PRE(M0_IN(op->op_sm.sm_state,
                     (M0_OS_STABLE, M0_OS_EXECUTED)));

        oc = bob_of(sop, struct m0_op_common, oc_op, &oc_bobtype);
        os = bob_of(oc, struct m0_op_sync, os_oc, &os_bobtype);

        /* Stores the target. */
        sreq = os->os_req;
        M0_ASSERT(sreq != NULL);
        rc = sync_request_target_add(sreq, SYNC_OP, op);
        if (rc != 0)
                return M0_ERR(rc);

        /* Adds service txid (if stx exists in the list, then merge.).*/
        rc = sync_request_stx_add(sreq, &op->op_pending_tx,
                                        &op->op_pending_tx_lock);

        return M0_RC(rc);
}
M0_EXPORTED(m0_sync_op_add);

int m0_entity_sync(struct m0_entity *ent)
{
        int                          rc;
        struct sync_request          sreq;
        struct sync_target          *tgt;
        struct m0_reqh_service_txid *stx;

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

        sync_request_init(&sreq);
        rc = sync_request_target_add(&sreq, SYNC_ENTITY, ent);
        if (rc != 0)
                return M0_ERR(rc);

        rc = sync_request_stx_add(
                &sreq, &ent->en_pending_tx, &ent->en_pending_tx_lock)?:
             sync_request_launch_and_wait(&sreq, M0_FSYNC_MODE_ACTIVE);
        m0_tl_teardown(spti, &sreq.sr_stxs, stx)
                m0_free(stx);
        m0_tl_teardown(sync_target, &sreq.sr_targets, tgt)
                m0_free(tgt);

        return M0_RC(rc);
}
M0_EXPORTED(m0_entity_sync);

int m0_sync(struct m0_client *m0c, bool wait)
{
        int                             rc;
        int                             saved_error = 0;
        struct m0_reqh_service_txid    *stx;
        struct m0_reqh_service_ctx     *iter;
        struct sync_request             sreq;
        struct sync_fop_wrapper        *sfw;

        M0_ENTRY();

        M0_PRE(si.si_post_rpc != NULL);
        M0_PRE(si.si_wait_for_reply != NULL);
        M0_PRE(si.si_fop_fini != NULL);

        sync_request_init(&sreq);

        /*
         *  loop over all services associated with this super block,
         *  send an fsync fop for those with pending transactions
         *
         *  fop sending loop
         */
        m0_tl_for(pools_common_svc_ctx, &m0c->m0c_pools_common.pc_svc_ctxs,
                  iter) {
                /*
                 * Send an fsync fop for iter->sc_max_pending_txt to iter.
                 */
                m0_mutex_lock(&iter->sc_max_pending_tx_lock);
                stx = &iter->sc_max_pending_tx;

                /*
                 * Check if this service has any pending transactions.
                 * Currently for fsync operations are supported only for
                 * ioservice and mdservice.
                 */
                if (stx->stx_tri.tri_txid == 0 ||
                    !M0_IN(stx->stx_service_ctx->sc_type,
                          (M0_CST_MDS, M0_CST_IOS))) {
                        m0_mutex_unlock(&iter->sc_max_pending_tx_lock);
                        continue;
                }

                /* Create and send a request */
                rc = sync_request_fop_send(&sreq, stx,
                                                  M0_FSYNC_MODE_ACTIVE,
                                                  true, &sfw);
                if (rc != 0) {
                        saved_error = rc;
                        m0_mutex_unlock(&iter->sc_max_pending_tx_lock);
                        break;
                } else {
                        /* Reset the rpc item ops to NULL. */
                        sfw->sfw_fop.f_item.ri_ops = NULL;
                        /* Add to list of pending fops */
                        spf_tlink_init_at(sfw, &sreq.sr_fops);
                }

                m0_mutex_unlock(&iter->sc_max_pending_tx_lock);
        } m0_tl_endfor;

        /*
         * At this point we may have sent some fops, but stopped when one
         * failed - collect all the replies before returning
         */

        /* reply receiving loop */
        m0_tl_teardown(spf, &sreq.sr_fops, sfw) {
                /* get and process the reply */
                rc = sync_reply_wait(sfw);
                saved_error = saved_error ? : rc;
        }

        M0_LEAVE();
        return (saved_error == 0) ? M0_RC(saved_error): M0_ERR(saved_error);
}
M0_EXPORTED(m0_sync);

M0_INTERNAL struct m0_entity *
m0__op_sync_entity(const struct m0_op *op)
{
        struct m0_op_sync   *os;
        struct sync_target  *stgt;
        struct m0_op_common *oc;

        M0_PRE(op != NULL);
        M0_PRE(op->op_code == M0_EO_SYNC);

        oc = bob_of(op, struct m0_op_common, oc_op, &oc_bobtype);
        M0_PRE(oc != NULL);
        os = M0_AMB(os, oc, os_oc);
        M0_PRE(os != NULL && os->os_req != NULL);

        stgt = sync_target_tlist_head(&os->os_req->sr_targets);
        M0_PRE(stgt != NULL);
        switch (stgt->srt_type) {
        case SYNC_ENTITY:
                return stgt->u.srt_ent;
        case SYNC_OP:
                return stgt->u.srt_op->op_entity;
        case SYNC_INSTANCE:
                break;
        default:
                M0_IMPOSSIBLE("Unknow type for SYNC request.");
        }

        return NULL;
}

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