path: root/rpc/rpc-lib/src/rpc-drc.c

/*
  Copyright (c) 2013 Red Hat, Inc. <http://www.redhat.com>
  This file is part of GlusterFS.

  This file is licensed to you under your choice of the GNU Lesser
  General Public License, version 3 or any later version (LGPLv3 or
  later), or the GNU General Public License, version 2 (GPLv2), in all
  cases as published by the Free Software Foundation.
*/

#include "rpcsvc.h"
#ifndef RPC_DRC_H
#include "rpc-drc.h"
#endif
#include <glusterfs/locking.h>
#include <glusterfs/hashfn.h>
#include <glusterfs/common-utils.h>
#include <glusterfs/statedump.h>
#include <glusterfs/mem-pool.h>

#include <netinet/in.h>
#include <unistd.h>

/**
 * rpcsvc_drc_op_destroy - Destroys the cached reply
 *
 * @param drc - the main drc structure
 * @param reply - the cached reply to destroy
 * @return NULL if reply is destroyed, reply otherwise
 */
static drc_cached_op_t *
rpcsvc_drc_op_destroy(rpcsvc_drc_globals_t *drc, drc_cached_op_t *reply)
{
    GF_ASSERT(drc);
    GF_ASSERT(reply);

    if (reply->state == DRC_OP_IN_TRANSIT)
        return reply;

    iobref_unref(reply->msg.iobref);
    if (reply->msg.rpchdr)
        GF_FREE(reply->msg.rpchdr);
    if (reply->msg.proghdr)
        GF_FREE(reply->msg.proghdr);
    if (reply->msg.progpayload)
        GF_FREE(reply->msg.progpayload);

    list_del(&reply->global_list);
    reply->client->op_count--;
    drc->op_count--;
    mem_put(reply);
    reply = NULL;

    return reply;
}

/**
 * rpcsvc_drc_op_rb_unref - This function is used in rb tree cleanup only
 *
 * @param reply - the cached reply to unref
 * @param drc - the main drc structure
 * @return void
 */
static void
rpcsvc_drc_rb_op_destroy(void *reply, void *drc)
{
    rpcsvc_drc_op_destroy(drc, (drc_cached_op_t *)reply);
}

/**
 * rpcsvc_remove_drc_client - Cleanup the drc client
 *
 * @param client - the drc client to be removed
 * @return void
 */
static void
rpcsvc_remove_drc_client(drc_client_t *client)
{
    rb_destroy(client->rbtree, rpcsvc_drc_rb_op_destroy);
    list_del(&client->client_list);
    GF_FREE(client);
}

/**
 * rpcsvc_client_lookup - Given a sockaddr_storage, find the client if it exists
 *
 * @param drc - the main drc structure
 * @param sockaddr - the network address of the client to be looked up
 * @return drc client if it exists, NULL otherwise
 */
static drc_client_t *
rpcsvc_client_lookup(rpcsvc_drc_globals_t *drc,
                     struct sockaddr_storage *sockaddr)
{
    drc_client_t *client = NULL;

    GF_ASSERT(drc);
    GF_ASSERT(sockaddr);

    if (list_empty(&drc->clients_head))
        return NULL;

    list_for_each_entry(client, &drc->clients_head, client_list)
    {
        if (gf_sock_union_equal_addr(&client->sock_union,
                                     (union gf_sock_union *)sockaddr))
            return client;
    }

    return NULL;
}

/**
 * drc_compare_reqs - Used by rbtree to determine if incoming req matches with
 *                    an existing node(cached reply) in rbtree
 *
 * @param item - pointer to the incoming req
 * @param rb_node_data - pointer to an rbtree node (cached reply)
 * @param param - drc pointer - unused here, but used in *op_destroy
 * @return 0 if req matches reply, else (req->xid - reply->xid)
 */
int
drc_compare_reqs(const void *item, const void *rb_node_data, void *param)
{
    int ret = -1;
    drc_cached_op_t *req = NULL;
    drc_cached_op_t *reply = NULL;

    GF_ASSERT(item);
    GF_ASSERT(rb_node_data);
    GF_ASSERT(param);

    req = (drc_cached_op_t *)item;
    reply = (drc_cached_op_t *)rb_node_data;

    ret = req->xid - reply->xid;
    if (ret != 0)
        return ret;

    if (req->prognum == reply->prognum && req->procnum == reply->procnum &&
        req->progversion == reply->progversion)
        return 0;

    return 1;
}

/**
 * drc_init_client_cache - initialize a drc client and its rb tree
 *
 * @param drc - the main drc structure
 * @param client - the drc client to be initialized
 * @return 0 on success, -1 on failure
 */
static int
drc_init_client_cache(rpcsvc_drc_globals_t *drc, drc_client_t *client)
{
    GF_ASSERT(drc);
    GF_ASSERT(client);

    client->rbtree = rb_create(drc_compare_reqs, drc, NULL);
    if (!client->rbtree) {
        gf_log(GF_RPCSVC, GF_LOG_DEBUG, "rb tree creation failed");
        return -1;
    }

    return 0;
}

/**
 * rpcsvc_get_drc_client - find the drc client with given sockaddr, else
 *                         allocate and initialize a new drc client
 *
 * @param drc - the main drc structure
 * @param sockaddr - network address of client
 * @return drc client on success, NULL on failure
 */
static drc_client_t *
rpcsvc_get_drc_client(rpcsvc_drc_globals_t *drc,
                      struct sockaddr_storage *sockaddr)
{
    drc_client_t *client = NULL;

    GF_ASSERT(drc);
    GF_ASSERT(sockaddr);

    client = rpcsvc_client_lookup(drc, sockaddr);
    if (client)
        goto out;

    /* if lookup fails, allocate cache for the new client */
    client = GF_CALLOC(1, sizeof(drc_client_t), gf_common_mt_drc_client_t);
    if (!client)
        goto out;

    client->ref = 0;
    client->sock_union = (union gf_sock_union) * sockaddr;
    client->op_count = 0;
    INIT_LIST_HEAD(&client->client_list);

    if (drc_init_client_cache(drc, client)) {
        gf_log(GF_RPCSVC, GF_LOG_DEBUG, "initialization of drc client failed");
        GF_FREE(client);
        client = NULL;
        goto out;
    }
    drc->client_count++;

    list_add(&client->client_list, &drc->clients_head);

out:
    return client;
}

/**
 * rpcsvc_need_drc - Determine if a request needs DRC service
 *
 * @param req - incoming request
 * @return 1 if DRC is needed for req, 0 otherwise
 */
int
rpcsvc_need_drc(rpcsvc_request_t *req)
{
    rpcsvc_actor_t *actor = NULL;
    rpcsvc_drc_globals_t *drc = NULL;

    GF_ASSERT(req);
    GF_ASSERT(req->svc);

    drc = req->svc->drc;

    if (!drc || drc->status == DRC_UNINITIATED)
        return 0;

    actor = rpcsvc_program_actor(req);
    if (!actor)
        return 0;

    return (actor->op_type == DRC_NON_IDEMPOTENT && drc->type != DRC_TYPE_NONE);
}

/**
 * rpcsvc_drc_client_ref - ref the drc client
 *
 * @param client - the drc client to ref
 * @return client
 */
static drc_client_t *
rpcsvc_drc_client_ref(drc_client_t *client)
{
    GF_ASSERT(client);
    client->ref++;
    return client;
}

/**
 * rpcsvc_drc_client_unref - unref the drc client, and destroy
 *                           the client on last unref
 *
 * @param drc - the main drc structure
 * @param client - the drc client to unref
 * @return NULL if it is the last unref, client otherwise
 */
static drc_client_t *
rpcsvc_drc_client_unref(rpcsvc_drc_globals_t *drc, drc_client_t *client)
{
    GF_ASSERT(drc);
    GF_ASSERT(client->ref);

    client->ref--;
    if (!client->ref) {
        drc->client_count--;
        rpcsvc_remove_drc_client(client);
        client = NULL;
    }

    return client;
}

/**
 * rpcsvc_drc_lookup - lookup a request to see if it is already cached
 *
 * @param req - incoming request
 * @return cached reply of req if found, NULL otherwise
 */
drc_cached_op_t *
rpcsvc_drc_lookup(rpcsvc_request_t *req)
{
    drc_client_t *client = NULL;
    drc_cached_op_t *reply = NULL;
    drc_cached_op_t new = {
        .xid = req->xid,
        .prognum = req->prognum,
        .progversion = req->progver,
        .procnum = req->procnum,
    };

    GF_ASSERT(req);

    if (!req->trans->drc_client) {
        client = rpcsvc_get_drc_client(req->svc->drc,
                                       &req->trans->peerinfo.sockaddr);
        if (!client)
            goto out;

        req->trans->drc_client = rpcsvc_drc_client_ref(client);
    }

    client = req->trans->drc_client;

    if (client->op_count == 0)
        goto out;

    reply = rb_find(client->rbtree, &new);

out:
    return reply;
}

/**
 * rpcsvc_send_cached_reply - send the cached reply for the incoming request
 *
 * @param req - incoming request (which is a duplicate in this case)
 * @param reply - the cached reply for req
 * @return 0 on successful reply submission, -1 or other non-zero value
 * otherwise
 */
int
rpcsvc_send_cached_reply(rpcsvc_request_t *req, drc_cached_op_t *reply)
{
    int ret = 0;

    GF_ASSERT(req);
    GF_ASSERT(reply);

    gf_log(GF_RPCSVC, GF_LOG_DEBUG,
           "sending cached reply: xid: %d, "
           "client: %s",
           req->xid, req->trans->peerinfo.identifier);

    rpcsvc_drc_client_ref(reply->client);
    ret = rpcsvc_transport_submit(
        req->trans, reply->msg.rpchdr, reply->msg.rpchdrcount,
        reply->msg.proghdr, reply->msg.proghdrcount, reply->msg.progpayload,
        reply->msg.progpayloadcount, reply->msg.iobref, req->trans_private);
    rpcsvc_drc_client_unref(req->svc->drc, reply->client);

    return ret;
}

/**
 * rpcsvc_cache_reply - cache the reply for the processed request 'req'
 *
 * @param req - processed request
 * @param iobref - iobref structure of the reply
 * @param rpchdr - rpc header of the reply
 * @param rpchdrcount - size of rpchdr
 * @param proghdr - program header of the reply
 * @param proghdrcount - size of proghdr
 * @param payload - payload of the reply if any
 * @param payloadcount - size of payload
 * @return 0 on success, -1 on failure
 */
int
rpcsvc_cache_reply(rpcsvc_request_t *req, struct iobref *iobref,
                   struct iovec *rpchdr, int rpchdrcount, struct iovec *proghdr,
                   int proghdrcount, struct iovec *payload, int payloadcount)
{
    int ret = -1;
    drc_cached_op_t *reply = NULL;

    GF_ASSERT(req);
    GF_ASSERT(req->reply);

    reply = req->reply;

    reply->state = DRC_OP_CACHED;

    reply->msg.iobref = iobref_ref(iobref);

    reply->msg.rpchdrcount = rpchdrcount;
    reply->msg.rpchdr = iov_dup(rpchdr, rpchdrcount);

    reply->msg.proghdrcount = proghdrcount;
    reply->msg.proghdr = iov_dup(proghdr, proghdrcount);

    reply->msg.progpayloadcount = payloadcount;
    if (payloadcount)
        reply->msg.progpayload = iov_dup(payload, payloadcount);

    //        rpcsvc_drc_client_unref (req->svc->drc, req->trans->drc_client);
    //        rpcsvc_drc_op_unref (req->svc->drc, reply);
    ret = 0;

    return ret;
}

/**
 * rpcsvc_vacate_drc_entries - free up some percentage of drc cache
 *                             based on the lru factor
 *
 * @param drc - the main drc structure
 * @return void
 */
static void
rpcsvc_vacate_drc_entries(rpcsvc_drc_globals_t *drc)
{
    uint32_t i = 0;
    uint32_t n = 0;
    drc_cached_op_t *reply = NULL;
    drc_cached_op_t *tmp = NULL;
    drc_client_t *client = NULL;

    GF_ASSERT(drc);

    n = drc->global_cache_size / drc->lru_factor;

    list_for_each_entry_safe_reverse(reply, tmp, &drc->cache_head, global_list)
    {
        /* Don't delete ops that are in transit */
        if (reply->state == DRC_OP_IN_TRANSIT)
            continue;

        client = reply->client;

        rb_delete(client->rbtree, reply);

        rpcsvc_drc_op_destroy(drc, reply);
        rpcsvc_drc_client_unref(drc, client);
        i++;
        if (i >= n)
            break;
    }
}

/**
 * rpcsvc_add_op_to_cache - insert the cached op into the client rbtree and drc
 * list
 *
 * @param drc - the main drc structure
 * @param reply - the op to be inserted
 * @return 0 on success, -1 on failure
 */
static int
rpcsvc_add_op_to_cache(rpcsvc_drc_globals_t *drc, drc_cached_op_t *reply)
{
    drc_client_t *client = NULL;
    drc_cached_op_t **tmp_reply = NULL;

    GF_ASSERT(drc);
    GF_ASSERT(reply);

    client = reply->client;

    /* cache is full, free up some space */
    if (drc->op_count >= drc->global_cache_size)
        rpcsvc_vacate_drc_entries(drc);

    tmp_reply = (drc_cached_op_t **)rb_probe(client->rbtree, reply);
    if (!tmp_reply) {
        /* mem alloc failed */
        return -1;
    } else if (*tmp_reply != reply) {
        /* should never happen */
        gf_log(GF_RPCSVC, GF_LOG_ERROR, "DRC failed to detect duplicates");
        return -1;
    }

    client->op_count++;
    list_add(&reply->global_list, &drc->cache_head);
    drc->op_count++;

    return 0;
}

/**
 * rpcsvc_cache_request - cache the in-transition incoming request
 *
 * @param req - incoming request
 * @return 0 on success, -1 on failure
 */
int
rpcsvc_cache_request(rpcsvc_request_t *req)
{
    int ret = -1;
    drc_client_t *client = NULL;
    drc_cached_op_t *reply = NULL;
    rpcsvc_drc_globals_t *drc = NULL;

    GF_ASSERT(req);

    drc = req->svc->drc;

    client = req->trans->drc_client;
    if (!client) {
        gf_log(GF_RPCSVC, GF_LOG_DEBUG, "drc client is NULL");
        goto out;
    }

    reply = mem_get0(drc->mempool);
    if (!reply)
        goto out;

    reply->client = rpcsvc_drc_client_ref(client);
    reply->xid = req->xid;
    reply->prognum = req->prognum;
    reply->progversion = req->progver;
    reply->procnum = req->procnum;
    reply->state = DRC_OP_IN_TRANSIT;
    req->reply = reply;
    INIT_LIST_HEAD(&reply->global_list);

    ret = rpcsvc_add_op_to_cache(drc, reply);
    if (ret) {
        req->reply = NULL;
        rpcsvc_drc_op_destroy(drc, reply);
        rpcsvc_drc_client_unref(drc, client);
        gf_log(GF_RPCSVC, GF_LOG_DEBUG, "Failed to add op to drc cache");
    }

out:
    return ret;
}

/**
 *
 * rpcsvc_drc_priv - function which dumps the drc state
 *
 * @param drc - the main drc structure
 * @return 0 on success, -1 on failure
 */
int32_t
rpcsvc_drc_priv(rpcsvc_drc_globals_t *drc)
{
    int i = 0;
    char key[GF_DUMP_MAX_BUF_LEN] = {0};
    drc_client_t *client = NULL;
    char ip[INET6_ADDRSTRLEN] = {0};

    if (!drc || drc->status == DRC_UNINITIATED) {
        gf_log(GF_RPCSVC, GF_LOG_DEBUG,
               "DRC is "
               "uninitialized, not dumping its state");
        return 0;
    }

    gf_proc_dump_add_section("rpc.drc");

    if (TRY_LOCK(&drc->lock))
        return -1;

    gf_proc_dump_build_key(key, "drc", "type");
    gf_proc_dump_write(key, "%d", drc->type);

    gf_proc_dump_build_key(key, "drc", "client_count");
    gf_proc_dump_write(key, "%d", drc->client_count);

    gf_proc_dump_build_key(key, "drc", "current_cache_size");
    gf_proc_dump_write(key, "%d", drc->op_count);

    gf_proc_dump_build_key(key, "drc", "max_cache_size");
    gf_proc_dump_write(key, "%d", drc->global_cache_size);

    gf_proc_dump_build_key(key, "drc", "lru_factor");
    gf_proc_dump_write(key, "%d", drc->lru_factor);

    gf_proc_dump_build_key(key, "drc", "duplicate_request_count");
    gf_proc_dump_write(key, "%" PRIu64, drc->cache_hits);

    gf_proc_dump_build_key(key, "drc", "in_transit_duplicate_requests");
    gf_proc_dump_write(key, "%" PRIu64, drc->intransit_hits);

    list_for_each_entry(client, &drc->clients_head, client_list)
    {
        gf_proc_dump_build_key(key, "client", "%d.ip-address", i);
        memset(ip, 0, INET6_ADDRSTRLEN);
        switch (client->sock_union.storage.ss_family) {
            case AF_INET:
                gf_proc_dump_write(
                    key, "%s",
                    inet_ntop(AF_INET, &client->sock_union.sin.sin_addr.s_addr,
                              ip, INET_ADDRSTRLEN));
                break;
            case AF_INET6:
                gf_proc_dump_write(
                    key, "%s",
                    inet_ntop(AF_INET6, &client->sock_union.sin6.sin6_addr, ip,
                              INET6_ADDRSTRLEN));
                break;
            default:
                gf_proc_dump_write(key, "%s", "N/A");
        }

        gf_proc_dump_build_key(key, "client", "%d.ref_count", i);
        gf_proc_dump_write(key, "%d", client->ref);
        gf_proc_dump_build_key(key, "client", "%d.op_count", i);
        gf_proc_dump_write(key, "%d", client->op_count);
        i++;
    }

    UNLOCK(&drc->lock);
    return 0;
}

/**
 * rpcsvc_drc_notify - function which is notified of RPC transport events
 *
 * @param svc - pointer to rpcsvc_t structure of the rpc
 * @param xl - pointer to the xlator
 * @param event - the event which triggered this notify
 * @param data - the transport structure
 * @return 0 on success, -1 on failure
 */
int
rpcsvc_drc_notify(rpcsvc_t *svc, void *xl, rpcsvc_event_t event, void *data)
{
    int ret = -1;
    rpc_transport_t *trans = NULL;
    drc_client_t *client = NULL;
    rpcsvc_drc_globals_t *drc = NULL;

    GF_ASSERT(svc);
    GF_ASSERT(svc->drc);
    GF_ASSERT(data);

    drc = svc->drc;

    if (drc->status == DRC_UNINITIATED || drc->type == DRC_TYPE_NONE)
        return 0;

    LOCK(&drc->lock);
    {
        trans = (rpc_transport_t *)data;
        client = rpcsvc_get_drc_client(drc, &trans->peerinfo.sockaddr);
        if (!client)
            goto unlock;

        switch (event) {
            case RPCSVC_EVENT_ACCEPT:
                trans->drc_client = rpcsvc_drc_client_ref(client);
                ret = 0;
                break;

            case RPCSVC_EVENT_DISCONNECT:
                ret = 0;
                if (list_empty(&drc->clients_head))
                    break;
                /* should be the last unref */
                trans->drc_client = NULL;
                rpcsvc_drc_client_unref(drc, client);
                break;

            default:
                break;
        }
    }
unlock:
    UNLOCK(&drc->lock);
    return ret;
}

/**
 * rpcsvc_drc_init - Initialize the duplicate request cache service
 *
 * @param svc - pointer to rpcsvc_t structure of the rpc
 * @param options - the options dictionary which configures drc
 * @return 0 on success, non-zero integer on failure
 */
int
rpcsvc_drc_init(rpcsvc_t *svc, dict_t *options)
{
    int ret = 0;
    uint32_t drc_type = 0;
    uint32_t drc_size = 0;
    uint32_t drc_factor = 0;
    rpcsvc_drc_globals_t *drc = NULL;

    GF_ASSERT(svc);
    GF_ASSERT(options);

    /* Toggle DRC on/off, when more drc types(persistent/cluster)
     * are added, we shouldn't treat this as boolean. */
    ret = dict_get_str_boolean(options, "nfs.drc", _gf_false);
    if (ret == -1) {
        gf_log(GF_RPCSVC, GF_LOG_INFO, "drc user options need second look");
        ret = _gf_false;
    }

    gf_log(GF_RPCSVC, GF_LOG_INFO, "DRC is turned %s", (ret ? "ON" : "OFF"));

    /*DRC off, nothing to do */
    if (ret == _gf_false)
        return (0);

    drc = GF_CALLOC(1, sizeof(rpcsvc_drc_globals_t),
                    gf_common_mt_drc_globals_t);
    if (!drc)
        return (-1);

    LOCK_INIT(&drc->lock);
    svc->drc = drc;

    /* Specify type of DRC to be used */
    ret = dict_get_uint32(options, "nfs.drc-type", &drc_type);
    if (ret) {
        gf_log(GF_RPCSVC, GF_LOG_DEBUG,
               "drc type not set. Continuing with default");
        drc_type = DRC_DEFAULT_TYPE;
    }

    /* Set the global cache size (no. of ops to cache) */
    ret = dict_get_uint32(options, "nfs.drc-size", &drc_size);
    if (ret) {
        gf_log(GF_RPCSVC, GF_LOG_DEBUG,
               "drc size not set. Continuing with default size");
        drc_size = DRC_DEFAULT_CACHE_SIZE;
    }

    LOCK(&drc->lock);

    drc->type = drc_type;
    drc->global_cache_size = drc_size;

    /* Mempool for cached ops */
    drc->mempool = mem_pool_new(drc_cached_op_t, drc->global_cache_size);
    if (!drc->mempool) {
        UNLOCK(&drc->lock);
        gf_log(GF_RPCSVC, GF_LOG_ERROR,
               "Failed to get mempool for DRC, drc-size: %d", drc_size);
        ret = -1;
        goto post_unlock;
    }

    /* What percent of cache to be evicted whenever it fills up */
    ret = dict_get_uint32(options, "nfs.drc-lru-factor", &drc_factor);
    if (ret) {
        gf_log(GF_RPCSVC, GF_LOG_DEBUG,
               "drc lru factor not set. Continuing with policy default");
        drc_factor = DRC_DEFAULT_LRU_FACTOR;
    }

    drc->lru_factor = (drc_lru_factor_t)drc_factor;

    INIT_LIST_HEAD(&drc->clients_head);
    INIT_LIST_HEAD(&drc->cache_head);

    ret = rpcsvc_register_notify(svc, rpcsvc_drc_notify, THIS);
    if (ret) {
        UNLOCK(&drc->lock);
        gf_log(GF_RPCSVC, GF_LOG_ERROR,
               "registration of drc_notify function failed");
        goto post_unlock;
    }

    drc->status = DRC_INITIATED;
    UNLOCK(&drc->lock);
    gf_log(GF_RPCSVC, GF_LOG_DEBUG, "drc init successful");
post_unlock:
    if (ret == -1) {
        if (drc->mempool) {
            mem_pool_destroy(drc->mempool);
            drc->mempool = NULL;
        }
        GF_FREE(drc);
        svc->drc = NULL;
    }
    return ret;
}

int
rpcsvc_drc_deinit(rpcsvc_t *svc)
{
    rpcsvc_drc_globals_t *drc = NULL;

    if (!svc)
        return (-1);

    drc = svc->drc;
    if (!drc)
        return (0);

    LOCK(&drc->lock);
    (void)rpcsvc_unregister_notify(svc, rpcsvc_drc_notify, THIS);
    if (drc->mempool) {
        mem_pool_destroy(drc->mempool);
        drc->mempool = NULL;
    }
    UNLOCK(&drc->lock);

    GF_FREE(drc);
    svc->drc = NULL;

    return (0);
}

int
rpcsvc_drc_reconfigure(rpcsvc_t *svc, dict_t *options)
{
    int ret = -1;
    gf_boolean_t enable_drc = _gf_false;
    rpcsvc_drc_globals_t *drc = NULL;
    uint32_t drc_size = 0;

    /* Input sanitization */
    if ((!svc) || (!options))
        return (-1);

    /* If DRC was not enabled before, Let rpcsvc_drc_init() to
     * take care of DRC initialization part.
     */
    drc = svc->drc;
    if (!drc) {
        return rpcsvc_drc_init(svc, options);
    }

    /* DRC was already enabled before. Going to be reconfigured. Check
     * if reconfigured options contain "nfs.drc" and "nfs.drc-size".
     *
     * NB: If DRC is "OFF", "drc-size" has no role to play.
     *     So, "drc-size" gets evaluated IFF DRC is "ON".
     *
     * If DRC is reconfigured,
     *     case 1: DRC is "ON"
     *         sub-case 1: drc-size remains same
     *              ACTION: Nothing to do.
     *         sub-case 2: drc-size just changed
     *              ACTION: rpcsvc_drc_deinit() followed by
     *                      rpcsvc_drc_init().
     *
     *     case 2: DRC is "OFF"
     *         ACTION: rpcsvc_drc_deinit()
     */
    ret = dict_get_str_boolean(options, "nfs.drc", _gf_false);
    if (ret < 0)
        ret = _gf_false;

    enable_drc = ret;
    gf_log(GF_RPCSVC, GF_LOG_INFO, "DRC is turned %s", (ret ? "ON" : "OFF"));

    /* case 1: DRC is "ON"*/
    if (enable_drc) {
        /* Fetch drc-size if reconfigured */
        if (dict_get_uint32(options, "nfs.drc-size", &drc_size))
            drc_size = DRC_DEFAULT_CACHE_SIZE;

        /* case 1: sub-case 1*/
        if (drc->global_cache_size == drc_size)
            return (0);

        /* case 1: sub-case 2*/
        (void)rpcsvc_drc_deinit(svc);
        return rpcsvc_drc_init(svc, options);
    }

    /* case 2: DRC is "OFF" */
    return rpcsvc_drc_deinit(svc);
}