8 #include <xtypes/xlock.h>
9 #include <xtypes/xhash.h>
10 #include <xseg/protocol.h>
16 #include <sys/syscall.h>
18 GCRY_THREAD_OPTION_PTHREAD_IMPL;
19 #define MF_LOAD (1 << 0)
20 #define MF_EXCLUSIVE (1 << 1)
21 #define MF_FORCE (1 << 2)
23 #define SHA256_DIGEST_SIZE 32
24 /* hex representation of sha256 value takes up double the sha256 size */
25 #define HEXLIFIED_SHA256_DIGEST_SIZE (SHA256_DIGEST_SIZE << 1)
27 #define block_size (1<<22) //FIXME this should be defined here?
28 #define objectsize_in_map (1 + XSEG_MAX_TARGETLEN) /* transparency byte + max object len */
29 #define mapheader_size (SHA256_DIGEST_SIZE + (sizeof(uint64_t)) ) /* magic hash value + volume size */
31 #define MF_OBJECT_EXIST (1 << 0)
32 #define MF_OBJECT_COPYING (1 << 1)
33 #define MF_OBJECT_WRITING (1 << 2)
34 #define MF_OBJECT_DELETING (1 << 3)
35 #define MF_OBJECT_DELETED (1 << 4)
36 #define MF_OBJECT_DESTROYED (1 << 5)
38 #define MF_OBJECT_NOT_READY (MF_OBJECT_COPYING|MF_OBJECT_WRITING|MF_OBJECT_DELETING)
40 char *magic_string = "This a magic string. Please hash me";
41 unsigned char magic_sha256[SHA256_DIGEST_SIZE]; /* sha256 hash value of magic string */
42 char zero_block[HEXLIFIED_SHA256_DIGEST_SIZE + 1]; /* hexlified sha256 hash value of a block full of zeros */
44 //dispatch_internal mapper states
53 typedef void (*cb_t)(struct peer_req *pr, struct xseg_request *req);
59 char object[XSEG_MAX_TARGETLEN + 1]; /* NULL terminated string */
66 #define MF_MAP_LOADING (1 << 0)
67 #define MF_MAP_DESTROYED (1 << 1)
68 #define MF_MAP_WRITING (1 << 2)
69 #define MF_MAP_DELETING (1 << 3)
70 #define MF_MAP_DROPPING_CACHE (1 << 4)
71 #define MF_MAP_EXCLUSIVE (1 << 5)
72 #define MF_MAP_OPENING (1 << 6)
73 #define MF_MAP_CLOSING (1 << 7)
75 #define MF_MAP_NOT_READY (MF_MAP_LOADING|MF_MAP_WRITING|MF_MAP_DELETING|\
76 MF_MAP_DROPPING_CACHE|MF_MAP_OPENING)
78 #define wait_on_pr(__pr, __condition__) \
79 while (__condition__){ \
81 __get_mapper_io(pr)->active = 0;\
82 XSEGLOG2(&lc, D, "Waiting on pr %lx, ta: %u", pr, ta); \
83 st_cond_wait(__pr->cond); \
86 #define wait_on_mapnode(__mn, __condition__) \
87 while (__condition__){ \
90 XSEGLOG2(&lc, D, "Waiting on map node %lx %s, waiters: %u, ta: %u", __mn, __mn->object, __mn->waiters, ta); \
91 st_cond_wait(__mn->cond); \
94 #define wait_on_map(__map, __condition__) \
95 while (__condition__){ \
98 XSEGLOG2(&lc, D, "Waiting on map %lx %s, waiters: %u, ta: %u", __map, __map->volume, __map->waiters, ta); \
99 st_cond_wait(__map->cond); \
102 #define signal_pr(__pr) \
104 if (!__get_mapper_io(pr)->active){\
106 XSEGLOG2(&lc, D, "Signaling pr %lx, ta: %u", pr, ta); \
107 __get_mapper_io(pr)->active = 1;\
108 st_cond_signal(__pr->cond); \
112 #define signal_map(__map) \
114 if (__map->waiters) { \
116 XSEGLOG2(&lc, D, "Signaling map %lx %s, waiters: %u, ta: %u", __map, __map->volume, __map->waiters, ta); \
118 st_cond_signal(__map->cond); \
122 #define signal_mapnode(__mn) \
124 if (__mn->waiters) { \
125 ta += __mn->waiters; \
126 XSEGLOG2(&lc, D, "Signaling map node %lx %s, waiters: %u, ta: %u", __mn, __mn->object, __mn->waiters, ta); \
128 st_cond_broadcast(__mn->cond); \
137 char volume[XSEG_MAX_TARGETLEN + 1]; /* NULL terminated string */
138 xhash_t *objects; /* obj_index --> map_node */
145 xport bportno; /* blocker that accesses data */
146 xport mbportno; /* blocker that accesses maps */
147 xhash_t *hashmaps; // hash_function(target) --> struct map
151 volatile uint32_t copyups; /* nr of copyups pending, issued by this mapper io */
152 xhash_t *copyups_nodes; /* hash map (xseg_request) --> (corresponding map_node of copied up object)*/
153 struct map_node *copyup_node;
154 volatile int err; /* error flag */
155 volatile uint64_t del_pending;
159 enum mapper_state state;
164 struct mapperd *mapper;
166 void print_map(struct map *m);
172 static inline struct mapperd * __get_mapperd(struct peerd *peer)
174 return (struct mapperd *) peer->priv;
177 static inline struct mapper_io * __get_mapper_io(struct peer_req *pr)
179 return (struct mapper_io *) pr->priv;
182 static inline uint64_t calc_map_obj(struct map *map)
186 uint64_t nr_objs = map->size / block_size;
187 if (map->size % block_size)
192 static uint32_t calc_nr_obj(struct xseg_request *req)
195 uint64_t rem_size = req->size;
196 uint64_t obj_offset = req->offset & (block_size -1); //modulo
197 uint64_t obj_size = (rem_size + obj_offset > block_size) ? block_size - obj_offset : rem_size;
198 rem_size -= obj_size;
199 while (rem_size > 0) {
200 obj_size = (rem_size > block_size) ? block_size : rem_size;
201 rem_size -= obj_size;
209 * Maps handling functions
211 //FIXME assert targetlen > 0
214 static struct map * find_map(struct mapperd *mapper, char *target, uint32_t targetlen)
217 struct map *m = NULL;
218 char buf[XSEG_MAX_TARGETLEN+1];
219 //assert targetlen <= XSEG_MAX_TARGETLEN
220 strncpy(buf, target, targetlen);
222 XSEGLOG2(&lc, D, "looking up map %s, len %u", buf, targetlen);
223 r = xhash_lookup(mapper->hashmaps, (xhashidx) buf, (xhashidx *) &m);
230 static int insert_map(struct mapperd *mapper, struct map *map)
234 if (find_map(mapper, map->volume, map->volumelen)){
235 XSEGLOG2(&lc, W, "Map %s found in hash maps", map->volume);
239 XSEGLOG2(&lc, D, "Inserting map %s, len: %d (map: %lx)",
240 map->volume, strlen(map->volume), (unsigned long) map);
241 r = xhash_insert(mapper->hashmaps, (xhashidx) map->volume, (xhashidx) map);
242 while (r == -XHASH_ERESIZE) {
243 xhashidx shift = xhash_grow_size_shift(mapper->hashmaps);
244 xhash_t *new_hashmap = xhash_resize(mapper->hashmaps, shift, NULL);
246 XSEGLOG2(&lc, E, "Cannot grow mapper->hashmaps to sizeshift %llu",
247 (unsigned long long) shift);
250 mapper->hashmaps = new_hashmap;
251 r = xhash_insert(mapper->hashmaps, (xhashidx) map->volume, (xhashidx) map);
257 static int remove_map(struct mapperd *mapper, struct map *map)
261 //assert no pending pr on map
263 r = xhash_delete(mapper->hashmaps, (xhashidx) map->volume);
264 while (r == -XHASH_ERESIZE) {
265 xhashidx shift = xhash_shrink_size_shift(mapper->hashmaps);
266 xhash_t *new_hashmap = xhash_resize(mapper->hashmaps, shift, NULL);
268 XSEGLOG2(&lc, E, "Cannot shrink mapper->hashmaps to sizeshift %llu",
269 (unsigned long long) shift);
272 mapper->hashmaps = new_hashmap;
273 r = xhash_delete(mapper->hashmaps, (xhashidx) map->volume);
279 static struct xseg_request * __close_map(struct peer_req *pr, struct map *map)
283 struct peerd *peer = pr->peer;
284 struct xseg_request *req;
285 struct mapperd *mapper = __get_mapperd(peer);
288 XSEGLOG2(&lc, I, "Closing map %s", map->volume);
290 req = xseg_get_request(peer->xseg, pr->portno, mapper->mbportno, X_ALLOC);
292 XSEGLOG2(&lc, E, "Cannot allocate request for map %s",
297 r = xseg_prep_request(peer->xseg, req, map->volumelen, block_size);
299 XSEGLOG2(&lc, E, "Cannot prepare request for map %s",
304 char *reqtarget = xseg_get_target(peer->xseg, req);
307 strncpy(reqtarget, map->volume, req->targetlen);
309 req->size = block_size;
311 r = xseg_set_req_data(peer->xseg, req, pr);
313 XSEGLOG2(&lc, E, "Cannot set request data for map %s",
317 p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
319 XSEGLOG2(&lc, E, "Cannot submit request for map %s",
323 r = xseg_signal(peer->xseg, p);
325 XSEGLOG2(&lc, I, "Map %s closing", map->volume);
329 xseg_get_req_data(peer->xseg, req, &dummy);
331 xseg_put_request(peer->xseg, req, pr->portno);
336 static int close_map(struct peer_req *pr, struct map *map)
339 struct xseg_request *req;
340 struct peerd *peer = pr->peer;
342 map->flags |= MF_MAP_CLOSING;
343 req = __close_map(pr, map);
346 wait_on_pr(pr, (!((req->state & XS_FAILED)||(req->state & XS_SERVED))));
347 map->flags &= ~MF_MAP_CLOSING;
348 err = req->state & XS_FAILED;
349 xseg_put_request(peer->xseg, req, pr->portno);
356 static int find_or_load_map(struct peerd *peer, struct peer_req *pr,
357 char *target, uint32_t targetlen, struct map **m)
359 struct mapperd *mapper = __get_mapperd(peer);
361 *m = find_map(mapper, target, targetlen);
363 XSEGLOG2(&lc, D, "Found map %s (%u)", (*m)->volume, (unsigned long) *m);
364 if ((*m)->flags & MF_MAP_NOT_READY) {
365 __xq_append_tail(&(*m)->pending, (xqindex) pr);
366 XSEGLOG2(&lc, I, "Map %s found and not ready", (*m)->volume);
368 //} else if ((*m)->flags & MF_MAP_DESTROYED){
372 XSEGLOG2(&lc, I, "Map %s found", (*m)->volume);
376 r = open_map(peer, pr, target, targetlen, 0);
383 * Object handling functions
386 struct map_node *find_object(struct map *map, uint64_t obj_index)
389 int r = xhash_lookup(map->objects, obj_index, (xhashidx *) &mn);
395 static int insert_object(struct map *map, struct map_node *mn)
397 //FIXME no find object first
398 int r = xhash_insert(map->objects, mn->objectidx, (xhashidx) mn);
399 if (r == -XHASH_ERESIZE) {
400 unsigned long shift = xhash_grow_size_shift(map->objects);
401 map->objects = xhash_resize(map->objects, shift, NULL);
404 r = xhash_insert(map->objects, mn->objectidx, (xhashidx) mn);
411 * map read/write functions
413 static inline void pithosmap_to_object(struct map_node *mn, unsigned char *buf)
416 //hexlify sha256 value
417 for (i = 0; i < SHA256_DIGEST_SIZE; i++) {
418 sprintf(mn->object+2*i, "%02x", buf[i]);
421 mn->object[SHA256_DIGEST_SIZE * 2] = 0;
422 mn->objectlen = SHA256_DIGEST_SIZE * 2;
423 mn->flags = MF_OBJECT_EXIST;
426 static inline void map_to_object(struct map_node *mn, char *buf)
431 mn->flags |= MF_OBJECT_EXIST;
432 memcpy(mn->object, buf+1, XSEG_MAX_TARGETLEN);
433 mn->object[XSEG_MAX_TARGETLEN] = 0;
434 mn->objectlen = strlen(mn->object);
437 static inline void object_to_map(char* buf, struct map_node *mn)
439 buf[0] = (mn->flags & MF_OBJECT_EXIST)? 1 : 0;
440 memcpy(buf+1, mn->object, mn->objectlen);
441 memset(buf+1+mn->objectlen, 0, XSEG_MAX_TARGETLEN - mn->objectlen); //zero out the rest of the buffer
444 static inline void mapheader_to_map(struct map *m, char *buf)
447 memcpy(buf + pos, magic_sha256, SHA256_DIGEST_SIZE);
448 pos += SHA256_DIGEST_SIZE;
449 memcpy(buf + pos, &m->size, sizeof(m->size));
450 pos += sizeof(m->size);
454 static struct xseg_request * object_write(struct peerd *peer, struct peer_req *pr,
455 struct map *map, struct map_node *mn)
458 struct mapperd *mapper = __get_mapperd(peer);
459 struct xseg_request *req = xseg_get_request(peer->xseg, pr->portno,
460 mapper->mbportno, X_ALLOC);
462 XSEGLOG2(&lc, E, "Cannot allocate request for object %s. \n\t"
464 mn->object, map->volume, (unsigned long long) mn->objectidx);
467 int r = xseg_prep_request(peer->xseg, req, map->volumelen, objectsize_in_map);
469 XSEGLOG2(&lc, E, "Cannot allocate request for object %s. \n\t"
471 mn->object, map->volume, (unsigned long long) mn->objectidx);
474 char *target = xseg_get_target(peer->xseg, req);
475 strncpy(target, map->volume, req->targetlen);
476 req->size = objectsize_in_map;
477 req->offset = mapheader_size + mn->objectidx * objectsize_in_map;
479 char *data = xseg_get_data(peer->xseg, req);
480 object_to_map(data, mn);
482 r = xseg_set_req_data(peer->xseg, req, pr);
484 XSEGLOG2(&lc, E, "Cannot set request data for object %s. \n\t"
486 mn->object, map->volume, (unsigned long long) mn->objectidx);
489 xport p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
491 XSEGLOG2(&lc, E, "Cannot submit request for object %s. \n\t"
493 mn->object, map->volume, (unsigned long long) mn->objectidx);
496 r = xseg_signal(peer->xseg, p);
498 XSEGLOG2(&lc, W, "Cannot signal port %u", p);
500 XSEGLOG2(&lc, I, "Writing object %s \n\t"
502 mn->object, map->volume, (unsigned long long) mn->objectidx);
507 xseg_get_req_data(peer->xseg, req, &dummy);
509 xseg_put_request(peer->xseg, req, pr->portno);
511 XSEGLOG2(&lc, E, "Object write for object %s failed. \n\t"
513 mn->object, map->volume, (unsigned long long) mn->objectidx);
517 static struct xseg_request * __write_map(struct peer_req* pr, struct map *map)
520 struct peerd *peer = pr->peer;
521 struct mapperd *mapper = __get_mapperd(peer);
523 uint64_t i, pos, max_objidx = calc_map_obj(map);
524 struct xseg_request *req = xseg_get_request(peer->xseg, pr->portno,
525 mapper->mbportno, X_ALLOC);
527 XSEGLOG2(&lc, E, "Cannot allocate request for map %s", map->volume);
530 int r = xseg_prep_request(peer->xseg, req, map->volumelen,
531 mapheader_size + max_objidx * objectsize_in_map);
533 XSEGLOG2(&lc, E, "Cannot prepare request for map %s", map->volume);
536 char *target = xseg_get_target(peer->xseg, req);
537 strncpy(target, map->volume, req->targetlen);
538 char *data = xseg_get_data(peer->xseg, req);
539 mapheader_to_map(map, data);
540 pos = mapheader_size;
542 req->size = req->datalen;
545 if (map->size % block_size)
547 for (i = 0; i < max_objidx; i++) {
548 mn = find_object(map, i);
550 XSEGLOG2(&lc, E, "Cannot find object %lli for map %s",
551 (unsigned long long) i, map->volume);
554 object_to_map(data+pos, mn);
555 pos += objectsize_in_map;
557 r = xseg_set_req_data(peer->xseg, req, pr);
559 XSEGLOG2(&lc, E, "Cannot set request data for map %s",
563 xport p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
565 XSEGLOG2(&lc, E, "Cannot submit request for map %s",
569 r = xseg_signal(peer->xseg, p);
571 XSEGLOG2(&lc, W, "Cannot signal port %u", p);
573 map->flags |= MF_MAP_WRITING;
574 XSEGLOG2(&lc, I, "Writing map %s", map->volume);
578 xseg_get_req_data(peer->xseg, req, &dummy);
580 xseg_put_request(peer->xseg, req, pr->portno);
582 XSEGLOG2(&lc, E, "Map write for map %s failed.", map->volume);
586 static int write_map(struct peer_req* pr, struct map *map)
589 struct peerd *peer = pr->peer;
590 map->flags |= MF_MAP_WRITING;
591 struct xseg_request *req = __write_map(pr, map);
594 wait_on_pr(pr, (!(req->state & XS_FAILED || req->state & XS_SERVED)));
595 if (req->state & XS_FAILED)
597 xseg_put_request(peer->xseg, req, pr->portno);
598 map->flags &= ~MF_MAP_WRITING;
602 static struct xseg_request * __load_map(struct peer_req *pr, struct map *m)
606 struct xseg_request *req;
607 struct peerd *peer = pr->peer;
608 struct mapperd *mapper = __get_mapperd(peer);
611 XSEGLOG2(&lc, I, "Loading ng map %s", m->volume);
613 req = xseg_get_request(peer->xseg, pr->portno, mapper->mbportno, X_ALLOC);
615 XSEGLOG2(&lc, E, "Cannot allocate request for map %s",
620 r = xseg_prep_request(peer->xseg, req, m->volumelen, block_size);
622 XSEGLOG2(&lc, E, "Cannot prepare request for map %s",
627 char *reqtarget = xseg_get_target(peer->xseg, req);
630 strncpy(reqtarget, m->volume, req->targetlen);
632 req->size = block_size;
634 r = xseg_set_req_data(peer->xseg, req, pr);
636 XSEGLOG2(&lc, E, "Cannot set request data for map %s",
640 p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
642 XSEGLOG2(&lc, E, "Cannot submit request for map %s",
646 r = xseg_signal(peer->xseg, p);
648 XSEGLOG2(&lc, I, "Map %s loading", m->volume);
652 xseg_get_req_data(peer->xseg, req, &dummy);
654 xseg_put_request(peer->xseg, req, pr->portno);
659 static int read_map (struct map *map, char *buf)
661 char nulls[SHA256_DIGEST_SIZE];
662 memset(nulls, 0, SHA256_DIGEST_SIZE);
664 int r = !memcmp(buf, nulls, SHA256_DIGEST_SIZE);
666 XSEGLOG2(&lc, D, "Read zeros");
670 //type 1, our type, type 0 pithos map
671 int type = !memcmp(buf, magic_sha256, SHA256_DIGEST_SIZE);
672 XSEGLOG2(&lc, I, "Type %d detected for map %s", type, map->volume);
675 struct map_node *map_node;
677 pos = SHA256_DIGEST_SIZE;
678 map->size = *(uint64_t *) (buf + pos);
679 pos += sizeof(uint64_t);
680 nr_objs = map->size / block_size;
681 if (map->size % block_size)
683 map_node = calloc(nr_objs, sizeof(struct map_node));
687 for (i = 0; i < nr_objs; i++) {
688 map_node[i].map = map;
689 map_node[i].objectidx = i;
690 map_node[i].waiters = 0;
692 map_node[i].cond = st_cond_new(); //FIXME err check;
693 map_to_object(&map_node[i], buf + pos);
694 pos += objectsize_in_map;
695 r = insert_object(map, &map_node[i]); //FIXME error check
699 uint64_t max_nr_objs = block_size/SHA256_DIGEST_SIZE;
700 map_node = calloc(max_nr_objs, sizeof(struct map_node));
703 for (i = 0; i < max_nr_objs; i++) {
704 if (!memcmp(buf+pos, nulls, SHA256_DIGEST_SIZE))
706 map_node[i].objectidx = i;
707 map_node[i].map = map;
708 map_node[i].waiters = 0;
710 map_node[i].cond = st_cond_new(); //FIXME err check;
711 pithosmap_to_object(&map_node[i], buf + pos);
712 pos += SHA256_DIGEST_SIZE;
713 r = insert_object(map, &map_node[i]); //FIXME error check
715 map->size = i * block_size;
718 XSEGLOG2(&lc, I, "Map read for map %s completed", map->volume);
721 //FIXME cleanup on error
724 static int load_map(struct peer_req *pr, struct map *map)
727 struct xseg_request *req;
728 struct peerd *peer = pr->peer;
729 map->flags |= MF_MAP_LOADING;
730 req = __load_map(pr, map);
733 wait_on_pr(pr, (!(req->state & XS_FAILED || req->state & XS_SERVED)));
734 map->flags &= ~MF_MAP_LOADING;
735 if (req->state & XS_FAILED){
736 XSEGLOG2(&lc, E, "Map load failed for map %s", map->volume);
737 xseg_put_request(peer->xseg, req, pr->portno);
740 r = read_map(map, xseg_get_data(peer->xseg, req));
741 xseg_put_request(peer->xseg, req, pr->portno);
745 static struct xseg_request * __open_map(struct peer_req *pr, struct map *m,
750 struct xseg_request *req;
751 struct peerd *peer = pr->peer;
752 struct mapperd *mapper = __get_mapperd(peer);
755 XSEGLOG2(&lc, I, "Opening map %s", m->volume);
757 req = xseg_get_request(peer->xseg, pr->portno, mapper->mbportno, X_ALLOC);
759 XSEGLOG2(&lc, E, "Cannot allocate request for map %s",
764 r = xseg_prep_request(peer->xseg, req, m->volumelen, block_size);
766 XSEGLOG2(&lc, E, "Cannot prepare request for map %s",
771 char *reqtarget = xseg_get_target(peer->xseg, req);
774 strncpy(reqtarget, m->volume, req->targetlen);
776 req->size = block_size;
778 if (!(flags & MF_FORCE))
779 req->flags = XF_NOSYNC;
780 r = xseg_set_req_data(peer->xseg, req, pr);
782 XSEGLOG2(&lc, E, "Cannot set request data for map %s",
786 p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
788 XSEGLOG2(&lc, E, "Cannot submit request for map %s",
792 r = xseg_signal(peer->xseg, p);
794 XSEGLOG2(&lc, I, "Map %s opening", m->volume);
798 xseg_get_req_data(peer->xseg, req, &dummy);
800 xseg_put_request(peer->xseg, req, pr->portno);
805 static int open_map(struct peer_req *pr, struct map *map, uint32_t flags)
808 struct xseg_request *req;
809 struct peerd *peer = pr->peer;
811 map->flags |= MF_MAP_OPENING;
812 req = __open_map(pr, map, flags);
815 wait_on_pr(pr, (!((req->state & XS_FAILED)||(req->state & XS_SERVED))));
816 map->flags &= ~MF_MAP_OPENING;
817 err = req->state & XS_FAILED;
818 xseg_put_request(peer->xseg, req, pr->portno);
822 map->flags |= MF_MAP_EXCLUSIVE;
830 static int __set_copyup_node(struct mapper_io *mio, struct xseg_request *req, struct map_node *mn)
834 r = xhash_insert(mio->copyups_nodes, (xhashidx) req, (xhashidx) mn);
835 if (r == -XHASH_ERESIZE) {
836 xhashidx shift = xhash_grow_size_shift(mio->copyups_nodes);
837 xhash_t *new_hashmap = xhash_resize(mio->copyups_nodes, shift, NULL);
840 mio->copyups_nodes = new_hashmap;
841 r = xhash_insert(mio->copyups_nodes, (xhashidx) req, (xhashidx) mn);
845 r = xhash_delete(mio->copyups_nodes, (xhashidx) req);
846 if (r == -XHASH_ERESIZE) {
847 xhashidx shift = xhash_shrink_size_shift(mio->copyups_nodes);
848 xhash_t *new_hashmap = xhash_resize(mio->copyups_nodes, shift, NULL);
851 mio->copyups_nodes = new_hashmap;
852 r = xhash_delete(mio->copyups_nodes, (xhashidx) req);
859 static struct map_node * __get_copyup_node(struct mapper_io *mio, struct xseg_request *req)
862 int r = xhash_lookup(mio->copyups_nodes, (xhashidx) req, (xhashidx *) &mn);
868 static struct xseg_request * copyup_object(struct peerd *peer, struct map_node *mn, struct peer_req *pr)
870 struct mapperd *mapper = __get_mapperd(peer);
871 struct mapper_io *mio = __get_mapper_io(pr);
872 struct map *map = mn->map;
877 //struct sha256_ctx sha256ctx;
878 uint32_t newtargetlen;
879 char new_target[XSEG_MAX_TARGETLEN + 1];
880 unsigned char buf[SHA256_DIGEST_SIZE]; //assert sha256_digest_size(32) <= MAXTARGETLEN
881 char new_object[XSEG_MAX_TARGETLEN + 20]; //20 is an arbitrary padding able to hold string representation of objectidx
882 strncpy(new_object, map->volume, map->volumelen);
883 sprintf(new_object + map->volumelen, "%u", mn->objectidx); //sprintf adds null termination
884 new_object[XSEG_MAX_TARGETLEN + 19] = 0;
886 gcry_md_hash_buffer(GCRY_MD_SHA256, buf, new_object, strlen(new_object));
887 for (i = 0; i < SHA256_DIGEST_SIZE; ++i)
888 sprintf (new_target + 2*i, "%02x", buf[i]);
889 newtargetlen = SHA256_DIGEST_SIZE * 2;
891 if (!strncmp(mn->object, zero_block, (mn->objectlen < HEXLIFIED_SHA256_DIGEST_SIZE)? mn->objectlen : HEXLIFIED_SHA256_DIGEST_SIZE))
892 goto copyup_zeroblock;
894 struct xseg_request *req = xseg_get_request(peer->xseg, pr->portno,
895 mapper->bportno, X_ALLOC);
897 XSEGLOG2(&lc, E, "Cannot get request for object %s", mn->object);
900 r = xseg_prep_request(peer->xseg, req, newtargetlen,
901 sizeof(struct xseg_request_copy));
903 XSEGLOG2(&lc, E, "Cannot prepare request for object %s", mn->object);
907 char *target = xseg_get_target(peer->xseg, req);
908 strncpy(target, new_target, req->targetlen);
910 struct xseg_request_copy *xcopy = (struct xseg_request_copy *) xseg_get_data(peer->xseg, req);
911 strncpy(xcopy->target, mn->object, mn->objectlen);
912 xcopy->targetlen = mn->objectlen;
915 req->size = block_size;
917 r = xseg_set_req_data(peer->xseg, req, pr);
919 XSEGLOG2(&lc, E, "Cannot set request data for object %s", mn->object);
922 r = __set_copyup_node(mio, req, mn);
923 p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
925 XSEGLOG2(&lc, E, "Cannot submit for object %s", mn->object);
928 xseg_signal(peer->xseg, p);
931 mn->flags |= MF_OBJECT_COPYING;
932 XSEGLOG2(&lc, I, "Copying up object %s \n\t to %s", mn->object, new_target);
936 r = __set_copyup_node(mio, req, NULL);
937 xseg_get_req_data(peer->xseg, req, &dummy);
939 xseg_put_request(peer->xseg, req, pr->portno);
941 XSEGLOG2(&lc, E, "Copying up object %s \n\t to %s failed", mn->object, new_target);
945 XSEGLOG2(&lc, I, "Copying up of zero block is not needed."
946 "Proceeding in writing the new object in map");
947 /* construct a tmp map_node for writing purposes */
948 struct map_node newmn = *mn;
949 newmn.flags = MF_OBJECT_EXIST;
950 strncpy(newmn.object, new_target, newtargetlen);
951 newmn.object[newtargetlen] = 0;
952 newmn.objectlen = newtargetlen;
953 newmn.objectidx = mn->objectidx;
954 req = object_write(peer, pr, map, &newmn);
955 r = __set_copyup_node(mio, req, mn);
957 XSEGLOG2(&lc, E, "Object write returned error for object %s"
958 "\n\t of map %s [%llu]",
959 mn->object, map->volume, (unsigned long long) mn->objectidx);
962 mn->flags |= MF_OBJECT_WRITING;
963 XSEGLOG2(&lc, I, "Object %s copy up completed. Pending writing.", mn->object);
967 static struct xseg_request * delete_object(struct peer_req *pr, struct map_node *mn)
970 struct peerd *peer = pr->peer;
971 struct mapperd *mapper = __get_mapperd(peer);
972 struct mapper_io *mio = __get_mapper_io(pr);
973 struct xseg_request *req = xseg_get_request(peer->xseg, pr->portno,
974 mapper->bportno, X_ALLOC);
975 XSEGLOG2(&lc, I, "Deleting mapnode %s", mn->object);
977 XSEGLOG2(&lc, E, "Cannot get request for object %s", mn->object);
980 int r = xseg_prep_request(peer->xseg, req, mn->objectlen, 0);
982 XSEGLOG2(&lc, E, "Cannot prep request for object %s", mn->object);
985 char *target = xseg_get_target(peer->xseg, req);
986 strncpy(target, mn->object, req->targetlen);
988 req->size = req->datalen;
990 r = xseg_set_req_data(peer->xseg, req, pr);
992 XSEGLOG2(&lc, E, "Cannot set req data for object %s", mn->object);
995 __set_copyup_node(mio, req, mn);
996 xport p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
998 XSEGLOG2(&lc, E, "Cannot submit request for object %s", mn->object);
1001 r = xseg_signal(peer->xseg, p);
1002 XSEGLOG2(&lc, I, "Object %s deletion pending", mn->object);
1006 xseg_get_req_data(peer->xseg, req, &dummy);
1008 xseg_put_request(peer->xseg, req, pr->portno);
1010 XSEGLOG2(&lc, I, "Object %s deletion failed", mn->object);
1014 static struct xseg_request * delete_map(struct peer_req *pr, struct map *map)
1017 struct peerd *peer = pr->peer;
1018 struct mapperd *mapper = __get_mapperd(peer);
1019 struct mapper_io *mio = __get_mapper_io(pr);
1020 struct xseg_request *req = xseg_get_request(peer->xseg, pr->portno,
1021 mapper->mbportno, X_ALLOC);
1022 XSEGLOG2(&lc, I, "Deleting map %s", map->volume);
1024 XSEGLOG2(&lc, E, "Cannot get request for map %s", map->volume);
1027 int r = xseg_prep_request(peer->xseg, req, map->volumelen, 0);
1029 XSEGLOG2(&lc, E, "Cannot prep request for map %s", map->volume);
1032 char *target = xseg_get_target(peer->xseg, req);
1033 strncpy(target, map->volume, req->targetlen);
1035 req->size = req->datalen;
1037 r = xseg_set_req_data(peer->xseg, req, pr);
1039 XSEGLOG2(&lc, E, "Cannot set req data for map %s", map->volume);
1042 __set_copyup_node(mio, req, NULL);
1043 xport p = xseg_submit(peer->xseg, req, pr->portno, X_ALLOC);
1045 XSEGLOG2(&lc, E, "Cannot submit request for map %s", map->volume);
1048 r = xseg_signal(peer->xseg, p);
1049 map->flags |= MF_MAP_DELETING;
1050 XSEGLOG2(&lc, I, "Map %s deletion pending", map->volume);
1054 xseg_get_req_data(peer->xseg, req, &dummy);
1056 xseg_put_request(peer->xseg, req, pr->portno);
1058 XSEGLOG2(&lc, E, "Map %s deletion failed", map->volume);
1063 static inline struct map_node * get_mapnode(struct map *map, uint32_t index)
1065 struct map_node *mn = find_object(map, index);
1071 static inline void put_mapnode(struct map_node *mn)
1076 st_cond_destroy(mn->cond);
1080 static inline void __get_map(struct map *map)
1085 static inline void put_map(struct map *map)
1087 struct map_node *mn;
1090 XSEGLOG2(&lc, I, "Freeing map %s", map->volume);
1093 for (i = 0; i < calc_map_obj(map); i++) {
1094 mn = get_mapnode(map, i);
1096 //make sure all pending operations on all objects are completed
1097 if (mn->flags & MF_OBJECT_NOT_READY){
1098 //this should never happen...
1099 wait_on_mapnode(mn, mn->flags & MF_OBJECT_NOT_READY);
1101 mn->flags &= MF_OBJECT_DESTROYED;
1102 put_mapnode(mn); //matchin mn->ref = 1 on mn init
1103 put_mapnode(mn); //matcing get_mapnode;
1104 //assert mn->ref == 0;
1107 mn = find_object(map, 0);
1110 XSEGLOG2(&lc, I, "Freed map %s", map->volume);
1115 static struct map * create_map(struct mapperd *mapper, char *name, uint32_t namelen)
1118 struct map *m = malloc(sizeof(struct map));
1120 XSEGLOG2(&lc, E, "Cannot allocate map ");
1124 strncpy(m->volume, name, namelen);
1125 m->volume[namelen] = 0;
1126 m->volumelen = namelen;
1128 m->objects = xhash_new(3, INTEGER);
1130 XSEGLOG2(&lc, E, "Cannot allocate object hashmap for map %s",
1136 m->cond = st_cond_new(); //FIXME err check;
1137 r = insert_map(mapper, m);
1139 XSEGLOG2(&lc, E, "Cannot insert map %s", m->volume);
1146 xhash_free(m->objects);
1148 XSEGLOG2(&lc, E, "failed to create map %s", m->volume);
1156 void deletion_cb(struct peer_req *pr, struct xseg_request *req)
1158 struct peerd *peer = pr->peer;
1159 struct mapperd *mapper = __get_mapperd(peer);
1161 struct mapper_io *mio = __get_mapper_io(pr);
1162 struct map_node *mn = __get_copyup_node(mio, req);
1165 if (req->state & XS_FAILED){
1169 xseg_put_request(peer->xseg, req, pr->portno);
1173 void copyup_cb(struct peer_req *pr, struct xseg_request *req)
1175 struct peerd *peer = pr->peer;
1176 struct mapperd *mapper = __get_mapperd(peer);
1178 struct mapper_io *mio = __get_mapper_io(pr);
1179 struct map_node *mn = __get_copyup_node(mio, req);
1181 XSEGLOG2(&lc, E, "Cannot get map node");
1184 __set_copyup_node(mio, req, NULL);
1186 if (req->state & XS_FAILED){
1187 XSEGLOG2(&lc, E, "Req failed");
1188 mn->flags &= ~MF_OBJECT_COPYING;
1189 mn->flags &= ~MF_OBJECT_WRITING;
1192 if (req->op == X_WRITE) {
1193 char *target = xseg_get_target(peer->xseg, req);
1195 //printf("handle object write replyi\n");
1196 __set_copyup_node(mio, req, NULL);
1197 //assert mn->flags & MF_OBJECT_WRITING
1198 mn->flags &= ~MF_OBJECT_WRITING;
1200 struct map_node tmp;
1201 char *data = xseg_get_data(peer->xseg, req);
1202 map_to_object(&tmp, data);
1203 mn->flags |= MF_OBJECT_EXIST;
1204 if (mn->flags != MF_OBJECT_EXIST){
1205 XSEGLOG2(&lc, E, "map node %s has wrong flags", mn->object);
1208 //assert mn->flags & MF_OBJECT_EXIST
1209 strncpy(mn->object, tmp.object, tmp.objectlen);
1210 mn->object[tmp.objectlen] = 0;
1211 mn->objectlen = tmp.objectlen;
1212 XSEGLOG2(&lc, I, "Object write of %s completed successfully", mn->object);
1215 } else if (req->op == X_COPY) {
1216 // issue write_object;
1217 mn->flags &= ~MF_OBJECT_COPYING;
1218 struct map *map = mn->map;
1220 XSEGLOG2(&lc, E, "Object %s has not map back pointer", mn->object);
1224 /* construct a tmp map_node for writing purposes */
1225 char *target = xseg_get_target(peer->xseg, req);
1226 struct map_node newmn = *mn;
1227 newmn.flags = MF_OBJECT_EXIST;
1228 strncpy(newmn.object, target, req->targetlen);
1229 newmn.object[req->targetlen] = 0;
1230 newmn.objectlen = req->targetlen;
1231 newmn.objectidx = mn->objectidx;
1232 struct xseg_request *xreq = object_write(peer, pr, map, &newmn);
1234 XSEGLOG2(&lc, E, "Object write returned error for object %s"
1235 "\n\t of map %s [%llu]",
1236 mn->object, map->volume, (unsigned long long) mn->objectidx);
1239 mn->flags |= MF_OBJECT_WRITING;
1240 __set_copyup_node (mio, xreq, mn);
1242 XSEGLOG2(&lc, I, "Object %s copy up completed. Pending writing.", mn->object);
1249 xseg_put_request(peer->xseg, req, pr->portno);
1254 XSEGLOG2(&lc, D, "Mio->copyups: %u", mio->copyups);
1263 struct map_node *mn;
1268 static int req2objs(struct peer_req *pr, struct map *map, int write)
1271 struct peerd *peer = pr->peer;
1272 struct mapper_io *mio = __get_mapper_io(pr);
1273 char *target = xseg_get_target(peer->xseg, pr->req);
1274 uint32_t nr_objs = calc_nr_obj(pr->req);
1275 uint64_t size = sizeof(struct xseg_reply_map) +
1276 nr_objs * sizeof(struct xseg_reply_map_scatterlist);
1277 uint32_t idx, i, ready;
1278 uint64_t rem_size, obj_index, obj_offset, obj_size;
1279 struct map_node *mn;
1281 XSEGLOG2(&lc, D, "Calculated %u nr_objs", nr_objs);
1283 /* get map_nodes of request */
1284 struct r2o *mns = malloc(sizeof(struct r2o)*nr_objs);
1286 XSEGLOG2(&lc, E, "Cannot allocate mns");
1290 rem_size = pr->req->size;
1291 obj_index = pr->req->offset / block_size;
1292 obj_offset = pr->req->offset & (block_size -1); //modulo
1293 obj_size = (obj_offset + rem_size > block_size) ? block_size - obj_offset : rem_size;
1294 mn = get_mapnode(map, obj_index);
1296 XSEGLOG2(&lc, E, "Cannot find obj_index %llu\n", (unsigned long long) obj_index);
1301 mns[idx].offset = obj_offset;
1302 mns[idx].size = obj_size;
1303 rem_size -= obj_size;
1304 while (rem_size > 0) {
1308 obj_size = (rem_size > block_size) ? block_size : rem_size;
1309 rem_size -= obj_size;
1310 mn = get_mapnode(map, obj_index);
1312 XSEGLOG2(&lc, E, "Cannot find obj_index %llu\n", (unsigned long long) obj_index);
1317 mns[idx].offset = obj_offset;
1318 mns[idx].size = obj_size;
1324 while (ready < (idx + 1)){
1326 for (i = 0; i < (idx+1); i++) {
1329 if (mn->flags & MF_OBJECT_NOT_READY) {
1331 wait_on_mapnode(mn, mn->flags & MF_OBJECT_NOT_READY);
1332 if (mn->flags & MF_OBJECT_DELETED){
1336 XSEGLOG2(&lc, E, "Mio-err, pending_copyups: %d", mio->copyups);
1344 else if (!(mn->flags & MF_OBJECT_EXIST)) {
1345 //calc new_target, copy up object
1346 if (copyup_object(peer, mn, pr) == NULL){
1347 XSEGLOG2(&lc, E, "Error in copy up object");
1360 while(mio->copyups > 0){
1361 mio->cb = copyup_cb;
1364 st_cond_wait(pr->cond);
1371 XSEGLOG2(&lc, E, "Mio->err");
1375 /* resize request to fit reply */
1376 char buf[XSEG_MAX_TARGETLEN];
1377 strncpy(buf, target, pr->req->targetlen);
1378 r = xseg_resize_request(peer->xseg, pr->req, pr->req->targetlen, size);
1380 XSEGLOG2(&lc, E, "Cannot resize request");
1383 target = xseg_get_target(peer->xseg, pr->req);
1384 strncpy(target, buf, pr->req->targetlen);
1386 /* structure reply */
1387 struct xseg_reply_map *reply = (struct xseg_reply_map *) xseg_get_data(peer->xseg, pr->req);
1388 reply->cnt = nr_objs;
1389 for (i = 0; i < (idx+1); i++) {
1390 strncpy(reply->segs[i].target, mns[i].mn->object, mns[i].mn->objectlen);
1391 reply->segs[i].targetlen = mns[i].mn->objectlen;
1392 reply->segs[i].offset = mns[i].offset;
1393 reply->segs[i].size = mns[i].size;
1396 for (i = 0; i < idx; i++) {
1397 put_mapnode(mns[i].mn);
1403 static int do_dropcache(struct peer_req *pr, struct map *map)
1405 struct map_node *mn;
1406 struct peerd *peer = pr->peer;
1407 struct mapperd *mapper = __get_mapperd(peer);
1409 XSEGLOG2(&lc, I, "Dropping cache for map %s", map->volume);
1410 map->flags |= MF_MAP_DROPPING_CACHE;
1411 for (i = 0; i < calc_map_obj(map); i++) {
1412 mn = get_mapnode(map, i);
1414 if (!(mn->flags & MF_OBJECT_DESTROYED)){
1415 //make sure all pending operations on all objects are completed
1416 if (mn->flags & MF_OBJECT_NOT_READY){
1417 wait_on_mapnode(mn, mn->flags & MF_OBJECT_NOT_READY);
1419 mn->flags &= MF_OBJECT_DESTROYED;
1424 map->flags &= ~MF_MAP_DROPPING_CACHE;
1425 map->flags |= MF_MAP_DESTROYED;
1426 remove_map(mapper, map);
1427 XSEGLOG2(&lc, I, "Dropping cache for map %s completed", map->volume);
1428 put_map(map); // put map here to destroy it (matches m->ref = 1 on map create)
1432 static int do_info(struct peer_req *pr, struct map *map)
1434 struct peerd *peer = pr->peer;
1435 struct xseg_reply_info *xinfo = (struct xseg_reply_info *) xseg_get_data(peer->xseg, pr->req);
1436 xinfo->size = map->size;
1441 static int do_close(struct peer_req *pr, struct map *map)
1443 // struct peerd *peer = pr->peer;
1444 // struct xseg_request *req;
1445 if (map->flags & MF_MAP_EXCLUSIVE)
1447 return do_dropcache(pr, map);
1450 static int do_destroy(struct peer_req *pr, struct map *map)
1453 struct peerd *peer = pr->peer;
1454 struct mapper_io *mio = __get_mapper_io(pr);
1455 struct map_node *mn;
1456 struct xseg_request *req;
1458 XSEGLOG2(&lc, I, "Destroying map %s", map->volume);
1459 map->flags |= MF_MAP_DELETING;
1460 req = delete_map(pr, map);
1463 wait_on_pr(pr, (!((req->state & XS_FAILED)||(req->state & XS_SERVED))));
1464 if (req->state & XS_FAILED){
1465 xseg_put_request(peer->xseg, req, pr->portno);
1466 map->flags &= ~MF_MAP_DELETING;
1469 xseg_put_request(peer->xseg, req, pr->portno);
1471 uint64_t nr_obj = calc_map_obj(map);
1472 uint64_t deleted = 0;
1473 while (deleted < nr_obj){
1475 for (i = 0; i < nr_obj; i++){
1476 mn = get_mapnode(map, i);
1478 if (!(mn->flags & MF_OBJECT_DESTROYED)){
1479 if (mn->flags & MF_OBJECT_EXIST){
1480 //make sure all pending operations on all objects are completed
1481 if (mn->flags & MF_OBJECT_NOT_READY){
1482 wait_on_mapnode(mn, mn->flags & MF_OBJECT_NOT_READY);
1484 req = delete_object(pr, mn);
1486 if (mio->del_pending){
1495 mn->flags &= MF_OBJECT_DESTROYED;
1502 mio->cb = deletion_cb;
1503 wait_on_pr(pr, mio->del_pending > 0);
1506 map->flags &= ~MF_MAP_DELETING;
1507 XSEGLOG2(&lc, I, "Destroyed map %s", map->volume);
1508 return do_close(pr, map);
1511 static int do_mapr(struct peer_req *pr, struct map *map)
1513 struct peerd *peer = pr->peer;
1514 int r = req2objs(pr, map, 0);
1516 XSEGLOG2(&lc, I, "Map r of map %s, range: %llu-%llu failed",
1518 (unsigned long long) pr->req->offset,
1519 (unsigned long long) (pr->req->offset + pr->req->size));
1522 XSEGLOG2(&lc, I, "Map r of map %s, range: %llu-%llu completed",
1524 (unsigned long long) pr->req->offset,
1525 (unsigned long long) (pr->req->offset + pr->req->size));
1526 XSEGLOG2(&lc, D, "Req->offset: %llu, req->size: %llu",
1527 (unsigned long long) pr->req->offset,
1528 (unsigned long long) pr->req->size);
1529 char buf[XSEG_MAX_TARGETLEN+1];
1530 struct xseg_reply_map *reply = (struct xseg_reply_map *) xseg_get_data(peer->xseg, pr->req);
1532 for (i = 0; i < reply->cnt; i++) {
1533 XSEGLOG2(&lc, D, "i: %d, reply->cnt: %u",i, reply->cnt);
1534 strncpy(buf, reply->segs[i].target, reply->segs[i].targetlen);
1535 buf[reply->segs[i].targetlen] = 0;
1536 XSEGLOG2(&lc, D, "%d: Object: %s, offset: %llu, size: %llu", i, buf,
1537 (unsigned long long) reply->segs[i].offset,
1538 (unsigned long long) reply->segs[i].size);
1543 static int do_mapw(struct peer_req *pr, struct map *map)
1545 struct peerd *peer = pr->peer;
1546 int r = req2objs(pr, map, 1);
1548 XSEGLOG2(&lc, I, "Map w of map %s, range: %llu-%llu failed",
1550 (unsigned long long) pr->req->offset,
1551 (unsigned long long) (pr->req->offset + pr->req->size));
1554 XSEGLOG2(&lc, I, "Map w of map %s, range: %llu-%llu completed",
1556 (unsigned long long) pr->req->offset,
1557 (unsigned long long) (pr->req->offset + pr->req->size));
1558 XSEGLOG2(&lc, D, "Req->offset: %llu, req->size: %llu",
1559 (unsigned long long) pr->req->offset,
1560 (unsigned long long) pr->req->size);
1561 char buf[XSEG_MAX_TARGETLEN+1];
1562 struct xseg_reply_map *reply = (struct xseg_reply_map *) xseg_get_data(peer->xseg, pr->req);
1564 for (i = 0; i < reply->cnt; i++) {
1565 XSEGLOG2(&lc, D, "i: %d, reply->cnt: %u",i, reply->cnt);
1566 strncpy(buf, reply->segs[i].target, reply->segs[i].targetlen);
1567 buf[reply->segs[i].targetlen] = 0;
1568 XSEGLOG2(&lc, D, "%d: Object: %s, offset: %llu, size: %llu", i, buf,
1569 (unsigned long long) reply->segs[i].offset,
1570 (unsigned long long) reply->segs[i].size);
1575 //here map is the parent map
1576 static int do_clone(struct peer_req *pr, struct map *map)
1579 FIXME check if clone map exists
1580 clonemap = get_map(pr, target, targetlen, MF_LOAD);
1582 do_dropcache(pr, clonemap); // drop map here, rely on get_map_function to drop
1583 // cache on non-exclusive opens or declare a NO_CACHE flag ?
1588 char buf[XSEG_MAX_TARGETLEN];
1589 struct peerd *peer = pr->peer;
1590 struct mapperd *mapper = __get_mapperd(peer);
1591 char *target = xseg_get_target(peer->xseg, pr->req);
1592 struct xseg_request_clone *xclone = (struct xseg_request_clone *) xseg_get_data(peer->xseg, pr->req);
1593 XSEGLOG2(&lc, I, "Cloning map %s", map->volume);
1594 struct map *clonemap = create_map(mapper, target, pr->req->targetlen);
1598 if (xclone->size == -1)
1599 clonemap->size = map->size;
1601 clonemap->size = xclone->size;
1602 if (clonemap->size < map->size){
1603 target = xseg_get_target(peer->xseg, pr->req);
1604 strncpy(buf, target, pr->req->targetlen);
1605 buf[pr->req->targetlen] = 0;
1606 XSEGLOG2(&lc, W, "Requested clone size (%llu) < map size (%llu)"
1607 "\n\t for requested clone %s",
1608 (unsigned long long) xclone->size,
1609 (unsigned long long) map->size, buf);
1613 //alloc and init map_nodes
1614 unsigned long c = clonemap->size/block_size + 1;
1615 struct map_node *map_nodes = calloc(c, sizeof(struct map_node));
1620 for (i = 0; i < clonemap->size/block_size + 1; i++) {
1621 struct map_node *mn = get_mapnode(map, i);
1623 strncpy(map_nodes[i].object, mn->object, mn->objectlen);
1624 map_nodes[i].objectlen = mn->objectlen;
1627 strncpy(map_nodes[i].object, zero_block, strlen(zero_block)); //this should be SHA256_DIGEST_SIZE *2
1628 map_nodes[i].objectlen = strlen(zero_block);
1630 map_nodes[i].object[map_nodes[i].objectlen] = 0; //NULL terminate
1631 map_nodes[i].flags = 0;
1632 map_nodes[i].objectidx = i;
1633 map_nodes[i].map = clonemap;
1634 map_nodes[i].ref = 1;
1635 map_nodes[i].waiters = 0;
1636 map_nodes[i].cond = st_cond_new(); //FIXME errcheck;
1637 r = insert_object(clonemap, &map_nodes[i]);
1639 XSEGLOG2(&lc, E, "Cannot insert object %d to map %s", i, clonemap->volume);
1643 r = write_map(pr, clonemap);
1645 XSEGLOG2(&lc, E, "Cannot write map %s", clonemap->volume);
1655 static int open_load_map(struct peer_req *pr, struct map *map, uint32_t flags)
1658 if (flags & MF_EXCLUSIVE){
1659 r = open_map(pr, map, flags);
1661 if (flags & MF_FORCE){
1668 r = load_map(pr, map);
1669 if (r < 0 && opened){
1675 struct map * get_map(struct peer_req *pr, char *name, uint32_t namelen, uint32_t flags)
1678 struct peerd *peer = pr->peer;
1679 struct mapperd *mapper = __get_mapperd(peer);
1680 struct map *map = find_map(mapper, name, namelen);
1682 if (flags & MF_LOAD){
1683 map = create_map(mapper, name, namelen);
1686 r = open_load_map(pr, map, flags);
1688 do_dropcache(pr, map);
1694 } else if (map->flags & MF_MAP_DESTROYED){
1702 static int map_action(int (action)(struct peer_req *pr, struct map *map),
1703 struct peer_req *pr, char *name, uint32_t namelen, uint32_t flags)
1705 //struct peerd *peer = pr->peer;
1708 map = get_map(pr, name, namelen, flags);
1711 if (map->flags & MF_MAP_NOT_READY){
1712 wait_on_map(map, (map->flags & MF_MAP_NOT_READY));
1716 int r = action(pr, map);
1717 //always drop cache if map not read exclusively
1718 if (!(map->flags & MF_MAP_EXCLUSIVE))
1719 do_dropcache(pr, map);
1720 //maybe capture ref before and compare here?
1728 void * handle_info(struct peer_req *pr)
1730 struct peerd *peer = pr->peer;
1731 char *target = xseg_get_target(peer->xseg, pr->req);
1732 int r = map_action(do_info, pr, target, pr->req->targetlen, MF_LOAD);
1741 void * handle_clone(struct peer_req *pr)
1744 struct peerd *peer = pr->peer;
1745 struct xseg_request_clone *xclone = (struct xseg_request_clone *) xseg_get_data(peer->xseg, pr->req);
1750 if (xclone->targetlen){
1751 r = map_action(do_clone, pr, xclone->target, xclone->targetlen, MF_LOAD);
1757 char *target = xseg_get_target(peer->xseg, pr->req);
1758 XSEGLOG2(&lc, I, "Creating volume");
1759 map = get_map(pr, target, pr->req->targetlen, MF_LOAD);
1761 XSEGLOG2(&lc, E, "Volume %s exists", map->volume);
1762 if (map->ref <= 2) //initial one + one ref from __get_map
1763 do_dropcache(pr, map); //we are the only ones usining this map. Drop the cache.
1764 put_map(map); //matches get_map
1768 //create a new empty map of size
1769 map = create_map(mapper, target, pr->req->targetlen);
1774 map->size = xclone->size;
1775 //populate_map with zero objects;
1776 uint64_t nr_objs = xclone->size / block_size;
1777 if (xclone->size % block_size)
1780 struct map_node *map_nodes = calloc(nr_objs, sizeof(struct map_node));
1782 do_dropcache(pr, map); //Since we just created the map, dropping cache should be sufficient.
1787 for (i = 0; i < nr_objs; i++) {
1788 strncpy(map_nodes[i].object, zero_block, strlen(zero_block)); //this should be SHA256_DIGEST_SIZE *2
1789 map_nodes[i].objectlen = strlen(zero_block);
1790 map_nodes[i].object[map_nodes[i].objectlen] = 0; //NULL terminate
1791 map_nodes[i].flags = 0;
1792 map_nodes[i].objectidx = i;
1793 map_nodes[i].map = map;
1794 map_nodes[i].ref = 1;
1795 map_nodes[i].waiters = 0;
1796 map_nodes[i].cond = st_cond_new(); //FIXME errcheck;
1797 r = insert_object(map, &map_nodes[i]);
1799 do_dropcache(pr, map);
1804 r = write_map(pr, map);
1806 XSEGLOG2(&lc, E, "Cannot write map %s", map->volume);
1807 do_dropcache(pr, map);
1810 XSEGLOG2(&lc, I, "Volume %s created", map->volume);
1812 do_dropcache(pr, map); //drop cache here for consistency
1824 void * handle_mapr(struct peer_req *pr)
1826 struct peerd *peer = pr->peer;
1827 char *target = xseg_get_target(peer->xseg, pr->req);
1828 int r = map_action(do_mapr, pr, target, pr->req->targetlen, MF_LOAD|MF_EXCLUSIVE);
1837 void * handle_mapw(struct peer_req *pr)
1839 struct peerd *peer = pr->peer;
1840 char *target = xseg_get_target(peer->xseg, pr->req);
1841 int r = map_action(do_mapw, pr, target, pr->req->targetlen, MF_LOAD|MF_EXCLUSIVE|MF_FORCE);
1846 XSEGLOG2(&lc, D, "Ta: %d", ta);
1851 void * handle_destroy(struct peer_req *pr)
1853 struct peerd *peer = pr->peer;
1854 char *target = xseg_get_target(peer->xseg, pr->req);
1855 int r = map_action(do_destroy, pr, target, pr->req->targetlen, MF_LOAD|MF_EXCLUSIVE|MF_FORCE);
1864 void * handle_close(struct peer_req *pr)
1866 struct peerd *peer = pr->peer;
1867 char *target = xseg_get_target(peer->xseg, pr->req);
1868 //here we do not want to load
1869 int r = map_action(do_close, pr, target, pr->req->targetlen, MF_EXCLUSIVE|MF_FORCE);
1878 int dispatch_accepted(struct peerd *peer, struct peer_req *pr,
1879 struct xseg_request *req)
1881 //struct mapperd *mapper = __get_mapperd(peer);
1882 struct mapper_io *mio = __get_mapper_io(pr);
1883 void *(*action)(struct peer_req *) = NULL;
1885 mio->state = ACCEPTED;
1888 switch (pr->req->op) {
1889 /* primary xseg operations of mapper */
1890 case X_CLONE: action = handle_clone; break;
1891 case X_MAPR: action = handle_mapr; break;
1892 case X_MAPW: action = handle_mapw; break;
1893 // case X_SNAPSHOT: handle_snap(peer, pr, req); break;
1894 case X_INFO: action = handle_info; break;
1895 case X_DELETE: action = handle_destroy; break;
1896 case X_CLOSE: action = handle_close; break;
1897 default: fprintf(stderr, "mydispatch: unknown up\n"); break;
1902 st_thread_create(action, pr, 0, 0);
1908 int dispatch(struct peerd *peer, struct peer_req *pr, struct xseg_request *req,
1909 enum dispatch_reason reason)
1911 struct mapperd *mapper = __get_mapperd(peer);
1913 struct mapper_io *mio = __get_mapper_io(pr);
1917 if (reason == dispatch_accept)
1918 dispatch_accepted(peer, pr, req);
1929 int custom_peer_init(struct peerd *peer, int argc, char *argv[])
1932 unsigned char buf[SHA256_DIGEST_SIZE];
1933 unsigned char *zero;
1935 gcry_control (GCRYCTL_SET_THREAD_CBS, &gcry_threads_pthread);
1937 /* Version check should be the very first call because it
1938 makes sure that important subsystems are intialized. */
1939 gcry_check_version (NULL);
1941 /* Disable secure memory. */
1942 gcry_control (GCRYCTL_DISABLE_SECMEM, 0);
1944 /* Tell Libgcrypt that initialization has completed. */
1945 gcry_control (GCRYCTL_INITIALIZATION_FINISHED, 0);
1947 /* calculate out magic sha hash value */
1948 gcry_md_hash_buffer(GCRY_MD_SHA256, magic_sha256, magic_string, strlen(magic_string));
1950 /* calculate zero block */
1951 //FIXME check hash value
1952 zero = malloc(block_size);
1953 memset(zero, 0, block_size);
1954 gcry_md_hash_buffer(GCRY_MD_SHA256, buf, zero, block_size);
1955 for (i = 0; i < SHA256_DIGEST_SIZE; ++i)
1956 sprintf(zero_block + 2*i, "%02x", buf[i]);
1957 printf("%s \n", zero_block);
1960 //FIXME error checks
1961 struct mapperd *mapperd = malloc(sizeof(struct mapperd));
1962 peer->priv = mapperd;
1964 mapper->hashmaps = xhash_new(3, STRING);
1966 for (i = 0; i < peer->nr_ops; i++) {
1967 struct mapper_io *mio = malloc(sizeof(struct mapper_io));
1968 mio->copyups_nodes = xhash_new(3, INTEGER);
1972 peer->peer_reqs[i].priv = mio;
1975 for (i = 0; i < argc; i++) {
1976 if (!strcmp(argv[i], "-bp") && (i+1) < argc){
1977 mapper->bportno = atoi(argv[i+1]);
1981 if (!strcmp(argv[i], "-mbp") && (i+1) < argc){
1982 mapper->mbportno = atoi(argv[i+1]);
1986 /* enforce only one thread */
1987 if (!strcmp(argv[i], "-t") && (i+1) < argc){
1988 int t = atoi(argv[i+1]);
1990 printf("ERROR: mapperd supports only one thread for the moment\nExiting ...\n");
1998 const struct sched_param param = { .sched_priority = 99 };
1999 sched_setscheduler(syscall(SYS_gettid), SCHED_FIFO, ¶m);
2007 void print_obj(struct map_node *mn)
2009 fprintf(stderr, "[%llu]object name: %s[%u] exists: %c\n",
2010 (unsigned long long) mn->objectidx, mn->object,
2011 (unsigned int) mn->objectlen,
2012 (mn->flags & MF_OBJECT_EXIST) ? 'y' : 'n');
2015 void print_map(struct map *m)
2017 uint64_t nr_objs = m->size/block_size;
2018 if (m->size % block_size)
2020 fprintf(stderr, "Volume name: %s[%u], size: %llu, nr_objs: %llu\n",
2021 m->volume, m->volumelen,
2022 (unsigned long long) m->size,
2023 (unsigned long long) nr_objs);
2025 struct map_node *mn;
2026 if (nr_objs > 1000000) //FIXME to protect against invalid volume size
2028 for (i = 0; i < nr_objs; i++) {
2029 mn = find_object(m, i);
2031 printf("object idx [%llu] not found!\n", (unsigned long long) i);
2039 void test_map(struct peerd *peer)
2042 //struct sha256_ctx sha256ctx;
2043 unsigned char buf[SHA256_DIGEST_SIZE];
2044 char buf_new[XSEG_MAX_TARGETLEN + 20];
2045 struct map *m = malloc(sizeof(struct map));
2046 strncpy(m->volume, "012345678901234567890123456789ab012345678901234567890123456789ab", XSEG_MAX_TARGETLEN + 1);
2047 m->volume[XSEG_MAX_TARGETLEN] = 0;
2048 strncpy(buf_new, m->volume, XSEG_MAX_TARGETLEN);
2049 buf_new[XSEG_MAX_TARGETLEN + 19] = 0;
2050 m->volumelen = XSEG_MAX_TARGETLEN;
2051 m->size = 100*block_size;
2052 m->objects = xhash_new(3, INTEGER);
2053 struct map_node *map_node = calloc(100, sizeof(struct map_node));
2054 for (i = 0; i < 100; i++) {
2055 sprintf(buf_new +XSEG_MAX_TARGETLEN, "%u", i);
2056 gcry_md_hash_buffer(GCRY_MD_SHA256, buf, buf_new, strlen(buf_new));
2058 for (j = 0; j < SHA256_DIGEST_SIZE; j++) {
2059 sprintf(map_node[i].object + 2*j, "%02x", buf[j]);
2061 map_node[i].objectidx = i;
2062 map_node[i].objectlen = XSEG_MAX_TARGETLEN;
2063 map_node[i].flags = MF_OBJECT_EXIST;
2064 ret = insert_object(m, &map_node[i]);
2067 char *data = malloc(block_size);
2068 mapheader_to_map(m, data);
2069 uint64_t pos = mapheader_size;
2071 for (i = 0; i < 100; i++) {
2072 map_node = find_object(m, i);
2074 printf("no object node %d \n", i);
2077 object_to_map(data+pos, map_node);
2078 pos += objectsize_in_map;
2082 struct map *m2 = malloc(sizeof(struct map));
2083 strncpy(m2->volume, "012345678901234567890123456789ab012345678901234567890123456789ab", XSEG_MAX_TARGETLEN +1);
2084 m->volume[XSEG_MAX_TARGETLEN] = 0;
2085 m->volumelen = XSEG_MAX_TARGETLEN;
2087 m2->objects = xhash_new(3, INTEGER);
2088 ret = read_map(peer, m2, data);
2091 int fd = open(m->volume, O_CREAT|O_WRONLY, S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH);
2093 while (sum < block_size) {
2094 r = write(fd, data + sum, block_size -sum);
2097 printf("write error\n");
2103 map_node = find_object(m, 0);