[linux-2.6-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 <linux/seq_file.h>
#include <linux/slab.h>
#include "internal.h"

atomic_t fscache_op_debug_id;
EXPORT_SYMBOL(fscache_op_debug_id);

static void fscache_operation_dummy_cancel(struct fscache_operation *op)
{
}

/**
 * fscache_operation_init - Do basic initialisation of an operation
 * @op: The operation to initialise
 * @release: The release function to assign
 *
 * Do basic initialisation of an operation.  The caller must still set flags,
 * object and processor if needed.
 */
void fscache_operation_init(struct fscache_cookie *cookie,
			    struct fscache_operation *op,
			    fscache_operation_processor_t processor,
			    fscache_operation_cancel_t cancel,
			    fscache_operation_release_t release)
{
	INIT_WORK(&op->work, fscache_op_work_func);
	atomic_set(&op->usage, 1);
	op->state = FSCACHE_OP_ST_INITIALISED;
	op->debug_id = atomic_inc_return(&fscache_op_debug_id);
	op->processor = processor;
	op->cancel = cancel ?: fscache_operation_dummy_cancel;
	op->release = release;
	INIT_LIST_HEAD(&op->pend_link);
	fscache_stat(&fscache_n_op_initialised);
	trace_fscache_op(cookie, op, fscache_op_init);
}
EXPORT_SYMBOL(fscache_operation_init);

/**
 * 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)
{
	struct fscache_cookie *cookie = op->object->cookie;
	
	_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);
	ASSERT(fscache_object_is_available(op->object));
	ASSERTCMP(atomic_read(&op->usage), >, 0);
	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);

	fscache_stat(&fscache_n_op_enqueue);
	switch (op->flags & FSCACHE_OP_TYPE) {
	case FSCACHE_OP_ASYNC:
		trace_fscache_op(cookie, op, fscache_op_enqueue_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:
		trace_fscache_op(cookie, op, fscache_op_enqueue_mythread);
		_debug("queue for caller's attention");
		break;
	default:
		pr_err("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)
{
	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_PENDING);

	op->state = FSCACHE_OP_ST_IN_PROGRESS;
	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);
	else
		trace_fscache_op(object->cookie, op, fscache_op_run);
	fscache_stat(&fscache_n_op_run);
}

/*
 * report an unexpected submission
 */
static void fscache_report_unexpected_submission(struct fscache_object *object,
						 struct fscache_operation *op,
						 const struct fscache_state *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, object->state->name);
	kdebug("objstate=%s [%s]", object->state->name, ostate->name);
	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 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)
{
	const struct fscache_state *ostate;
	unsigned long flags;
	int ret;

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

	trace_fscache_op(object->cookie, op, fscache_op_submit_ex);

	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
	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();

	op->state = FSCACHE_OP_ST_PENDING;
	flags = READ_ONCE(object->flags);
	if (unlikely(!(flags & BIT(FSCACHE_OBJECT_IS_LIVE)))) {
		fscache_stat(&fscache_n_op_rejected);
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else if (unlikely(fscache_cache_is_broken(object))) {
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		ret = -EIO;
	} else if (flags & BIT(FSCACHE_OBJECT_IS_AVAILABLE)) {
		op->object = object;
		object->n_ops++;
		object->n_exclusive++;	/* reads and writes must wait */

		if (object->n_in_progress > 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 (flags & BIT(FSCACHE_OBJECT_IS_LOOKED_UP)) {
		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 if (flags & BIT(FSCACHE_OBJECT_KILLED_BY_CACHE)) {
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else {
		fscache_report_unexpected_submission(object, op, ostate);
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	}

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

/*
 * 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)
{
	const struct fscache_state *ostate;
	unsigned long flags;
	int ret;

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

	trace_fscache_op(object->cookie, op, fscache_op_submit);

	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
	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();

	op->state = FSCACHE_OP_ST_PENDING;
	flags = READ_ONCE(object->flags);
	if (unlikely(!(flags & BIT(FSCACHE_OBJECT_IS_LIVE)))) {
		fscache_stat(&fscache_n_op_rejected);
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else if (unlikely(fscache_cache_is_broken(object))) {
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		ret = -EIO;
	} else if (flags & BIT(FSCACHE_OBJECT_IS_AVAILABLE)) {
		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 (flags & BIT(FSCACHE_OBJECT_IS_LOOKED_UP)) {
		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 (flags & BIT(FSCACHE_OBJECT_KILLED_BY_CACHE)) {
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else {
		fscache_report_unexpected_submission(object, op, ostate);
		ASSERT(!fscache_object_is_active(object));
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		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.  The 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,
		      bool cancel_in_progress_op)
{
	struct fscache_object *object = op->object;
	bool put = false;
	int ret;

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

	trace_fscache_op(object->cookie, op, fscache_op_cancel);

	ASSERTCMP(op->state, >=, FSCACHE_OP_ST_PENDING);
	ASSERTCMP(op->state, !=, FSCACHE_OP_ST_CANCELLED);
	ASSERTCMP(atomic_read(&op->usage), >, 0);

	spin_lock(&object->lock);

	ret = -EBUSY;
	if (op->state == FSCACHE_OP_ST_PENDING) {
		ASSERT(!list_empty(&op->pend_link));
		list_del_init(&op->pend_link);
		put = true;

		fscache_stat(&fscache_n_op_cancelled);
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		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);
		ret = 0;
	} else if (op->state == FSCACHE_OP_ST_IN_PROGRESS && cancel_in_progress_op) {
		ASSERTCMP(object->n_in_progress, >, 0);
		if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
			object->n_exclusive--;
		object->n_in_progress--;
		if (object->n_in_progress == 0)
			fscache_start_operations(object);

		fscache_stat(&fscache_n_op_cancelled);
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;
		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);
		ret = 0;
	}

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

/*
 * Cancel all pending operations on an object
 */
void fscache_cancel_all_ops(struct fscache_object *object)
{
	struct fscache_operation *op;

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

	spin_lock(&object->lock);

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

		trace_fscache_op(object->cookie, op, fscache_op_cancel_all);

		ASSERTCMP(op->state, ==, FSCACHE_OP_ST_PENDING);
		op->cancel(op);
		op->state = FSCACHE_OP_ST_CANCELLED;

		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);
		cond_resched_lock(&object->lock);
	}

	spin_unlock(&object->lock);
	_leave("");
}

/*
 * Record the completion or cancellation of an in-progress operation.
 */
void fscache_op_complete(struct fscache_operation *op, bool cancelled)
{
	struct fscache_object *object = op->object;

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

	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);
	ASSERTCMP(object->n_in_progress, >, 0);
	ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
		    object->n_exclusive, >, 0);
	ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
		    object->n_in_progress, ==, 1);

	spin_lock(&object->lock);

	if (!cancelled) {
		trace_fscache_op(object->cookie, op, fscache_op_completed);
		op->state = FSCACHE_OP_ST_COMPLETE;
	} else {
		op->cancel(op);
		trace_fscache_op(object->cookie, op, fscache_op_cancelled);
		op->state = FSCACHE_OP_ST_CANCELLED;
	}

	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
		object->n_exclusive--;
	object->n_in_progress--;
	if (object->n_in_progress == 0)
		fscache_start_operations(object);

	spin_unlock(&object->lock);
	_leave("");
}
EXPORT_SYMBOL(fscache_op_complete);

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

	trace_fscache_op(op->object ? op->object->cookie : NULL, op, fscache_op_put);

	_debug("PUT OP");
	ASSERTIFCMP(op->state != FSCACHE_OP_ST_INITIALISED &&
		    op->state != FSCACHE_OP_ST_COMPLETE,
		    op->state, ==, FSCACHE_OP_ST_CANCELLED);

	fscache_stat(&fscache_n_op_release);

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

	object = op->object;
	if (likely(object)) {
		if (test_bit(FSCACHE_OP_DEC_READ_CNT, &op->flags))
			atomic_dec(&object->n_reads);
		if (test_bit(FSCACHE_OP_UNUSE_COOKIE, &op->flags))
			fscache_unuse_cookie(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;
		}

		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;
		trace_fscache_op(object->cookie, op, fscache_op_gc);

		spin_lock(&object->lock);

		_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);
		ASSERTCMP(op->state, ==, FSCACHE_OP_ST_DEAD);

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

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

	trace_fscache_op(op->object->cookie, op, fscache_op_work);

	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
d3b97ca4 DH	23	static void fscache_operation_dummy_cancel(struct fscache_operation *op)
	24	{
	25	}
	26
1339ec98 DH	27	/**
	28	* fscache_operation_init - Do basic initialisation of an operation
	29	* @op: The operation to initialise
	30	* @release: The release function to assign
	31	*
	32	* Do basic initialisation of an operation. The caller must still set flags,
	33	* object and processor if needed.
	34	*/
08c2e3d0 DH	35	void fscache_operation_init(struct fscache_cookie *cookie,
08c2e3d0 DH	36	struct fscache_operation *op,
1339ec98	37	fscache_operation_processor_t processor,
d3b97ca4	38	fscache_operation_cancel_t cancel,
1339ec98 DH	39	fscache_operation_release_t release)
	40	{
	41	INIT_WORK(&op->work, fscache_op_work_func);
	42	atomic_set(&op->usage, 1);
	43	op->state = FSCACHE_OP_ST_INITIALISED;
	44	op->debug_id = atomic_inc_return(&fscache_op_debug_id);
	45	op->processor = processor;
d3b97ca4	46	op->cancel = cancel ?: fscache_operation_dummy_cancel;
1339ec98 DH	47	op->release = release;
1339ec98 DH	48	INIT_LIST_HEAD(&op->pend_link);
03cdd0e4	49	fscache_stat(&fscache_n_op_initialised);
08c2e3d0	50	trace_fscache_op(cookie, op, fscache_op_init);
1339ec98 DH	51	}
	52	EXPORT_SYMBOL(fscache_operation_init);
	53
952efe7b DH	54	/**
	55	* fscache_enqueue_operation - Enqueue an operation for processing
	56	* @op: The operation to enqueue
	57	*
	58	* Enqueue an operation for processing by the FS-Cache thread pool.
	59	*
	60	* This will get its own ref on the object.
	61	*/
	62	void fscache_enqueue_operation(struct fscache_operation *op)
	63	{
08c2e3d0 DH	64	struct fscache_cookie *cookie = op->object->cookie;
08c2e3d0 DH	65
952efe7b DH	66	_enter("{OBJ%x OP%x,%u}",
	67	op->object->debug_id, op->debug_id, atomic_read(&op->usage));
	68
5753c441	69	ASSERT(list_empty(&op->pend_link));
952efe7b	70	ASSERT(op->processor != NULL);
493f7bc1	71	ASSERT(fscache_object_is_available(op->object));
952efe7b	72	ASSERTCMP(atomic_read(&op->usage), >, 0);
9f10523f	73	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);
952efe7b	74
5753c441 DH	75	fscache_stat(&fscache_n_op_enqueue);
5753c441 DH	76	switch (op->flags & FSCACHE_OP_TYPE) {
8af7c124	77	case FSCACHE_OP_ASYNC:
08c2e3d0	78	trace_fscache_op(cookie, op, fscache_op_enqueue_async);
8af7c124	79	_debug("queue async");
5753c441	80	atomic_inc(&op->usage);
8af7c124	81	if (!queue_work(fscache_op_wq, &op->work))
5753c441 DH	82	fscache_put_operation(op);
5753c441 DH	83	break;
5753c441	84	case FSCACHE_OP_MYTHREAD:
08c2e3d0	85	trace_fscache_op(cookie, op, fscache_op_enqueue_mythread);
5753c441 DH	86	_debug("queue for caller's attention");
	87	break;
	88	default:
36dfd116	89	pr_err("Unexpected op type %lx", op->flags);
5753c441 DH	90	BUG();
5753c441 DH	91	break;
952efe7b DH	92	}
	93	}
	94	EXPORT_SYMBOL(fscache_enqueue_operation);
	95
	96	/*
	97	* start an op running
	98	*/
	99	static void fscache_run_op(struct fscache_object *object,
	100	struct fscache_operation *op)
	101	{
9f10523f DH	102	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_PENDING);
	103
	104	op->state = FSCACHE_OP_ST_IN_PROGRESS;
952efe7b DH	105	object->n_in_progress++;
	106	if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
	107	wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
	108	if (op->processor)
	109	fscache_enqueue_operation(op);
08c2e3d0 DH	110	else
08c2e3d0 DH	111	trace_fscache_op(object->cookie, op, fscache_op_run);
952efe7b DH	112	fscache_stat(&fscache_n_op_run);
	113	}
	114
3c305984 DH	115	/*
	116	* report an unexpected submission
	117	*/
	118	static void fscache_report_unexpected_submission(struct fscache_object *object,
	119	struct fscache_operation *op,
	120	const struct fscache_state *ostate)
	121	{
	122	static bool once_only;
	123	struct fscache_operation *p;
	124	unsigned n;
	125
	126	if (once_only)
	127	return;
	128	once_only = true;
	129
	130	kdebug("unexpected submission OP%x [OBJ%x %s]",
	131	op->debug_id, object->debug_id, object->state->name);
	132	kdebug("objstate=%s [%s]", object->state->name, ostate->name);
	133	kdebug("objflags=%lx", object->flags);
	134	kdebug("objevent=%lx [%lx]", object->events, object->event_mask);
	135	kdebug("ops=%u inp=%u exc=%u",
	136	object->n_ops, object->n_in_progress, object->n_exclusive);
	137
	138	if (!list_empty(&object->pending_ops)) {
	139	n = 0;
	140	list_for_each_entry(p, &object->pending_ops, pend_link) {
	141	ASSERTCMP(p->object, ==, object);
	142	kdebug("%p %p", op->processor, op->release);
	143	n++;
	144	}
	145
	146	kdebug("n=%u", n);
	147	}
	148
	149	dump_stack();
	150	}
	151
952efe7b DH	152	/*
	153	* submit an exclusive operation for an object
	154	* - other ops are excluded from running simultaneously with this one
	155	* - this gets any extra refs it needs on an op
	156	*/
	157	int fscache_submit_exclusive_op(struct fscache_object *object,
	158	struct fscache_operation *op)
	159	{
30ceec62 DH	160	const struct fscache_state *ostate;
30ceec62 DH	161	unsigned long flags;
8d76349d DH	162	int ret;
8d76349d DH	163
952efe7b DH	164	_enter("{OBJ%x OP%x},", object->debug_id, op->debug_id);
952efe7b DH	165
08c2e3d0 DH	166	trace_fscache_op(object->cookie, op, fscache_op_submit_ex);
08c2e3d0 DH	167
9f10523f DH	168	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
	169	ASSERTCMP(atomic_read(&op->usage), >, 0);
	170
952efe7b DH	171	spin_lock(&object->lock);
	172	ASSERTCMP(object->n_ops, >=, object->n_in_progress);
	173	ASSERTCMP(object->n_ops, >=, object->n_exclusive);
5753c441	174	ASSERT(list_empty(&op->pend_link));
952efe7b	175
30ceec62 DH	176	ostate = object->state;
	177	smp_rmb();
	178
9f10523f	179	op->state = FSCACHE_OP_ST_PENDING;
30ceec62 DH	180	flags = READ_ONCE(object->flags);
	181	if (unlikely(!(flags & BIT(FSCACHE_OBJECT_IS_LIVE)))) {
	182	fscache_stat(&fscache_n_op_rejected);
d3b97ca4	183	op->cancel(op);
30ceec62 DH	184	op->state = FSCACHE_OP_ST_CANCELLED;
	185	ret = -ENOBUFS;
	186	} else if (unlikely(fscache_cache_is_broken(object))) {
d3b97ca4	187	op->cancel(op);
30ceec62 DH	188	op->state = FSCACHE_OP_ST_CANCELLED;
	189	ret = -EIO;
	190	} else if (flags & BIT(FSCACHE_OBJECT_IS_AVAILABLE)) {
952efe7b DH	191	op->object = object;
	192	object->n_ops++;
	193	object->n_exclusive++; /* reads and writes must wait */
	194
9f10523f	195	if (object->n_in_progress > 0) {
952efe7b DH	196	atomic_inc(&op->usage);
	197	list_add_tail(&op->pend_link, &object->pending_ops);
	198	fscache_stat(&fscache_n_op_pend);
	199	} else if (!list_empty(&object->pending_ops)) {
	200	atomic_inc(&op->usage);
	201	list_add_tail(&op->pend_link, &object->pending_ops);
	202	fscache_stat(&fscache_n_op_pend);
	203	fscache_start_operations(object);
	204	} else {
	205	ASSERTCMP(object->n_in_progress, ==, 0);
	206	fscache_run_op(object, op);
	207	}
	208
	209	/* need to issue a new write op after this */
	210	clear_bit(FSCACHE_OBJECT_PENDING_WRITE, &object->flags);
8d76349d	211	ret = 0;
30ceec62	212	} else if (flags & BIT(FSCACHE_OBJECT_IS_LOOKED_UP)) {
952efe7b DH	213	op->object = object;
	214	object->n_ops++;
	215	object->n_exclusive++; /* reads and writes must wait */
	216	atomic_inc(&op->usage);
	217	list_add_tail(&op->pend_link, &object->pending_ops);
	218	fscache_stat(&fscache_n_op_pend);
8d76349d	219	ret = 0;
6515d1db	220	} else if (flags & BIT(FSCACHE_OBJECT_KILLED_BY_CACHE)) {
d3b97ca4	221	op->cancel(op);
6515d1db DH	222	op->state = FSCACHE_OP_ST_CANCELLED;
6515d1db DH	223	ret = -ENOBUFS;
952efe7b	224	} else {
30ceec62	225	fscache_report_unexpected_submission(object, op, ostate);
d3b97ca4	226	op->cancel(op);
30ceec62 DH	227	op->state = FSCACHE_OP_ST_CANCELLED;
30ceec62 DH	228	ret = -ENOBUFS;
952efe7b DH	229	}
	230
	231	spin_unlock(&object->lock);
8d76349d	232	return ret;
952efe7b DH	233	}
952efe7b DH	234
952efe7b DH	235	/*
	236	* submit an operation for an object
	237	* - objects may be submitted only in the following states:
	238	* - during object creation (write ops may be submitted)
	239	* - whilst the object is active
	240	* - after an I/O error incurred in one of the two above states (op rejected)
	241	* - this gets any extra refs it needs on an op
	242	*/
	243	int fscache_submit_op(struct fscache_object *object,
	244	struct fscache_operation *op)
	245	{
caaef690	246	const struct fscache_state *ostate;
30ceec62	247	unsigned long flags;
952efe7b DH	248	int ret;
	249
	250	_enter("{OBJ%x OP%x},{%u}",
	251	object->debug_id, op->debug_id, atomic_read(&op->usage));
	252
08c2e3d0 DH	253	trace_fscache_op(object->cookie, op, fscache_op_submit);
08c2e3d0 DH	254
9f10523f	255	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
952efe7b DH	256	ASSERTCMP(atomic_read(&op->usage), >, 0);
	257
	258	spin_lock(&object->lock);
	259	ASSERTCMP(object->n_ops, >=, object->n_in_progress);
	260	ASSERTCMP(object->n_ops, >=, object->n_exclusive);
5753c441	261	ASSERT(list_empty(&op->pend_link));
952efe7b DH	262
	263	ostate = object->state;
	264	smp_rmb();
	265
9f10523f	266	op->state = FSCACHE_OP_ST_PENDING;
30ceec62 DH	267	flags = READ_ONCE(object->flags);
	268	if (unlikely(!(flags & BIT(FSCACHE_OBJECT_IS_LIVE)))) {
	269	fscache_stat(&fscache_n_op_rejected);
d3b97ca4	270	op->cancel(op);
30ceec62 DH	271	op->state = FSCACHE_OP_ST_CANCELLED;
	272	ret = -ENOBUFS;
	273	} else if (unlikely(fscache_cache_is_broken(object))) {
d3b97ca4	274	op->cancel(op);
30ceec62 DH	275	op->state = FSCACHE_OP_ST_CANCELLED;
	276	ret = -EIO;
	277	} else if (flags & BIT(FSCACHE_OBJECT_IS_AVAILABLE)) {
952efe7b DH	278	op->object = object;
	279	object->n_ops++;
	280
	281	if (object->n_exclusive > 0) {
	282	atomic_inc(&op->usage);
	283	list_add_tail(&op->pend_link, &object->pending_ops);
	284	fscache_stat(&fscache_n_op_pend);
	285	} else if (!list_empty(&object->pending_ops)) {
	286	atomic_inc(&op->usage);
	287	list_add_tail(&op->pend_link, &object->pending_ops);
	288	fscache_stat(&fscache_n_op_pend);
	289	fscache_start_operations(object);
	290	} else {
	291	ASSERTCMP(object->n_exclusive, ==, 0);
	292	fscache_run_op(object, op);
	293	}
	294	ret = 0;
30ceec62	295	} else if (flags & BIT(FSCACHE_OBJECT_IS_LOOKED_UP)) {
952efe7b DH	296	op->object = object;
	297	object->n_ops++;
	298	atomic_inc(&op->usage);
	299	list_add_tail(&op->pend_link, &object->pending_ops);
	300	fscache_stat(&fscache_n_op_pend);
	301	ret = 0;
6515d1db	302	} else if (flags & BIT(FSCACHE_OBJECT_KILLED_BY_CACHE)) {
d3b97ca4	303	op->cancel(op);
6515d1db DH	304	op->state = FSCACHE_OP_ST_CANCELLED;
6515d1db DH	305	ret = -ENOBUFS;
30ceec62	306	} else {
952efe7b DH	307	fscache_report_unexpected_submission(object, op, ostate);
952efe7b DH	308	ASSERT(!fscache_object_is_active(object));
d3b97ca4	309	op->cancel(op);
9f10523f	310	op->state = FSCACHE_OP_ST_CANCELLED;
952efe7b	311	ret = -ENOBUFS;
952efe7b DH	312	}
	313
	314	spin_unlock(&object->lock);
	315	return ret;
	316	}
	317
	318	/*
	319	* queue an object for withdrawal on error, aborting all following asynchronous
	320	* operations
	321	*/
	322	void fscache_abort_object(struct fscache_object *object)
	323	{
	324	_enter("{OBJ%x}", object->debug_id);
	325
	326	fscache_raise_event(object, FSCACHE_OBJECT_EV_ERROR);
	327	}
	328
	329	/*
dcfae32f DH	330	* Jump start the operation processing on an object. The caller must hold
dcfae32f DH	331	* object->lock.
952efe7b DH	332	*/
	333	void fscache_start_operations(struct fscache_object *object)
	334	{
	335	struct fscache_operation *op;
	336	bool stop = false;
	337
	338	while (!list_empty(&object->pending_ops) && !stop) {
	339	op = list_entry(object->pending_ops.next,
	340	struct fscache_operation, pend_link);
	341
	342	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
	343	if (object->n_in_progress > 0)
	344	break;
	345	stop = true;
	346	}
	347	list_del_init(&op->pend_link);
5753c441	348	fscache_run_op(object, op);
952efe7b DH	349
	350	/* the pending queue was holding a ref on the object */
	351	fscache_put_operation(op);
	352	}
	353
	354	ASSERTCMP(object->n_in_progress, <=, object->n_ops);
	355
	356	_debug("woke %d ops on OBJ%x",
	357	object->n_in_progress, object->debug_id);
	358	}
	359
5753c441 DH	360	/*
	361	* cancel an operation that's pending on an object
	362	*/
91c7fbbf	363	int fscache_cancel_op(struct fscache_operation *op,
418b7eb9	364	bool cancel_in_progress_op)
5753c441 DH	365	{
5753c441 DH	366	struct fscache_object *object = op->object;
418b7eb9	367	bool put = false;
5753c441 DH	368	int ret;
	369
	370	_enter("OBJ%x OP%x}", op->object->debug_id, op->debug_id);
	371
08c2e3d0 DH	372	trace_fscache_op(object->cookie, op, fscache_op_cancel);
08c2e3d0 DH	373
9f10523f DH	374	ASSERTCMP(op->state, >=, FSCACHE_OP_ST_PENDING);
	375	ASSERTCMP(op->state, !=, FSCACHE_OP_ST_CANCELLED);
	376	ASSERTCMP(atomic_read(&op->usage), >, 0);
	377
5753c441 DH	378	spin_lock(&object->lock);
	379
	380	ret = -EBUSY;
9f10523f DH	381	if (op->state == FSCACHE_OP_ST_PENDING) {
9f10523f DH	382	ASSERT(!list_empty(&op->pend_link));
5753c441	383	list_del_init(&op->pend_link);
418b7eb9	384	put = true;
d3b97ca4	385
418b7eb9	386	fscache_stat(&fscache_n_op_cancelled);
d3b97ca4	387	op->cancel(op);
418b7eb9 DH	388	op->state = FSCACHE_OP_ST_CANCELLED;
	389	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
	390	object->n_exclusive--;
	391	if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
	392	wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
	393	ret = 0;
	394	} else if (op->state == FSCACHE_OP_ST_IN_PROGRESS && cancel_in_progress_op) {
73c04a47 DH	395	ASSERTCMP(object->n_in_progress, >, 0);
	396	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
	397	object->n_exclusive--;
	398	object->n_in_progress--;
	399	if (object->n_in_progress == 0)
	400	fscache_start_operations(object);
	401
418b7eb9	402	fscache_stat(&fscache_n_op_cancelled);
d3b97ca4	403	op->cancel(op);
9f10523f	404	op->state = FSCACHE_OP_ST_CANCELLED;
5753c441 DH	405	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
	406	object->n_exclusive--;
	407	if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
	408	wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
5753c441 DH	409	ret = 0;
	410	}
	411
418b7eb9 DH	412	if (put)
418b7eb9 DH	413	fscache_put_operation(op);
5753c441 DH	414	spin_unlock(&object->lock);
	415	_leave(" = %d", ret);
	416	return ret;
	417	}
	418
ef778e7a DH	419	/*
	420	* Cancel all pending operations on an object
	421	*/
	422	void fscache_cancel_all_ops(struct fscache_object *object)
	423	{
	424	struct fscache_operation *op;
	425
	426	_enter("OBJ%x", object->debug_id);
	427
	428	spin_lock(&object->lock);
	429
	430	while (!list_empty(&object->pending_ops)) {
	431	op = list_entry(object->pending_ops.next,
	432	struct fscache_operation, pend_link);
	433	fscache_stat(&fscache_n_op_cancelled);
	434	list_del_init(&op->pend_link);
	435
08c2e3d0 DH	436	trace_fscache_op(object->cookie, op, fscache_op_cancel_all);
08c2e3d0 DH	437
ef778e7a	438	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_PENDING);
d3b97ca4	439	op->cancel(op);
ef778e7a DH	440	op->state = FSCACHE_OP_ST_CANCELLED;
	441
	442	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
	443	object->n_exclusive--;
	444	if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
	445	wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
	446	fscache_put_operation(op);
	447	cond_resched_lock(&object->lock);
	448	}
	449
	450	spin_unlock(&object->lock);
	451	_leave("");
	452	}
	453
9f10523f	454	/*
1f372dff	455	* Record the completion or cancellation of an in-progress operation.
9f10523f	456	*/
1f372dff	457	void fscache_op_complete(struct fscache_operation *op, bool cancelled)
9f10523f DH	458	{
	459	struct fscache_object *object = op->object;
	460
	461	_enter("OBJ%x", object->debug_id);
	462
	463	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);
	464	ASSERTCMP(object->n_in_progress, >, 0);
	465	ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
	466	object->n_exclusive, >, 0);
	467	ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
	468	object->n_in_progress, ==, 1);
	469
	470	spin_lock(&object->lock);
	471
d3b97ca4	472	if (!cancelled) {
08c2e3d0	473	trace_fscache_op(object->cookie, op, fscache_op_completed);
d3b97ca4 DH	474	op->state = FSCACHE_OP_ST_COMPLETE;
	475	} else {
	476	op->cancel(op);
08c2e3d0	477	trace_fscache_op(object->cookie, op, fscache_op_cancelled);
d3b97ca4 DH	478	op->state = FSCACHE_OP_ST_CANCELLED;
d3b97ca4 DH	479	}
9f10523f DH	480
	481	if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
	482	object->n_exclusive--;
	483	object->n_in_progress--;
	484	if (object->n_in_progress == 0)
	485	fscache_start_operations(object);
	486
	487	spin_unlock(&object->lock);
	488	_leave("");
	489	}
	490	EXPORT_SYMBOL(fscache_op_complete);
	491
952efe7b DH	492	/*
	493	* release an operation
	494	* - queues pending ops if this is the last in-progress op
	495	*/
	496	void fscache_put_operation(struct fscache_operation *op)
	497	{
	498	struct fscache_object *object;
	499	struct fscache_cache *cache;
	500
	501	_enter("{OBJ%x OP%x,%d}",
	502	op->object->debug_id, op->debug_id, atomic_read(&op->usage));
	503
	504	ASSERTCMP(atomic_read(&op->usage), >, 0);
	505
	506	if (!atomic_dec_and_test(&op->usage))
	507	return;
	508
402cb8dd	509	trace_fscache_op(op->object ? op->object->cookie : NULL, op, fscache_op_put);
08c2e3d0	510
952efe7b	511	_debug("PUT OP");
a39caadf DH	512	ASSERTIFCMP(op->state != FSCACHE_OP_ST_INITIALISED &&
a39caadf DH	513	op->state != FSCACHE_OP_ST_COMPLETE,
9f10523f	514	op->state, ==, FSCACHE_OP_ST_CANCELLED);
952efe7b DH	515
	516	fscache_stat(&fscache_n_op_release);
	517
	518	if (op->release) {
	519	op->release(op);
	520	op->release = NULL;
	521	}
4a47132f	522	op->state = FSCACHE_OP_ST_DEAD;
952efe7b DH	523
952efe7b DH	524	object = op->object;
a39caadf DH	525	if (likely(object)) {
	526	if (test_bit(FSCACHE_OP_DEC_READ_CNT, &op->flags))
	527	atomic_dec(&object->n_reads);
	528	if (test_bit(FSCACHE_OP_UNUSE_COOKIE, &op->flags))
	529	fscache_unuse_cookie(object);
	530
	531	/* now... we may get called with the object spinlock held, so we
	532	* complete the cleanup here only if we can immediately acquire the
	533	* lock, and defer it otherwise */
	534	if (!spin_trylock(&object->lock)) {
	535	_debug("defer put");
	536	fscache_stat(&fscache_n_op_deferred_release);
	537
	538	cache = object->cache;
	539	spin_lock(&cache->op_gc_list_lock);
	540	list_add_tail(&op->pend_link, &cache->op_gc_list);
	541	spin_unlock(&cache->op_gc_list_lock);
	542	schedule_work(&cache->op_gc);
	543	_leave(" [defer]");
	544	return;
	545	}
952efe7b	546
a39caadf DH	547	ASSERTCMP(object->n_ops, >, 0);
	548	object->n_ops--;
	549	if (object->n_ops == 0)
	550	fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
952efe7b	551
a39caadf	552	spin_unlock(&object->lock);
952efe7b DH	553	}
952efe7b DH	554
952efe7b DH	555	kfree(op);
	556	_leave(" [done]");
	557	}
	558	EXPORT_SYMBOL(fscache_put_operation);
	559
	560	/*
	561	* garbage collect operations that have had their release deferred
	562	*/
	563	void fscache_operation_gc(struct work_struct *work)
	564	{
	565	struct fscache_operation *op;
	566	struct fscache_object *object;
	567	struct fscache_cache *cache =
	568	container_of(work, struct fscache_cache, op_gc);
	569	int count = 0;
	570
	571	_enter("");
	572
	573	do {
	574	spin_lock(&cache->op_gc_list_lock);
	575	if (list_empty(&cache->op_gc_list)) {
	576	spin_unlock(&cache->op_gc_list_lock);
	577	break;
	578	}
	579
	580	op = list_entry(cache->op_gc_list.next,
	581	struct fscache_operation, pend_link);
	582	list_del(&op->pend_link);
	583	spin_unlock(&cache->op_gc_list_lock);
	584
	585	object = op->object;
08c2e3d0 DH	586	trace_fscache_op(object->cookie, op, fscache_op_gc);
08c2e3d0 DH	587
9f10523f	588	spin_lock(&object->lock);
952efe7b DH	589
	590	_debug("GC DEFERRED REL OBJ%x OP%x",
	591	object->debug_id, op->debug_id);
	592	fscache_stat(&fscache_n_op_gc);
	593
	594	ASSERTCMP(atomic_read(&op->usage), ==, 0);
9f10523f	595	ASSERTCMP(op->state, ==, FSCACHE_OP_ST_DEAD);
952efe7b DH	596
	597	ASSERTCMP(object->n_ops, >, 0);
	598	object->n_ops--;
	599	if (object->n_ops == 0)
	600	fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
	601
	602	spin_unlock(&object->lock);
9f10523f	603	kfree(op);
952efe7b DH	604
	605	} while (count++ < 20);
	606
	607	if (!list_empty(&cache->op_gc_list))
	608	schedule_work(&cache->op_gc);
	609
	610	_leave("");
	611	}
	612
	613	/*
8af7c124 TH	614	* execute an operation using fs_op_wq to provide processing context -
8af7c124 TH	615	* the caller holds a ref to this object, so we don't need to hold one
952efe7b	616	*/
8af7c124	617	void fscache_op_work_func(struct work_struct *work)
952efe7b DH	618	{
952efe7b DH	619	struct fscache_operation *op =
8af7c124	620	container_of(work, struct fscache_operation, work);
952efe7b DH	621	unsigned long start;
	622
	623	_enter("{OBJ%x OP%x,%d}",
	624	op->object->debug_id, op->debug_id, atomic_read(&op->usage));
	625
08c2e3d0 DH	626	trace_fscache_op(op->object->cookie, op, fscache_op_work);
08c2e3d0 DH	627
952efe7b DH	628	ASSERT(op->processor != NULL);
	629	start = jiffies;
	630	op->processor(op);
	631	fscache_hist(fscache_ops_histogram, start);
8af7c124	632	fscache_put_operation(op);
952efe7b DH	633
	634	_leave("");
	635	}