[mirror_ubuntu-artful-kernel.git] / fs / fscache / operation.c

/* FS-Cache worker operation management routines
 *
 * Copyright (C) 2008 Red Hat, Inc. All Rights Reserved.
 * Written by David Howells (dhowells@redhat.com)
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version
 * 2 of the License, or (at your option) any later version.
 *
 * See Documentation/filesystems/caching/operations.txt
 */

#define FSCACHE_DEBUG_LEVEL OPERATION
#include <linux/module.h>
#include <linux/seq_file.h>
#include <linux/slab.h>
#include "internal.h"

atomic_t fscache_op_debug_id;
EXPORT_SYMBOL(fscache_op_debug_id);

/**
 * fscache_enqueue_operation - Enqueue an operation for processing
 * @op: The operation to enqueue
 *
 * Enqueue an operation for processing by the FS-Cache thread pool.
 *
 * This will get its own ref on the object.
 */
void fscache_enqueue_operation(struct fscache_operation *op)
{
	_enter("{OBJ%x OP%x,%u}",
	       op->object->debug_id, op->debug_id, atomic_read(&op->usage));

	ASSERT(list_empty(&op->pend_link));
	ASSERT(op->processor != NULL);
	ASSERTCMP(op->object->state, >=, FSCACHE_OBJECT_AVAILABLE);
	ASSERTCMP(atomic_read(&op->usage), >, 0);

	fscache_stat(&fscache_n_op_enqueue);
	switch (op->flags & FSCACHE_OP_TYPE) {
	case FSCACHE_OP_ASYNC:
		_debug("queue async");
		atomic_inc(&op->usage);
		if (!queue_work(fscache_op_wq, &op->work))
			fscache_put_operation(op);
		break;
	case FSCACHE_OP_MYTHREAD:
		_debug("queue for caller's attention");
		break;
	default:
		printk(KERN_ERR "FS-Cache: Unexpected op type %lx",
		       op->flags);
		BUG();
		break;
	}
}
EXPORT_SYMBOL(fscache_enqueue_operation);

/*
 * start an op running
 */
static void fscache_run_op(struct fscache_object *object,
			   struct fscache_operation *op)
{
	object->n_in_progress++;
	if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
		wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
	if (op->processor)
		fscache_enqueue_operation(op);
	fscache_stat(&fscache_n_op_run);
}

/*
 * submit an exclusive operation for an object
 * - other ops are excluded from running simultaneously with this one
 * - this gets any extra refs it needs on an op
 */
int fscache_submit_exclusive_op(struct fscache_object *object,
				struct fscache_operation *op)
{
	int ret;

	_enter("{OBJ%x OP%x},", object->debug_id, op->debug_id);

	spin_lock(&object->lock);
	ASSERTCMP(object->n_ops, >=, object->n_in_progress);
	ASSERTCMP(object->n_ops, >=, object->n_exclusive);
	ASSERT(list_empty(&op->pend_link));

	ret = -ENOBUFS;
	if (fscache_object_is_active(object)) {
		op->object = object;
		object->n_ops++;
		object->n_exclusive++;	/* reads and writes must wait */

		if (object->n_ops > 1) {
			atomic_inc(&op->usage);
			list_add_tail(&op->pend_link, &object->pending_ops);
			fscache_stat(&fscache_n_op_pend);
		} else if (!list_empty(&object->pending_ops)) {
			atomic_inc(&op->usage);
			list_add_tail(&op->pend_link, &object->pending_ops);
			fscache_stat(&fscache_n_op_pend);
			fscache_start_operations(object);
		} else {
			ASSERTCMP(object->n_in_progress, ==, 0);
			fscache_run_op(object, op);
		}

		/* need to issue a new write op after this */
		clear_bit(FSCACHE_OBJECT_PENDING_WRITE, &object->flags);
		ret = 0;
	} else if (object->state == FSCACHE_OBJECT_CREATING) {
		op->object = object;
		object->n_ops++;
		object->n_exclusive++;	/* reads and writes must wait */
		atomic_inc(&op->usage);
		list_add_tail(&op->pend_link, &object->pending_ops);
		fscache_stat(&fscache_n_op_pend);
		ret = 0;
	} else {
		/* not allowed to submit ops in any other state */
		BUG();
	}

	spin_unlock(&object->lock);
	return ret;
}

/*
 * report an unexpected submission
 */
static void fscache_report_unexpected_submission(struct fscache_object *object,
						 struct fscache_operation *op,
						 unsigned long ostate)
{
	static bool once_only;
	struct fscache_operation *p;
	unsigned n;

	if (once_only)
		return;
	once_only = true;

	kdebug("unexpected submission OP%x [OBJ%x %s]",
	       op->debug_id, object->debug_id,
	       fscache_object_states[object->state]);
	kdebug("objstate=%s [%s]",
	       fscache_object_states[object->state],
	       fscache_object_states[ostate]);
	kdebug("objflags=%lx", object->flags);
	kdebug("objevent=%lx [%lx]", object->events, object->event_mask);
	kdebug("ops=%u inp=%u exc=%u",
	       object->n_ops, object->n_in_progress, object->n_exclusive);

	if (!list_empty(&object->pending_ops)) {
		n = 0;
		list_for_each_entry(p, &object->pending_ops, pend_link) {
			ASSERTCMP(p->object, ==, object);
			kdebug("%p %p", op->processor, op->release);
			n++;
		}

		kdebug("n=%u", n);
	}

	dump_stack();
}

/*
 * submit an operation for an object
 * - objects may be submitted only in the following states:
 *   - during object creation (write ops may be submitted)
 *   - whilst the object is active
 *   - after an I/O error incurred in one of the two above states (op rejected)
 * - this gets any extra refs it needs on an op
 */
int fscache_submit_op(struct fscache_object *object,
		      struct fscache_operation *op)
{
	unsigned long ostate;
	int ret;

	_enter("{OBJ%x OP%x},{%u}",
	       object->debug_id, op->debug_id, atomic_read(&op->usage));

	ASSERTCMP(atomic_read(&op->usage), >, 0);

	spin_lock(&object->lock);
	ASSERTCMP(object->n_ops, >=, object->n_in_progress);
	ASSERTCMP(object->n_ops, >=, object->n_exclusive);
	ASSERT(list_empty(&op->pend_link));

	ostate = object->state;
	smp_rmb();

	if (fscache_object_is_active(object)) {
		op->object = object;
		object->n_ops++;

		if (object->n_exclusive > 0) {
			atomic_inc(&op->usage);
			list_add_tail(&op->pend_link, &object->pending_ops);
			fscache_stat(&fscache_n_op_pend);
		} else if (!list_empty(&object->pending_ops)) {
			atomic_inc(&op->usage);
			list_add_tail(&op->pend_link, &object->pending_ops);
			fscache_stat(&fscache_n_op_pend);
			fscache_start_operations(object);
		} else {
			ASSERTCMP(object->n_exclusive, ==, 0);
			fscache_run_op(object, op);
		}
		ret = 0;
	} else if (object->state == FSCACHE_OBJECT_CREATING) {
		op->object = object;
		object->n_ops++;
		atomic_inc(&op->usage);
		list_add_tail(&op->pend_link, &object->pending_ops);
		fscache_stat(&fscache_n_op_pend);
		ret = 0;
	} else if (object->state == FSCACHE_OBJECT_DYING ||
		   object->state == FSCACHE_OBJECT_LC_DYING ||
		   object->state == FSCACHE_OBJECT_WITHDRAWING) {
		fscache_stat(&fscache_n_op_rejected);
		ret = -ENOBUFS;
	} else if (!test_bit(FSCACHE_IOERROR, &object->cache->flags)) {
		fscache_report_unexpected_submission(object, op, ostate);
		ASSERT(!fscache_object_is_active(object));
		ret = -ENOBUFS;
	} else {
		ret = -ENOBUFS;
	}

	spin_unlock(&object->lock);
	return ret;
}

/*
 * queue an object for withdrawal on error, aborting all following asynchronous
 * operations
 */
void fscache_abort_object(struct fscache_object *object)
{
	_enter("{OBJ%x}", object->debug_id);

	fscache_raise_event(object, FSCACHE_OBJECT_EV_ERROR);
}

/*
 * jump start the operation processing on an object
 * - caller must hold object->lock
 */
void fscache_start_operations(struct fscache_object *object)
{
	struct fscache_operation *op;
	bool stop = false;

	while (!list_empty(&object->pending_ops) && !stop) {
		op = list_entry(object->pending_ops.next,
				struct fscache_operation, pend_link);

		if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
			if (object->n_in_progress > 0)
				break;
			stop = true;
		}
		list_del_init(&op->pend_link);
		fscache_run_op(object, op);

		/* the pending queue was holding a ref on the object */
		fscache_put_operation(op);
	}

	ASSERTCMP(object->n_in_progress, <=, object->n_ops);

	_debug("woke %d ops on OBJ%x",
	       object->n_in_progress, object->debug_id);
}

/*
 * cancel an operation that's pending on an object
 */
int fscache_cancel_op(struct fscache_operation *op)
{
	struct fscache_object *object = op->object;
	int ret;

	_enter("OBJ%x OP%x}", op->object->debug_id, op->debug_id);

	spin_lock(&object->lock);

	ret = -EBUSY;
	if (!list_empty(&op->pend_link)) {
		fscache_stat(&fscache_n_op_cancelled);
		list_del_init(&op->pend_link);
		object->n_ops--;
		if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
			object->n_exclusive--;
		if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
			wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
		fscache_put_operation(op);
		ret = 0;
	}

	spin_unlock(&object->lock);
	_leave(" = %d", ret);
	return ret;
}

/*
 * release an operation
 * - queues pending ops if this is the last in-progress op
 */
void fscache_put_operation(struct fscache_operation *op)
{
	struct fscache_object *object;
	struct fscache_cache *cache;

	_enter("{OBJ%x OP%x,%d}",
	       op->object->debug_id, op->debug_id, atomic_read(&op->usage));

	ASSERTCMP(atomic_read(&op->usage), >, 0);

	if (!atomic_dec_and_test(&op->usage))
		return;

	_debug("PUT OP");
	if (test_and_set_bit(FSCACHE_OP_DEAD, &op->flags))
		BUG();

	fscache_stat(&fscache_n_op_release);

	if (op->release) {
		op->release(op);
		op->release = NULL;
	}

	object = op->object;

	if (test_bit(FSCACHE_OP_DEC_READ_CNT, &op->flags))
		atomic_dec(&object->n_reads);

	/* now... we may get called with the object spinlock held, so we
	 * complete the cleanup here only if we can immediately acquire the
	 * lock, and defer it otherwise */
	if (!spin_trylock(&object->lock)) {
		_debug("defer put");
		fscache_stat(&fscache_n_op_deferred_release);

		cache = object->cache;
		spin_lock(&cache->op_gc_list_lock);
		list_add_tail(&op->pend_link, &cache->op_gc_list);
		spin_unlock(&cache->op_gc_list_lock);
		schedule_work(&cache->op_gc);
		_leave(" [defer]");
		return;
	}

	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
		ASSERTCMP(object->n_exclusive, >, 0);
		object->n_exclusive--;
	}

	ASSERTCMP(object->n_in_progress, >, 0);
	object->n_in_progress--;
	if (object->n_in_progress == 0)
		fscache_start_operations(object);

	ASSERTCMP(object->n_ops, >, 0);
	object->n_ops--;
	if (object->n_ops == 0)
		fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);

	spin_unlock(&object->lock);

	kfree(op);
	_leave(" [done]");
}
EXPORT_SYMBOL(fscache_put_operation);

/*
 * garbage collect operations that have had their release deferred
 */
void fscache_operation_gc(struct work_struct *work)
{
	struct fscache_operation *op;
	struct fscache_object *object;
	struct fscache_cache *cache =
		container_of(work, struct fscache_cache, op_gc);
	int count = 0;

	_enter("");

	do {
		spin_lock(&cache->op_gc_list_lock);
		if (list_empty(&cache->op_gc_list)) {
			spin_unlock(&cache->op_gc_list_lock);
			break;
		}

		op = list_entry(cache->op_gc_list.next,
				struct fscache_operation, pend_link);
		list_del(&op->pend_link);
		spin_unlock(&cache->op_gc_list_lock);

		object = op->object;

		_debug("GC DEFERRED REL OBJ%x OP%x",
		       object->debug_id, op->debug_id);
		fscache_stat(&fscache_n_op_gc);

		ASSERTCMP(atomic_read(&op->usage), ==, 0);

		spin_lock(&object->lock);
		if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
			ASSERTCMP(object->n_exclusive, >, 0);
			object->n_exclusive--;
		}

		ASSERTCMP(object->n_in_progress, >, 0);
		object->n_in_progress--;
		if (object->n_in_progress == 0)
			fscache_start_operations(object);

		ASSERTCMP(object->n_ops, >, 0);
		object->n_ops--;
		if (object->n_ops == 0)
			fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);

		spin_unlock(&object->lock);

	} while (count++ < 20);

	if (!list_empty(&cache->op_gc_list))
		schedule_work(&cache->op_gc);

	_leave("");
}

/*
 * execute an operation using fs_op_wq to provide processing context -
 * the caller holds a ref to this object, so we don't need to hold one
 */
void fscache_op_work_func(struct work_struct *work)
{
	struct fscache_operation *op =
		container_of(work, struct fscache_operation, work);
	unsigned long start;

	_enter("{OBJ%x OP%x,%d}",
	       op->object->debug_id, op->debug_id, atomic_read(&op->usage));

	ASSERT(op->processor != NULL);
	start = jiffies;
	op->processor(op);
	fscache_hist(fscache_ops_histogram, start);
	fscache_put_operation(op);

	_leave("");
}
Commit	Line	Data
952efe7b DH	1	/* FS-Cache worker operation management routines
	2	*
	3	* Copyright (C) 2008 Red Hat, Inc. All Rights Reserved.
	4	* Written by David Howells (dhowells@redhat.com)
	5	*
	6	* This program is free software; you can redistribute it and/or
	7	* modify it under the terms of the GNU General Public License
	8	* as published by the Free Software Foundation; either version
	9	* 2 of the License, or (at your option) any later version.
	10	*
	11	* See Documentation/filesystems/caching/operations.txt
	12	*/
	13
	14	#define FSCACHE_DEBUG_LEVEL OPERATION
	15	#include <linux/module.h>
440f0aff	16	#include <linux/seq_file.h>
5a0e3ad6	17	#include <linux/slab.h>
952efe7b DH	18	#include "internal.h"
	19
	20	atomic_t fscache_op_debug_id;
	21	EXPORT_SYMBOL(fscache_op_debug_id);
	22
	23	/**
	24	* fscache_enqueue_operation - Enqueue an operation for processing
	25	* @op: The operation to enqueue
	26	*
	27	* Enqueue an operation for processing by the FS-Cache thread pool.
	28	*
	29	* This will get its own ref on the object.
	30	*/
	31	void fscache_enqueue_operation(struct fscache_operation *op)
	32	{
	33	_enter("{OBJ%x OP%x,%u}",
	34	op->object->debug_id, op->debug_id, atomic_read(&op->usage));
	35
5753c441	36	ASSERT(list_empty(&op->pend_link));
952efe7b DH	37	ASSERT(op->processor != NULL);
	38	ASSERTCMP(op->object->state, >=, FSCACHE_OBJECT_AVAILABLE);
	39	ASSERTCMP(atomic_read(&op->usage), >, 0);
	40
5753c441 DH	41	fscache_stat(&fscache_n_op_enqueue);
5753c441 DH	42	switch (op->flags & FSCACHE_OP_TYPE) {
8af7c124 TH	43	case FSCACHE_OP_ASYNC:
8af7c124 TH	44	_debug("queue async");
5753c441	45	atomic_inc(&op->usage);
8af7c124	46	if (!queue_work(fscache_op_wq, &op->work))
5753c441 DH	47	fscache_put_operation(op);
5753c441 DH	48	break;
5753c441 DH	49	case FSCACHE_OP_MYTHREAD:
	50	_debug("queue for caller's attention");
	51	break;
	52	default:
	53	printk(KERN_ERR "FS-Cache: Unexpected op type %lx",
	54	op->flags);
	55	BUG();
	56	break;
952efe7b DH	57	}
	58	}
	59	EXPORT_SYMBOL(fscache_enqueue_operation);
	60
	61	/*
	62	* start an op running
	63	*/
	64	static void fscache_run_op(struct fscache_object *object,
	65	struct fscache_operation *op)
	66	{
	67	object->n_in_progress++;
	68	if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
	69	wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
	70	if (op->processor)
	71	fscache_enqueue_operation(op);
	72	fscache_stat(&fscache_n_op_run);
	73	}
	74
	75	/*
	76	* submit an exclusive operation for an object
	77	* - other ops are excluded from running simultaneously with this one
	78	* - this gets any extra refs it needs on an op
	79	*/
	80	int fscache_submit_exclusive_op(struct fscache_object *object,
	81	struct fscache_operation *op)
	82	{
	83	int ret;
	84
	85	_enter("{OBJ%x OP%x},", object->debug_id, op->debug_id);
	86
	87	spin_lock(&object->lock);
	88	ASSERTCMP(object->n_ops, >=, object->n_in_progress);
	89	ASSERTCMP(object->n_ops, >=, object->n_exclusive);
5753c441	90	ASSERT(list_empty(&op->pend_link));
952efe7b DH	91
	92	ret = -ENOBUFS;
	93	if (fscache_object_is_active(object)) {
	94	op->object = object;
	95	object->n_ops++;
	96	object->n_exclusive++; /* reads and writes must wait */
	97
ba28b93a	98	if (object->n_ops > 1) {
952efe7b DH	99	atomic_inc(&op->usage);
	100	list_add_tail(&op->pend_link, &object->pending_ops);
	101	fscache_stat(&fscache_n_op_pend);
	102	} else if (!list_empty(&object->pending_ops)) {
	103	atomic_inc(&op->usage);
	104	list_add_tail(&op->pend_link, &object->pending_ops);
	105	fscache_stat(&fscache_n_op_pend);
	106	fscache_start_operations(object);
	107	} else {
	108	ASSERTCMP(object->n_in_progress, ==, 0);
	109	fscache_run_op(object, op);
	110	}
	111
	112	/* need to issue a new write op after this */
	113	clear_bit(FSCACHE_OBJECT_PENDING_WRITE, &object->flags);
	114	ret = 0;
	115	} else if (object->state == FSCACHE_OBJECT_CREATING) {
	116	op->object = object;
	117	object->n_ops++;
	118	object->n_exclusive++; /* reads and writes must wait */
	119	atomic_inc(&op->usage);
	120	list_add_tail(&op->pend_link, &object->pending_ops);
	121	fscache_stat(&fscache_n_op_pend);
	122	ret = 0;
	123	} else {
	124	/* not allowed to submit ops in any other state */
	125	BUG();
	126	}
	127
	128	spin_unlock(&object->lock);
	129	return ret;
	130	}
	131
	132	/*
	133	* report an unexpected submission
	134	*/
	135	static void fscache_report_unexpected_submission(struct fscache_object *object,
	136	struct fscache_operation *op,
	137	unsigned long ostate)
	138	{
	139	static bool once_only;
	140	struct fscache_operation *p;
	141	unsigned n;
	142
	143	if (once_only)
	144	return;
	145	once_only = true;
	146
	147	kdebug("unexpected submission OP%x [OBJ%x %s]",
	148	op->debug_id, object->debug_id,
	149	fscache_object_states[object->state]);
	150	kdebug("objstate=%s [%s]",
	151	fscache_object_states[object->state],
	152	fscache_object_states[ostate]);
	153	kdebug("objflags=%lx", object->flags);
	154	kdebug("objevent=%lx [%lx]", object->events, object->event_mask);
	155	kdebug("ops=%u inp=%u exc=%u",
	156	object->n_ops, object->n_in_progress, object->n_exclusive);
	157
	158	if (!list_empty(&object->pending_ops)) {
	159	n = 0;
	160	list_for_each_entry(p, &object->pending_ops, pend_link) {
	161	ASSERTCMP(p->object, ==, object);
	162	kdebug("%p %p", op->processor, op->release);
163	n++;
164	}
165
166	kdebug("n=%u", n);
167	}
168
169	dump_stack();
170	}
171
172	/*
173	* submit an operation for an object
174	* - objects may be submitted only in the following states:
175	* - during object creation (write ops may be submitted)
176	* - whilst the object is active
177	* - after an I/O error incurred in one of the two above states (op rejected)
178	* - this gets any extra refs it needs on an op
179	*/
180	int fscache_submit_op(struct fscache_object *object,
181	struct fscache_operation *op)
182	{
183	unsigned long ostate;
184	int ret;
185
186	_enter("{OBJ%x OP%x},{%u}",
187	object->debug_id, op->debug_id, atomic_read(&op->usage));
188
189	ASSERTCMP(atomic_read(&op->usage), >, 0);
190
191	spin_lock(&object->lock);
192	ASSERTCMP(object->n_ops, >=, object->n_in_progress);
193	ASSERTCMP(object->n_ops, >=, object->n_exclusive);
5753c441	194	ASSERT(list_empty(&op->pend_link));
952efe7b DH	195
	196	ostate = object->state;
	197	smp_rmb();
	198
	199	if (fscache_object_is_active(object)) {
	200	op->object = object;
	201	object->n_ops++;
	202
	203	if (object->n_exclusive > 0) {
	204	atomic_inc(&op->usage);
	205	list_add_tail(&op->pend_link, &object->pending_ops);
	206	fscache_stat(&fscache_n_op_pend);
	207	} else if (!list_empty(&object->pending_ops)) {
	208	atomic_inc(&op->usage);
	209	list_add_tail(&op->pend_link, &object->pending_ops);
	210	fscache_stat(&fscache_n_op_pend);
	211	fscache_start_operations(object);
	212	} else {
	213	ASSERTCMP(object->n_exclusive, ==, 0);
	214	fscache_run_op(object, op);
	215	}
	216	ret = 0;
	217	} else if (object->state == FSCACHE_OBJECT_CREATING) {
	218	op->object = object;
	219	object->n_ops++;
	220	atomic_inc(&op->usage);
	221	list_add_tail(&op->pend_link, &object->pending_ops);
	222	fscache_stat(&fscache_n_op_pend);
	223	ret = 0;
e3d4d28b DH	224	} else if (object->state == FSCACHE_OBJECT_DYING \|\|
	225	object->state == FSCACHE_OBJECT_LC_DYING \|\|
	226	object->state == FSCACHE_OBJECT_WITHDRAWING) {
	227	fscache_stat(&fscache_n_op_rejected);
	228	ret = -ENOBUFS;
952efe7b DH	229	} else if (!test_bit(FSCACHE_IOERROR, &object->cache->flags)) {
	230	fscache_report_unexpected_submission(object, op, ostate);
	231	ASSERT(!fscache_object_is_active(object));
	232	ret = -ENOBUFS;
	233	} else {
	234	ret = -ENOBUFS;
	235	}
	236
	237	spin_unlock(&object->lock);
	238	return ret;
	239	}
	240
	241	/*
	242	* queue an object for withdrawal on error, aborting all following asynchronous
	243	* operations
	244	*/
	245	void fscache_abort_object(struct fscache_object *object)
	246	{
	247	_enter("{OBJ%x}", object->debug_id);
	248
	249	fscache_raise_event(object, FSCACHE_OBJECT_EV_ERROR);
	250	}
	251
	252	/*
	253	* jump start the operation processing on an object
	254	* - caller must hold object->lock
	255	*/
	256	void fscache_start_operations(struct fscache_object *object)
	257	{
	258	struct fscache_operation *op;
	259	bool stop = false;
	260
	261	while (!list_empty(&object->pending_ops) && !stop) {
	262	op = list_entry(object->pending_ops.next,
	263	struct fscache_operation, pend_link);
	264
	265	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
	266	if (object->n_in_progress > 0)
	267	break;
	268	stop = true;
	269	}
	270	list_del_init(&op->pend_link);
5753c441	271	fscache_run_op(object, op);
952efe7b DH	272
	273	/* the pending queue was holding a ref on the object */
	274	fscache_put_operation(op);
	275	}
	276
	277	ASSERTCMP(object->n_in_progress, <=, object->n_ops);
	278
	279	_debug("woke %d ops on OBJ%x",
	280	object->n_in_progress, object->debug_id);
	281	}
	282
5753c441 DH	283	/*
	284	* cancel an operation that's pending on an object
	285	*/
	286	int fscache_cancel_op(struct fscache_operation *op)
	287	{
	288	struct fscache_object *object = op->object;
	289	int ret;
	290
	291	_enter("OBJ%x OP%x}", op->object->debug_id, op->debug_id);
	292
	293	spin_lock(&object->lock);
	294
	295	ret = -EBUSY;
	296	if (!list_empty(&op->pend_link)) {
	297	fscache_stat(&fscache_n_op_cancelled);
	298	list_del_init(&op->pend_link);
	299	object->n_ops--;
	300	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
	301	object->n_exclusive--;
	302	if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
	303	wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
	304	fscache_put_operation(op);
	305	ret = 0;
	306	}
	307
	308	spin_unlock(&object->lock);
	309	_leave(" = %d", ret);
	310	return ret;
	311	}
	312
952efe7b DH	313	/*
	314	* release an operation
	315	* - queues pending ops if this is the last in-progress op
	316	*/
	317	void fscache_put_operation(struct fscache_operation *op)
	318	{
	319	struct fscache_object *object;
	320	struct fscache_cache *cache;
	321
	322	_enter("{OBJ%x OP%x,%d}",
	323	op->object->debug_id, op->debug_id, atomic_read(&op->usage));
	324
	325	ASSERTCMP(atomic_read(&op->usage), >, 0);
	326
	327	if (!atomic_dec_and_test(&op->usage))
	328	return;
	329
	330	_debug("PUT OP");
	331	if (test_and_set_bit(FSCACHE_OP_DEAD, &op->flags))
	332	BUG();
	333
	334	fscache_stat(&fscache_n_op_release);
	335
	336	if (op->release) {
	337	op->release(op);
	338	op->release = NULL;
	339	}
	340
	341	object = op->object;
	342
4fbf4291 DH	343	if (test_bit(FSCACHE_OP_DEC_READ_CNT, &op->flags))
	344	atomic_dec(&object->n_reads);
	345
952efe7b DH	346	/* now... we may get called with the object spinlock held, so we
	347	* complete the cleanup here only if we can immediately acquire the
	348	* lock, and defer it otherwise */
	349	if (!spin_trylock(&object->lock)) {
	350	_debug("defer put");
	351	fscache_stat(&fscache_n_op_deferred_release);
	352
	353	cache = object->cache;
	354	spin_lock(&cache->op_gc_list_lock);
	355	list_add_tail(&op->pend_link, &cache->op_gc_list);
	356	spin_unlock(&cache->op_gc_list_lock);
	357	schedule_work(&cache->op_gc);
	358	_leave(" [defer]");
	359	return;
	360	}
	361
	362	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
	363	ASSERTCMP(object->n_exclusive, >, 0);
	364	object->n_exclusive--;
	365	}
	366
	367	ASSERTCMP(object->n_in_progress, >, 0);
	368	object->n_in_progress--;
	369	if (object->n_in_progress == 0)
	370	fscache_start_operations(object);
	371
	372	ASSERTCMP(object->n_ops, >, 0);
	373	object->n_ops--;
	374	if (object->n_ops == 0)
	375	fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
	376
	377	spin_unlock(&object->lock);
	378
	379	kfree(op);
	380	_leave(" [done]");
	381	}
	382	EXPORT_SYMBOL(fscache_put_operation);
	383
	384	/*
	385	* garbage collect operations that have had their release deferred
	386	*/
	387	void fscache_operation_gc(struct work_struct *work)
	388	{
	389	struct fscache_operation *op;
	390	struct fscache_object *object;
	391	struct fscache_cache *cache =
	392	container_of(work, struct fscache_cache, op_gc);
	393	int count = 0;
	394
	395	_enter("");
	396
	397	do {
	398	spin_lock(&cache->op_gc_list_lock);
	399	if (list_empty(&cache->op_gc_list)) {
	400	spin_unlock(&cache->op_gc_list_lock);
	401	break;
	402	}
	403
	404	op = list_entry(cache->op_gc_list.next,
	405	struct fscache_operation, pend_link);
	406	list_del(&op->pend_link);
	407	spin_unlock(&cache->op_gc_list_lock);
	408
	409	object = op->object;
410
411	_debug("GC DEFERRED REL OBJ%x OP%x",
412	object->debug_id, op->debug_id);
413	fscache_stat(&fscache_n_op_gc);
414
415	ASSERTCMP(atomic_read(&op->usage), ==, 0);
416
417	spin_lock(&object->lock);
418	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
419	ASSERTCMP(object->n_exclusive, >, 0);
420	object->n_exclusive--;
421	}
422
423	ASSERTCMP(object->n_in_progress, >, 0);
424	object->n_in_progress--;
425	if (object->n_in_progress == 0)
426	fscache_start_operations(object);
427
428	ASSERTCMP(object->n_ops, >, 0);
429	object->n_ops--;
430	if (object->n_ops == 0)
431	fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
432
433	spin_unlock(&object->lock);
434
435	} while (count++ < 20);
436
437	if (!list_empty(&cache->op_gc_list))
438	schedule_work(&cache->op_gc);
439
440	_leave("");
441	}
442
443	/*
8af7c124 TH	444	* execute an operation using fs_op_wq to provide processing context -
8af7c124 TH	445	* the caller holds a ref to this object, so we don't need to hold one
952efe7b	446	*/
8af7c124	447	void fscache_op_work_func(struct work_struct *work)
952efe7b DH	448	{
952efe7b DH	449	struct fscache_operation *op =
8af7c124	450	container_of(work, struct fscache_operation, work);
952efe7b DH	451	unsigned long start;
	452
	453	_enter("{OBJ%x OP%x,%d}",
	454	op->object->debug_id, op->debug_id, atomic_read(&op->usage));
	455
	456	ASSERT(op->processor != NULL);
	457	start = jiffies;
	458	op->processor(op);
	459	fscache_hist(fscache_ops_histogram, start);
8af7c124	460	fscache_put_operation(op);
952efe7b DH	461
	462	_leave("");
	463	}