[linux-block.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 "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(op->processor != NULL);
	ASSERTCMP(op->object->state, >=, FSCACHE_OBJECT_AVAILABLE);
	ASSERTCMP(atomic_read(&op->usage), >, 0);

	if (list_empty(&op->pend_link)) {
		switch (op->flags & FSCACHE_OP_TYPE) {
		case FSCACHE_OP_FAST:
			_debug("queue fast");
			atomic_inc(&op->usage);
			if (!schedule_work(&op->fast_work))
				fscache_put_operation(op);
			break;
		case FSCACHE_OP_SLOW:
			_debug("queue slow");
			slow_work_enqueue(&op->slow_work);
			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;
		}
		fscache_stat(&fscache_n_op_enqueue);
	}
}
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);

	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 > 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_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);

	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 (!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);
		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);

		/* 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);
}

/*
 * 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;

	/* 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("");
}

/*
 * allow the slow work item processor to get a ref on an operation
 */
static int fscache_op_get_ref(struct slow_work *work)
{
	struct fscache_operation *op =
		container_of(work, struct fscache_operation, slow_work);

	atomic_inc(&op->usage);
	return 0;
}

/*
 * allow the slow work item processor to discard a ref on an operation
 */
static void fscache_op_put_ref(struct slow_work *work)
{
	struct fscache_operation *op =
		container_of(work, struct fscache_operation, slow_work);

	fscache_put_operation(op);
}

/*
 * execute an operation using the slow thread pool to provide processing context
 * - the caller holds a ref to this object, so we don't need to hold one
 */
static void fscache_op_execute(struct slow_work *work)
{
	struct fscache_operation *op =
		container_of(work, struct fscache_operation, slow_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);

	_leave("");
}

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