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