[deliverable/linux.git] / drivers / md / persistent-data / dm-block-manager.c

/*
 * Copyright (C) 2011 Red Hat, Inc.
 *
 * This file is released under the GPL.
 */
#include "dm-block-manager.h"
#include "dm-persistent-data-internal.h"
#include "../dm-bufio.h"

#include <linux/crc32c.h>
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/rwsem.h>
#include <linux/device-mapper.h>
#include <linux/stacktrace.h>

#define DM_MSG_PREFIX "block manager"

/*----------------------------------------------------------------*/

/*
 * This is a read/write semaphore with a couple of differences.
 *
 * i) There is a restriction on the number of concurrent read locks that
 * may be held at once.  This is just an implementation detail.
 *
 * ii) Recursive locking attempts are detected and return EINVAL.  A stack
 * trace is also emitted for the previous lock acquisition.
 *
 * iii) Priority is given to write locks.
 */
#define MAX_HOLDERS 4
#define MAX_STACK 10

typedef unsigned long stack_entries[MAX_STACK];

struct block_lock {
	spinlock_t lock;
	__s32 count;
	struct list_head waiters;
	struct task_struct *holders[MAX_HOLDERS];

#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	struct stack_trace traces[MAX_HOLDERS];
	stack_entries entries[MAX_HOLDERS];
#endif
};

struct waiter {
	struct list_head list;
	struct task_struct *task;
	int wants_write;
};

static unsigned __find_holder(struct block_lock *lock,
			      struct task_struct *task)
{
	unsigned i;

	for (i = 0; i < MAX_HOLDERS; i++)
		if (lock->holders[i] == task)
			break;

	BUG_ON(i == MAX_HOLDERS);
	return i;
}

/* call this *after* you increment lock->count */
static void __add_holder(struct block_lock *lock, struct task_struct *task)
{
	unsigned h = __find_holder(lock, NULL);
#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	struct stack_trace *t;
#endif

	get_task_struct(task);
	lock->holders[h] = task;

#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	t = lock->traces + h;
	t->nr_entries = 0;
	t->max_entries = MAX_STACK;
	t->entries = lock->entries[h];
	t->skip = 2;
	save_stack_trace(t);
#endif
}

/* call this *before* you decrement lock->count */
static void __del_holder(struct block_lock *lock, struct task_struct *task)
{
	unsigned h = __find_holder(lock, task);
	lock->holders[h] = NULL;
	put_task_struct(task);
}

static int __check_holder(struct block_lock *lock)
{
	unsigned i;
#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	static struct stack_trace t;
	static stack_entries entries;
#endif

	for (i = 0; i < MAX_HOLDERS; i++) {
		if (lock->holders[i] == current) {
			DMERR("recursive lock detected in metadata");
#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
			DMERR("previously held here:");
			print_stack_trace(lock->traces + i, 4);

			DMERR("subsequent acquisition attempted here:");
			t.nr_entries = 0;
			t.max_entries = MAX_STACK;
			t.entries = entries;
			t.skip = 3;
			save_stack_trace(&t);
			print_stack_trace(&t, 4);
#endif
			return -EINVAL;
		}
	}

	return 0;
}

static void __wait(struct waiter *w)
{
	for (;;) {
		set_task_state(current, TASK_UNINTERRUPTIBLE);

		if (!w->task)
			break;

		schedule();
	}

	set_task_state(current, TASK_RUNNING);
}

static void __wake_waiter(struct waiter *w)
{
	struct task_struct *task;

	list_del(&w->list);
	task = w->task;
	smp_mb();
	w->task = NULL;
	wake_up_process(task);
}

/*
 * We either wake a few readers or a single writer.
 */
static void __wake_many(struct block_lock *lock)
{
	struct waiter *w, *tmp;

	BUG_ON(lock->count < 0);
	list_for_each_entry_safe(w, tmp, &lock->waiters, list) {
		if (lock->count >= MAX_HOLDERS)
			return;

		if (w->wants_write) {
			if (lock->count > 0)
				return; /* still read locked */

			lock->count = -1;
			__add_holder(lock, w->task);
			__wake_waiter(w);
			return;
		}

		lock->count++;
		__add_holder(lock, w->task);
		__wake_waiter(w);
	}
}

static void bl_init(struct block_lock *lock)
{
	int i;

	spin_lock_init(&lock->lock);
	lock->count = 0;
	INIT_LIST_HEAD(&lock->waiters);
	for (i = 0; i < MAX_HOLDERS; i++)
		lock->holders[i] = NULL;
}

static int __available_for_read(struct block_lock *lock)
{
	return lock->count >= 0 &&
		lock->count < MAX_HOLDERS &&
		list_empty(&lock->waiters);
}

static int bl_down_read(struct block_lock *lock)
{
	int r;
	struct waiter w;

	spin_lock(&lock->lock);
	r = __check_holder(lock);
	if (r) {
		spin_unlock(&lock->lock);
		return r;
	}

	if (__available_for_read(lock)) {
		lock->count++;
		__add_holder(lock, current);
		spin_unlock(&lock->lock);
		return 0;
	}

	get_task_struct(current);

	w.task = current;
	w.wants_write = 0;
	list_add_tail(&w.list, &lock->waiters);
	spin_unlock(&lock->lock);

	__wait(&w);
	put_task_struct(current);
	return 0;
}

static int bl_down_read_nonblock(struct block_lock *lock)
{
	int r;

	spin_lock(&lock->lock);
	r = __check_holder(lock);
	if (r)
		goto out;

	if (__available_for_read(lock)) {
		lock->count++;
		__add_holder(lock, current);
		r = 0;
	} else
		r = -EWOULDBLOCK;

out:
	spin_unlock(&lock->lock);
	return r;
}

static void bl_up_read(struct block_lock *lock)
{
	spin_lock(&lock->lock);
	BUG_ON(lock->count <= 0);
	__del_holder(lock, current);
	--lock->count;
	if (!list_empty(&lock->waiters))
		__wake_many(lock);
	spin_unlock(&lock->lock);
}

static int bl_down_write(struct block_lock *lock)
{
	int r;
	struct waiter w;

	spin_lock(&lock->lock);
	r = __check_holder(lock);
	if (r) {
		spin_unlock(&lock->lock);
		return r;
	}

	if (lock->count == 0 && list_empty(&lock->waiters)) {
		lock->count = -1;
		__add_holder(lock, current);
		spin_unlock(&lock->lock);
		return 0;
	}

	get_task_struct(current);
	w.task = current;
	w.wants_write = 1;

	/*
	 * Writers given priority. We know there's only one mutator in the
	 * system, so ignoring the ordering reversal.
	 */
	list_add(&w.list, &lock->waiters);
	spin_unlock(&lock->lock);

	__wait(&w);
	put_task_struct(current);

	return 0;
}

static void bl_up_write(struct block_lock *lock)
{
	spin_lock(&lock->lock);
	__del_holder(lock, current);
	lock->count = 0;
	if (!list_empty(&lock->waiters))
		__wake_many(lock);
	spin_unlock(&lock->lock);
}

static void report_recursive_bug(dm_block_t b, int r)
{
	if (r == -EINVAL)
		DMERR("recursive acquisition of block %llu requested.",
		      (unsigned long long) b);
}

/*----------------------------------------------------------------*/

/*
 * Block manager is currently implemented using dm-bufio.  struct
 * dm_block_manager and struct dm_block map directly onto a couple of
 * structs in the bufio interface.  I want to retain the freedom to move
 * away from bufio in the future.  So these structs are just cast within
 * this .c file, rather than making it through to the public interface.
 */
static struct dm_buffer *to_buffer(struct dm_block *b)
{
	return (struct dm_buffer *) b;
}

dm_block_t dm_block_location(struct dm_block *b)
{
	return dm_bufio_get_block_number(to_buffer(b));
}
EXPORT_SYMBOL_GPL(dm_block_location);

void *dm_block_data(struct dm_block *b)
{
	return dm_bufio_get_block_data(to_buffer(b));
}
EXPORT_SYMBOL_GPL(dm_block_data);

struct buffer_aux {
	struct dm_block_validator *validator;
	struct block_lock lock;
	int write_locked;
};

static void dm_block_manager_alloc_callback(struct dm_buffer *buf)
{
	struct buffer_aux *aux = dm_bufio_get_aux_data(buf);
	aux->validator = NULL;
	bl_init(&aux->lock);
}

static void dm_block_manager_write_callback(struct dm_buffer *buf)
{
	struct buffer_aux *aux = dm_bufio_get_aux_data(buf);
	if (aux->validator) {
		aux->validator->prepare_for_write(aux->validator, (struct dm_block *) buf,
			 dm_bufio_get_block_size(dm_bufio_get_client(buf)));
	}
}

/*----------------------------------------------------------------
 * Public interface
 *--------------------------------------------------------------*/
struct dm_block_manager {
	struct dm_bufio_client *bufio;
	bool read_only:1;
};

struct dm_block_manager *dm_block_manager_create(struct block_device *bdev,
						 unsigned block_size,
						 unsigned cache_size,
						 unsigned max_held_per_thread)
{
	int r;
	struct dm_block_manager *bm;

	bm = kmalloc(sizeof(*bm), GFP_KERNEL);
	if (!bm) {
		r = -ENOMEM;
		goto bad;
	}

	bm->bufio = dm_bufio_client_create(bdev, block_size, max_held_per_thread,
					   sizeof(struct buffer_aux),
					   dm_block_manager_alloc_callback,
					   dm_block_manager_write_callback);
	if (IS_ERR(bm->bufio)) {
		r = PTR_ERR(bm->bufio);
		kfree(bm);
		goto bad;
	}

	bm->read_only = false;

	return bm;

bad:
	return ERR_PTR(r);
}
EXPORT_SYMBOL_GPL(dm_block_manager_create);

void dm_block_manager_destroy(struct dm_block_manager *bm)
{
	dm_bufio_client_destroy(bm->bufio);
	kfree(bm);
}
EXPORT_SYMBOL_GPL(dm_block_manager_destroy);

unsigned dm_bm_block_size(struct dm_block_manager *bm)
{
	return dm_bufio_get_block_size(bm->bufio);
}
EXPORT_SYMBOL_GPL(dm_bm_block_size);

dm_block_t dm_bm_nr_blocks(struct dm_block_manager *bm)
{
	return dm_bufio_get_device_size(bm->bufio);
}

static int dm_bm_validate_buffer(struct dm_block_manager *bm,
				 struct dm_buffer *buf,
				 struct buffer_aux *aux,
				 struct dm_block_validator *v)
{
	if (unlikely(!aux->validator)) {
		int r;
		if (!v)
			return 0;
		r = v->check(v, (struct dm_block *) buf, dm_bufio_get_block_size(bm->bufio));
		if (unlikely(r)) {
			DMERR_LIMIT("%s validator check failed for block %llu", v->name,
				    (unsigned long long) dm_bufio_get_block_number(buf));
			return r;
		}
		aux->validator = v;
	} else {
		if (unlikely(aux->validator != v)) {
			DMERR_LIMIT("validator mismatch (old=%s vs new=%s) for block %llu",
				    aux->validator->name, v ? v->name : "NULL",
				    (unsigned long long) dm_bufio_get_block_number(buf));
			return -EINVAL;
		}
	}

	return 0;
}
int dm_bm_read_lock(struct dm_block_manager *bm, dm_block_t b,
		    struct dm_block_validator *v,
		    struct dm_block **result)
{
	struct buffer_aux *aux;
	void *p;
	int r;

	p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
	if (unlikely(IS_ERR(p)))
		return PTR_ERR(p);

	aux = dm_bufio_get_aux_data(to_buffer(*result));
	r = bl_down_read(&aux->lock);
	if (unlikely(r)) {
		dm_bufio_release(to_buffer(*result));
		report_recursive_bug(b, r);
		return r;
	}

	aux->write_locked = 0;

	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	if (unlikely(r)) {
		bl_up_read(&aux->lock);
		dm_bufio_release(to_buffer(*result));
		return r;
	}

	return 0;
}
EXPORT_SYMBOL_GPL(dm_bm_read_lock);

int dm_bm_write_lock(struct dm_block_manager *bm,
		     dm_block_t b, struct dm_block_validator *v,
		     struct dm_block **result)
{
	struct buffer_aux *aux;
	void *p;
	int r;

	if (bm->read_only)
		return -EPERM;

	p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
	if (unlikely(IS_ERR(p)))
		return PTR_ERR(p);

	aux = dm_bufio_get_aux_data(to_buffer(*result));
	r = bl_down_write(&aux->lock);
	if (r) {
		dm_bufio_release(to_buffer(*result));
		report_recursive_bug(b, r);
		return r;
	}

	aux->write_locked = 1;

	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	if (unlikely(r)) {
		bl_up_write(&aux->lock);
		dm_bufio_release(to_buffer(*result));
		return r;
	}

	return 0;
}
EXPORT_SYMBOL_GPL(dm_bm_write_lock);

int dm_bm_read_try_lock(struct dm_block_manager *bm,
			dm_block_t b, struct dm_block_validator *v,
			struct dm_block **result)
{
	struct buffer_aux *aux;
	void *p;
	int r;

	p = dm_bufio_get(bm->bufio, b, (struct dm_buffer **) result);
	if (unlikely(IS_ERR(p)))
		return PTR_ERR(p);
	if (unlikely(!p))
		return -EWOULDBLOCK;

	aux = dm_bufio_get_aux_data(to_buffer(*result));
	r = bl_down_read_nonblock(&aux->lock);
	if (r < 0) {
		dm_bufio_release(to_buffer(*result));
		report_recursive_bug(b, r);
		return r;
	}
	aux->write_locked = 0;

	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	if (unlikely(r)) {
		bl_up_read(&aux->lock);
		dm_bufio_release(to_buffer(*result));
		return r;
	}

	return 0;
}

int dm_bm_write_lock_zero(struct dm_block_manager *bm,
			  dm_block_t b, struct dm_block_validator *v,
			  struct dm_block **result)
{
	int r;
	struct buffer_aux *aux;
	void *p;

	if (bm->read_only)
		return -EPERM;

	p = dm_bufio_new(bm->bufio, b, (struct dm_buffer **) result);
	if (unlikely(IS_ERR(p)))
		return PTR_ERR(p);

	memset(p, 0, dm_bm_block_size(bm));

	aux = dm_bufio_get_aux_data(to_buffer(*result));
	r = bl_down_write(&aux->lock);
	if (r) {
		dm_bufio_release(to_buffer(*result));
		return r;
	}

	aux->write_locked = 1;
	aux->validator = v;

	return 0;
}
EXPORT_SYMBOL_GPL(dm_bm_write_lock_zero);

int dm_bm_unlock(struct dm_block *b)
{
	struct buffer_aux *aux;
	aux = dm_bufio_get_aux_data(to_buffer(b));

	if (aux->write_locked) {
		dm_bufio_mark_buffer_dirty(to_buffer(b));
		bl_up_write(&aux->lock);
	} else
		bl_up_read(&aux->lock);

	dm_bufio_release(to_buffer(b));

	return 0;
}
EXPORT_SYMBOL_GPL(dm_bm_unlock);

int dm_bm_flush(struct dm_block_manager *bm)
{
	if (bm->read_only)
		return -EPERM;

	return dm_bufio_write_dirty_buffers(bm->bufio);
}
EXPORT_SYMBOL_GPL(dm_bm_flush);

void dm_bm_prefetch(struct dm_block_manager *bm, dm_block_t b)
{
	dm_bufio_prefetch(bm->bufio, b, 1);
}

bool dm_bm_is_read_only(struct dm_block_manager *bm)
{
	return bm->read_only;
}
EXPORT_SYMBOL_GPL(dm_bm_is_read_only);

void dm_bm_set_read_only(struct dm_block_manager *bm)
{
	bm->read_only = true;
}
EXPORT_SYMBOL_GPL(dm_bm_set_read_only);

void dm_bm_set_read_write(struct dm_block_manager *bm)
{
	bm->read_only = false;
}
EXPORT_SYMBOL_GPL(dm_bm_set_read_write);

u32 dm_bm_checksum(const void *data, size_t len, u32 init_xor)
{
	return crc32c(~(u32) 0, data, len) ^ init_xor;
}
EXPORT_SYMBOL_GPL(dm_bm_checksum);

/*----------------------------------------------------------------*/

MODULE_LICENSE("GPL");
MODULE_AUTHOR("Joe Thornber <dm-devel@redhat.com>");
MODULE_DESCRIPTION("Immutable metadata library for dm");

/*----------------------------------------------------------------*/
Commit	Line	Data
3241b1d3 JT	1	/*
	2	* Copyright (C) 2011 Red Hat, Inc.
	3	*
	4	* This file is released under the GPL.
	5	*/
	6	#include "dm-block-manager.h"
	7	#include "dm-persistent-data-internal.h"
	8	#include "../dm-bufio.h"
	9
	10	#include <linux/crc32c.h>
	11	#include <linux/module.h>
	12	#include <linux/slab.h>
	13	#include <linux/rwsem.h>
	14	#include <linux/device-mapper.h>
	15	#include <linux/stacktrace.h>
	16
	17	#define DM_MSG_PREFIX "block manager"
	18
	19	/----------------------------------------------------------------/
	20
	21	/*
	22	* This is a read/write semaphore with a couple of differences.
	23	*
	24	* i) There is a restriction on the number of concurrent read locks that
	25	* may be held at once. This is just an implementation detail.
	26	*
	27	* ii) Recursive locking attempts are detected and return EINVAL. A stack
83f0d77a	28	* trace is also emitted for the previous lock acquisition.
3241b1d3 JT	29	*
	30	* iii) Priority is given to write locks.
	31	*/
	32	#define MAX_HOLDERS 4
	33	#define MAX_STACK 10
	34
	35	typedef unsigned long stack_entries[MAX_STACK];
	36
	37	struct block_lock {
	38	spinlock_t lock;
	39	__s32 count;
	40	struct list_head waiters;
	41	struct task_struct *holders[MAX_HOLDERS];
	42
	43	#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	44	struct stack_trace traces[MAX_HOLDERS];
	45	stack_entries entries[MAX_HOLDERS];
	46	#endif
	47	};
	48
	49	struct waiter {
	50	struct list_head list;
	51	struct task_struct *task;
	52	int wants_write;
	53	};
	54
	55	static unsigned __find_holder(struct block_lock *lock,
	56	struct task_struct *task)
	57	{
	58	unsigned i;
	59
	60	for (i = 0; i < MAX_HOLDERS; i++)
	61	if (lock->holders[i] == task)
	62	break;
	63
	64	BUG_ON(i == MAX_HOLDERS);
	65	return i;
	66	}
	67
	68	/* call this after you increment lock->count */
	69	static void __add_holder(struct block_lock lock, struct task_struct task)
	70	{
	71	unsigned h = __find_holder(lock, NULL);
	72	#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	73	struct stack_trace *t;
	74	#endif
	75
	76	get_task_struct(task);
	77	lock->holders[h] = task;
	78
	79	#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	80	t = lock->traces + h;
	81	t->nr_entries = 0;
	82	t->max_entries = MAX_STACK;
	83	t->entries = lock->entries[h];
	84	t->skip = 2;
	85	save_stack_trace(t);
	86	#endif
	87	}
	88
	89	/* call this before you decrement lock->count */
	90	static void __del_holder(struct block_lock lock, struct task_struct task)
	91	{
	92	unsigned h = __find_holder(lock, task);
93	lock->holders[h] = NULL;
94	put_task_struct(task);
95	}
96
97	static int __check_holder(struct block_lock *lock)
98	{
99	unsigned i;
100	#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
101	static struct stack_trace t;
102	static stack_entries entries;
103	#endif
104
105	for (i = 0; i < MAX_HOLDERS; i++) {
106	if (lock->holders[i] == current) {
10343180	107	DMERR("recursive lock detected in metadata");
3241b1d3 JT	108	#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
	109	DMERR("previously held here:");
	110	print_stack_trace(lock->traces + i, 4);
	111
83f0d77a	112	DMERR("subsequent acquisition attempted here:");
3241b1d3 JT	113	t.nr_entries = 0;
	114	t.max_entries = MAX_STACK;
	115	t.entries = entries;
	116	t.skip = 3;
	117	save_stack_trace(&t);
	118	print_stack_trace(&t, 4);
	119	#endif
	120	return -EINVAL;
	121	}
	122	}
	123
	124	return 0;
	125	}
	126
	127	static void __wait(struct waiter *w)
	128	{
	129	for (;;) {
	130	set_task_state(current, TASK_UNINTERRUPTIBLE);
	131
	132	if (!w->task)
	133	break;
	134
	135	schedule();
	136	}
	137
	138	set_task_state(current, TASK_RUNNING);
	139	}
	140
	141	static void __wake_waiter(struct waiter *w)
	142	{
	143	struct task_struct *task;
	144
	145	list_del(&w->list);
	146	task = w->task;
	147	smp_mb();
	148	w->task = NULL;
	149	wake_up_process(task);
	150	}
	151
	152	/*
	153	* We either wake a few readers or a single writer.
	154	*/
	155	static void __wake_many(struct block_lock *lock)
	156	{
	157	struct waiter w, tmp;
	158
	159	BUG_ON(lock->count < 0);
	160	list_for_each_entry_safe(w, tmp, &lock->waiters, list) {
	161	if (lock->count >= MAX_HOLDERS)
	162	return;
	163
	164	if (w->wants_write) {
	165	if (lock->count > 0)
	166	return; /* still read locked */
	167
	168	lock->count = -1;
	169	__add_holder(lock, w->task);
	170	__wake_waiter(w);
	171	return;
	172	}
	173
	174	lock->count++;
	175	__add_holder(lock, w->task);
	176	__wake_waiter(w);
177	}
178	}
179
180	static void bl_init(struct block_lock *lock)
181	{
182	int i;
183
184	spin_lock_init(&lock->lock);
185	lock->count = 0;
186	INIT_LIST_HEAD(&lock->waiters);
187	for (i = 0; i < MAX_HOLDERS; i++)
188	lock->holders[i] = NULL;
189	}
190
191	static int __available_for_read(struct block_lock *lock)
192	{
193	return lock->count >= 0 &&
194	lock->count < MAX_HOLDERS &&
195	list_empty(&lock->waiters);
196	}
197
198	static int bl_down_read(struct block_lock *lock)
199	{
200	int r;
201	struct waiter w;
202
203	spin_lock(&lock->lock);
204	r = __check_holder(lock);
205	if (r) {
206	spin_unlock(&lock->lock);
207	return r;
208	}
209
210	if (__available_for_read(lock)) {
211	lock->count++;
212	__add_holder(lock, current);
213	spin_unlock(&lock->lock);
214	return 0;
215	}
216
217	get_task_struct(current);
218
219	w.task = current;
220	w.wants_write = 0;
221	list_add_tail(&w.list, &lock->waiters);
222	spin_unlock(&lock->lock);
223
224	__wait(&w);
225	put_task_struct(current);
226	return 0;
227	}
228
229	static int bl_down_read_nonblock(struct block_lock *lock)
230	{
231	int r;
232
233	spin_lock(&lock->lock);
234	r = __check_holder(lock);
235	if (r)
236	goto out;
237
238	if (__available_for_read(lock)) {
239	lock->count++;
240	__add_holder(lock, current);
241	r = 0;
242	} else
243	r = -EWOULDBLOCK;
244
245	out:
246	spin_unlock(&lock->lock);
247	return r;
248	}
249
250	static void bl_up_read(struct block_lock *lock)
251	{
252	spin_lock(&lock->lock);
253	BUG_ON(lock->count <= 0);
254	__del_holder(lock, current);
255	--lock->count;
256	if (!list_empty(&lock->waiters))
257	__wake_many(lock);
258	spin_unlock(&lock->lock);
259	}
260
261	static int bl_down_write(struct block_lock *lock)
262	{
263	int r;
264	struct waiter w;
265
266	spin_lock(&lock->lock);
267	r = __check_holder(lock);
268	if (r) {
269	spin_unlock(&lock->lock);
270	return r;
271	}
272
273	if (lock->count == 0 && list_empty(&lock->waiters)) {
274	lock->count = -1;
275	__add_holder(lock, current);
276	spin_unlock(&lock->lock);
277	return 0;
278	}
279
280	get_task_struct(current);
281	w.task = current;
282	w.wants_write = 1;
283
284	/*
285	* Writers given priority. We know there's only one mutator in the
286	* system, so ignoring the ordering reversal.
287	*/
288	list_add(&w.list, &lock->waiters);
289	spin_unlock(&lock->lock);
290
291	__wait(&w);
292	put_task_struct(current);
293
294	return 0;
295	}
296
297	static void bl_up_write(struct block_lock *lock)
298	{
299	spin_lock(&lock->lock);
300	__del_holder(lock, current);
301	lock->count = 0;
302	if (!list_empty(&lock->waiters))
303	__wake_many(lock);
304	spin_unlock(&lock->lock);
305	}
306
307	static void report_recursive_bug(dm_block_t b, int r)
308	{
309	if (r == -EINVAL)
310	DMERR("recursive acquisition of block %llu requested.",
311	(unsigned long long) b);
312	}
313
314	/----------------------------------------------------------------/
315
316	/*
317	* Block manager is currently implemented using dm-bufio. struct
318	* dm_block_manager and struct dm_block map directly onto a couple of
319	* structs in the bufio interface. I want to retain the freedom to move
320	* away from bufio in the future. So these structs are just cast within
321	* this .c file, rather than making it through to the public interface.
322	*/
323	static struct dm_buffer to_buffer(struct dm_block b)
324	{
325	return (struct dm_buffer *) b;
326	}
327
3241b1d3 JT	328	dm_block_t dm_block_location(struct dm_block *b)
	329	{
	330	return dm_bufio_get_block_number(to_buffer(b));
	331	}
	332	EXPORT_SYMBOL_GPL(dm_block_location);
	333
	334	void dm_block_data(struct dm_block b)
	335	{
	336	return dm_bufio_get_block_data(to_buffer(b));
	337	}
	338	EXPORT_SYMBOL_GPL(dm_block_data);
	339
	340	struct buffer_aux {
	341	struct dm_block_validator *validator;
	342	struct block_lock lock;
	343	int write_locked;
	344	};
	345
	346	static void dm_block_manager_alloc_callback(struct dm_buffer *buf)
	347	{
	348	struct buffer_aux *aux = dm_bufio_get_aux_data(buf);
	349	aux->validator = NULL;
	350	bl_init(&aux->lock);
	351	}
	352
	353	static void dm_block_manager_write_callback(struct dm_buffer *buf)
	354	{
	355	struct buffer_aux *aux = dm_bufio_get_aux_data(buf);
	356	if (aux->validator) {
	357	aux->validator->prepare_for_write(aux->validator, (struct dm_block *) buf,
	358	dm_bufio_get_block_size(dm_bufio_get_client(buf)));
	359	}
	360	}
	361
	362	/*----------------------------------------------------------------
	363	* Public interface
	364	--------------------------------------------------------------/
51a0f659 JT	365	struct dm_block_manager {
51a0f659 JT	366	struct dm_bufio_client *bufio;
31097557	367	bool read_only:1;
51a0f659 JT	368	};
51a0f659 JT	369
3241b1d3 JT	370	struct dm_block_manager dm_block_manager_create(struct block_device bdev,
	371	unsigned block_size,
	372	unsigned cache_size,
	373	unsigned max_held_per_thread)
	374	{
51a0f659 JT	375	int r;
	376	struct dm_block_manager *bm;
	377
	378	bm = kmalloc(sizeof(*bm), GFP_KERNEL);
	379	if (!bm) {
	380	r = -ENOMEM;
	381	goto bad;
	382	}
	383
	384	bm->bufio = dm_bufio_client_create(bdev, block_size, max_held_per_thread,
	385	sizeof(struct buffer_aux),
	386	dm_block_manager_alloc_callback,
	387	dm_block_manager_write_callback);
	388	if (IS_ERR(bm->bufio)) {
	389	r = PTR_ERR(bm->bufio);
	390	kfree(bm);
	391	goto bad;
	392	}
	393
31097557 JT	394	bm->read_only = false;
31097557 JT	395
51a0f659 JT	396	return bm;
	397
	398	bad:
	399	return ERR_PTR(r);
3241b1d3 JT	400	}
	401	EXPORT_SYMBOL_GPL(dm_block_manager_create);
	402
	403	void dm_block_manager_destroy(struct dm_block_manager *bm)
	404	{
51a0f659 JT	405	dm_bufio_client_destroy(bm->bufio);
51a0f659 JT	406	kfree(bm);
3241b1d3 JT	407	}
	408	EXPORT_SYMBOL_GPL(dm_block_manager_destroy);
	409
	410	unsigned dm_bm_block_size(struct dm_block_manager *bm)
	411	{
51a0f659	412	return dm_bufio_get_block_size(bm->bufio);
3241b1d3 JT	413	}
	414	EXPORT_SYMBOL_GPL(dm_bm_block_size);
	415
	416	dm_block_t dm_bm_nr_blocks(struct dm_block_manager *bm)
	417	{
51a0f659	418	return dm_bufio_get_device_size(bm->bufio);
3241b1d3 JT	419	}
	420
	421	static int dm_bm_validate_buffer(struct dm_block_manager *bm,
	422	struct dm_buffer *buf,
	423	struct buffer_aux *aux,
	424	struct dm_block_validator *v)
	425	{
	426	if (unlikely(!aux->validator)) {
	427	int r;
	428	if (!v)
	429	return 0;
51a0f659	430	r = v->check(v, (struct dm_block *) buf, dm_bufio_get_block_size(bm->bufio));
a5bd968a	431	if (unlikely(r)) {
89ddeb8c MS	432	DMERR_LIMIT("%s validator check failed for block %llu", v->name,
89ddeb8c MS	433	(unsigned long long) dm_bufio_get_block_number(buf));
3241b1d3	434	return r;
a5bd968a	435	}
3241b1d3 JT	436	aux->validator = v;
	437	} else {
	438	if (unlikely(aux->validator != v)) {
89ddeb8c MS	439	DMERR_LIMIT("validator mismatch (old=%s vs new=%s) for block %llu",
	440	aux->validator->name, v ? v->name : "NULL",
	441	(unsigned long long) dm_bufio_get_block_number(buf));
3241b1d3 JT	442	return -EINVAL;
	443	}
	444	}
	445
	446	return 0;
	447	}
	448	int dm_bm_read_lock(struct dm_block_manager *bm, dm_block_t b,
	449	struct dm_block_validator *v,
	450	struct dm_block **result)
	451	{
	452	struct buffer_aux *aux;
	453	void *p;
	454	int r;
	455
51a0f659	456	p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	457	if (unlikely(IS_ERR(p)))
	458	return PTR_ERR(p);
	459
	460	aux = dm_bufio_get_aux_data(to_buffer(*result));
	461	r = bl_down_read(&aux->lock);
	462	if (unlikely(r)) {
	463	dm_bufio_release(to_buffer(*result));
	464	report_recursive_bug(b, r);
	465	return r;
	466	}
	467
	468	aux->write_locked = 0;
	469
	470	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	471	if (unlikely(r)) {
	472	bl_up_read(&aux->lock);
	473	dm_bufio_release(to_buffer(*result));
	474	return r;
	475	}
	476
	477	return 0;
	478	}
	479	EXPORT_SYMBOL_GPL(dm_bm_read_lock);
	480
	481	int dm_bm_write_lock(struct dm_block_manager *bm,
	482	dm_block_t b, struct dm_block_validator *v,
	483	struct dm_block **result)
	484	{
	485	struct buffer_aux *aux;
	486	void *p;
	487	int r;
	488
31097557 JT	489	if (bm->read_only)
	490	return -EPERM;
	491
51a0f659	492	p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	493	if (unlikely(IS_ERR(p)))
	494	return PTR_ERR(p);
	495
	496	aux = dm_bufio_get_aux_data(to_buffer(*result));
	497	r = bl_down_write(&aux->lock);
	498	if (r) {
	499	dm_bufio_release(to_buffer(*result));
	500	report_recursive_bug(b, r);
	501	return r;
	502	}
	503
	504	aux->write_locked = 1;
	505
	506	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	507	if (unlikely(r)) {
	508	bl_up_write(&aux->lock);
	509	dm_bufio_release(to_buffer(*result));
	510	return r;
	511	}
	512
	513	return 0;
	514	}
	515	EXPORT_SYMBOL_GPL(dm_bm_write_lock);
	516
	517	int dm_bm_read_try_lock(struct dm_block_manager *bm,
	518	dm_block_t b, struct dm_block_validator *v,
	519	struct dm_block **result)
	520	{
	521	struct buffer_aux *aux;
	522	void *p;
	523	int r;
	524
51a0f659	525	p = dm_bufio_get(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	526	if (unlikely(IS_ERR(p)))
	527	return PTR_ERR(p);
	528	if (unlikely(!p))
	529	return -EWOULDBLOCK;
	530
	531	aux = dm_bufio_get_aux_data(to_buffer(*result));
	532	r = bl_down_read_nonblock(&aux->lock);
	533	if (r < 0) {
	534	dm_bufio_release(to_buffer(*result));
	535	report_recursive_bug(b, r);
	536	return r;
	537	}
	538	aux->write_locked = 0;
	539
	540	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	541	if (unlikely(r)) {
	542	bl_up_read(&aux->lock);
	543	dm_bufio_release(to_buffer(*result));
	544	return r;
	545	}
	546
	547	return 0;
	548	}
	549
	550	int dm_bm_write_lock_zero(struct dm_block_manager *bm,
	551	dm_block_t b, struct dm_block_validator *v,
	552	struct dm_block **result)
	553	{
	554	int r;
	555	struct buffer_aux *aux;
	556	void *p;
	557
31097557 JT	558	if (bm->read_only)
	559	return -EPERM;
	560
51a0f659	561	p = dm_bufio_new(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	562	if (unlikely(IS_ERR(p)))
	563	return PTR_ERR(p);
	564
	565	memset(p, 0, dm_bm_block_size(bm));
	566
	567	aux = dm_bufio_get_aux_data(to_buffer(*result));
	568	r = bl_down_write(&aux->lock);
	569	if (r) {
	570	dm_bufio_release(to_buffer(*result));
	571	return r;
	572	}
	573
	574	aux->write_locked = 1;
	575	aux->validator = v;
	576
	577	return 0;
	578	}
384ef0e6	579	EXPORT_SYMBOL_GPL(dm_bm_write_lock_zero);
3241b1d3 JT	580
	581	int dm_bm_unlock(struct dm_block *b)
	582	{
	583	struct buffer_aux *aux;
	584	aux = dm_bufio_get_aux_data(to_buffer(b));
	585
	586	if (aux->write_locked) {
	587	dm_bufio_mark_buffer_dirty(to_buffer(b));
	588	bl_up_write(&aux->lock);
	589	} else
	590	bl_up_read(&aux->lock);
	591
	592	dm_bufio_release(to_buffer(b));
	593
	594	return 0;
	595	}
	596	EXPORT_SYMBOL_GPL(dm_bm_unlock);
	597
a9d45396	598	int dm_bm_flush(struct dm_block_manager *bm)
3241b1d3	599	{
31097557 JT	600	if (bm->read_only)
	601	return -EPERM;
	602
51a0f659	603	return dm_bufio_write_dirty_buffers(bm->bufio);
3241b1d3	604	}
a9d45396	605	EXPORT_SYMBOL_GPL(dm_bm_flush);
3241b1d3	606
04f17c80 JT	607	void dm_bm_prefetch(struct dm_block_manager *bm, dm_block_t b)
	608	{
	609	dm_bufio_prefetch(bm->bufio, b, 1);
	610	}
	611
49f154c7 MS	612	bool dm_bm_is_read_only(struct dm_block_manager *bm)
	613	{
	614	return bm->read_only;
	615	}
	616	EXPORT_SYMBOL_GPL(dm_bm_is_read_only);
	617
31097557 JT	618	void dm_bm_set_read_only(struct dm_block_manager *bm)
	619	{
	620	bm->read_only = true;
	621	}
	622	EXPORT_SYMBOL_GPL(dm_bm_set_read_only);
	623
9b7aaa64 JT	624	void dm_bm_set_read_write(struct dm_block_manager *bm)
	625	{
	626	bm->read_only = false;
	627	}
	628	EXPORT_SYMBOL_GPL(dm_bm_set_read_write);
	629
3241b1d3 JT	630	u32 dm_bm_checksum(const void *data, size_t len, u32 init_xor)
	631	{
	632	return crc32c(~(u32) 0, data, len) ^ init_xor;
	633	}
	634	EXPORT_SYMBOL_GPL(dm_bm_checksum);
	635
	636	/----------------------------------------------------------------/
	637
	638	MODULE_LICENSE("GPL");
	639	MODULE_AUTHOR("Joe Thornber <dm-devel@redhat.com>");
	640	MODULE_DESCRIPTION("Immutable metadata library for dm");
	641
	642	/----------------------------------------------------------------/