/xseg/peers/user/cached.c - Diff - Archipelago - Greek Research and Technology Network's projects

Revision d2d79e63 xseg/peers/user/cached.c

     #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>
     /* bucket states */
     #define INVALID 0
     #define LOADING 1
     #define VALID   2
     #define DIRTY   3
     #define WRITING 4
     #define bucket_readable(__status) \
     	(__status == VALID || __status == DIRTY || __status == WRITING)
     /* write policies */
     #define WRITETHROUGH 1
     #define WRITEBACK    2
     struct cache_entry {
     	xlock lock;
     	uint32_t ref;
     	unsigned char *data;
     	char name[XSEG_MAX_TARGETLEN + 1];
     	uint32_t *status //len: object_size/bucket_size
     };
     /* object states */
     #define INVALIDATED (1 << 0)
     /* cio states */
     #define CIO_FAILED	1
     #define CIO_ACCEPTED	2
     #define CIO_READING	3
     #define BUCKET_SIZE_QUANTUM 4096
     struct cache_io {
     	uint32_t state;
     	xcache_handler h;
     	uint32_t pending_reqs;
     	struct work work;
     };
     struct cache {
     	xlock lock;
     struct cached {
     	xport bportno;
     	uint32_t cache_size;
     	struct xcache *cache;
     	uint64_t max_req_size;
     	uint64_t object_size;
     	uint64_t bucket_per_object;
     	uint32_t bucket_size;
     	uint32_t buckets_per_object;
     	int write_policy;
     	struct xq free_nodes;
     	xhash_t *entries;
     	struct cache_entry *nodes;
     	//scheduler
     };
     struct cached {
     	xport bp;
     	uint64_t max_req_size;
     	struct cache_io *ios;
     	struct xq free_ios;
     	//scheduler
     struct ce {
     	unsigned char *data;
     	uint32_t *status;
     	struct xwaitq *waitq;
     	uint32_t flags;
     	struct xlock lock;
     	struct xworkq workq;
     };
-...
     	return (struct cache_io *) pr->priv;
+    }
     static uint32_t __get_bucket(struct cached *cache, uint64_t offset)
+    {
     	return (offset / cache->bucket_size);
+    }
     void cache_entry_get(struct cache *cache, xqindex idx)
     static int is_loading(void *arg)
+    {
     	struct cache_entry *ce = cache->entries[idx];
     	ce->ref++;
     	uint32_t *status = (uint32_t *)arg;
     	return (*status == LOADING);
+    }
     void cache_entry_put()
     static int rw_range(struct peerd *peer, struct peer_req *pr, int action,
     		uint32_t start, uint32_t end)
+    {
     	struct cache_entry *ce = cache->entries[idx];
     	ce->ref--;
     	if (!ce->ref){
     		//write dirty buckets
     		//if no pending
     		//  free cache entry
+    	}
     	struct cached *cached = __get_cached(peer);
     	struct cache_io *cio = __get_cache_io(pr);
     	struct ce *ce = get_cache_entry(cached->cache, cio->h);
     	return 0;
+    }
     int cache_entry_init(struct cache *cache, xqindex idx, char *name)
     int on_init(void *c, void *e)
+    {
     	struct cache_entry *ce = cache->entries[idx];
     	xlock_release(&ce->lock);
     	ce->ref = 1;
     	memset(ce->data, 0, cache->object_size);
     	for (i = 0; i < cache->bucket_per_object; i++) {
     	uint32_t i;
     	struct cached *cached = (struct cached *)c;
     	struct ce *ce = (struct ce *)e;
     	ce->flags = 0;
     	memset(ce->data, 0, cached->object_size);
     	for (i = 0; i < cached->buckets_per_object; i++) {
     		ce->status[i] = INVALID;
+    	}
     	strncpy(ce->name, name, XSEG_MAX_TARGETLEN);
     	ce->name[XSEG_MAX_TARGETLEN] = 0;
     	xlock_release(&ce->lock);
     	return 0;
+    }
     int cache_init(struct cache *cache, uint32_t cache_size, uint64_t object_size,
     		uint64_t bucket_size, int write_policy)
     void on_put(void *c, void *e)
+    {
     	unsigned long nr_nodes = cache_size * 2;
     	xlock_release(&cache->lock);
     	cache->object_size = object_size;
     	cache->bucket_per_object = object_size/bucket_size;
     	cache->write_policy = write_policy;
     	cache->size = cache_size;
     	if (!xq_alloc_seq(&cache->free_nodes, nr_nodes, nr_nodes)){
     		return -1;
     	struct cached *cached = (struct cached *)c;
     	struct ce *ce = (struct ce *)e;
     	//since we are the last referrer to the cache entry
     	//no lock is needed.
     	uint32_t start, end, i = 0;
     	if (cached->write_policy == WRITETHROUGH || ce->flags & INVALIDATED)
     		return;
     	//write all dirty buckets.
     	while(i < cached->buckets_per_object){
     		if (ce->status[i] != DIRTY){
     			i++;
     			continue;
+    		}
     		start = i;
     		while (i < cached->buckets_per_object &&
     			(i-start)*cached->bucket_size < cached->max_req_size &&
     				ce->status[i] == DIRTY){
     			i++;
+    		}
     		end = i;
     		//problem: no assocciated pr
     		//maybe put one in cache entry
     		rw_range(cached, ce, 1, start, end);
+    	}
     	cache->entries = xhash_new(shift, cache_size, STRING);
     	if (!cache->entries){
     		return -1;
+    }
     void * init_node(void *c)
+    {
     	int i;
     	//TODO err check
     	struct cached *cached = (struct cached *)c;
     	struct ce *ce = malloc(sizeof(struct ce));
     	xlock_release(&ce->lock);
     	ce->data = malloc(sizeof(unsigned char) * cached->object_size);
     	ce->status = malloc(sizeof(uint32_t) * cached->buckets_per_object);
     	ce->waitq = malloc(sizeof(struct xwaitq) * cached->buckets_per_object);
     	for (i = 0; i < cached->buckets_per_object; i++) {
     		xwaitq_init(&ce->waitq[i], is_loading, &ce->status[i],
     				XWAIT_SIGNAL_ONE);
+    	}
     	cache->nodes = malloc(sizeof(struct cache_entry) * nr_nodes);
     	if (!cache->nodes){
     		return -1;
     	xworkq_init(&ce->workq, &ce->lock, 0);
     	return ce;
+    }
     struct xcache_ops c_ops = {
     	.on_init = on_init,
     	.on_put  = on_put,
     	.on_node_init = init_node
     };
     static uint32_t __get_next_invalid(struct ce *ce, uint32_t start,
     					uint32_t limit)
+    {
     	uint32_t end = start+1;
     	while (end <= limit && ce->status[end] == INVALID)
     		end++;
     	return end;
+    }
     static void cached_fail(struct peerd *peer, struct peer_req *pr)
+    {
     	struct cached *cached = __get_cached(peer);
     	struct cache_io *cio = __get_cache_io(pr);
     	if (cio->h != NoEntry){
     		xcache_put(cached->cache, cio->h);
+    	}
     	cio->h = NoEntry;
     	fail(peer, pr);
+    }
     	return 0;
     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);
+    }
     xqindex cache_lookup(struct cache *cache, char *name)
     static void handle_read(void *arg);
     //is this necessary?
     static void status_changed(void *arg)
+    {
     	return Noneidx;
     	/*
     	 * In this context we hold a reference to the cache entry.
+    	 *
     	 * This function gets called only after the bucket at which the
     	 * current peer_req is waiting, has finished loading of failed.
+    	 *
     	 * Assumptions:
     	 * 	Each pr waits only at one bucket at any time. That means that
     	 * 	under no circumstances, this function get called simutaneously
     	 * 	for the same pr.
     	 */
     	struct peer_req *pr = (struct peer_req *)arg;
     	struct peerd *peer = pr->peer;
     	struct cached *cached = __get_cached(peer);
     	struct cache_io *cio = __get_cache_io(pr);
     	struct ce *ce = get_cache_entry(cached->cache, cio->h);
     	if (xworkq_enqueue(&ce->workq, handle_read, (void *)pr) < 0){
     		//FAIL or mark as failed ? are we the last?
     		if (cio->pending_reqs){
     			// cannot put here, since there are outstanding reqs to
     			// be received.
     			// Simply mark pr as failed.
     			cio->state = CIO_FAILED;
     		} else {
     			//safe to fail here, since there is no pending action on
     			//this pr.
     			cached_fail(peer, pr);
+    		}
+    	}
+    }
     int __cache_insert(struct cache *cache, xqindex *idx)
     static void handle_read(void *arg)
+    {
     	return -1;
     	/*
     	 * In this context we hold a reference to the cache entry and
     	 * the assocciated lock
     	 */
     	struct peer_req *pr = (struct peer_req *)arg;
     	struct peerd *peer = pr->peer;
     	struct cached *cached = __get_cached(peer);
     	struct cache_io *cio = __get_cache_io(pr);
     	struct xseg_request *req = pr->req;
     	struct ce *ce = get_cache_entry(cached->cache, cio->h);
     	uint32_t start_bucket, end_bucket, next;
     	uint32_t i, b, limit;
     	uint32_t pending_buckets = 0;
     	if (cio->state == CIO_FAILED)
     		goto out;
     	b = __get_bucket(cached, req->offset);
     	limit = __get_bucket(cached, req->offset + req->size);
     	//assert limit < cached->object_size
     	for (i = b; i < limit; i++) {
     		if (bucket_readable(ce->status[i]))
     			continue;
     		if (ce->status[i] != LOADING){
     			start_bucket = i;
     			next = __get_next_invalid(ce, start_bucket, limit);
     			end_bucket = next -1;
     			i = next;
     			if (rw_range(peer, pr, 0, start_bucket, end_bucket) < 0){
     				cio->state = CIO_FAILED;
     				break;
+    			}
     			cio->pending_reqs++;
     			cio->state =  CIO_READING;
+    		}
     		pending_buckets++;;
+    	}
     	if (pending_buckets) {
     		/* Do not put cache entry yet */
     		cio->work.job_fn = handle_read;
     		cio->work.job = pr;
     		/* wait on the last bucket */
     		xwaitq_enqueue(&ce->waitq[end_bucket], &cio->work);
     		return;
+    	}
     out:
     	if (cio->state == CIO_FAILED){
     		if (!cio->pending_reqs)
     			cached_fail(peer, pr);
+    	}
     	else{
     		//serve req;
     		cached_complete(peer, pr);
+    	}
     	return;
+    }
     int cache_insert(struct cache *cache, xqindex *idx)
     static void handle_write(void *arg)
+    {
     	//if writeback
     	//  	for each bucket
     	//		write all buckets
     	//		mark them as dirty
     	//	cache_put(h)
     	//	complete
     	//else
     	//	send write to blocker
     	/*
     	 * In this context we hold a reference to the cache entry and
     	 * the assocciated lock
     	 */
     	int r;
     	xlock_acquire(&cache->lock, 1);
     	r = __cache_insert(cache, idx);
     	xlock_release(&cache->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 = get_cache_entry(cached->cache, cio->h);
     	(void)ce;
     	if (cached->write_policy == WRITETHROUGH){
     		//send write to blocker
     		//return
     	} else if (cached->write_policy == WRITEBACK) {
     		//for each bucket
     		//	write all buckets
     		//	mark them as dirty
     		r = 0;
     	} else {
     		r = -1;
+    	}
     out:
     	if (r < 0)
     		cached_fail(peer, pr);
     	else
     		cached_complete(peer, pr);
     	return;
+    }
     static int handle_readwrite(struct peerd *peer, struct peer_req *pr)
+    {
     	int r = -1;
     	struct ce *ce;
     	struct cached *cached = __get_cached(peer);
     	char name[XSEG_MAX_TARGETLEN + 1];
     	struct xseg_request *req = pr->req;
     	char *target = xseg_get_target(peer->xseg, req);
     	xcache_handler h = NoEntry;
     	strncpy(name, target, req->targetlen);
     	name[XSEG_MAX_TARGETLEN] = 0;
     	h = xcache_lookup(cached->cache, name);
     	if (h == NoEntry){
     		h = xcache_alloc_init(cached->cache, name);
     		if (h == NoEntry){
     			goto out;
+    		}
     		r = xcache_insert(cached->cache, h);
     		if (r < 0){
     			goto out;
+    		}
+    	}
     	ce = (struct ce *)get_cache_entry(cached->cache, h);
     	if (!ce){
     		r = -1;
     		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;
     		goto out;
+    	}
     out:
     	if (r < 0){
     		cached_fail(peer, pr);
+    	}
     	return r;
+    }
     int __cache_remove(struct cache *cache, xqindex *idx)
     struct req_completion{
     	struct peer_req *pr;
     	struct xseg_request *req;
     };
     static void complete_read(void *arg)
+    {
     	struct cache_entry *ce = cache->entries[idx];
     	return xhash_delete(cache->entries, (xhashidx)ce->name);
     	/*
     	 * 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 = get_cache_entry(cached->cache, cio->h);
     	uint32_t start, end, i;
     	int success;
     	char *data = xseg_get_data(peer->xseg, req);
     	/*
     	 * Synchronize pending_reqs of the cache_io here, since each cache_io
     	 * refers to only one object, and therefore we can use the object lock
     	 * to synchronize between receive contextes.
     	 */
     	cio->pending_reqs--;
     	success = (req->state == XS_SERVED && req->serviced == req->size);
     	if (!success)
     		cio->state = CIO_FAILED;
     	//assert (req->offset % cached->bucket_size) == 0;
     	//assert ((req->offset+req->serviced) % cached->bucket_size) == 0;
     	start = __get_bucket(cached, req->offset);
     	end = __get_bucket(cached, req->offset + req->serviced);
     	for (i = start; i < end; i++) {
     		if (ce->status[i] == LOADING){
     			if (success){
     				memcpy(ce->data+(i*cached->bucket_size), data,
     						cached->bucket_size);
     				ce->status[i] = VALID;
+    			}
     			else {
     				//reset status
     				ce->status[i] = INVALID;
+    			}
     			xwaitq_signal(&ce->waitq[i]);
+    		}
+    	}
     	free(rc);
+    }
     int cache_remove(struct cache *cache, xqindex *idx)
     void complete_write(void *arg)
+    {
     	int r;
     	xlock_acquire(&cache->lock, 1);
     	r = __cache_remove(cache, idx);
     	xlock_release(&cache->lock);
     	return r;
     	//for each bucket
     	//	if WRITETHROUGH
     	//		copy data to bucket
     	//		mark as valid
     	//	else if WRITEBACK
     	//		if status writing
     	//			mark as valid
     	//
     	/*
     	 * In this context we hold a reference to the cache entry and
     	 * the assocciated lock
     	 */
     	return;
+    }
     int cache_invalidate(struct cache *cache, char *name)
     static int handle_receive_read(struct peerd *peer, struct peer_req *pr,
     			struct xseg_request *req)
+    {
     	/*
     	 * Should be rentrant
     	 */
     	struct cached *cached = __get_cached(peer);
     	struct cache_io *cio = __get_cache_io(pr);
     	struct ce *ce = get_cache_entry(cached->cache, cio->h);
     	struct req_completion *rc;
     	rc = malloc(sizeof(struct req_completion));
     	rc->pr = pr;
     	rc->req = req;
     	if (xworkq_enqueue(&ce->workq, complete_read, (void *)rc) < 0){
     		free(rc);
     		//TODO WHAT?
+    	}
     	return 0;
+    }
     static int handle_receive_write(struct peerd *peer, struct peer_req *pr)
+    {
     	//enqueue_work
     	return 0;
+    }
     static int handle_delete(struct peerd *peer, struct peer_req *pr)
+    {
     	//h = cache_lookup
     	//if h
     	//	cio->h = h
     	//
     	//send delete to blocker
     	return 0;
+    }
     static int handle_receive_delete(struct peerd *peer, struct peer_req *pr)
+    {
     	//if success
     	//	if cio->h
     	//		//this should not write any dirty data
     	//		xcache_remove(h)
     	return 0;
+    }
     static int forward_req(struct peerd *peer, struct peer_req *pr)
+    {
     	//get request
     	//hijack target and data
     	//submit
     	return 0;
+    }
     static int handle_receive(struct peerd *peer, struct peer_req *pr,
     			struct xseg_request *req)
+    {
     	//if not read/write/delete
     	//	put req;
     	//	complete or fail pr;
     	return 0;
+    }
     int dispatch(struct peerd *peer, struct peer_req *pr, struct xseg_request *req,
     		enum dispatch_reason reason)
+    {
     	struct cached *cacher = __get_cached(peer);
     	(void) cacher;
     	struct cached *cached = __get_cached(peer);
     	(void) cached;
     	struct cache_io *cio = __get_cache_io(pr);
     	(void) cio;
     	switch reason {
     	switch (reason) {
     		case dispatch_accept:
     			cio->state = CIO_ACCEPTED;
     			break;
     		case dispatch_receive:
     			break;
-...
     	int i;
     	//FIXME error checks
     	struct cacherd *cacherd = malloc(sizeof(struct cacherd));
     	peer->priv = cacherd;
     	cacher = cacherd;
     	cacher->hashmaps = xhash_new(3, STRING);
     	struct cached *cached = malloc(sizeof(struct cached));
     	cached->cache = malloc(sizeof(struct xcache));
     	xcache_init(cached->cache, cached->cache_size, &c_ops, peer);
     	peer->priv = cached;
     	for (i = 0; i < peer->nr_ops; i++) {
     		struct mapper_io *mio = malloc(sizeof(struct mapper_io));
     		mio->copyups_nodes = xhash_new(3, INTEGER);
     		mio->copyups = 0;
     		mio->err = 0;
     		mio->active = 0;
     		struct cache_io *mio = malloc(sizeof(struct cache_io));
     		peer->peer_reqs[i].priv = mio;
+    	}
     	mapper->bportno = -1;
     	mapper->mbportno = -1;
     	uint32_t cache_size;
     	unsigned long bucket_size;
     	unsigned long object_size;
     	unsigned long max_req_size;
     	cached->bportno = -1;
     	cached->cache_size = -1;
     	cached->max_req_size = -1;
     	cached->object_size = -1;
     	cached->bucket_size = -1;
     	cached->write_policy = WRITETHROUGH;
     	BEGIN_READ_ARGS(argc, argv);
     	READ_ARG_ULONG("-bp", mapper->bportno);
     	READ_ARG_ULONG("-mbp", mapper->mbportno);
     	READ_ARG_ULONG("-bp", cached->bportno);
     	READ_ARG_ULONG("-cs", cached->cache_size);
     	READ_ARG_ULONG("-mrs", max_req_size);
     	READ_ARG_ULONG("-os", object_size); // In buckets
     	READ_ARG_ULONG("-bs", bucket_size); // In BUCKET_SIZE_QUANTUM
     //	READ_ARG_ULONG("-wcp", cached->write_policy);
     	END_READ_ARGS();
     	if (mapper->bportno == -1){
     	cached->bucket_size = BUCKET_SIZE_QUANTUM;
     	cached->max_req_size = 512 * 1024;
     	cached->object_size = 4 * 1024 * 1024;
     	if (cached->bportno == -1){
     		XSEGLOG2(&lc, E, "Portno for blocker must be provided");
     		usage(argv[0]);
     		return -1;
+    	}
     	if (mapper->mbportno == -1){
     		XSEGLOG2(&lc, E, "Portno for mblocker must be provided");
     		usage(argv[0]);
     		return -1;
+    	}
     	cached->buckets_per_object = cached->object_size / cached->bucket_size;
     	return 0;
+    }
     void custom_peer_finalize(struct peerd *peer)
+    {
     	//write dirty objects
     	//or cache_close(cached->cache);
     	return;
+    }

Also available in: Unified diff