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(__bucket_status) \

        (__bucket_status == VALID || __bucket_status == DIRTY || __bucket_status == WRITING)

/* write policies */

#define WRITETHROUGH 1

#define WRITEBACK    2

#define WRITE_POLICY(__wcp)                                        \

        __wcp == WRITETHROUGH        ? "writethrough"        :        \

        __wcp == WRITEBACK        ? "writeback"                 :        \

        "undefined"

/* cio states */

#define CIO_FAILED        1

#define CIO_ACCEPTED        2

#define CIO_READING        3

/* ce states */

#define CE_READY        1

#define CE_WRITING        2

#define CE_FAILED        3

#define CE_INVALIDATED        4

#define CE_NOT_READY        5

#define CE_EVICTED        6

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

        uint32_t bucket_size; /*In bytes*/

        uint32_t buckets_per_object;

        xport bportno;

        int write_policy;

        struct xworkq workq;

        //scheduler

};

struct ce {

        unsigned char *data;

        uint32_t *bucket_status;        /* bucket_status of each bucket */

        struct xwaitq *bucket_waitq;        /* waitq of each bucket */

        uint32_t status;                /* cache entry status */

        struct xlock lock;                /* cache entry lock */

        struct xworkq workq;                /* workq of the cache entry */

        struct xwaitq waitq;                /* waitq of the cache entry */

        struct peer_req pr;

        struct ce *evicted_by;

};

struct req_completion{

        struct peer_req *pr;

        struct xseg_request *req;

};

struct eviction_arg {

        struct peerd *peer;

        struct ce *evicted;

        struct ce *new_entry;

};

/*

 * Helper functions

 */

#define MIN(__a__, __b__) ((__a__ < __b__) ? __a__ : __b__)

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 inline uint64_t __quantize(uint64_t num, uint32_t quantum)

{

        quantum--;

        return num & (uint64_t)(~quantum);

}

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 *bucket_status = (uint32_t *)arg;

        return (*bucket_status != LOADING);

}

static int cache_entry_ready(void *arg)

{

        struct ce *ce = (struct ce *)arg;

        return (ce->status == CE_READY);

}

static void __set_buckets_status(struct ce *ce, uint32_t start, uint32_t end,

                uint32_t status)

{

        uint32_t i;

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

                ce->bucket_status[i] = status;

        }

}

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;

}

static void signal_waitq(void *q, void *arg)

{

        struct xwaitq *waitq = (struct xwaitq *)arg;

        xwaitq_signal(waitq);

}

/*

 * serve_req is called only when all the requested buckets are readable.

 */

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

{

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

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

        struct xseg *xseg = peer->xseg;

        struct xseg_request *req = pr->req;

        char *req_data = xseg_get_data(xseg, req);

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

        req->serviced = req->size;

        //assert req->serviced <= req->datalen

        memcpy(req_data, ce->data + req->offset, req->serviced);

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

        return 0;

}

/*

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

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

 *

 * read/write an object range in buckets.

 *

 * Associate the request with the given pr.

 *

 */

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

                xcache_handler h, uint32_t start, uint32_t end)

{

        struct cached *cached = __get_cached(peer);

        //struct cache_io *cio = __get_cache_io(pr);

        struct xseg_request *req;

        struct xseg *xseg = peer->xseg;

        xport srcport = pr->portno;

        xport dstport = cached->bportno;

        xport p;

        char *req_target, *req_data;

        int r;

        char *target;

        uint32_t targetlen;

        struct ce *ce;

        target = xcache_get_name(cached->cache, h);

        targetlen = strlen(target);

        ce = xcache_get_entry(cached->cache, h);

        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);

        if (write)

                req->op = X_WRITE;

        else

                req->op = X_READ;

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

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

        if (r < 0) {

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

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

                goto put_xseg_request;

        }

        req_target = xseg_get_target(xseg, req);

        strncpy(req_target, target, targetlen);

        if (req->op == X_WRITE) {

                 req_data = xseg_get_data(xseg, req);

                 memcpy(req_data, ce->data+req->offset, req->size);

        } else {

        }

        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;

        struct cache_io *ce_cio = ce->pr.priv;

        ce->status = CE_READY;

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

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

                ce->bucket_status[i] = INVALID;

        }

        ce_cio->h = NoEntry;

        ce_cio->pending_reqs = 0;

        //FIXME what here?

        ce_cio->state = CIO_ACCEPTED;

        ce->evicted_by = NULL;

        xlock_release(&ce->lock);

        return 0;

}

/*

 * Insertion/eviction process with writeback policy:

 *

 * Insertion evicts a cache entry.

 * On eviction gets called.

 *         If cache entry is invalidated then we have no write back to do.

 * Else we issue a write dirty pages work serialized on the object lock.

 *

 * Now all pending operations are serialized on the object work queue.

 * Since there are no read hazards with the dirty buckets write, they pose no

 * problem. But a pending write does, since it will not be written back to the

 * permament storage media when cache entry is put. So, for any pending write

 * operations on the evicted cache entry, we must treat them with write through

 * policy.

 *

 * When all pending operations on the evicted cache entry are done, on_put gets

 * called and the waiting cache entry can continue.

 */

void eviction_work(void *wq, void *arg)

{

        /*

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

         * the new cache entry.

         * Evicted cache entry lock is also held.

         */

        struct eviction_arg *earg = (struct eviction_arg *)arg;

        struct peerd *peer = earg->peer;;

        struct cached *cached = peer->priv;

        struct ce *ce_evicted = earg->evicted;

        struct ce *ce_new = earg->new_entry;

        struct cache_io *ce_cio = ce_evicted->pr.priv;

        uint32_t start, end, i = 0;

        /* recheck status here, in case there was a race */

        if (ce_evicted->status == CE_INVALIDATED){

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

                return;

        }

        /* write all dirty buckets */

        while(i < cached->buckets_per_object){

                if (ce_evicted->bucket_status[i] != DIRTY){

                        i++;

                        continue;

                }

                start = i;

                while (i < cached->buckets_per_object &&

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

                                ce_evicted->bucket_status[i] == DIRTY){

                        i++;

                }

                end = i-1;

                if (rw_range(peer, &ce_evicted->pr, 1, ce_cio->h, start, end) < 0){

                        ce_cio->state = CIO_FAILED;

                }

                __set_buckets_status(ce_evicted, start, end, WRITING);

                ce_cio->pending_reqs++;

        }

        /*

         * If no request was issued, eviction is now completed.

         * Notify the waiter, if failed.

         */

        if (!ce_cio->pending_reqs){

                if (ce_cio->state == CIO_FAILED){

                        ce_new->status = CE_FAILED;

                        ce_evicted->evicted_by = NULL;

                        xworkq_enqueue(&cached->workq, signal_waitq, &ce_new->waitq);

                } else {

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

                }

        }

        free(earg);

}

int on_evict(void *c, void *evicted, void *new_entry)

{

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

        struct cached *cached = peer->priv;

        struct ce *ce_evicted = (struct ce *)evicted;

        struct ce *ce_new = (struct ce *)new_entry;

        struct cache_io *ce_cio = (struct cache_io *)ce_evicted->pr.priv;

        struct eviction_arg *earg;

        /*

         * Since write policy doesn't change and after a cache entry gets

         * invalidated, it cannot never be valid again, it is safe to procceed

         * without the cache entry lock.

         */

        XSEGLOG2(&lc, D, "Evicting cache entry %p", evicted);

        ce_evicted->evicted_by = ce_new;

        if (cached->write_policy != WRITEBACK ||

                        ce_evicted->status == CE_INVALIDATED){

                return 0;

        }

        ce_evicted->status = CE_EVICTED;

        ce_new->status = CE_NOT_READY;

        earg = malloc(sizeof(struct eviction_arg));

        if (!earg){

                goto out_error;

        }

        earg->peer = (struct peerd *)peer;

        earg->evicted = ce_evicted;

        earg->new_entry = ce_new;

        /*

         * In all other cases, we should have the cache entry lock before

         * proceeding.

         */

        if (xworkq_enqueue(&ce_evicted->workq, eviction_work, (void *)earg) < 0){

                goto out_error_free;

        }

        xworkq_signal(&ce_evicted->workq);

        return 1;

out_error_free:

        free(earg);

out_error:

        /*

         * In case of an eviction error, do not put the old cache entry, but

         * mark as failed the new one.

         */

        ce_new->status = CE_FAILED;

        ce_evicted->evicted_by = NULL;

        xworkq_enqueue(&cached->workq, signal_waitq, &ce_new->waitq);

        return -1;

}

void on_put(void *c, void *e)

{

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

        //no lock is needed.

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

        struct cached *cached = peer->priv;

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

        struct cache_io *ce_cio = (struct cache_io *)ce->pr.priv;

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

        if (ce->evicted_by){

                ce->evicted_by->status = CE_READY;

                xworkq_enqueue(&cached->workq, signal_waitq, &ce->evicted_by->waitq);

        }

}

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->bucket_status = malloc(sizeof(uint32_t) * cached->buckets_per_object);

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

        ce->pr.priv = malloc(sizeof(struct cache_io));

        if (!ce->data || !ce->bucket_status || !ce->bucket_waitq || !ce->pr.priv)

                goto ce_fields_fail;

        ce->pr.peer = peer;

        ce->pr.portno = peer->portno_start;

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

                xwaitq_init(&ce->bucket_waitq[i], is_not_loading,

                                &ce->bucket_status[i], XWAIT_SIGNAL_ONE);

        }

        xwaitq_init(&ce->waitq, cache_entry_ready, ce, XWAIT_SIGNAL_ONE);

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

        return ce;

ce_fields_fail:

        free(ce->data);

        free(ce->bucket_status);

        free(ce->bucket_waitq);

        free(ce);

ce_fail:

        perror("malloc");

        return NULL;

}

struct xcache_ops c_ops = {

        .on_init = on_init,

        .on_evict = on_evict,

        .on_put  = on_put,

        .on_node_init = init_node

};

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

                                        uint32_t limit)

{

        uint32_t end = start + 1;

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

                end++;

        return (end - 1);

}

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 *q, void *arg);

static void handle_write(void *q, void *arg);

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

                        struct xseg_request *req);

static void bucket_status_changed(void *q, 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 = xcache_get_entry(cached->cache, cio->h);

        void (*fn)(void *q, void *arg);

        if (pr->req->op == X_WRITE){

                fn = handle_write;

        } else {

                fn = handle_read;

        }

        if (xworkq_enqueue(&ce->workq, fn, (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);

                }

                return;

        }

        xworkq_signal(&ce->workq);

}

/*

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

 *

 * If unreadable buckets exist, it waits on the last unreadable bucket.

 */

static void handle_read(void *q, 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 = xcache_get_entry(cached->cache, cio->h);

        uint32_t start_bucket, end_bucket;

        uint32_t i, b, limit;

        uint32_t pending_buckets = 0;

        XSEGLOG2(&lc, D, "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 - 1);

        //assert limit < cached->object_size

        XSEGLOG2(&lc, D, "Start: %lu, Limit %lu", b, limit );

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

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

                        continue;

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

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

                        start_bucket = i;

                        end_bucket = __get_last_invalid(ce, start_bucket,

                                MIN(limit, cached->max_req_size / cached->bucket_size));

                        i = end_bucket + 1;

                        if (rw_range(peer, pr, 0, cio->h, start_bucket, end_bucket) < 0){

                                cio->state = CIO_FAILED;

                                break;

                        }

                        __set_buckets_status(ce, start_bucket, end_bucket, LOADING);

                        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 = bucket_status_changed;

                cio->work.job = pr;

                /* wait on the last bucket */

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

                return;

        }

out:

        if (cio->state == CIO_FAILED){

                if (!cio->pending_reqs)

                        cached_fail(peer, pr);

        } else {

                if (serve_req(peer, pr)) {

                        XSEGLOG2(&lc, E,"Serve of request failed\n");

                        cached_fail(peer, pr);

                }

                cached_complete(peer, pr);

        }

        return;

}

static void handle_write(void *q, 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 ce *ce = xcache_get_entry(cached->cache, cio->h);

        struct xseg_request *req = pr->req;

        char *req_data;

        uint32_t start_bucket, end_bucket, last_read_bucket = -1;

        uint32_t i;

        uint64_t data_len, data_start;

        uint64_t first_bucket_offset = req->offset % cached->bucket_size;

        //what about FUA?

        //

        //if invalidated, fake complete

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

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

        /*

         * In case of a misaligned write, if the start, end buckets of the write

         * are invalid, we have to read them before continue with the write.

         */

        if (ce->bucket_status[start_bucket] == INVALID && first_bucket_offset){

                if (rw_range(peer, pr, 0, cio->h, start_bucket, start_bucket) < 0){

                        cio->state = CIO_FAILED;

                        goto out;

                }

                __set_buckets_status(ce, start_bucket, start_bucket, LOADING);

                cio->pending_reqs++;

                last_read_bucket = start_bucket;

        }

        if (ce->bucket_status[end_bucket] == INVALID &&

                        (req->offset + req->size) % cached->bucket_size){

                if (rw_range(peer, pr, 0, cio->h, end_bucket, end_bucket) < 0){

                        cio->state = CIO_FAILED;

                        goto out;

                }

                __set_buckets_status(ce, end_bucket, end_bucket, LOADING);

                cio->pending_reqs++;

                last_read_bucket = end_bucket;

        }

        if (last_read_bucket != -1){

                cio->work.job_fn = bucket_status_changed;

                cio->work.job = pr;

                /* wait on the last read bucket */

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

                xwaitq_enqueue(&ce->bucket_waitq[last_read_bucket], &cio->work);

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

                return;

        }

        req_data = xseg_get_data(peer->xseg, pr->req);

        if (cached->write_policy == WRITETHROUGH){

                // || ce->status == CE_EVICTED

                //forward_req but make sure we don't put it on receive

                if (forward_req(peer, pr, pr->req) < 0){

                        cio->state = CIO_FAILED;

                        goto out;

                }

                return;

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

                /*

                 * for each bucket

                 *         write all buckets

                 *         mark them as dirty

                 */

                /* special care for the first bucket */

                data_start = cached->bucket_size * start_bucket + first_bucket_offset;

                data_len = cached->bucket_size - first_bucket_offset;

                memcpy(&ce->data[data_start], req_data, data_len);

                ce->bucket_status[start_bucket] = DIRTY;

                /* and the rest */

                for (i = start_bucket + 1; i <= end_bucket; i++) {

                        data_start = cached->bucket_size * (i-start_bucket) +

                                first_bucket_offset;

                        if (data_start + cached->bucket_size <= req->size)

                                data_len = cached->bucket_size;

                        else

                                data_len = req->size - data_start;

                        memcpy(&ce->data[i * cached->bucket_size],

                                &req_data[data_start], data_len);

                        ce->bucket_status[i] = DIRTY;

                }

        } else {

                cio->state = CIO_FAILED;

                XSEGLOG2(&lc, E, "Invalid cache write policy");

        }

out:

        if (cio->state == CIO_FAILED){

                if (!cio->pending_reqs)

                        cached_fail(peer, pr);

        }

        else{

                cached_complete(peer, pr);

        }

        return;

}

/*

 * Assert cache entry is ready.

 *

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

 * target's cache_entry.

 */

static void handle_readwrite_post(void *q, void *arg)

{

        /*

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

         */

        struct ce *ce;

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

        struct peerd *peer = pr->peer;

        struct cached *cached = __get_cached(peer);

        struct cache_io *ce_cio, *cio = __get_cache_io(pr);

        struct xseg_request *req = pr->req;

        int r = 0;

        ce = xcache_get_entry(cached->cache, cio->h); 

        if (ce->status != CE_READY){

                XSEGLOG2(&lc, E, "Cache entry %p has status %u", ce, ce->status);

                r = -1;

                //FIXME defer request ?

                goto out;

        }

        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);

        }

out:

        if (r < 0){

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

                cached_fail(peer, pr);

        } else {

                xworkq_signal(&ce->workq);

        }

}

/*

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

 *

 * Then, we wait until the returned cache entry is ready.

 */

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

{

        struct ce *ce;

        struct cached *cached = __get_cached(peer);

        struct cache_io *ce_cio, *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;

        //TODO: assert req->size != 0 --> complete req

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

        //NOTE: What about writes with req->size = 0?

        XSEGLOG2(&lc, D, "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){

                XSEGLOG2(&lc, D, "Cache miss\n");

                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;

                }

        } else {

                XSEGLOG2(&lc, D, "Cache hit\n");

        }

        ce = (struct ce *)xcache_get_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;

        ce_cio = ce->pr.priv;

        ce_cio->h = h;

        /* wait for the cache_entry to be ready */

        cio->work.job_fn = handle_readwrite_post;

        cio->work.job = pr;

        r = xwaitq_enqueue(&ce->waitq, &cio->work);

out:

        if (r < 0){

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

                cached_fail(peer, pr);

        }

        return r;

}

/*

 * complete_read is called when we receive a reply from a request issued by

 * rw_range. The process mentioned below applies only to buckets previously

 * marked as LOADING:

 *

 * If all requested buckets are serviced, we mark these buckets as VALID.

 * If not, we mark serviced buckets as VALID, non-serviced buckets as INVALID

 * and the cio is failed

 *

 * At any point when a bucket bucket_status changes, we signal the respective waitq.

 */

static void complete_read(void *q, void *arg)

{

        /*

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

         * the assocciated lock

         */

        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 = xcache_get_entry(cached->cache, cio->h);

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

        uint32_t start, end_serviced, end_size, i;

        int success;

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

        XSEGLOG2(&lc, D, "Target: %s.\n Serviced vs total: %lu/%lu.\n",

                        xseg_get_target(peer->xseg, req), req->serviced, req->size);

        /*

         * 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--;

        /* TODO: Assert req->size != 0 ? */

        if (!req->size){

                XSEGLOG2(&lc, E, "BUG: zero sized read");

                cio->state = CIO_FAILED;

                goto out;

        }

        /* Check if request has been correctly served */

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

        if (!success)

                cio->state = CIO_FAILED;

        if (success && req->serviced % cached->bucket_size) {

                XSEGLOG2(&lc, E, "Misaligned successful read. Failing.");

                success = 0;

                cio->state = CIO_FAILED;

        }

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

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

        if (success)

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

        else

                end_serviced = start;

        XSEGLOG2(&lc, D,"Stats: \n"

                        "start        = %lu\n"

                        "end_serviced = %lu\n"

                        "end_size     = %lu\n",

                        start, end_serviced, end_size);

        /* Check serviced buckets */

        for (i = start; i <= end_serviced && success; i++) {

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

                        continue;

                XSEGLOG2(&lc, D, "Bucket %lu loading and reception successful\n",i);

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

                ce->bucket_status[i] = VALID;

                xworkq_enqueue(&cached->workq, signal_waitq, &ce->bucket_waitq[i]);

        }

        /* Check non-serviced buckets */

        for (; i <= end_size; i++) {

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

                        continue;

                XSEGLOG2(&lc, D, "Bucket %lu loading but reception unsuccessful\n", i);

                ce->bucket_status[i] = INVALID;

                xworkq_enqueue(&cached->workq, signal_waitq, &ce->bucket_waitq[i]);

        }

out:

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

        free(rc);

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

}

static void complete_write_through(struct peerd *peer, struct peer_req *pr,

                                        struct xseg_request *req)

{

        /*

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

         * the assocciated lock

         */

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct ce *ce;

        uint32_t start, end_serviced, end_requested, i;

        uint64_t data_start, data_len, first_bucket_offset;

        char *req_data;

        int success;

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

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

        ce = xcache_get_entry(cached->cache, cio->h);

        if (!success){

                XSEGLOG2(&lc, E, "Write failed");

                cio->state = CIO_FAILED;

                goto out;

        }

        /*

         * If the cache entry was evicted or invalidated, then there is no point

         * updating the cache entry buckets.

         * If there is a request on a loading bucket, it should complete by the

         * completion of the read request. Otherwise there is a serious bug.

         */

        if (ce->status == CE_EVICTED || ce->status == CE_INVALIDATED)

                goto out;

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

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

//        end_requested = __get_bucket(cached, req->offset + req->size - 1);

        first_bucket_offset = req->offset % cached->bucket_size;

        req_data = xseg_get_data(peer->xseg, req);

        /*

         * for each serviced bucket

         *        copy data to bucket

         *        mark as valid

         *        signal_waitq

         */

        data_start = start * cached->bucket_size + first_bucket_offset;

        data_len = cached->bucket_size - first_bucket_offset;

        memcpy(&ce->data[data_start], req_data, data_len);

        ce->bucket_status[start] = VALID;

        xworkq_enqueue(&cached->workq, signal_waitq, &ce->bucket_waitq[start]);

        for (i = start+1; i <= end_serviced; i++) {

                data_start = cached->bucket_size * (i - start) +

                        first_bucket_offset;

                if (data_start + cached->bucket_size <= req->size)

                        data_len = cached->bucket_size;

                else

                        data_len = req->size - data_start;

                memcpy(&ce->data[cached->bucket_size * i], &req_data[data_start],

                                data_len);

                ce->bucket_status[i] = VALID;

                xworkq_enqueue(&cached->workq, signal_waitq, &ce->bucket_waitq[i]);

        }

out:

        /*

         * Here we do not put request, because we forwarded the original request.

         */

        if (cio->pending_reqs){

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

                return;

        }

        /*

         * This pr, was triggered in response to an accepted request and

         * therefore must be completed or failed.

         */

        if (cio->state == CIO_FAILED)

                cached_fail(peer, pr);

        else

                cached_complete(peer, pr);

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

        return;

}

static void complete_write_back(struct peerd *peer, struct peer_req *pr,

                struct xseg_request *req)

{

        /*

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

         * the assocciated lock

         */

        struct cached *cached = __get_cached(peer);

        struct cache_io *cio = __get_cache_io(pr);

        struct ce *ce;

        uint32_t start, end_serviced, end_requested, i;

        uint64_t first_bucket_offset;

        char *req_data;

        int success;

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

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

        ce = xcache_get_entry(cached->cache, cio->h);

        if (!success){

                XSEGLOG2(&lc, E, "Write failed");

                cio->state = CIO_FAILED;

                goto out;

        }

        /*

         * If the cache entry was evicted or invalidated, then there is no point

         * updating the cache entry buckets.

         * If there is a request on a loading bucket, it should complete by the

         * completion of the read request. Otherwise there is a serious bug.

         */

        if (ce->status == CE_EVICTED || ce->status == CE_INVALIDATED)

                goto out;

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

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

//        end_requested = __get_bucket(cached, req->offset + req->size - 1);

        first_bucket_offset = req->offset % cached->bucket_size;

        req_data = xseg_get_data(peer->xseg, req);

        /*

         * for each serviced bucket

         *        if bucket_status writing

         *                mark as valid

         *                signal bucket

         */

        /* should we signal with the cache entry lock held ? */

        if (ce->bucket_status[start] == WRITING){

                ce->bucket_status[start] = VALID;

                xworkq_enqueue(&cached->workq, signal_waitq, &ce->bucket_waitq[start]);

        }

        ce->bucket_status[start] = VALID;

        for (i = start+1; i <= end_serviced; i++) {

                if (ce->bucket_status[i] == WRITING){

                        ce->bucket_status[i] = VALID;

                        xworkq_enqueue(&cached->workq, signal_waitq, &ce->bucket_waitq[i]);

                }

        }

out:

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

        if (cio->pending_reqs){

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

                return;

        }

        /*

         * This write was a result of a cache eviction or (in future), an

         * explicit cache flush, and therefore, we have nothing to respond. We

         * do however have to put the cache entry and/or notify waiter of the

         * evicted entry.

         */

        if (ce->status == CE_EVICTED){

                if (cio->state == CIO_FAILED){

                        ce->evicted_by->status = CE_FAILED;

                        xworkq_enqueue(&cached->workq, signal_waitq, &ce->evicted_by->waitq);

                        ce->evicted_by = NULL;

                }

        } else {

                /* a flush request */

                //TODO notify the rest of the flushing process

        }

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

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

        return;

}

void complete_write(void *q, void *arg)

{

        /*

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

         * the assocciated lock

         */

        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);

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

        free(rc);

        /*

         * 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--;

        if (cached->write_policy == WRITETHROUGH){

                complete_write_through(peer, pr, req);

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

                complete_write_back(peer, pr, req);

        }

        return;

}

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

                        struct xseg_request *req)

{

        XSEGLOG2(&lc, D, "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 = xcache_get_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?

        }

        xworkq_signal(&ce->workq);

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

        return 0;

}

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