Archipelago

/*

 * Copyright 2013 GRNET S.A. All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or

 * without modification, are permitted provided that the following

 * conditions are met:

 *

 *   1. Redistributions of source code must retain the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer.

 *   2. Redistributions in binary form must reproduce the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer in the documentation and/or other materials

 *      provided with the distribution.

 *

 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 * POSSIBILITY OF SUCH DAMAGE.

 *

 * The views and conclusions contained in the software and

 * documentation are those of the authors and should not be

 * interpreted as representing official policies, either expressed

 * or implied, of GRNET S.A.

 */

#include <stdio.h>

#include <unistd.h>

#include <sys/types.h>

#include <pthread.h>

#include <xseg/xseg.h>

#include <peer.h>

#include <time.h>

#include <xtypes/xlock.h>

#include <xtypes/xq.h>

#include <xtypes/xhash.h>

#include <xtypes/xworkq.h>

#include <xtypes/xwaitq.h>

#include <xseg/protocol.h>

#include <xtypes/xcache.h>

#define MAX_ARG_LEN 12

/* bucket states */

#define INVALID 0

#define LOADING 1

#define VALID   2

#define DIRTY   3

#define WRITING 4

#define bucket_readable(__status) \

        (__status == VALID || __status == DIRTY || __status == WRITING)

/* write policies */

/*FIXME: Shouldn't they be on a flag?*/

#define WRITETHROUGH 1

#define WRITEBACK    2

#define WRITE_POLICY(__wcp)                                                        \

        __wcp == WRITETHROUGH ? "writethrough" :                \

        __wcp == WRITEBACK        ? "writeback" :                                \

        "undefined"

/* object states */

#define INVALIDATED (1 << 0)

/* cio states */

#define CIO_FAILED        1

#define CIO_ACCEPTED        2

#define CIO_READING        3

#define BUCKET_SIZE_QUANTUM 4096

struct cache_io {

        uint32_t state;

        xcache_handler h;

        uint32_t pending_reqs;

        struct work work;

};

struct cached {

        struct xcache *cache;

        uint64_t cache_size; /*Number of objects*/

        uint64_t max_req_size;

        uint32_t object_size; /*Bytes*/ /*TODO: change this to buckets_per_object*/

        uint32_t bucket_size; /*In bytes*/

        uint32_t buckets_per_object;

        xport bportno;

        int write_policy;

        //scheduler

};

struct ce {

        unsigned char *data;

        uint32_t *status;

        struct xwaitq *waitq;

        uint32_t flags;

        struct xlock lock;

        struct xworkq workq;

        struct peer_req pr;

};

/*

 * Helper functions

 */

static inline struct cached * __get_cached(struct peerd *peer)

{

        return (struct cached *) peer->priv;

}

static inline struct cache_io * __get_cache_io(struct peer_req *pr)

{

        return (struct cache_io *) pr->priv;

}

static inline uint32_t __calculate_size(struct cached *cached,

                uint32_t start, uint32_t end)

{

        return (end - start + 1) * cached->bucket_size;

}

static inline uint32_t __calculate_offset(struct cached *cached,

                uint32_t start)

{

        return start * cached->bucket_size;

}

static uint32_t __get_bucket(struct cached *cache, uint64_t offset)

{

        return (offset / cache->bucket_size);

}

static int is_not_loading(void *arg)

{

        uint32_t *status = (uint32_t *)arg;

        return (*status != LOADING);

}

static void print_cached(struct cached *cached)

{

        if (!cached) {

                XSEGLOG2(&lc, W, "Struct cached is NULL\n");

                return;

        }

        XSEGLOG2(&lc, I, "Struct cached fields:\n"

                        "                     cache        = %p\n"

                        "                     cache_size   = %lu\n"

                        "                     max_req_size = %lu\n"

                        "                     object_size  = %lu\n"

                        "                     bucket_size  = %lu\n"

                        "                     bucks_per_obj= %lu\n"

                        "                     Bportno      = %d\n"

                        "                     write_policy = %s\n",

                        cached->cache, cached->cache_size, cached->max_req_size,

                        cached->object_size, cached->bucket_size,

                        cached->buckets_per_object, cached->bportno,

                        WRITE_POLICY(cached->write_policy));

}

int read_write_policy(char *write_policy)

{

        if (strcmp(write_policy, "writethrough") == 0)

                return WRITETHROUGH;

        if (strcmp(write_policy, "writeback") == 0)

                return WRITEBACK;

        return -1;

}

/*

 * Convert string to size in bytes.

 * If syntax is invalid, return 0. Values such as zero and non-integer

 * multiples of segment's page size should not be accepted.

 */

uint64_t str2num(char *str)

{

        char *unit;

        uint64_t num;

        num = strtoll(str, &unit, 10);

        if (strlen(unit) > 1) //Invalid syntax

                return 0;

        else if (strlen(unit) < 1) //Plain number in bytes

                return num;

        switch (*unit) {

                case 'g':

                case 'G':

                        num *= 1024;

                case 'm':

                case 'M':

                        num *= 1024;

                case 'k':

                case 'K':

                        num *= 1024;

                        break;

                default:

                        num = 0;

        }

        return num;

}

/*

 * rw_range handles the issuing of requests to the blocker. Usually called when

 * we need to read(write) data from(to) slower media.

 */

static int rw_range(struct peerd *peer, struct peer_req *pr, int action,

                uint32_t start, uint32_t end)

{

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct ce *ce = get_cache_entry(cached->cache, cio->h);

        struct xseg_request *req;

        struct xseg_request *req_old = pr->req;

        struct xseg *xseg = peer->xseg;

        xport srcport = pr->portno;

        xport dstport = cached->bportno;

        xport p;

        char *req_target;

        int r;

        uint32_t i;

        req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);

        if (!req) {

                XSEGLOG2(&lc, W, "Cannot get request\n");

                return -1;

        }

        req->size = __calculate_size(cached, start, end);

        req->offset = __calculate_offset(cached, start);

        req->op = req_old->op;

        //Allocate enough space for the data and the target's name

        r = xseg_prep_request(xseg, req, req_old->targetlen, req->size);

        if (r < 0) {

                XSEGLOG2(&lc, W, "Cannot prepare request! (%lu, %llu)\n",

                                req_old->targetlen, (unsigned long long)req->size);

                goto put_xseg_request;

        }

        req_target = xseg_get_target(xseg, req);

        char *req_old_target = xseg_get_target(xseg, req_old);

        strncpy(req_target, req_old_target, req_old->targetlen);

        if (req->op == X_WRITE) {

                /*

                 * //Paste data

                 * req_data = xseg_get_data(xseg, req);

                 * memcpy(req_data, req_old->data, size);

                 */

        } else {

                for (i=start; i<=end; i++){

                        ce->status[i] = LOADING;

                }

        }

        r = xseg_set_req_data(xseg, req, pr);

        if (r < 0) {

                XSEGLOG2(&lc, W, "Cannot set request data\n");

                goto put_xseg_request;

        }

        p = xseg_submit(xseg, req, srcport, X_ALLOC);

        if (p == NoPort) {

                XSEGLOG2(&lc, W, "Cannot submit request\n");

                goto out_unset_data;

        }

        r = xseg_signal(xseg, p);

        return 0;

out_unset_data:

        xseg_set_req_data(xseg, req, NULL);

put_xseg_request:

        if (xseg_put_request(xseg, req, srcport))

                XSEGLOG2(&lc, W, "Cannot put request\n");

        return -1;

}

int on_init(void *c, void *e)

{

        uint32_t i;

        struct peerd *peer = (struct peerd *)c;

        struct cached *cached = peer->priv;

        struct ce *ce = (struct ce *)e;

        ce->flags = 0;

        memset(ce->data, 0, cached->object_size);

        for (i = 0; i < cached->buckets_per_object; i++) {

                ce->status[i] = INVALID;

        }

        xlock_release(&ce->lock);

        return 0;

}

void on_put(void *c, void *e)

{

        struct peerd *peer = (struct peerd *)c;

        struct cached *cached = peer->priv;

        struct ce *ce = (struct ce *)e;

        //since we are the last referrer to the cache entry

        //no lock is needed.

        XSEGLOG2(&lc, D, "Putting cache entry %p", e);

        uint32_t start, end, i = 0;

        if (cached->write_policy == WRITETHROUGH || ce->flags & INVALIDATED)

                return;

        //write all dirty buckets.

        while(i < cached->buckets_per_object){

                if (ce->status[i] != DIRTY){

                        i++;

                        continue;

                }

                start = i;

                while (i < cached->buckets_per_object &&

                        (i-start)*cached->bucket_size < cached->max_req_size &&

                                ce->status[i] == DIRTY){

                        i++;

                }

                end = i;

                //problem: no assocciated pr

                //maybe put one in cache entry

                //rw_range(cached, ce, 1, start, end);

        }

}

void * init_node(void *c)

{

        int i;

        struct peerd *peer = (struct peerd *)c;

        struct cached *cached = peer->priv;

        struct ce *ce = malloc(sizeof(struct ce));

        if (!ce)

                goto ce_fail;

        xlock_release(&ce->lock);

        ce->data = malloc(sizeof(unsigned char) * cached->object_size);

        ce->status = malloc(sizeof(uint32_t) * cached->buckets_per_object);

        ce->waitq = malloc(sizeof(struct xwaitq) * cached->buckets_per_object);

        if (!ce->data || !ce->status || !ce->waitq)

                goto ce_fields_fail;

        ce->pr.peer = peer;

        for (i = 0; i < cached->buckets_per_object; i++) {

                xwaitq_init(&ce->waitq[i], is_not_loading, &ce->status[i],

                                XWAIT_SIGNAL_ONE);

        }

        xworkq_init(&ce->workq, &ce->lock, 0);

        return ce;

ce_fields_fail:

        free(ce->data);

        free(ce->status);

        free(ce->waitq);

        free(ce);

ce_fail:

        perror("malloc");

        return NULL;

}

struct xcache_ops c_ops = {

        .on_init = on_init,

        .on_put  = on_put,

        .on_node_init = init_node

};

static uint32_t __get_next_invalid(struct ce *ce, uint32_t start,

                                        uint32_t limit)

{

        uint32_t end = start + 1;

        while (end <= limit && ce->status[end] == INVALID)

                end++;

        return end;

}

static void cached_fail(struct peerd *peer, struct peer_req *pr)

{

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        if (cio->h != NoEntry){

                xcache_put(cached->cache, cio->h);

        }

        cio->h = NoEntry;

        fail(peer, pr);

}

static void cached_complete(struct peerd *peer, struct peer_req *pr)

{

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        if (cio->h != NoEntry){

                xcache_put(cached->cache, cio->h);

        }

        cio->h = NoEntry;

        complete(peer, pr);

}

static void handle_read(void *arg);

//is this necessary?

static void status_changed(void *arg)

{

        /*

         * In this context we hold a reference to the cache entry.

         *

         * This function gets called only after the bucket at which the

         * current peer_req is waiting, has finished loading of failed.

         *

         * Assumptions:

         *         Each pr waits only at one bucket at any time. That means that

         *         under no circumstances, this function get called simutaneously

         *         for the same pr.

         */

        struct peer_req *pr = (struct peer_req *)arg;

        struct peerd *peer = pr->peer;

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct ce *ce = get_cache_entry(cached->cache, cio->h);

        if (xworkq_enqueue(&ce->workq, handle_read, (void *)pr) < 0){

                //FAIL or mark as failed ? are we the last?

                if (cio->pending_reqs){

                        // cannot put here, since there are outstanding reqs to

                        // be received.

                        // Simply mark pr as failed.

                        cio->state = CIO_FAILED;

                } else {

                        //safe to fail here, since there is no pending action on

                        //this pr.

                        cached_fail(peer, pr);

                }

        }

}

/*

 * handle_read reads all buckets within a given request's range.

 * If a bucket is:

 * VALID || DIRTY || WRITING: it's good to read.

 * INVALID: we have to issue a request (via blocker) to read it from slower

 *          media.

 * LOADING: We have to wait (on a waitq) for the slower media to answer our

 *          previous request.

 */

static void handle_read(void *arg)

{

        /*

         * In this context we hold a reference to the cache entry and

         * the assocciated lock

         */

        struct peer_req *pr = (struct peer_req *)arg;

        struct peerd *peer = pr->peer;

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct xseg_request *req = pr->req;

        struct ce *ce = get_cache_entry(cached->cache, cio->h);

        uint32_t start_bucket, end_bucket, next;

        uint32_t i, b, limit;

        uint32_t pending_buckets = 0;

        XSEGLOG2(&lc, E, "Handle read started for %p (ce: %p)", pr, ce );

        if (cio->state == CIO_FAILED)

                goto out;

        b = __get_bucket(cached, req->offset);

        limit = __get_bucket(cached, req->offset + req->size);

        //assert limit < cached->object_size

        for (i = b; i <= limit; i++) {

                if (bucket_readable(ce->status[i]))

                        continue;

                if (ce->status[i] != LOADING){

                        XSEGLOG2(&lc, I, "Found invalid bucket %lu\n", i);

                        start_bucket = i;

                        end_bucket = __get_next_invalid(ce, start_bucket, limit) - 1;

                        i = end_bucket;

                        if (rw_range(peer, pr, 0, start_bucket, end_bucket) < 0){

                                cio->state = CIO_FAILED;

                                break;

                        }

                        cio->pending_reqs++;

                        cio->state =  CIO_READING;

                }

                pending_buckets++;

        }

        if (pending_buckets) {

                XSEGLOG2(&lc, D, "Pending buckets exists: %u\n", pending_buckets);

                /* Do not put cache entry yet */

                cio->work.job_fn = handle_read;

                cio->work.job = pr;

                /* wait on the last bucket */

                XSEGLOG2(&lc, I, "Enqueuing cio %p in waitq (fn: handle_read).\n", cio);

                xwaitq_enqueue(&ce->waitq[end_bucket], &cio->work);

                XSEGLOG2(&lc, I, "Handle_read returned after enqueuing cio %p in waitq.\n", cio);

                return;

        }

out:

        if (cio->state == CIO_FAILED){

                if (!cio->pending_reqs)

                        cached_fail(peer, pr);

        }

        else{

                //serve req;

                cached_complete(peer, pr);

        }

        return;

}

static void handle_write(void *arg)

{

        //if writeback

        //          for each bucket

        //                write all buckets

        //                mark them as dirty

        //        cache_put(h)

        //        complete

        //else

        //        send write to blocker

        /*

         * In this context we hold a reference to the cache entry and

         * the assocciated lock

         */

        int r;

        struct peer_req *pr = (struct peer_req *)arg;

        struct peerd *peer = pr->peer;

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct ce *ce = get_cache_entry(cached->cache, cio->h);

        (void)ce;

        if (cached->write_policy == WRITETHROUGH){

                //send write to blocker

                //return

        } else if (cached->write_policy == WRITEBACK) {

                //for each bucket

                //        write all buckets

                //        mark them as dirty

                r = 0;

        } else {

                r = -1;

        }

out:

        if (r < 0)

                cached_fail(peer, pr);

        else

                cached_complete(peer, pr);

        return;

}

/*

 * handle_readwrite is called when we accept a request.

 * Its purpose is to find a handler associated with the request's target

 * object (partial cache hit), or to allocate a new one (cache_miss) and insert

 * it in xcache.

 *

 * Depending on the op type, a handler function is enqueued in the workq of the

 * target's cache_entry.

 */

static int handle_readwrite(struct peerd *peer, struct peer_req *pr)

{

        struct ce *ce;

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct xseg_request *req = pr->req;

        char name[XSEG_MAX_TARGETLEN + 1];

        char *target;

        int r = -1;

        xcache_handler h = NoEntry, nh;

        XSEGLOG2(&lc, I, "Started\n");

        target = xseg_get_target(peer->xseg, req);

        strncpy(name, target, req->targetlen);

        name[req->targetlen] = 0;

        XSEGLOG2(&lc, D, "Target is %s\n", name);

        h = xcache_lookup(cached->cache, name);

        if (h == NoEntry){

                h = xcache_alloc_init(cached->cache, name);

                if (h == NoEntry){

                        XSEGLOG2(&lc, E, "Could not allocate cache entry");

                        goto out;

                }

                nh = xcache_insert(cached->cache, h);

                if (nh == NoEntry){

                        xcache_put(cached->cache, h);

                        XSEGLOG2(&lc, E, "Could not insert cache entry");

                        goto out;

                }

                if (nh != h){

                        /* if insert returns another cache entry than the one we

                         * allocated and requested to be inserted, then

                         * someone else beat us to the insertion of a cache

                         * entry assocciated with the same name. Use this cache

                         * entry instead and put the one we allocated.

                         */

                        xcache_put(cached->cache, h);

                        h = nh;

                }

        }

        ce = (struct ce *)get_cache_entry(cached->cache, h);

        if (!ce){

                r = -1;

                XSEGLOG2(&lc, E, "Received cache entry handler %lu but no cache entry", h);

                goto out;

        }

        cio->h = h;

        if (req->op == X_WRITE)

                r = xworkq_enqueue(&ce->workq, handle_write, (void *)pr);

        else if (req->op == X_READ)

                r = xworkq_enqueue(&ce->workq, handle_read, (void *)pr);

        else {

                r = -1;

                XSEGLOG2(&lc, E, "Invalid op %u", req->op);

                goto out;

        }

out:

        if (r < 0){

                XSEGLOG2(&lc, E, "Failing pr %p", pr);

                cached_fail(peer, pr);

        }

        return r;

}

struct req_completion{

        struct peer_req *pr;

        struct xseg_request *req;

};

static void complete_read(void *arg)

{

        /*

         * In this context we hold a reference to the cache entry and

         * the assocciated lock

         */

        XSEGLOG2(&lc, I, "Started\n");

        struct req_completion *rc = (struct req_completion *)arg;

        struct peer_req *pr = rc->pr;

        struct xseg_request *req = rc->req;

        struct peerd *peer = pr->peer;

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct ce *ce = get_cache_entry(cached->cache, cio->h);

        uint32_t start, end, i;

        int success;

        char *data = xseg_get_data(peer->xseg, req);

        /*

         * Synchronize pending_reqs of the cache_io here, since each cache_io

         * refers to only one object, and therefore we can use the object lock

         * to synchronize between receive contextes.

         */

        cio->pending_reqs--;

        success = (req->state == XS_SERVED && req->serviced == req->size);

        if (!success)

                cio->state = CIO_FAILED;

        //assert (req->offset % cached->bucket_size) == 0;

        //assert ((req->offset+req->serviced) % cached->bucket_size) == 0;

        start = __get_bucket(cached, req->offset);

        end = __get_bucket(cached, req->offset + req->serviced);

        for (i = start; i <= end; i++) {

                if (ce->status[i] == LOADING){

                        if (success){

                                XSEGLOG2(&lc, I,

                                                "Bucket %lu loading and reception successful\n", i);

                                memcpy(ce->data+(i*cached->bucket_size), data,

                                                cached->bucket_size);

                                ce->status[i] = VALID;

                        }

                        else {

                                XSEGLOG2(&lc, I,

                                                "Bucket %lu loading but reception unsuccessful\n", i);

                                //reset status

                                XSEGLOG2(&lc, E, "Before ce %p, i %lu, ce->status[i] %u", ce, i, ce->status[i]);

                                ce->status[i] = INVALID;

                                XSEGLOG2(&lc, E, "After ce %p, i %lu, ce->status[i] %u", ce, i, ce->status[i]);

                        }

                        xwaitq_signal(&ce->waitq[i]);

                }

        }

        xseg_put_request(peer->xseg, rc->req, pr->portno);

        free(rc);

        XSEGLOG2(&lc, I, "Finished\n");

}

void complete_write(void *arg)

{

        //for each bucket

        //        if WRITETHROUGH

        //                copy data to bucket

        //                mark as valid

        //        else if WRITEBACK

        //                if status writing

        //                        mark as valid

        //

        /*

         * In this context we hold a reference to the cache entry and

         * the assocciated lock

         */

        return;

}

static int handle_receive_read(struct peerd *peer, struct peer_req *pr,

                        struct xseg_request *req)

{

        XSEGLOG2(&lc, I, "Started\n");

        /*

         * Should be rentrant

         */

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        /*assert there is a handler for received cio*/

        struct ce *ce = get_cache_entry(cached->cache, cio->h);

        struct req_completion *rc;

        rc = malloc(sizeof(struct req_completion));

        if (!rc) {

                perror("malloc");

                return -1;

        }

        rc->pr = pr;

        rc->req = req;

        if (xworkq_enqueue(&ce->workq, complete_read, (void *)rc) < 0){

                free(rc);

                return -1;

                //TODO WHAT?

        }

        XSEGLOG2(&lc, I, "Finished\n");

        return 0;

}

static int handle_receive_write(struct peerd *peer, struct peer_req *pr)

{

        //enqueue_work

        return 0;

}

static int handle_delete(struct peerd *peer, struct peer_req *pr)

{

        //h = cache_lookup

        //if h

        //        cio->h = h

        //

        //send delete to blocker

        return 0;

}

static int handle_receive_delete(struct peerd *peer, struct peer_req *pr)

{

        //if success

        //        if cio->h

        //                //this should not write any dirty data

        //                xcache_remove(h)

        return 0;

}

static int forward_req(struct peerd *peer, struct peer_req *pr)

{

        /*

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        xport p;

        p = xseg_forward(peer->xseg, req, pr->portno, X_ALLOC);

        xseg_signal(peer->xseg, p);

        */

        return 0;

}

static int handle_receive(struct peerd *peer, struct peer_req *pr,

                        struct xseg_request *req)

{

        //if not read/write/delete

        //        put req;

        //        complete or fail pr;

        int r;

        xport p;

        switch (req->op){

                case X_READ: r = handle_receive_read(peer, pr, req); break;

//                case X_WRITE: r = handle_receive_write(peer, pr, req); break;

//                case X_DELETE: r = handle_receive_delete(peer, pr, req); break;

                default:

                        p = xseg_respond(peer->xseg, req, pr->portno, X_ALLOC);

                        if (p == NoPort)

                                xseg_put_request(peer->xseg, req, pr->portno);

                        break;

        }

        return 0;

}

int dispatch(struct peerd *peer, struct peer_req *pr, struct xseg_request *req,

                enum dispatch_reason reason)

{

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        switch (reason) {

                case dispatch_accept:

                        XSEGLOG2(&lc, D, "In dispatch accept");

                        if (req->op == X_READ || req->op == X_WRITE) {

                                /*We cache only read/write requests*/

                                cio->state = CIO_ACCEPTED;

                                handle_readwrite(peer, pr);

                        } else {

                                /*FIXME: Other requests should be forwarded to blocker*/

                                fail(peer, pr);

                        }

                        break;

                case dispatch_receive:

                        XSEGLOG2(&lc, D, "In dispatch receive");

                        handle_receive(peer, pr, req);

                        break;

                case dispatch_internal:

                default:

                        XSEGLOG2(&lc, E, "Invalid dispatch reason");

        }

        return 0;

}

int custom_peer_init(struct peerd *peer, int argc, char *argv[])

{

        int i;

        char bucket_size[MAX_ARG_LEN + 1];

        char object_size[MAX_ARG_LEN + 1];

        char max_req_size[MAX_ARG_LEN + 1];

        char write_policy[MAX_ARG_LEN + 1];

        long bportno = -1;

        long cache_size = -1;

        bucket_size[0] = 0;

        object_size[0] = 0;

        max_req_size[0] = 0;

        write_policy[0] = 0;

        /*Allocate enough space for needed structs*/

        struct cached *cached = malloc(sizeof(struct cached));

        if (!cached) {

                perror("malloc");

                goto fail;

        }

        cached->cache = malloc(sizeof(struct xcache));

        if (!cached->cache) {

                perror("malloc");

                goto cache_fail;

        }

        peer->priv = cached;

        for (i = 0; i < peer->nr_ops; i++) {

                struct cache_io *cio = malloc(sizeof(struct cache_io));

                if (!cio) {

                        perror("malloc");

                        goto cio_fail;

                }

                cio->h = NoEntry;

                cio->pending_reqs = 0;

                peer->peer_reqs[i].priv = cio;

        }

        /*Read arguments*/

        BEGIN_READ_ARGS(argc, argv);

        READ_ARG_ULONG("-bp", bportno);

        READ_ARG_ULONG("-cs", cache_size);

        READ_ARG_STRING("-mrs", max_req_size, MAX_ARG_LEN);

        READ_ARG_STRING("-os", object_size, MAX_ARG_LEN);

        READ_ARG_STRING("-bs", bucket_size, MAX_ARG_LEN);

        READ_ARG_STRING("-wcp", write_policy, MAX_ARG_LEN);

        END_READ_ARGS();

        /*Parse arguments for:*/

        /*Bucket size*/

        if (!bucket_size[0]) {

                cached->bucket_size = BUCKET_SIZE_QUANTUM; /*Default value*/

        } else {

                cached->bucket_size = str2num(bucket_size);

                if (!cached->bucket_size) {

                        XSEGLOG2(&lc, E, "Invalid syntax: -bs %s\n", bucket_size);

                        goto arg_fail;

                }

                if (cached->bucket_size % BUCKET_SIZE_QUANTUM) {

                        XSEGLOG2(&lc, E, "Misaligned bucket size: %s\n", bucket_size);

                        goto arg_fail;

                }

        }

        /*Object size*/

        if (!object_size[0])

                strcpy(object_size, "4M"); /*Default value*/

        cached->object_size = str2num(object_size);

        if (!cached->object_size) {

                XSEGLOG2(&lc, E, "Invalid syntax: -os %s\n", object_size);

                goto arg_fail;

        }

        if (cached->object_size % cached->bucket_size) {

                XSEGLOG2(&lc, E, "Misaligned object size: %s\n", object_size);

                goto arg_fail;

        }

        /*Max request size*/

        if (!max_req_size[0]) {

                XSEGLOG2(&lc, E, "Maximum request size must be provided\n");

                goto arg_fail;

        }

        cached->max_req_size = str2num(max_req_size);

        if (!cached->max_req_size) {

                XSEGLOG2(&lc, E, "Invalid syntax: -mrs %s\n", max_req_size);

                goto arg_fail;

        }

        if (cached->max_req_size % BUCKET_SIZE_QUANTUM) {

                XSEGLOG2(&lc, E, "Misaligned maximum request size: %s\n",

                                max_req_size);

                goto arg_fail;

        }

        /*Cache size*/

        if (cache_size < 0) {

                XSEGLOG2(&lc, E, "Cache size must be provided\n");

                goto arg_fail;

        }

        cached->cache_size = cache_size;

        /*Blocker port*/

        if (bportno < 0){

                XSEGLOG2(&lc, E, "Blocker port must be provided");

                goto arg_fail;

        }

        cached->bportno = bportno;

        /*Write policy*/

        if (!write_policy[0]) {

                XSEGLOG2(&lc, E, "Write policy must be provided");

                goto arg_fail;

        }

        cached->write_policy = read_write_policy(write_policy);

        if (cached->write_policy < 0) {

                XSEGLOG2(&lc, E, "Invalid syntax: -wcp %s\n", write_policy);

                goto arg_fail;

        }

        cached->buckets_per_object = cached->object_size / cached->bucket_size;

        print_cached(cached);

        /*Initialize xcache*/

        xcache_init(cached->cache, cached->cache_size, &c_ops, XCACHE_LRU_HEAP,

                        peer);

        return 0;

arg_fail:

        custom_peer_usage();

cio_fail:

        for (i = 0; i < peer->nr_ops && peer->peer_reqs[i].priv != NULL; i++)

                free(peer->peer_reqs[i].priv);

        free(cached->cache);

cache_fail:

        free(cached);

fail:

        return -1;

}

void custom_peer_finalize(struct peerd *peer)

{

        //write dirty objects

        //or cache_close(cached->cache);

        return;

}

void custom_peer_usage()

{

        fprintf(stderr, "Custom peer options: \n"

                        "  --------------------------------------------\n"

                        "    -cs       | None    | Number of objects to cache\n"

                        "    -mrs      | None    | Max request size\n"

                        "    -os       | 4M      | Object size\n"

                        "    -bs       | 4K      | Bucket size\n"

                        "    -bp       | None    | Blocker port\n"

                        "    -wcp      | None    | Write policy [writethrough|writeback]\n"

                        "\n");

}