Archipelago

#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/xhash.h>

#include <xseg/protocol.h>

#include <sys/stat.h>

#include <fcntl.h>

#include <gcrypt.h>

#include <errno.h>

#include <sched.h>

#include <sys/syscall.h>

GCRY_THREAD_OPTION_PTHREAD_IMPL;

#define MF_LOAD         (1 << 0)

#define MF_EXCLUSIVE         (1 << 1)

#define MF_FORCE         (1 << 2)

#define SHA256_DIGEST_SIZE 32

/* hex representation of sha256 value takes up double the sha256 size */

#define HEXLIFIED_SHA256_DIGEST_SIZE (SHA256_DIGEST_SIZE << 1)

#define block_size (1<<22) //FIXME this should be defined here?

#define objectsize_in_map (1 + XSEG_MAX_TARGETLEN) /* transparency byte + max object len */

#define mapheader_size (SHA256_DIGEST_SIZE + (sizeof(uint64_t)) ) /* magic hash value  + volume size */

#define MF_OBJECT_EXIST                (1 << 0)

#define MF_OBJECT_COPYING        (1 << 1)

#define MF_OBJECT_WRITING        (1 << 2)

#define MF_OBJECT_DELETING        (1 << 3)

#define MF_OBJECT_DELETED        (1 << 4)

#define MF_OBJECT_DESTROYED        (1 << 5)

#define MF_OBJECT_NOT_READY        (MF_OBJECT_COPYING|MF_OBJECT_WRITING|MF_OBJECT_DELETING)

char *magic_string = "This a magic string. Please hash me";

unsigned char magic_sha256[SHA256_DIGEST_SIZE];        /* sha256 hash value of magic string */

char zero_block[HEXLIFIED_SHA256_DIGEST_SIZE + 1]; /* hexlified sha256 hash value of a block full of zeros */

//dispatch_internal mapper states

enum mapper_state {

        ACCEPTED = 0,

        WRITING = 1,

        COPYING = 2,

        DELETING = 3,

        DROPPING_CACHE = 4

};

typedef void (*cb_t)(struct peer_req *pr, struct xseg_request *req);

struct map_node {

        uint32_t flags;

        uint32_t objectidx;

        uint32_t objectlen;

        char object[XSEG_MAX_TARGETLEN + 1];         /* NULL terminated string */

        struct map *map;

        uint32_t ref;

        uint32_t waiters;

        st_cond_t cond;

};

#define MF_MAP_LOADING                (1 << 0)

#define MF_MAP_DESTROYED        (1 << 1)

#define MF_MAP_WRITING                (1 << 2)

#define MF_MAP_DELETING                (1 << 3)

#define MF_MAP_DROPPING_CACHE        (1 << 4)

#define MF_MAP_EXCLUSIVE        (1 << 5)

#define MF_MAP_OPENING                (1 << 6)

#define MF_MAP_CLOSING                (1 << 7)

#define MF_MAP_NOT_READY        (MF_MAP_LOADING|MF_MAP_WRITING|MF_MAP_DELETING|\

                                        MF_MAP_DROPPING_CACHE|MF_MAP_OPENING)

#define wait_on_pr(__pr, __condition__)         \

        while (__condition__){                        \

                ta--;                                \

                __get_mapper_io(pr)->active = 0;\

                XSEGLOG2(&lc, D, "Waiting on pr %lx, ta: %u",  pr, ta); \

                st_cond_wait(__pr->cond);        \

        }

#define wait_on_mapnode(__mn, __condition__)        \

        while (__condition__){                        \

                ta--;                                \

                __mn->waiters++;                \

                XSEGLOG2(&lc, D, "Waiting on map node %lx %s, waiters: %u, ta: %u",  __mn, __mn->object, __mn->waiters, ta); \

                st_cond_wait(__mn->cond);        \

        }

#define wait_on_map(__map, __condition__)        \

        while (__condition__){                        \

                ta--;                                \

                __map->waiters++;                \

                XSEGLOG2(&lc, D, "Waiting on map %lx %s, waiters: %u, ta: %u",  __map, __map->volume, __map->waiters, ta); \

                st_cond_wait(__map->cond);        \

        }

#define signal_pr(__pr)                                \

        do {                                         \

                if (!__get_mapper_io(pr)->active){\

                        ta++;                        \

                        XSEGLOG2(&lc, D, "Signaling  pr %lx, ta: %u",  pr, ta); \

                        __get_mapper_io(pr)->active = 1;\

                        st_cond_signal(__pr->cond);        \

                }                                \

        }while(0)

#define signal_map(__map)                        \

        do {                                         \

                if (__map->waiters) {                \

                        ta += 1;                \

                        XSEGLOG2(&lc, D, "Signaling map %lx %s, waiters: %u, ta: %u",  __map, __map->volume, __map->waiters, ta); \

                        __map->waiters--;        \

                        st_cond_signal(__map->cond);        \

                }                                \

        }while(0)

#define signal_mapnode(__mn)                        \

        do {                                         \

                if (__mn->waiters) {                \

                        ta += __mn->waiters;        \

                        XSEGLOG2(&lc, D, "Signaling map node %lx %s, waiters: %u, ta: %u",  __mn, __mn->object, __mn->waiters, ta); \

                        __mn->waiters = 0;        \

                        st_cond_broadcast(__mn->cond);        \

                }                                \

        }while(0)

struct map {

        uint32_t flags;

        uint64_t size;

        uint32_t volumelen;

        char volume[XSEG_MAX_TARGETLEN + 1]; /* NULL terminated string */

        xhash_t *objects;         /* obj_index --> map_node */

        uint32_t ref;

        uint32_t waiters;

        st_cond_t cond;

};

struct mapperd {

        xport bportno;                /* blocker that accesses data */

        xport mbportno;                /* blocker that accesses maps */

        xhash_t *hashmaps; // hash_function(target) --> struct map

};

struct mapper_io {

        volatile uint32_t copyups;        /* nr of copyups pending, issued by this mapper io */

        xhash_t *copyups_nodes;                /* hash map (xseg_request) --> (corresponding map_node of copied up object)*/

        struct map_node *copyup_node;

        volatile int err;                        /* error flag */

        volatile uint64_t del_pending;

        uint64_t delobj;

        uint64_t dcobj;

        cb_t cb;

        enum mapper_state state;

        volatile int active;

};

/* global vars */

struct mapperd *mapper;

void print_map(struct map *m);

/*

 * Helper functions

 */

static inline struct mapperd * __get_mapperd(struct peerd *peer)

{

        return (struct mapperd *) peer->priv;

}

static inline struct mapper_io * __get_mapper_io(struct peer_req *pr)

{

        return (struct mapper_io *) pr->priv;

}

static inline uint64_t calc_map_obj(struct map *map)

{

        if (map->size == -1)

                return 0;

        uint64_t nr_objs = map->size / block_size;

        if (map->size % block_size)

                nr_objs++;

        return nr_objs;

}

static uint32_t calc_nr_obj(struct xseg_request *req)

{

        unsigned int r = 1;

        uint64_t rem_size = req->size;

        uint64_t obj_offset = req->offset & (block_size -1); //modulo

        uint64_t obj_size =  (rem_size + obj_offset > block_size) ? block_size - obj_offset : rem_size;

        rem_size -= obj_size;

        while (rem_size > 0) {

                obj_size = (rem_size > block_size) ? block_size : rem_size;

                rem_size -= obj_size;

                r++;

        }

        return r;

}

/*

 * Maps handling functions

 */

//FIXME assert targetlen > 0

static struct map * find_map(struct mapperd *mapper, char *target, uint32_t targetlen)

{

        int r;

        struct map *m = NULL;

        char buf[XSEG_MAX_TARGETLEN+1];

        //assert targetlen <= XSEG_MAX_TARGETLEN

        strncpy(buf, target, targetlen);

        buf[targetlen] = 0;

        XSEGLOG2(&lc, D, "looking up map %s, len %u", buf, targetlen);

        r = xhash_lookup(mapper->hashmaps, (xhashidx) buf, (xhashidx *) &m);

        if (r < 0)

                return NULL;

        return m;

}

static int insert_map(struct mapperd *mapper, struct map *map)

{

        int r = -1;

        if (find_map(mapper, map->volume, map->volumelen)){

                XSEGLOG2(&lc, W, "Map %s found in hash maps", map->volume);

                goto out;

        }

        XSEGLOG2(&lc, D, "Inserting map %s, len: %d (map: %lx)", 

                        map->volume, strlen(map->volume), (unsigned long) map);

        r = xhash_insert(mapper->hashmaps, (xhashidx) map->volume, (xhashidx) map);

        while (r == -XHASH_ERESIZE) {

                xhashidx shift = xhash_grow_size_shift(mapper->hashmaps);

                xhash_t *new_hashmap = xhash_resize(mapper->hashmaps, shift, NULL);

                if (!new_hashmap){

                        XSEGLOG2(&lc, E, "Cannot grow mapper->hashmaps to sizeshift %llu",

                                        (unsigned long long) shift);

                        goto out;

                }

                mapper->hashmaps = new_hashmap;

                r = xhash_insert(mapper->hashmaps, (xhashidx) map->volume, (xhashidx) map);

        }

out:

        return r;

}

static int remove_map(struct mapperd *mapper, struct map *map)

{

        int r = -1;

        //assert no pending pr on map

        r = xhash_delete(mapper->hashmaps, (xhashidx) map->volume);

        while (r == -XHASH_ERESIZE) {

                xhashidx shift = xhash_shrink_size_shift(mapper->hashmaps);

                xhash_t *new_hashmap = xhash_resize(mapper->hashmaps, shift, NULL);

                if (!new_hashmap){

                        XSEGLOG2(&lc, E, "Cannot shrink mapper->hashmaps to sizeshift %llu",

                                        (unsigned long long) shift);

                        goto out;

                }

                mapper->hashmaps = new_hashmap;

                r = xhash_delete(mapper->hashmaps, (xhashidx) map->volume);

        }

out:

        return r;

}

static struct xseg_request * __close_map(struct peer_req *pr, struct map *map)

{

        int r;

        xport p;

        struct peerd *peer = pr->peer;

        struct xseg_request *req;

        struct mapperd *mapper = __get_mapperd(peer);

        void *dummy;

        XSEGLOG2(&lc, I, "Closing map %s", map->volume);

        req = xseg_get_request(peer->xseg, pr->portno, mapper->mbportno, X_ALLOC);

        if (!req){

                XSEGLOG2(&lc, E, "Cannot allocate request for map %s",

                                map->volume);

                goto out_err;

        }

        r = xseg_prep_request(peer->xseg, req, map->volumelen, block_size);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot prepare request for map %s",

                                map->volume);

                goto out_put;

        }

        char *reqtarget = xseg_get_target(peer->xseg, req);

        if (!reqtarget)

                goto out_put;

        strncpy(reqtarget, map->volume, req->targetlen);

        req->op = X_CLOSE;

        req->size = block_size;

        req->offset = 0;

        r = xseg_set_req_data(peer->xseg, req, pr);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot set request data for map %s",

                                map->volume);

                goto out_put;

        }

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

        if (p == NoPort){ 

                XSEGLOG2(&lc, E, "Cannot submit request for map %s",

                                map->volume);

                goto out_unset;

        }

        r = xseg_signal(peer->xseg, p);

        XSEGLOG2(&lc, I, "Map %s closing", map->volume);

        return req;

out_unset:

        xseg_get_req_data(peer->xseg, req, &dummy);

out_put:

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

out_err:

        return NULL;

}

static int close_map(struct peer_req *pr, struct map *map)

{

        int err;

        struct xseg_request *req;

        struct peerd *peer = pr->peer;

        map->flags |= MF_MAP_CLOSING;

        req = __close_map(pr, map);

        if (!req)

                return -1;

        wait_on_pr(pr, (!((req->state & XS_FAILED)||(req->state & XS_SERVED))));

        map->flags &= ~MF_MAP_CLOSING;

        err = req->state & XS_FAILED;

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

        if (err)

                return -1;

        return 0;

}

/*

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

                                char *target, uint32_t targetlen, struct map **m)

{

        struct mapperd *mapper = __get_mapperd(peer);

        int r;

        *m = find_map(mapper, target, targetlen);

        if (*m) {

                XSEGLOG2(&lc, D, "Found map %s (%u)", (*m)->volume, (unsigned long) *m);

                if ((*m)->flags & MF_MAP_NOT_READY) {

                        __xq_append_tail(&(*m)->pending, (xqindex) pr);

                        XSEGLOG2(&lc, I, "Map %s found and not ready", (*m)->volume);

                        return MF_PENDING;

                //} else if ((*m)->flags & MF_MAP_DESTROYED){

                //        return -1;

                // 

                }else {

                        XSEGLOG2(&lc, I, "Map %s found", (*m)->volume);

                        return 0;

                }

        }

        r = open_map(peer, pr, target, targetlen, 0);

        if (r < 0)

                return -1; //error

        return MF_PENDING;        

}

*/

/*

 * Object handling functions

 */

struct map_node *find_object(struct map *map, uint64_t obj_index)

{

        struct map_node *mn;

        int r = xhash_lookup(map->objects, obj_index, (xhashidx *) &mn);

        if (r < 0)

                return NULL;

        return mn;

}

static int insert_object(struct map *map, struct map_node *mn)

{

        //FIXME no find object first

        int r = xhash_insert(map->objects, mn->objectidx, (xhashidx) mn);

        if (r == -XHASH_ERESIZE) {

                unsigned long shift = xhash_grow_size_shift(map->objects);

                map->objects = xhash_resize(map->objects, shift, NULL);

                if (!map->objects)

                        return -1;

                r = xhash_insert(map->objects, mn->objectidx, (xhashidx) mn);

        }

        return r;

}

/*

 * map read/write functions 

 */

static inline void pithosmap_to_object(struct map_node *mn, unsigned char *buf)

{

        int i;

        //hexlify sha256 value

        for (i = 0; i < SHA256_DIGEST_SIZE; i++) {

                sprintf(mn->object+2*i, "%02x", buf[i]);

        }

        mn->object[SHA256_DIGEST_SIZE * 2] = 0;

        mn->objectlen = SHA256_DIGEST_SIZE * 2;

        mn->flags = MF_OBJECT_EXIST;

}

static inline void map_to_object(struct map_node *mn, char *buf)

{

        char c = buf[0];

        mn->flags = 0;

        if (c)

                mn->flags |= MF_OBJECT_EXIST;

        memcpy(mn->object, buf+1, XSEG_MAX_TARGETLEN);

        mn->object[XSEG_MAX_TARGETLEN] = 0;

        mn->objectlen = strlen(mn->object);

}

static inline void object_to_map(char* buf, struct map_node *mn)

{

        buf[0] = (mn->flags & MF_OBJECT_EXIST)? 1 : 0;

        memcpy(buf+1, mn->object, mn->objectlen);

        memset(buf+1+mn->objectlen, 0, XSEG_MAX_TARGETLEN - mn->objectlen); //zero out the rest of the buffer

}

static inline void mapheader_to_map(struct map *m, char *buf)

{

        uint64_t pos = 0;

        memcpy(buf + pos, magic_sha256, SHA256_DIGEST_SIZE);

        pos += SHA256_DIGEST_SIZE;

        memcpy(buf + pos, &m->size, sizeof(m->size));

        pos += sizeof(m->size);

}

static struct xseg_request * object_write(struct peerd *peer, struct peer_req *pr, 

                                struct map *map, struct map_node *mn)

{

        void *dummy;

        struct mapperd *mapper = __get_mapperd(peer);

        struct xseg_request *req = xseg_get_request(peer->xseg, pr->portno,

                                                        mapper->mbportno, X_ALLOC);

        if (!req){

                XSEGLOG2(&lc, E, "Cannot allocate request for object %s. \n\t"

                                "(Map: %s [%llu]",

                                mn->object, map->volume, (unsigned long long) mn->objectidx);

                goto out_err;

        }

        int r = xseg_prep_request(peer->xseg, req, map->volumelen, objectsize_in_map);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot allocate request for object %s. \n\t"

                                "(Map: %s [%llu]",

                                mn->object, map->volume, (unsigned long long) mn->objectidx);

                goto out_put;

        }

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

        strncpy(target, map->volume, req->targetlen);

        req->size = objectsize_in_map;

        req->offset = mapheader_size + mn->objectidx * objectsize_in_map;

        req->op = X_WRITE;

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

        object_to_map(data, mn);

        r = xseg_set_req_data(peer->xseg, req, pr);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot set request data for object %s. \n\t"

                                "(Map: %s [%llu]",

                                mn->object, map->volume, (unsigned long long) mn->objectidx);

                goto out_put;

        }

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

        if (p == NoPort){

                XSEGLOG2(&lc, E, "Cannot submit request for object %s. \n\t"

                                "(Map: %s [%llu]",

                                mn->object, map->volume, (unsigned long long) mn->objectidx);

                goto out_unset;

        }

        r = xseg_signal(peer->xseg, p);

        if (r < 0)

                XSEGLOG2(&lc, W, "Cannot signal port %u", p);

        XSEGLOG2(&lc, I, "Writing object %s \n\t"

                        "Map: %s [%llu]",

                        mn->object, map->volume, (unsigned long long) mn->objectidx);

        return req;

out_unset:

        xseg_get_req_data(peer->xseg, req, &dummy);

out_put:

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

out_err:

        XSEGLOG2(&lc, E, "Object write for object %s failed. \n\t"

                        "(Map: %s [%llu]",

                        mn->object, map->volume, (unsigned long long) mn->objectidx);

        return NULL;

}

static struct xseg_request * __write_map(struct peer_req* pr, struct map *map)

{

        void *dummy;

        struct peerd *peer = pr->peer;

        struct mapperd *mapper = __get_mapperd(peer);

        struct map_node *mn;

        uint64_t i, pos, max_objidx = calc_map_obj(map);

        struct xseg_request *req = xseg_get_request(peer->xseg, pr->portno, 

                                                        mapper->mbportno, X_ALLOC);

        if (!req){

                XSEGLOG2(&lc, E, "Cannot allocate request for map %s", map->volume);

                goto out_err;

        }

        int r = xseg_prep_request(peer->xseg, req, map->volumelen, 

                                        mapheader_size + max_objidx * objectsize_in_map);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot prepare request for map %s", map->volume);

                goto out_put;

        }

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

        strncpy(target, map->volume, req->targetlen);

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

        mapheader_to_map(map, data);

        pos = mapheader_size;

        req->op = X_WRITE;

        req->size = req->datalen;

        req->offset = 0;

        if (map->size % block_size)

                max_objidx++;

        for (i = 0; i < max_objidx; i++) {

                mn = find_object(map, i);

                if (!mn){

                        XSEGLOG2(&lc, E, "Cannot find object %lli for map %s",

                                        (unsigned long long) i, map->volume);

                        goto out_put;

                }

                object_to_map(data+pos, mn);

                pos += objectsize_in_map;

        }

        r = xseg_set_req_data(peer->xseg, req, pr);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot set request data for map %s",

                                map->volume);

                goto out_put;

        }

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

        if (p == NoPort){

                XSEGLOG2(&lc, E, "Cannot submit request for map %s",

                                map->volume);

                goto out_unset;

        }

        r = xseg_signal(peer->xseg, p);

        if (r < 0)

                XSEGLOG2(&lc, W, "Cannot signal port %u", p);

        map->flags |= MF_MAP_WRITING;

        XSEGLOG2(&lc, I, "Writing map %s", map->volume);

        return req;

out_unset:

        xseg_get_req_data(peer->xseg, req, &dummy);

out_put:

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

out_err:

        XSEGLOG2(&lc, E, "Map write for map %s failed.", map->volume);

        return NULL;

}

static int write_map(struct peer_req* pr, struct map *map)

{

        int r = 0;

        struct peerd *peer = pr->peer;

        map->flags |= MF_MAP_WRITING;

        struct xseg_request *req = __write_map(pr, map);

        if (!req)

                return -1;

        wait_on_pr(pr, (!(req->state & XS_FAILED || req->state & XS_SERVED)));

        if (req->state & XS_FAILED)

                r = -1;

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

        map->flags &= ~MF_MAP_WRITING;

        return r;

}

static struct xseg_request * __load_map(struct peer_req *pr, struct map *m)

{

        int r;

        xport p;

        struct xseg_request *req;

        struct peerd *peer = pr->peer;

        struct mapperd *mapper = __get_mapperd(peer);

        void *dummy;

        XSEGLOG2(&lc, I, "Loading ng map %s", m->volume);

        req = xseg_get_request(peer->xseg, pr->portno, mapper->mbportno, X_ALLOC);

        if (!req){

                XSEGLOG2(&lc, E, "Cannot allocate request for map %s",

                                m->volume);

                goto out_fail;

        }

        r = xseg_prep_request(peer->xseg, req, m->volumelen, block_size);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot prepare request for map %s",

                                m->volume);

                goto out_put;

        }

        char *reqtarget = xseg_get_target(peer->xseg, req);

        if (!reqtarget)

                goto out_put;

        strncpy(reqtarget, m->volume, req->targetlen);

        req->op = X_READ;

        req->size = block_size;

        req->offset = 0;

        r = xseg_set_req_data(peer->xseg, req, pr);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot set request data for map %s",

                                m->volume);

                goto out_put;

        }

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

        if (p == NoPort){ 

                XSEGLOG2(&lc, E, "Cannot submit request for map %s",

                                m->volume);

                goto out_unset;

        }

        r = xseg_signal(peer->xseg, p);

        XSEGLOG2(&lc, I, "Map %s loading", m->volume);

        return req;

out_unset:

        xseg_get_req_data(peer->xseg, req, &dummy);

out_put:

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

out_fail:

        return NULL;

}

static int read_map (struct map *map, char *buf)

{

        char nulls[SHA256_DIGEST_SIZE];

        memset(nulls, 0, SHA256_DIGEST_SIZE);

        int r = !memcmp(buf, nulls, SHA256_DIGEST_SIZE);

        if (r) {

                XSEGLOG2(&lc, D, "Read zeros");

                //read error;

                return -1;

        }

        //type 1, our type, type 0 pithos map

        int type = !memcmp(buf, magic_sha256, SHA256_DIGEST_SIZE);

        XSEGLOG2(&lc, I, "Type %d detected for map %s", type, map->volume);

        uint64_t pos;

        uint64_t i, nr_objs;

        struct map_node *map_node;

        if (type) {

                pos = SHA256_DIGEST_SIZE;

                map->size = *(uint64_t *) (buf + pos);

                pos += sizeof(uint64_t);

                nr_objs = map->size / block_size;

                if (map->size % block_size)

                        nr_objs++;

                map_node = calloc(nr_objs, sizeof(struct map_node));

                if (!map_node)

                        return -1;

                for (i = 0; i < nr_objs; i++) {

                        map_node[i].map = map;

                        map_node[i].objectidx = i;

                        map_node[i].waiters = 0;

                        map_node[i].ref = 1;

                        map_node[i].cond = st_cond_new(); //FIXME err check;

                        map_to_object(&map_node[i], buf + pos);

                        pos += objectsize_in_map;

                        r = insert_object(map, &map_node[i]); //FIXME error check

                }

        } else {

                pos = 0;

                uint64_t max_nr_objs = block_size/SHA256_DIGEST_SIZE;

                map_node = calloc(max_nr_objs, sizeof(struct map_node));

                if (!map_node)

                        return -1;

                for (i = 0; i < max_nr_objs; i++) {

                        if (!memcmp(buf+pos, nulls, SHA256_DIGEST_SIZE))

                                break;

                        map_node[i].objectidx = i;

                        map_node[i].map = map;

                        map_node[i].waiters = 0;

                        map_node[i].ref = 1;

                        map_node[i].cond = st_cond_new(); //FIXME err check;

                        pithosmap_to_object(&map_node[i], buf + pos);

                        pos += SHA256_DIGEST_SIZE; 

                        r = insert_object(map, &map_node[i]); //FIXME error check

                }

                map->size = i * block_size; 

        }

        print_map(map);

        XSEGLOG2(&lc, I, "Map read for map %s completed", map->volume);

        return 0;

        //FIXME cleanup on error

}

static int load_map(struct peer_req *pr, struct map *map)

{

        int r = 0;

        struct xseg_request *req;

        struct peerd *peer = pr->peer;

        map->flags |= MF_MAP_LOADING;

        req = __load_map(pr, map);

        if (!req)

                return -1;

        wait_on_pr(pr, (!(req->state & XS_FAILED || req->state & XS_SERVED)));

        map->flags &= ~MF_MAP_LOADING;

        if (req->state & XS_FAILED){

                XSEGLOG2(&lc, E, "Map load failed for map %s", map->volume);

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

                return -1;

        }

        r = read_map(map, xseg_get_data(peer->xseg, req));

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

        return r;

}

static struct xseg_request * __open_map(struct peer_req *pr, struct map *m)

{

        int r;

        xport p;

        struct xseg_request *req;

        struct peerd *peer = pr->peer;

        struct mapperd *mapper = __get_mapperd(peer);

        void *dummy;

        XSEGLOG2(&lc, I, "Opening map %s", m->volume);

        req = xseg_get_request(peer->xseg, pr->portno, mapper->mbportno, X_ALLOC);

        if (!req){

                XSEGLOG2(&lc, E, "Cannot allocate request for map %s",

                                m->volume);

                goto out_fail;

        }

        r = xseg_prep_request(peer->xseg, req, m->volumelen, block_size);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot prepare request for map %s",

                                m->volume);

                goto out_put;

        }

        char *reqtarget = xseg_get_target(peer->xseg, req);

        if (!reqtarget)

                goto out_put;

        strncpy(reqtarget, m->volume, req->targetlen);

        req->op = X_OPEN;

        req->size = block_size;

        req->offset = 0;

        r = xseg_set_req_data(peer->xseg, req, pr);

        if (r < 0){

                XSEGLOG2(&lc, E, "Cannot set request data for map %s",

                                m->volume);

                goto out_put;

        }

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

        if (p == NoPort){ 

                XSEGLOG2(&lc, E, "Cannot submit request for map %s",

                                m->volume);

                goto out_unset;

        }

        r = xseg_signal(peer->xseg, p);

        XSEGLOG2(&lc, I, "Map %s opening", m->volume);

        return req;

out_unset:

        xseg_get_req_data(peer->xseg, req, &dummy);

out_put:

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

out_fail:

        return NULL;

}

static int open_map(struct peer_req *pr, struct map *map)

{

        int err;

        struct xseg_request *req;

        struct peerd *peer = pr->peer;

        map->flags |= MF_MAP_OPENING;

        req = __open_map(pr, map);

        if (!req)

                return -1;

        wait_on_pr(pr, (!((req->state & XS_FAILED)||(req->state & XS_SERVED))));

        map->flags &= ~MF_MAP_OPENING;

        err = req->state & XS_FAILED;

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

        if (err)

                return -1;

        else 

                map->flags |= MF_MAP_EXCLUSIVE;

        return 0;

}

/*

 * copy up functions

 */

static int __set_copyup_node(struct mapper_io *mio, struct xseg_request *req, struct map_node *mn)

{

        int r = 0;

        if (mn){

                r = xhash_insert(mio->copyups_nodes, (xhashidx) req, (xhashidx) mn);

                if (r == -XHASH_ERESIZE) {

                        xhashidx shift = xhash_grow_size_shift(mio->copyups_nodes);

                        xhash_t *new_hashmap = xhash_resize(mio->copyups_nodes, shift, NULL);

                        if (!new_hashmap)

                                goto out;

                        mio->copyups_nodes = new_hashmap;

                        r = xhash_insert(mio->copyups_nodes, (xhashidx) req, (xhashidx) mn);

                }

        }

        else {

                r = xhash_delete(mio->copyups_nodes, (xhashidx) req);

                if (r == -XHASH_ERESIZE) {

                        xhashidx shift = xhash_shrink_size_shift(mio->copyups_nodes);

                        xhash_t *new_hashmap = xhash_resize(mio->copyups_nodes, shift, NULL);

                        if (!new_hashmap)

                                goto out;

                        mio->copyups_nodes = new_hashmap;

                        r = xhash_delete(mio->copyups_nodes, (xhashidx) req);

                }

        }

out:

        return r;

}

static struct map_node * __get_copyup_node(struct mapper_io *mio, struct xseg_request *req)

{

        struct map_node *mn;

        int r = xhash_lookup(mio->copyups_nodes, (xhashidx) req, (xhashidx *) &mn);

        if (r < 0)

                return NULL;

        return mn;

}

static struct xseg_request * copyup_object(struct peerd *peer, struct map_node *mn, struct peer_req *pr)

{

        struct mapperd *mapper = __get_mapperd(peer);

        struct mapper_io *mio = __get_mapper_io(pr);

        struct map *map = mn->map;