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

	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->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else if (unlikely(fscache_cache_is_broken(object))) {
		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->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else {
		fscache_report_unexpected_submission(object, op, ostate);
		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));

	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->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else if (unlikely(fscache_cache_is_broken(object))) {
		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->state = FSCACHE_OP_ST_CANCELLED;
		ret = -ENOBUFS;
	} else {
		fscache_report_unexpected_submission(object, op, ostate);
		ASSERT(!fscache_object_is_active(object));
		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,
		      void (*do_cancel)(struct fscache_operation *),
		      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);

	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);
		if (do_cancel)
			do_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) {
		fscache_stat(&fscache_n_op_cancelled);
		if (do_cancel)
			do_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);

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

	op->state = cancelled ?
		FSCACHE_OP_ST_CANCELLED : FSCACHE_OP_ST_COMPLETE;

	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;

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

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

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