source_dock.c

Go to the documentation of this file.

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

#include <unistd.h>
#include "lib/memory.h"
#include "fdmi/fdmi.h"
#include "fdmi/source_dock.h"
#include "fdmi/source_dock_internal.h"
#include "fdmi/service.h"

/* fdmi registered sources list declaration */
M0_TL_DESCR_DEFINE(fdmi_src_dock_src_list,
                   "fdmi src list",
                   M0_INTERNAL,
                   struct m0_fdmi_src_ctx,
                   fsc_linkage, fsc_magic,
                   M0_FDMI_SRC_DOCK_SRC_CTX_MAGIC,
                   M0_FDMI_SRC_DOCK_SRC_CTX_HEAD_MAGIC);

M0_TL_DEFINE(fdmi_src_dock_src_list, M0_INTERNAL, struct m0_fdmi_src_ctx);

/*
 * FDMI source records list declaration.  This list keeps all records that are
 * waiting for filter processing. When records are processed, they are removed
 * from this list and added to the inflight list.
 *
 * NOTE. 'release' reply is handled in a different way, when release comes in
 * the record is looked up in a different way.  This list is only needed for
 * record resend operations, not for release.
 */
M0_TL_DESCR_DEFINE(fdmi_record_list, "fdmi rec list", M0_INTERNAL,
                   struct m0_fdmi_src_rec, fsr_linkage, fsr_magic,
                   M0_FDMI_SRC_DOCK_REC_MAGIC,
                   M0_FDMI_SRC_DOCK_REC_HEAD_MAGIC);

M0_TL_DEFINE(fdmi_record_list, M0_INTERNAL, struct m0_fdmi_src_rec);

/*
 * Inflight FDMI records. When records are processed, they are moved here.
 * When 'release' is received, ths record is removed from this list.
 *
 * We have a timeout timer waking up the fom periodically, checking
 * if any record in this list has not been acknowledged in a specified
 * time. If yes, we have to give up (release refcount on it and let the
 * dtx done). If not, resend it for processing.
 */
M0_TL_DESCR_DEFINE(fdmi_record_inflight, "fdmi rec inflight", M0_INTERNAL,
                   struct m0_fdmi_src_rec, fsr_linkage, fsr_magic,
                   M0_FDMI_SRC_DOCK_REC_MAGIC,
                   M0_FDMI_SRC_DOCK_REC_HEAD_MAGIC);

M0_TL_DEFINE(fdmi_record_inflight, M0_INTERNAL, struct m0_fdmi_src_rec);

/* Matched filters list declaration */
M0_TL_DESCR_DEFINE(fdmi_matched_filter_list, "fdmi matched filter list",
                   M0_INTERNAL, struct m0_conf_fdmi_filter,
                   ff_linkage, ff_magic,
                   M0_FDMI_SRC_DOCK_MATCHED_FILTER_MAGIC,
                   M0_FDMI_SRC_DOCK_MATCHED_FILTER_HEAD_MAGIC);

M0_TL_DEFINE(fdmi_matched_filter_list, M0_INTERNAL, struct m0_conf_fdmi_filter);

M0_INTERNAL void m0_fdmi_source_dock_init(struct m0_fdmi_src_dock *src_dock)
{
        M0_ENTRY();
        M0_SET0(src_dock);
        fdmi_src_dock_src_list_tlist_init(&src_dock->fsdc_src_list);
        fdmi_record_list_tlist_init(&src_dock->fsdc_posted_rec_list);
        fdmi_record_inflight_tlist_init(&src_dock->fsdc_rec_inflight);
        m0_mutex_init(&src_dock->fsdc_list_mutex);
        src_dock->fsdc_instance_id = 0;
        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi_source_dock_fini(struct m0_fdmi_src_dock *src_dock)
{
        struct m0_fdmi_src_ctx *src_ctx;

        M0_ENTRY();

        /* Deinitialize FDMI source dock instance */

        m0_tl_teardown(fdmi_src_dock_src_list,
                       &src_dock->fsdc_src_list, src_ctx) {
                M0_ASSERT(src_ctx->fsc_registered == false);
                m0_free(src_ctx);
        }

        fdmi_src_dock_src_list_tlist_fini(&src_dock->fsdc_src_list);

        /* Posted record list is handled by FOM and should be empty here */
        fdmi_record_list_tlist_fini(&src_dock->fsdc_posted_rec_list);
        fdmi_record_inflight_tlist_fini(&src_dock->fsdc_rec_inflight);
        m0_mutex_fini(&src_dock->fsdc_list_mutex);

        M0_LEAVE();
}

M0_INTERNAL bool m0_fdmi__record_is_valid(struct m0_fdmi_src_rec *src_rec)
{
        return  _0C(src_rec != NULL) &&
                _0C(src_rec->fsr_src != NULL) &&
                _0C(src_rec->fsr_src_ctx != NULL) &&
                _0C(src_rec->fsr_magic == M0_FDMI_SRC_DOCK_REC_MAGIC);
}

static void src_rec_free(struct m0_ref *ref)
{
        struct m0_fdmi_src_rec *src_rec = M0_AMB(src_rec, ref, fsr_ref);

        M0_ENTRY("ref %p", ref);
        M0_PRE(m0_fdmi__record_is_valid(src_rec));
        M0_PRE(!fdmi_record_list_tlink_is_in(src_rec));

        M0_LOG(M0_DEBUG, "last ref dropped on fdmi rec id " U128X_F,
                         U128_P(&src_rec->fsr_rec_id));

        /* Finalize list and link. @src_rec is still alive. */
        m0_fdmi__record_deinit(src_rec);

        m0_fdmi__fs_end(src_rec);

        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi__record_init(struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("src_rec %p", src_rec);
        M0_ASSERT(src_rec != NULL);

        fdmi_record_list_tlink_init(src_rec);
        fdmi_matched_filter_list_tlist_init(&src_rec->fsr_filter_list);
        src_rec->fsr_src_ctx =
                m0_fdmi__src_ctx_get(src_rec->fsr_src->fs_type_id);
        if (src_rec->fsr_src_ctx == NULL) {
                M0_LOG(M0_ERROR, "FDMI context for record type %x is not found.",
                       src_rec->fsr_src->fs_type_id);
        }

        src_rec->fsr_init_time = m0_time_now();

        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi__record_deinit(struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("rec_src %p", src_rec);
        M0_PRE(m0_fdmi__record_is_valid(src_rec));

        fdmi_matched_filter_list_tlist_fini(&src_rec->fsr_filter_list);
        fdmi_record_list_tlink_fini(src_rec);
        src_rec->fsr_rec_id = M0_UINT128(0, 0);

        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi__rec_id_gen(struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("src_rec %p", src_rec);
        M0_PRE(m0_fdmi__record_is_valid(src_rec));
        M0_PRE(src_rec->fsr_src != NULL);
        M0_PRE(src_rec->fsr_rec_id.u_hi == 0 && src_rec->fsr_rec_id.u_lo == 0);

        if (m0_fdmi_src_dock_get()->fsdc_instance_id == 0) {
#ifndef __KERNEL__
                uint64_t seed = m0_time_now() ^ getpid();
#else
                uint64_t seed = m0_time_now();
#endif
                uint64_t id = 0;

                /* this loop is to make sure ID is never 0. */
                while (id == 0) {
                        id = m0_rnd64(&seed);
                        seed++;
                }
                m0_fdmi_src_dock_get()->fsdc_instance_id = id;
        }

        M0_ASSERT(sizeof(src_rec) <= sizeof(uint64_t));
        src_rec->fsr_rec_id = M0_UINT128(
                m0_fdmi_src_dock_get()->fsdc_instance_id,
                (uint64_t)src_rec
                );
        M0_LOG(M0_DEBUG, "new FDMI record id = "U128X_F,
               U128_P(&src_rec->fsr_rec_id));
        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi__enqueue_locked(struct m0_fdmi_src_rec *src_rec)
{
        struct m0_fdmi_src_dock *src_dock = m0_fdmi_src_dock_get();

        M0_ASSERT(m0_mutex_is_locked(&src_dock->fsdc_list_mutex));
        M0_ASSERT(src_rec != NULL && src_rec->fsr_src != NULL);

        M0_LOG(M0_DEBUG, "Adding into record list id = "U128X_F,
                         U128_P(&src_rec->fsr_rec_id));
        fdmi_record_list_tlist_add_tail(
                        &src_dock->fsdc_posted_rec_list, src_rec);
        m0_fdmi__src_dock_fom_wakeup(&src_dock->fsdc_sd_fom);
}

M0_INTERNAL void m0_fdmi__enqueue(struct m0_fdmi_src_rec *src_rec)
{
        struct m0_fdmi_src_dock *src_dock = m0_fdmi_src_dock_get();

        M0_ASSERT(src_rec != NULL && src_rec->fsr_src != NULL);

        m0_mutex_lock(&src_dock->fsdc_list_mutex);
        m0_fdmi__enqueue_locked(src_rec);
        m0_mutex_unlock(&src_dock->fsdc_list_mutex);
}

M0_INTERNAL void m0_fdmi__record_post(struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("src %p, fdmi_data %p, rec_id "U128X_F_SAFE,
                 (src_rec ? src_rec->fsr_src : (void*)(-1)),
                 (src_rec ? src_rec->fsr_data : (void*)(-1)),
                 U128_P_SAFE_EX(src_rec, &src_rec->fsr_rec_id));

        M0_ASSERT(src_rec != NULL && src_rec->fsr_src != NULL);

        m0_fdmi__record_init(src_rec);
        m0_fdmi__rec_id_gen(src_rec);

        /* Init ref cnt to 1, to be droppped when it is processed/released. */
        m0_ref_init(&src_rec->fsr_ref, 1, src_rec_free);
        m0_fdmi__fs_get(src_rec);
        M0_LOG(M0_DEBUG, "fdmi rec id=" U128X_F " ref cnt=%d",
                         U128_P(&src_rec->fsr_rec_id),
                         (int)m0_ref_read(&src_rec->fsr_ref));

        m0_fdmi__enqueue(src_rec);

        M0_LEAVE();
}

M0_INTERNAL int m0_fdmi_source_alloc(enum m0_fdmi_rec_type_id type_id,
                                     struct m0_fdmi_src     **src)
{
        struct m0_fdmi_src_ctx *src_ctx;

        M0_ALLOC_PTR(src_ctx);
        if (src_ctx == NULL) {
                *src = NULL;
                return -ENOMEM;
        }
        src_ctx->fsc_src.fs_type_id = type_id;
        *src = &src_ctx->fsc_src;

        return 0;
}

M0_INTERNAL void m0_fdmi_source_free(struct m0_fdmi_src *src)
{
        struct m0_fdmi_src_ctx *src_ctx;
        M0_PRE(src != NULL);
        src_ctx = container_of(src, struct m0_fdmi_src_ctx, fsc_src);
        M0_ASSERT(!src_ctx->fsc_registered);
        fdmi_src_dock_src_list_tlist_remove(src_ctx);
        m0_free(src_ctx);
}

M0_INTERNAL int m0_fdmi_source_register(struct m0_fdmi_src *src)
{
        struct m0_fdmi_src_ctx  *src_ctx;
        struct m0_fdmi_src_dock *src_dock = m0_fdmi_src_dock_get();

        M0_PRE(src != NULL);
        M0_PRE(src->fs_node_eval != NULL);
        M0_PRE(src->fs_encode != NULL);
        M0_PRE(m0_tl_find(fdmi_src_dock_src_list, src_ctx,
                          &src_dock->fsdc_src_list,
                          (src->fs_type_id == src_ctx->fsc_src.fs_type_id &&
                           src_ctx->fsc_registered)) == NULL);
        M0_ENTRY();
        src_ctx = container_of(src, struct m0_fdmi_src_ctx, fsc_src);
        src_ctx->fsc_registered = true;
        fdmi_src_dock_src_list_tlink_init_at(
                src_ctx,
                &src_dock->fsdc_src_list);

        src->fs_record_post = m0_fdmi__record_post;

        return M0_RC(0);
}

M0_INTERNAL void m0_fdmi_source_deregister(struct m0_fdmi_src *src)
{
        struct m0_fdmi_src_dock *src_dock = m0_fdmi_src_dock_get();
        struct m0_fdmi_src_ctx  *src_ctx;
        struct m0_fdmi_src_rec  *src_rec;

        M0_ENTRY("src = %p", src);
        M0_PRE(src != NULL);

        m0_mutex_lock(&src_dock->fsdc_list_mutex);

        src_ctx = m0_tl_find(
                fdmi_src_dock_src_list, src_ctx,
                &m0_fdmi_src_dock_get()->fsdc_src_list,
                (src->fs_type_id == src_ctx->fsc_src.fs_type_id &&
                 src_ctx->fsc_registered));

        M0_POST(src_ctx != NULL);

        src_ctx->fsc_registered = false;
        src->fs_record_post     = NULL;

        m0_tlist_for(&fdmi_record_list_tl, &src_dock->fsdc_posted_rec_list,
                     src_rec) {
                M0_ASSERT(m0_fdmi__record_is_valid(src_rec));
                if (src_rec->fsr_src == src) {
                        fdmi_record_list_tlist_remove(src_rec);
                        m0_fdmi__record_deinit(src_rec);
                }
        } m0_tlist_endfor;

        m0_mutex_unlock(&src_dock->fsdc_list_mutex);

        M0_LEAVE();
}

M0_INTERNAL struct m0_fdmi_src_ctx *
m0_fdmi__src_ctx_get(enum m0_fdmi_rec_type_id  src_type_id)
{
        struct m0_fdmi_src_ctx  *ret;
        struct m0_fdmi_src_dock *src_dock = m0_fdmi_src_dock_get();

        M0_ENTRY("src_dock=%p, src_type_id=%d", src_dock, src_type_id);

        ret = m0_tl_find(fdmi_src_dock_src_list, ret, &src_dock->fsdc_src_list,
                         ret->fsc_src.fs_type_id == src_type_id);

        M0_LEAVE("ret=%p", ret);
        return ret;
}

M0_INTERNAL void m0_fdmi__fs_get(struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("src_rec=%p", src_rec);
        M0_ASSERT(m0_fdmi__record_is_valid(src_rec));

        M0_LOG(M0_DEBUG, "fs get for "U128X_F,
                         U128_P(&src_rec->fsr_rec_id));
        if (src_rec->fsr_src->fs_get != NULL) {
                src_rec->fsr_src->fs_get(src_rec);
        }

        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi__fs_put(struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("src_rec=%p", src_rec);
        M0_ASSERT(m0_fdmi__record_is_valid(src_rec));

        M0_LOG(M0_DEBUG, "fs put for "U128X_F,
                         U128_P(&src_rec->fsr_rec_id));
        if (src_rec->fsr_src->fs_put != NULL)
                src_rec->fsr_src->fs_put(src_rec);

        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi__fs_begin(
        struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("src_rec = %p", src_rec);
        M0_ASSERT(m0_fdmi__record_is_valid(src_rec));

        if (src_rec->fsr_src->fs_begin != NULL)
                src_rec->fsr_src_ctx->fsc_src.fs_begin(src_rec);

        M0_LEAVE();
}

M0_INTERNAL void m0_fdmi__fs_end(struct m0_fdmi_src_rec *src_rec)
{
        M0_ENTRY("src_rec = %p", src_rec);
        M0_ASSERT(m0_fdmi__record_is_valid(src_rec));

        if (src_rec->fsr_src->fs_end != NULL)
                src_rec->fsr_src->fs_end(src_rec);

        M0_LEAVE();
}

M0_INTERNAL enum m0_fdmi_rec_type_id
m0_fdmi__sd_rec_type_id_get(struct m0_fdmi_src_rec *src_rec)
{
        M0_ASSERT(m0_fdmi__record_is_valid(src_rec));
        return src_rec->fsr_src->fs_type_id;
}

M0_INTERNAL int m0_fdmi__handle_release(struct m0_uint128 *fdmi_rec_id)
{
        struct m0_fdmi_src_rec  *src_rec;
        uint64_t                 expected;
        uint64_t                 actual;
        struct m0_fdmi_src_dock *src_dock = m0_fdmi_src_dock_get();
        bool                     waiting;

        M0_PRE(fdmi_rec_id != NULL);

        M0_ENTRY("fdmi_rec_id "U128X_F_SAFE, U128_P_SAFE(fdmi_rec_id));

        /*
         * First, validate that this rec_id was generated by this running
         * instance (see m0_fdmi__rec_id_gen).
         */
        actual = fdmi_rec_id->u_hi;
        expected = m0_fdmi_src_dock_get()->fsdc_instance_id;
        if (actual != expected) {
                M0_LOG(M0_WARN,
                       "Failed rc=1. Received release-record aimed at "
                       "another source dock instance: u_hi = %" PRIx64 ", "
                       "expected %"PRIx64, actual, expected);
                /*
                 * @todo FDMI Phase 2 -- post addb event about invalid
                 * fdmi_record_id.
                 */
                return M0_RC(1);
        }

        /*
         * Lower 64 bit of fdmi_record_id are a pointer to
         * src_rec.  See m0_fdmi__rec_id_gen.
         */
        src_rec = (void *)fdmi_rec_id->u_lo;
        if (src_rec == NULL ||
            src_rec->fsr_magic != M0_FDMI_SRC_DOCK_REC_MAGIC ||
            src_rec->fsr_rec_id.u_hi != fdmi_rec_id->u_hi    ||
            src_rec->fsr_rec_id.u_lo != fdmi_rec_id->u_lo) {
                /*
                 * @todo Phase 2.  In Phase 1, we release transaction BEFORE
                 * this 'release' even comes.  So in Phase 1 we always end up
                 * in this code branch.  But in Phase 2, it will all change.
                 */
                M0_LOG(M0_WARN,
                       "Attempt to release non-existent FDMI record with ID "
                       U128X_F_SAFE" (magic %ld)", U128_P_SAFE(fdmi_rec_id),
                       (src_rec ? src_rec->fsr_magic : 0));

                /*
                 * @todo FDMI Phase 2 -- post addb event about invalid
                 * fdmi_record_id.
                 */
                return M0_RC(0);
        }

        m0_mutex_lock(&src_dock->fsdc_list_mutex);
        if ((waiting = fdmi_record_inflight_tlink_is_in(src_rec)))
                fdmi_record_inflight_tlist_remove(src_rec);
        m0_mutex_unlock(&src_dock->fsdc_list_mutex);
        if (waiting) {
                M0_LOG(M0_DEBUG, "src_rec =" U128X_F " ref cnt:%d",
                                 U128_P(&src_rec->fsr_rec_id),
                                 (int)m0_ref_read(&src_rec->fsr_ref) - 1);
                m0_ref_put(&src_rec->fsr_ref);
                m0_fdmi__fs_put(src_rec);
        }
        /* @todo Phase 2: clear map <fdmi record id, endpoint>. */
        return M0_RC(0);
}

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