[sheepdog] [PATCH v2 7/9] object cache: reclaim cached objects when cache reaches the max size

Liu Yuan namei.unix at gmail.com
Thu Jul 26 05:27:24 CEST 2012


On 07/25/2012 08:15 PM, levin li wrote:
> +#define list_for_each_entry_revert_safe_rcu(pos, n, head, member) \
> +	for (pos = cds_list_entry(rcu_dereference((head)->prev), \
> +				  typeof(*pos), member), \
> +	     n = cds_list_entry(rcu_dereference(pos->member.prev),\
> +				typeof(*pos), member); \
> +				&pos->member != (head); \
> +	     pos = n, n = cds_list_entry(rcu_dereference(pos->member.prev),\
> +					 typeof(*pos), member))
>  

This should be placed in list.h

>  struct global_cache {
>  	uint64_t cache_size;
> @@ -50,6 +64,7 @@ struct global_cache {
>  
>  struct object_cache {
>  	uint32_t vid;
> +	int flushing;

uint8_t is enough.

> +
> +static int check_cache_status(struct object_cache *oc,
> +			      struct object_cache_entry *entry)
> +{
> +	int refcnt = uatomic_read(&entry->refcnt);
> +
> +	if (cache_is_flushing(oc)) {
> +		dprintf("cache %" PRIx32 " is flushing, don't reclaim it.\n",
> +			oc->vid);
> +		return SD_RES_CACHE_FLUSHING;
> +	}
> +
> +	/* If entry is being accessed, we don't reclaim it */
> +	if (refcnt > 0) {

uatomic_read(&entry->refcnt); should be placed in if clause.

> +		dprintf("cache object %" PRIx32 "(%08" PRIx32 ") "
> +			"can't be reclaimed, refcnt: %d\n",
> +			oc->vid, entry->idx, refcnt);
> +		return SD_RES_CACHE_REFERENCING;
> +	}
> +
> +	return SD_RES_SUCCESS;
> +}
> +
> +static int reclaim_object(struct object_cache_entry *entry)
> +{
> +	struct object_cache *oc = entry->oc;
> +	uint32_t idx = entry->idx;
> +	int ret = SD_RES_SUCCESS;
> +
> +	pthread_rwlock_wrlock(&oc->lock);
> +	dprintf("reclaiming /%06"PRIx32"/%08"PRIx32", cache_size: %ld\n",
> +		oc->vid, idx, uatomic_read(&sys_cache.cache_size));
> +
> +	ret = check_cache_status(oc, entry);
> +	if (ret != SD_RES_SUCCESS)
> +		goto out;
> +
> +	if (entry_is_dirty(entry)) {
> +		uint64_t bmap = entry->bmap;
> +		int create = entry->flags & ENTRY_CREATE_BIT;
> +
> +		entry->bmap = 0;
> +		del_from_dirty_tree_and_list(entry, &oc->dirty_tree);
> +		pthread_rwlock_unlock(&oc->lock);
> +
> +		ret = push_cache_object(oc->vid, idx, bmap, create);
> +
> +		pthread_rwlock_wrlock(&oc->lock);
> +		if (ret == SD_RES_SUCCESS) {
> +			entry->flags &= ~ENTRY_CREATE_BIT;
> +			/* dirty again */
> +			if (entry_is_dirty(entry)) {
> +				dprintf("object cache is dirty again %06" PRIx32 "\n", idx);
> +				ret = SD_RES_CACHE_REFERENCING;
> +				goto out;
> +			}
> +		} else {
> +			/* still dirty */
> +			entry->bmap |= bmap;
> +			ret = SD_RES_CACHE_REFERENCING;
> +			goto out;
> +		}
> +
> +		/* Now we get lock again, check cache status again */
> +		ret = check_cache_status(oc, entry);
> +		if (ret != SD_RES_SUCCESS)
> +			goto out;
> +	}
> +
> +	/* Mark the entry as being reclaimed */
> +	entry_start_reclaiming(entry);

rename entry_start_reclaiming as mark_entry_dirty() instead of adding
another comment.

> +
> +	ret = remove_cache_object(oc, idx);
> +	if (ret == SD_RES_SUCCESS)
> +		del_from_object_tree_and_list(entry, &oc->object_tree);
> +out:
> +	pthread_rwlock_unlock(&oc->lock);
> +	return ret;
> +}
> +
> +static void reclaim_work(struct work *work)
> +{
> +	struct object_cache_entry *entry, *n;
> +	int ret;
> +
> +	if (node_in_recovery())
> +		return;
> +
> +	list_for_each_entry_revert_safe_rcu(entry, n,
> +		       &sys_cache.cache_lru_list, lru_list) {
> +		/* Reclaim cache to 80% of max size */
> +		if (uatomic_read(&sys_cache.cache_size) <=
> +		    sys->cache_size * 8 / 10)
> +			break;
> +
> +		ret = reclaim_object(entry);

check if (ret == SD_RES_CACHE_FLUSHING) first, then we don't need else
clause.

> +		if (ret == SD_RES_SUCCESS) {
> +			unsigned data_length;
> +			if (idx_has_vdi_bit(entry->idx))
> +				data_length = SD_INODE_SIZE / 1024;
> +			else
> +				data_length = SD_DATA_OBJ_SIZE / 1024;
> +

Why 1024? can't we simply cache_size - data_length?

> +			uatomic_sub(&sys_cache.cache_size, data_length);
> +			free(entry);
> +		} else if (ret == SD_RES_CACHE_FLUSHING)
> +			/* If cache is flushing, stop reclaiming. */
> +			break;
> +	}
> +
> +	dprintf("cache reclaim complete\n");
> +}
> +
> +static void reclaim_done(struct work *work)
> +{
> +	uatomic_set(&sys_cache.reclaiming, 0);
> +	free(work);
> +}
> +
>  static struct object_cache_entry *
>  dirty_tree_insert(struct object_cache *oc, uint32_t idx,
>  		  uint64_t bmap, int create)
> @@ -183,6 +406,8 @@ dirty_tree_insert(struct object_cache *oc, uint32_t idx,
>  		else {
>  			/* already has this entry, merge bmap */
>  			entry->bmap |= bmap;
> +			if (create)
> +				entry->flags |= ENTRY_CREATE_BIT;
>  			return entry;
>  		}
>  	}
> @@ -192,7 +417,8 @@ dirty_tree_insert(struct object_cache *oc, uint32_t idx,
>  		return NULL;
>  
>  	entry->bmap |= bmap;
> -	entry->flags |= ENTRY_CREATE_BIT;
> +	if (create)
> +		entry->flags |= ENTRY_CREATE_BIT;
>  	rb_link_node(&entry->dirty_node, parent, p);
>  	rb_insert_color(&entry->dirty_node, &oc->dirty_tree);
>  	list_add(&entry->list, &oc->dirty_list);
> @@ -200,26 +426,6 @@ dirty_tree_insert(struct object_cache *oc, uint32_t idx,
>  	return entry;
>  }
>  
> -static struct object_cache_entry *dirty_tree_search(struct rb_root *root,
> -						    uint32_t idx)
> -{
> -	struct rb_node *n = root->rb_node;
> -	struct object_cache_entry *t;
> -
> -	while (n) {
> -		t = rb_entry(n, struct object_cache_entry, dirty_node);
> -
> -		if (idx < t->idx)
> -			n = n->rb_left;
> -		else if (idx > t->idx)
> -			n = n->rb_right;
> -		else
> -			return t; /* found it */
> -	}
> -
> -	return NULL;
> -}
> -
>  static int create_dir_for(uint32_t vid)
>  {
>  	int ret = 0;
> @@ -257,6 +463,7 @@ not_found:
>  	if (create) {
>  		cache = xzalloc(sizeof(*cache));
>  		cache->vid = vid;
> +		cache->object_tree = RB_ROOT;
>  		create_dir_for(vid);
>  
>  		cache->dirty_tree = RB_ROOT;
> @@ -271,14 +478,6 @@ out:
>  	return cache;
>  }
>  
> -static inline void
> -del_from_dirty_tree_and_list(struct object_cache_entry *entry,
> -			     struct rb_root *dirty_tree)
> -{
> -	rb_erase(&entry->dirty_node, dirty_tree);
> -	list_del(&entry->list);
> -}
> -
>  /* Caller should hold the oc->lock */
>  static inline void
>  add_to_dirty_tree_and_list(struct object_cache *oc, uint32_t idx,
> @@ -289,6 +488,9 @@ add_to_dirty_tree_and_list(struct object_cache *oc, uint32_t idx,
>  	if (!entry)
>  		panic("Can not find object entry %" PRIx32 "\n", idx);
>  
> +	if (cache_is_reclaiming())

should be renamed as cache_in_reclaim()

> +		return;
> +
>  	/* If cache isn't in reclaiming, move it
>  	 * to the head of lru list */
>  	cds_list_del_rcu(&entry->lru_list);
> @@ -321,6 +523,9 @@ static void add_to_object_cache(struct object_cache *oc, uint32_t idx)
>  	entry->idx = idx;
>  	CDS_INIT_LIST_HEAD(&entry->lru_list);
>  
> +	dprintf("cache object for vdi %" PRIx32 ", idx %08" PRIx32 "added\n",
> +		oc->vid, idx);
> +
>  	pthread_rwlock_wrlock(&oc->lock);
>  	old = object_cache_insert(&oc->object_tree, entry);
>  	if (!old) {
> @@ -331,20 +536,55 @@ static void add_to_object_cache(struct object_cache *oc, uint32_t idx)
>  		entry = old;
>  	}
>  	pthread_rwlock_unlock(&oc->lock);
> +
> +	dprintf("sys_cache.cache_size %" PRIx64 ", sys->cache_size %" PRIx64 "\n",
> +		uatomic_read(&sys_cache.cache_size), sys->cache_size);
> +	if (sys->cache_size &&
> +	    uatomic_read(&sys_cache.cache_size) > sys->cache_size &&
> +	    !cache_is_reclaiming()) {
> +		struct work *work = xzalloc(sizeof(struct work));
> +		uatomic_set(&sys_cache.reclaiming, 1);
> +		work->fn = reclaim_work;
> +		work->done = reclaim_done;
> +		queue_work(sys->reclaim_wqueue, work);
> +	}
> +}
> +
> +static inline int cache_sanity_check(struct object_cache *oc, uint32_t idx,
> +				     struct object_cache_entry **entry)
> +{
> +	struct object_cache_entry *ent;
> +
> +	ent = object_tree_search(&oc->object_tree, idx);
> +
> +	if (!ent || entry_is_reclaiming(ent))
> +		return SD_RES_NO_CACHE;
> +
> +	if (entry)
> +		*entry = ent;
> +
> +	return SD_RES_SUCCESS;
>  }

better renamed as get_cache_entry() which inc refcnt inside if
successful and add another helper find_cache_entry() to find the entry,
put_cache_entry() to dec the refcnt (other uninit work if any).

	get_cache_entry() {
 		entry = find_cache_entry();
		...
		inc refcnt..
		...
	}

Then find_cache_entry can be used by object_cache_lookup() too.

>  
>  static int object_cache_lookup(struct object_cache *oc, uint32_t idx,
>  			       int create)
>  {
>  	struct strbuf buf;
> -	int fd, ret = 0, flags = def_open_flags;
> +	int fd, ret = SD_RES_SUCCESS, flags = def_open_flags;
> +	unsigned data_length;
> +
> +	if (!create) {
> +		pthread_rwlock_wrlock(&oc->lock);
> +		ret = cache_sanity_check(oc, idx, NULL);
> +		pthread_rwlock_unlock(&oc->lock);
> +		return ret;
> +	}
>  
>  	strbuf_init(&buf, PATH_MAX);
>  	strbuf_addstr(&buf, cache_dir);
>  	strbuf_addf(&buf, "/%06"PRIx32"/%08"PRIx32, oc->vid, idx);
>  
> -	if (create)
> -		flags |= O_CREAT | O_TRUNC;
> +	flags |= O_CREAT | O_TRUNC;
>  
>  	fd = open(buf.buf, flags, def_fmode);
>  	if (fd < 0) {
> @@ -352,26 +592,22 @@ static int object_cache_lookup(struct object_cache *oc, uint32_t idx,
>  		goto out;
>  	}
>  
> -	if (create) {
> -		unsigned data_length;
> -
> -		if (idx_has_vdi_bit(idx))
> -			data_length = SD_INODE_SIZE;
> -		else
> -			data_length = SD_DATA_OBJ_SIZE;
> +	if (idx_has_vdi_bit(idx))
> +		data_length = SD_INODE_SIZE;
> +	else
> +		data_length = SD_DATA_OBJ_SIZE;
>  
> -		ret = prealloc(fd, data_length);
> -		if (ret != SD_RES_SUCCESS)
> -			ret = -1;
> -		else {
> -			uint64_t bmap = UINT64_MAX;
> +	ret = prealloc(fd, data_length);
> +	if (ret != SD_RES_SUCCESS)
> +		ret = -1;
> +	else {
> +		uint64_t bmap = UINT64_MAX;
>  
> -			add_to_object_cache(oc, idx);
> +		add_to_object_cache(oc, idx);
>  
> -			pthread_rwlock_wrlock(&oc->lock);
> -			add_to_dirty_tree_and_list(oc, idx, bmap, 1);
> -			pthread_rwlock_unlock(&oc->lock);
> -		}
> +		pthread_rwlock_wrlock(&oc->lock);
> +		add_to_dirty_tree_and_list(oc, idx, bmap, 1);
> +		pthread_rwlock_unlock(&oc->lock);
>  	}
>  	close(fd);
>  out:
> @@ -651,20 +887,23 @@ static int object_cache_push(struct object_cache *oc)
>  		pthread_rwlock_rdlock(&oc->lock);
>  		bmap = entry->bmap;
>  		create = entry->flags & ENTRY_CREATE_BIT;
> +		entry->bmap = 0;
>  		pthread_rwlock_unlock(&oc->lock);
>  
>  		ret = push_cache_object(oc->vid, entry->idx, bmap, create);
> -		if (ret != SD_RES_SUCCESS)
> -			goto push_failed;
>  
>  		pthread_rwlock_wrlock(&oc->lock);
> +		if (ret != SD_RES_SUCCESS) {
> +			entry->bmap |= bmap;
> +			goto push_failed;
> +		}
> +		entry->flags &= ~ENTRY_CREATE_BIT;
>  		del_from_dirty_tree_and_list(entry, &oc->dirty_tree);
>  		pthread_rwlock_unlock(&oc->lock);
>  	}
>  	return ret;
>  
>  push_failed:
> -	pthread_rwlock_wrlock(&oc->lock);
>  	list_splice_init(&inactive_dirty_list, &oc->dirty_list);
>  	pthread_rwlock_unlock(&oc->lock);
>  
> @@ -681,10 +920,10 @@ int object_is_cached(uint64_t oid)
>  	if (!cache)
>  		return 0;
>  
> -	if (object_cache_lookup(cache, idx, 0) < 0)
> -		return 0;
> +	if (object_cache_lookup(cache, idx, 0) == SD_RES_SUCCESS)
> +		return 1;
>  	else
> -		return 1; /* found it */
> +		return 0;
>  }
>  
>  void object_cache_delete(uint32_t vid)
> @@ -716,6 +955,52 @@ void object_cache_delete(uint32_t vid)
>  
>  }
>  
> +static void object_cache_flush_begin(struct object_cache *oc)
> +{
> +	pthread_rwlock_wrlock(&oc->lock);
> +	oc->flushing = 1;
> +	pthread_rwlock_unlock(&oc->lock);
> +}
> +
> +static void object_cache_flush_end(struct object_cache *oc)
> +{
> +	pthread_rwlock_wrlock(&oc->lock);
> +	oc->flushing = 0;
> +	pthread_rwlock_unlock(&oc->lock);
> +}

Why can't flushing be atomic? Better fold these two helpers into callers.

> +
> +static int object_cache_access_begin(struct object_cache *cache, uint32_t idx,
> +				     struct object_cache_entry **entry)
> +{
> +	struct object_cache_entry *ent;
> +	int ret = SD_RES_SUCCESS;
> +
> +	if (entry)
> +		*entry = NULL;
> +
> +	pthread_rwlock_rdlock(&cache->lock);
> +	ret = cache_sanity_check(cache, idx, &ent);
> +	if (ret != SD_RES_SUCCESS) {
> +		/* The cache entry may be reclaimed, so try again. */
> +		pthread_rwlock_unlock(&cache->lock);
> +		dprintf("cache sanity error %d\n", ret);
> +		return ret;
> +	}
> +
> +	uatomic_inc(&ent->refcnt);
> +	pthread_rwlock_unlock(&cache->lock);
> +
> +	if (entry)
> +		*entry = ent;
> +
> +	return ret;
> +}
> +

It is cleaner to call access_begin/end() in read/write_cache_object().

> +static void object_cache_access_end(struct object_cache_entry *entry)
> +{
> +	uatomic_dec(&entry->refcnt);
> +}
> +
>  static int object_cache_flush_and_delete(struct object_cache *oc)
>  {
>  	DIR *dir;
> @@ -726,6 +1011,8 @@ static int object_cache_flush_and_delete(struct object_cache *oc)
>  	struct strbuf p;
>  	int ret = 0;
>  
> +	object_cache_flush_begin(oc);
> +
>  	strbuf_init(&p, PATH_MAX);
>  	strbuf_addstr(&p, cache_dir);
>  	strbuf_addf(&p, "/%06"PRIx32, vid);
> @@ -754,6 +1041,8 @@ static int object_cache_flush_and_delete(struct object_cache *oc)
>  	}
>  
>  	object_cache_delete(vid);
> +
> +	object_cache_flush_end(oc);
>  out:
>  	strbuf_release(&p);
>  	return ret;
> @@ -777,10 +1066,10 @@ int bypass_object_cache(struct request *req)
>  			/* For read requet, we can read cache if any */
>  			uint32_t idx = object_cache_oid_to_idx(oid);
>  
> -			if (object_cache_lookup(cache, idx, 0) < 0)
> -				return 1;
> -			else
> +			if (object_cache_lookup(cache, idx, 0) == 0)
>  				return 0;
> +			else
> +				return 1;
>  		}
>  	}
>  
> @@ -811,37 +1100,42 @@ int object_cache_handle_request(struct request *req)
>  	if (req->rq.opcode == SD_OP_CREATE_AND_WRITE_OBJ)
>  		create = 1;
>  
> -	if (object_cache_lookup(cache, idx, create) < 0) {
> +retry:
> +	ret = object_cache_lookup(cache, idx, create);
> +	if (ret == SD_RES_NO_CACHE) {
>  		ret = object_cache_pull(cache, idx);
>  		if (ret != SD_RES_SUCCESS)
>  			return ret;
>  	}
>  
> -	pthread_rwlock_rdlock(&cache->lock);
> -	entry = object_tree_search(&cache->object_tree, idx);
> -	pthread_rwlock_unlock(&cache->lock);
> +	ret = object_cache_access_begin(cache, idx, &entry);
> +	if (ret != SD_RES_SUCCESS) {
> +		dprintf("retrying... %d\n", ret);
> +		goto retry;
> +	}
>  
>  	if (hdr->flags & SD_FLAG_CMD_WRITE) {
>  		ret = write_cache_object(cache->vid, idx, req->data,
>  					 hdr->data_length, hdr->obj.offset);
>  		if (ret != SD_RES_SUCCESS)
> -			goto out;
> +			goto err;
>  		update_cache_entry(cache, idx, hdr->data_length,
>  				   hdr->obj.offset);
>  	} else {
>  		ret = read_cache_object(cache->vid, idx, req->data,
>  					hdr->data_length, hdr->obj.offset);
>  		if (ret != SD_RES_SUCCESS)
> -			goto out;
> +			goto err;
>  		req->rp.data_length = hdr->data_length;
>  
> -		if (entry) {
> +		if (entry && !cache_is_reclaiming()) {
>  			cds_list_del_rcu(&entry->lru_list);
>  			cds_list_add_rcu(&entry->lru_list,
>  					 &sys_cache.cache_lru_list);
>  		}
>  	}
> -out:
> +err:
> +	object_cache_access_end(entry);
>  	return ret;
>  }
>  
> @@ -851,13 +1145,23 @@ int object_cache_write(uint64_t oid, char *data, unsigned int datalen,
>  	uint32_t vid = oid_to_vid(oid);
>  	uint32_t idx = object_cache_oid_to_idx(oid);
>  	struct object_cache *cache;
> +	struct object_cache_entry *entry;
>  	int ret;
>  
>  	cache = find_object_cache(vid, 0);
>  
> +	dprintf("cache object write %" PRIx32 "\n", idx);
> +
> +	ret = object_cache_access_begin(cache, idx, &entry);
> +	if (ret != SD_RES_SUCCESS)
> +		panic("cache object %" PRIx32 " doesn't exist\n", idx);
> +
>  	ret = write_cache_object(vid, idx, data, datalen, offset);
>  	if (ret == SD_RES_SUCCESS)
>  		update_cache_entry(cache, idx, datalen, offset);
> +
> +	object_cache_access_end(entry);
> +
>  	return ret;
>  }
>  
> @@ -866,20 +1170,40 @@ int object_cache_read(uint64_t oid, char *data, unsigned int datalen,
>  {
>  	uint32_t vid = oid_to_vid(oid);
>  	uint32_t idx = object_cache_oid_to_idx(oid);
> +	struct object_cache *cache;
> +	struct object_cache_entry *entry;
> +	int ret;
> +
> +	cache = find_object_cache(vid, 0);
> +
> +	dprintf("cache object read %" PRIx32 "\n", idx);
> +
> +	ret = object_cache_access_begin(cache, idx, &entry);
> +	if (ret != SD_RES_SUCCESS)
> +		panic("cache object %" PRIx32 " doesn't exist\n", idx);
>  
> -	return read_cache_object(vid, idx, data, datalen, offset);
> +	ret = read_cache_object(vid, idx, data, datalen, offset);
> +
> +	object_cache_access_end(entry);
> +
> +	return ret;
>  }
>  
>  int object_cache_flush_vdi(struct request *req)
>  {
>  	uint32_t vid = oid_to_vid(req->rq.obj.oid);
>  	struct object_cache *cache;
> +	int ret;
>  
>  	cache = find_object_cache(vid, 0);
>  	if (!cache)
>  		return SD_RES_SUCCESS;
>  
> -	return object_cache_push(cache);
> +	object_cache_flush_begin(cache);
> +	ret = object_cache_push(cache);
> +	object_cache_flush_end(cache);
> +
> +	return ret;
>  }
>  
>  int object_cache_flush_and_del(struct request *req)
> @@ -888,8 +1212,10 @@ int object_cache_flush_and_del(struct request *req)
>  	struct object_cache *cache;
>  
>  	cache = find_object_cache(vid, 0);
> +
>  	if (cache && object_cache_flush_and_delete(cache) < 0)
>  		return SD_RES_EIO;
> +
>  	return SD_RES_SUCCESS;
>  }
>  
> @@ -1003,6 +1329,7 @@ int object_cache_init(const char *p)
>  
>  	CDS_INIT_LIST_HEAD(&sys_cache.cache_lru_list);
>  	uatomic_set(&sys_cache.cache_size, 0);
> +	uatomic_set(&sys_cache.reclaiming, 0);
>  
>  	ret = load_existing_cache();
>  err:
> diff --git a/sheep/sheep.c b/sheep/sheep.c
> index 8b78669..29270e9 100644
> --- a/sheep/sheep.c
> +++ b/sheep/sheep.c
> @@ -359,8 +359,9 @@ int main(int argc, char **argv)
>  	sys->deletion_wqueue = init_work_queue("deletion", true);
>  	sys->block_wqueue = init_work_queue("block", true);
>  	sys->sockfd_wqueue = init_work_queue("sockfd", true);
> +	sys->reclaim_wqueue = init_work_queue("reclaim", true);
>  	if (!sys->gateway_wqueue || !sys->io_wqueue ||!sys->recovery_wqueue ||
> -	    !sys->deletion_wqueue || !sys->block_wqueue)
> +	    !sys->deletion_wqueue || !sys->block_wqueue || !sys->reclaim_wqueue)
>  		exit(1);
>  
>  	ret = trace_init();
> diff --git a/sheep/sheep_priv.h b/sheep/sheep_priv.h
> index fe61411..2090d67 100644
> --- a/sheep/sheep_priv.h
> +++ b/sheep/sheep_priv.h
> @@ -125,6 +125,7 @@ struct cluster_info {
>  	struct work_queue *recovery_wqueue;
>  	struct work_queue *block_wqueue;
>  	struct work_queue *sockfd_wqueue;
> +	struct work_queue *reclaim_wqueue;
>  };
>  
>  struct siocb {
> diff --git a/sheep/store.c b/sheep/store.c
> index a05822d..4a65c9e 100644
> --- a/sheep/store.c
> +++ b/sheep/store.c
> @@ -492,9 +492,13 @@ int write_object(uint64_t oid, char *data, unsigned int datalen,
>  	struct sd_req hdr;
>  	int ret;
>  
> +retry:
>  	if (sys->enable_write_cache && object_is_cached(oid)) {
>  		ret = object_cache_write(oid, data, datalen, offset,
>  					 flags, create);
> +		if (ret == SD_RES_NO_CACHE)
> +			goto retry;
> +
>  		if (ret != 0) {
>  			eprintf("write cache failed %"PRIx64" %"PRIx32"\n",
>  				oid, ret);
> @@ -529,8 +533,12 @@ int read_object(uint64_t oid, char *data, unsigned int datalen,
>  	struct sd_req hdr;
>  	int ret;
>  
> +retry:
>  	if (sys->enable_write_cache && object_is_cached(oid)) {
>  		ret = object_cache_read(oid, data, datalen, offset);
> +		if (ret == SD_RES_NO_CACHE)
> +			goto retry;
> +
>  		if (ret != SD_RES_SUCCESS) {
>  			eprintf("try forward read %"PRIx64" %"PRIx32"\n",
>  				oid, ret);
> 




More information about the sheepdog mailing list