rpc: duplicate request cache for nfs

Duplicate request cache provides a mechanism for detecting
duplicate rpc requests from clients. DRC caches replies
and on duplicate requests, sends the cached reply instead of
re-processing the request.

Change-Id: I3d62a6c4aa86c92bf61f1038ca62a1a46bf1c303
BUG: 847624
Signed-off-by: Rajesh Amaravathi <rajesh@redhat.com>
Reviewed-on: http://review.gluster.org/4049
Reviewed-by: Rajesh Joseph <rjoseph@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

1 files changed, 811 insertions, 0 deletions

diff --git a/rpc/rpc-lib/src/rpc-drc.c b/rpc/rpc-lib/src/rpc-drc.c
new file mode 100644
index 00000000000..66d07cfe6bd
--- /dev/null
+++ b/rpc/rpc-lib/src/rpc-drc.c
@@ -0,0 +1,811 @@
+/*
+  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.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "rpcsvc.h"
+#ifndef RPC_DRC_H
+#include "rpc-drc.h"
+#endif
+#include "locking.h"
+#include "hashfn.h"
+#include "common-utils.h"
+#include "statedump.h"
+#include "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;
+        rpcsvc_request_t *req      = NULL;
+        drc_cached_op_t  *reply    = NULL;
+
+        GF_ASSERT (item);
+        GF_ASSERT (rb_node_data);
+        GF_ASSERT (param);
+
+        req = (rpcsvc_request_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->progver == reply->progversion)
+                return 0;
+
+        return 1;
+}
+
+/**
+ * drc_rb_calloc - used by rbtree api to allocate memory for nodes
+ *
+ * @param allocator - the libavl_allocator structure used by rbtree
+ * @param size - not needed by this function
+ * @return pointer to new cached reply (node in rbtree)
+ */
+static void *
+drc_rb_calloc (struct libavl_allocator *allocator, size_t size)
+{
+        rpcsvc_drc_globals_t *drc = NULL;
+
+        /* get the drc pointer by simple typecast, since allocator
+         * is the first member of rpcsvc_drc_globals_t
+         */
+        drc = (rpcsvc_drc_globals_t *)allocator;
+
+        return mem_get (drc->mempool);
+}
+
+/**
+ * drc_rb_free - used by rbtree api to free a node
+ *
+ * @param a - the libavl_allocator structure used by rbtree api
+ * @param block - node that needs to be freed
+ * @return void
+ */
+static void
+drc_rb_free (struct libavl_allocator *a, void *block)
+{
+        mem_put (block);
+}
+
+/**
+ * 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);
+
+        drc->allocator.libavl_malloc = drc_rb_calloc;
+        drc->allocator.libavl_free = drc_rb_free;
+
+        client->rbtree = rb_create (drc_compare_reqs, drc,
+                                    (struct libavl_allocator *)drc);
+        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;
+
+        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;
+
+        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 = client;
+        }
+
+        client = rpcsvc_drc_client_ref (req->trans->drc_client);
+
+        if (client->op_count == 0)
+                goto out;
+
+        reply = rb_find (client->rbtree, req);
+
+ out:
+        if (client)
+                rpcsvc_drc_client_unref (req->svc->drc, client);
+
+        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;
+
+                (void *)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 != reply) {
+                /* should never happen */
+                gf_log (GF_RPCSVC, GF_LOG_ERROR,
+                        "DRC failed to detect duplicates");
+                return -1;
+        } else if (*tmp_reply == NULL) {
+                /* mem alloc failed */
+                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_get (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;
+
+        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, "%d", drc->cache_hits);
+
+        gf_proc_dump_build_key (key, "drc", "in_transit_duplicate_requests");
+        gf_proc_dump_write (key, "%d", 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 out;
+
+        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 */
+                rpcsvc_drc_client_unref (drc, client);
+                trans->drc_client = NULL;
+                break;
+
+        default:
+                break;
+        }
+
+ out:
+        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);
+
+        if (!svc->drc) {
+                drc = GF_CALLOC (1, sizeof (rpcsvc_drc_globals_t),
+                                 gf_common_mt_drc_globals_t);
+                if (!drc)
+                        return -1;
+
+                svc->drc = drc;
+                LOCK_INIT (&drc->lock);
+        } else {
+                drc = svc->drc;
+        }
+
+        LOCK (&drc->lock);
+        if (drc->type != DRC_TYPE_NONE) {
+                ret = 0;
+                goto out;
+        }
+
+        /* 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_true;
+        }
+
+        if (ret == _gf_false) {
+                /* drc off */
+                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "DRC is off");
+                ret = 0;
+                goto out;
+        }
+
+        /* 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;
+        }
+
+        drc->type = drc_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;
+        }
+
+        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) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get mempool for"
+                        " DRC, drc-size: %d", drc->global_cache_size);
+                ret = -1;
+                goto out;
+        }
+
+        /* 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) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR,
+                        "registration of drc_notify function failed");
+                goto out;
+        }
+
+        gf_log (GF_RPCSVC, GF_LOG_DEBUG, "drc init successful");
+        drc->status = DRC_INITIATED;
+
+ out:
+        UNLOCK (&drc->lock);
+        if (ret == -1) {
+                if (drc->mempool) {
+                        mem_pool_destroy (drc->mempool);
+                        drc->mempool = NULL;
+                }
+                GF_FREE (drc);
+                svc->drc = NULL;
+        }
+        return ret;
+}