item.h

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


#pragma once

#ifndef __MOTR_RPC_ITEM_H__
#define __MOTR_RPC_ITEM_H__

#include "lib/types.h"             /* m0_bcount_t */
#include "lib/tlist.h"
#include "lib/list.h"
#include "lib/time.h"
#include "sm/sm.h"                 /* m0_sm */
#include "rpc/onwire.h"            /* m0_rpc_item_header2 */

/* Imports */
struct m0_rpc_session;
struct m0_bufvec_cursor;
struct m0_rpc_frm;
struct m0_rpc_machine;

/* Forward declarations */
struct m0_rpc_item_ops;
struct m0_rpc_item_type;

enum m0_rpc_item_priority {
        M0_RPC_ITEM_PRIO_MIN,
        M0_RPC_ITEM_PRIO_MID,
        M0_RPC_ITEM_PRIO_MAX,
        M0_RPC_ITEM_PRIO_NR
};

enum m0_rpc_item_state {
        M0_RPC_ITEM_UNINITIALISED,
        M0_RPC_ITEM_INITIALISED,
        M0_RPC_ITEM_ENQUEUED,
        M0_RPC_ITEM_URGENT,
        M0_RPC_ITEM_SENDING,
        M0_RPC_ITEM_SENT,
        M0_RPC_ITEM_WAITING_FOR_REPLY,
        M0_RPC_ITEM_REPLIED,
        M0_RPC_ITEM_ACCEPTED,
        M0_RPC_ITEM_FAILED,
        M0_RPC_ITEM_NR_STATES,
};

enum m0_rpc_item_dir {
        M0_RPC_ITEM_INCOMING,
        M0_RPC_ITEM_OUTGOING,
};

enum m0_rpc_item_resend {
        M0_RPC_ITEM_RESEND_INTERVAL =   1, /* in secs */
        M0_RPC_ITEM_REPLY_CACHE_TMO = 100, /* refer to M0T1FS_RPC_TIMEOUT */
        M0_RPC_ITEM_REQ_CACHE_TMO   = 100, /* refer to M0T1FS_RPC_TIMEOUT */
};

struct m0_rpc_item {
/* Public fields: write-once/read */

        enum m0_rpc_item_priority        ri_prio;

        uint64_t                         ri_ha_epoch;

        m0_time_t                        ri_deadline;
        m0_time_t                        ri_resend_interval;
        uint64_t                         ri_nr_sent_max;
        struct m0_rpc_session           *ri_session;
        const struct m0_rpc_item_ops    *ri_ops;
        uint32_t                         ri_flags;

/* Public fields: read only */

        struct m0_rpc_machine           *ri_rmachine;
        int32_t                          ri_error;
        struct m0_rpc_item              *ri_reply;

/* Private fields: */
        struct m0_sm_timeout             ri_deadline_timeout;
        struct m0_sm_timer               ri_timer;
        struct m0_sm                     ri_sm;
        uint32_t                         ri_nr_sent;
        struct m0_rpc_item              *ri_pending_reply;

        uint64_t                         ri_cookid;

        struct m0_rpc_item_header2       ri_header;

        size_t                           ri_size;
        const struct m0_rpc_item_type   *ri_type;
        m0_time_t                        ri_rpc_time;
        struct m0_tl                     ri_compound_items;
        struct m0_tlink                  ri_field;
        struct m0_tlink                  ri_iq_link;
        struct m0_tlink                  ri_plink;
        struct m0_rpc_packet            *ri_packet;
        struct m0_tl                    *ri_itemq;
        struct m0_rpc_frm               *ri_frm;
        struct m0_tlink                  ri_pending_link;
        uint64_t                         ri_magic;
        struct m0_tlink                  ri_cache_link;
        struct m0_tlink                  ri_xid_link;
        bool                             ri_xid_assigned_here;
        m0_time_t                        ri_cache_deadline;
};

enum m0_rpc_item_flags {
        M0_RIF_DUP  = 1 << 0,
        M0_RIF_REPLIED = 1 << 1,
};

struct m0_rpc_item_ops {
        void (*rio_sent)(struct m0_rpc_item *item);
        void (*rio_replied)(struct m0_rpc_item *item);
};

void m0_rpc_item_init(struct m0_rpc_item *item,
                      const struct m0_rpc_item_type *itype);

void m0_rpc_item_fini(struct m0_rpc_item *item);

void m0_rpc_item_get(struct m0_rpc_item *item);

void m0_rpc_item_put(struct m0_rpc_item *item);
void m0_rpc_item_put_lock(struct m0_rpc_item *item);

M0_EXTERN m0_bcount_t m0_rpc_item_onwire_header_size;
M0_EXTERN m0_bcount_t m0_rpc_item_onwire_footer_size;
m0_bcount_t m0_rpc_item_payload_size(struct m0_rpc_item *item);
m0_bcount_t m0_rpc_item_size(struct m0_rpc_item *item);

M0_INTERNAL
bool m0_rpc_item_max_payload_exceeded(struct m0_rpc_item    *item,
                                      struct m0_rpc_session *session);

int m0_rpc_item_timedwait(struct m0_rpc_item *item,
                          uint64_t states, m0_time_t timeout);

int m0_rpc_item_wait_for_reply(struct m0_rpc_item *item,
                               m0_time_t timeout);
void m0_rpc_item_cancel(struct m0_rpc_item *item);

void m0_rpc_item_cancel_nolock(struct m0_rpc_item *item);

void m0_rpc_item_cancel_init(struct m0_rpc_item *item);

int32_t m0_rpc_item_error(const struct m0_rpc_item *item);

struct m0_rpc_item_type_ops {
        m0_bcount_t (*rito_payload_size)(const struct m0_rpc_item *item);

        bool (*rito_eq)(const struct m0_rpc_item *i1,
                        const struct m0_rpc_item *i2);

        void (*rito_io_coalesce)(struct m0_rpc_item *head,
                                 struct m0_list *list, uint64_t size);

        int (*rito_encode)(const struct m0_rpc_item_type *item_type,
                           struct m0_rpc_item *item,
                           struct m0_bufvec_cursor *cur);
        int (*rito_decode)(const struct m0_rpc_item_type *item_type,
                           struct m0_rpc_item **item,
                           struct m0_bufvec_cursor *cur);

        bool (*rito_try_merge)(struct m0_rpc_item *container,
                               struct m0_rpc_item *component,
                               m0_bcount_t         limit);

        void (*rito_item_get)(struct m0_rpc_item *item);
        void (*rito_item_put)(struct m0_rpc_item *item);

        int (*rito_deliver)(struct m0_rpc_machine *rpcmach,
                            struct m0_rpc_item *item);
};

enum m0_rpc_item_type_flags {
        M0_RPC_ITEM_TYPE_REQUEST = (1 << 0),
        M0_RPC_ITEM_TYPE_REPLY   = (1 << 1),
        M0_RPC_ITEM_TYPE_ONEWAY  = (1 << 2),
        M0_RPC_ITEM_TYPE_MUTABO  = (1 << 3),

        M0_RPC_MUTABO_REQ = M0_RPC_ITEM_TYPE_MUTABO | M0_RPC_ITEM_TYPE_REQUEST
};

struct m0_rpc_item_type {
        uint32_t                           rit_opcode;
        const struct m0_rpc_item_type_ops *rit_ops;
        uint64_t                           rit_flags;
        struct m0_tlink                    rit_linkage;
        uint64_t                           rit_magic;
        struct m0_sm_conf                  rit_incoming_conf;
        struct m0_sm_conf                  rit_outgoing_conf;
};

M0_INTERNAL void m0_rpc_item_type_register(struct m0_rpc_item_type *item_type);

M0_INTERNAL void m0_rpc_item_type_deregister(struct m0_rpc_item_type
                                             *item_type);

M0_INTERNAL struct m0_rpc_item_type *m0_rpc_item_type_lookup(uint32_t opcode);

M0_INTERNAL const char *
m0_rpc_item_remote_ep_addr(const struct m0_rpc_item *item);

M0_INTERNAL const char *m0_rpc_item_opname(const struct m0_rpc_item *item);

M0_INTERNAL const char *
m0_rpc_item_type_name(const struct m0_rpc_item_type *item_type);

enum {
        RIC_HASH_MASK = 0xff,
        RIC_HASH_SIZE = RIC_HASH_MASK + 1,
};

struct m0_rpc_item_cache {
        struct m0_tl    *ric_items;
        struct m0_mutex *ric_lock;
};

M0_INTERNAL int m0_rpc_item_cache_init(struct m0_rpc_item_cache *ic,
                                       struct m0_mutex          *lock);
M0_INTERNAL void m0_rpc_item_cache_fini(struct m0_rpc_item_cache *ic);
M0_INTERNAL bool m0_rpc_item_cache__invariant(struct m0_rpc_item_cache *ic);

M0_INTERNAL bool m0_rpc_item_cache_add(struct m0_rpc_item_cache *ic,
                                       struct m0_rpc_item       *item,
                                       m0_time_t                 deadline);
M0_INTERNAL void m0_rpc_item_cache_del(struct m0_rpc_item_cache *ic,
                                       uint64_t                  xid);
M0_INTERNAL struct m0_rpc_item *
m0_rpc_item_cache_lookup(struct m0_rpc_item_cache *ic, uint64_t xid);

M0_INTERNAL void m0_rpc_item_cache_purge(struct m0_rpc_item_cache *ic);
M0_INTERNAL void m0_rpc_item_cache_clear(struct m0_rpc_item_cache *ic);

M0_INTERNAL void m0_rpc_item_xid_list_init(struct m0_rpc_session *session);
M0_INTERNAL void m0_rpc_item_xid_list_fini(struct m0_rpc_session *session);

M0_INTERNAL void m0_rpc_item_pending_cache_init(struct m0_rpc_session *session);
M0_INTERNAL void m0_rpc_item_pending_cache_fini(struct m0_rpc_session *session);

M0_INTERNAL void m0_rpc_item_pending_cache_add( struct m0_rpc_item *item);
M0_INTERNAL void m0_rpc_item_pending_cache_del(struct m0_rpc_item *item);

#define ITEM_FMT "%p [%s/%s(%u)]"
#define ITEM_ARG(item) (item), item_kind(item), m0_rpc_item_opname(item), \
        ((item)->ri_type != NULL ? (item)->ri_type->rit_opcode : 0)

#endif

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