[linux-2.6-block.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 aquisition.
 *
 * 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 pool metadata");
#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
			DMERR("previously held here:");
			print_stack_trace(lock->traces + i, 4);

			DMERR("subsequent aquisition 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))
			return r;
		aux->validator = v;
	} else {
		if (unlikely(aux->validator != v)) {
			DMERR("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_and_unlock(struct dm_block_manager *bm,
			   struct dm_block *superblock)
{
	int r;

	if (bm->read_only)
		return -EPERM;

	r = dm_bufio_write_dirty_buffers(bm->bufio);
	if (unlikely(r)) {
		dm_bm_unlock(superblock);
		return r;
	}

	dm_bm_unlock(superblock);

	return dm_bufio_write_dirty_buffers(bm->bufio);
}

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

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
	28	* trace is also emitted for the previous lock aquisition.
	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) {
107	DMERR("recursive lock detected in pool metadata");
108	#ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
109	DMERR("previously held here:");
110	print_stack_trace(lock->traces + i, 4);
111
112	DMERR("subsequent aquisition attempted here:");
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));
3241b1d3 JT	431	if (unlikely(r))
	432	return r;
	433	aux->validator = v;
	434	} else {
	435	if (unlikely(aux->validator != v)) {
	436	DMERR("validator mismatch (old=%s vs new=%s) for block %llu",
	437	aux->validator->name, v ? v->name : "NULL",
	438	(unsigned long long)
	439	dm_bufio_get_block_number(buf));
	440	return -EINVAL;
	441	}
	442	}
	443
	444	return 0;
	445	}
	446	int dm_bm_read_lock(struct dm_block_manager *bm, dm_block_t b,
	447	struct dm_block_validator *v,
	448	struct dm_block **result)
	449	{
	450	struct buffer_aux *aux;
	451	void *p;
	452	int r;
	453
51a0f659	454	p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	455	if (unlikely(IS_ERR(p)))
	456	return PTR_ERR(p);
	457
	458	aux = dm_bufio_get_aux_data(to_buffer(*result));
	459	r = bl_down_read(&aux->lock);
	460	if (unlikely(r)) {
	461	dm_bufio_release(to_buffer(*result));
	462	report_recursive_bug(b, r);
	463	return r;
	464	}
	465
	466	aux->write_locked = 0;
	467
	468	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	469	if (unlikely(r)) {
	470	bl_up_read(&aux->lock);
	471	dm_bufio_release(to_buffer(*result));
	472	return r;
	473	}
	474
	475	return 0;
	476	}
	477	EXPORT_SYMBOL_GPL(dm_bm_read_lock);
	478
	479	int dm_bm_write_lock(struct dm_block_manager *bm,
	480	dm_block_t b, struct dm_block_validator *v,
	481	struct dm_block **result)
	482	{
	483	struct buffer_aux *aux;
	484	void *p;
	485	int r;
	486
31097557 JT	487	if (bm->read_only)
	488	return -EPERM;
	489
51a0f659	490	p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	491	if (unlikely(IS_ERR(p)))
	492	return PTR_ERR(p);
	493
	494	aux = dm_bufio_get_aux_data(to_buffer(*result));
	495	r = bl_down_write(&aux->lock);
	496	if (r) {
	497	dm_bufio_release(to_buffer(*result));
	498	report_recursive_bug(b, r);
	499	return r;
	500	}
	501
	502	aux->write_locked = 1;
	503
	504	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	505	if (unlikely(r)) {
	506	bl_up_write(&aux->lock);
	507	dm_bufio_release(to_buffer(*result));
	508	return r;
	509	}
	510
	511	return 0;
	512	}
	513	EXPORT_SYMBOL_GPL(dm_bm_write_lock);
	514
	515	int dm_bm_read_try_lock(struct dm_block_manager *bm,
	516	dm_block_t b, struct dm_block_validator *v,
	517	struct dm_block **result)
	518	{
	519	struct buffer_aux *aux;
	520	void *p;
	521	int r;
	522
51a0f659	523	p = dm_bufio_get(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	524	if (unlikely(IS_ERR(p)))
	525	return PTR_ERR(p);
	526	if (unlikely(!p))
	527	return -EWOULDBLOCK;
	528
	529	aux = dm_bufio_get_aux_data(to_buffer(*result));
	530	r = bl_down_read_nonblock(&aux->lock);
	531	if (r < 0) {
	532	dm_bufio_release(to_buffer(*result));
	533	report_recursive_bug(b, r);
	534	return r;
	535	}
	536	aux->write_locked = 0;
	537
	538	r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
	539	if (unlikely(r)) {
	540	bl_up_read(&aux->lock);
	541	dm_bufio_release(to_buffer(*result));
	542	return r;
	543	}
	544
	545	return 0;
	546	}
	547
	548	int dm_bm_write_lock_zero(struct dm_block_manager *bm,
	549	dm_block_t b, struct dm_block_validator *v,
	550	struct dm_block **result)
	551	{
	552	int r;
	553	struct buffer_aux *aux;
	554	void *p;
	555
31097557 JT	556	if (bm->read_only)
	557	return -EPERM;
	558
51a0f659	559	p = dm_bufio_new(bm->bufio, b, (struct dm_buffer **) result);
3241b1d3 JT	560	if (unlikely(IS_ERR(p)))
	561	return PTR_ERR(p);
	562
	563	memset(p, 0, dm_bm_block_size(bm));
	564
	565	aux = dm_bufio_get_aux_data(to_buffer(*result));
	566	r = bl_down_write(&aux->lock);
	567	if (r) {
	568	dm_bufio_release(to_buffer(*result));
	569	return r;
	570	}
	571
	572	aux->write_locked = 1;
	573	aux->validator = v;
	574
	575	return 0;
	576	}
384ef0e6	577	EXPORT_SYMBOL_GPL(dm_bm_write_lock_zero);
3241b1d3 JT	578
	579	int dm_bm_unlock(struct dm_block *b)
	580	{
	581	struct buffer_aux *aux;
	582	aux = dm_bufio_get_aux_data(to_buffer(b));
	583
	584	if (aux->write_locked) {
	585	dm_bufio_mark_buffer_dirty(to_buffer(b));
	586	bl_up_write(&aux->lock);
	587	} else
	588	bl_up_read(&aux->lock);
	589
	590	dm_bufio_release(to_buffer(b));
	591
	592	return 0;
	593	}
	594	EXPORT_SYMBOL_GPL(dm_bm_unlock);
	595
3241b1d3 JT	596	int dm_bm_flush_and_unlock(struct dm_block_manager *bm,
	597	struct dm_block *superblock)
	598	{
	599	int r;
	600
31097557 JT	601	if (bm->read_only)
	602	return -EPERM;
	603
51a0f659	604	r = dm_bufio_write_dirty_buffers(bm->bufio);
8d44c98a JT	605	if (unlikely(r)) {
8d44c98a JT	606	dm_bm_unlock(superblock);
3241b1d3	607	return r;
8d44c98a	608	}
3241b1d3 JT	609
	610	dm_bm_unlock(superblock);
	611
51a0f659	612	return dm_bufio_write_dirty_buffers(bm->bufio);
3241b1d3 JT	613	}
3241b1d3 JT	614
31097557 JT	615	void dm_bm_set_read_only(struct dm_block_manager *bm)
	616	{
	617	bm->read_only = true;
	618	}
	619	EXPORT_SYMBOL_GPL(dm_bm_set_read_only);
	620
3241b1d3 JT	621	u32 dm_bm_checksum(const void *data, size_t len, u32 init_xor)
	622	{
	623	return crc32c(~(u32) 0, data, len) ^ init_xor;
	624	}
	625	EXPORT_SYMBOL_GPL(dm_bm_checksum);
	626
	627	/----------------------------------------------------------------/
	628
	629	MODULE_LICENSE("GPL");
	630	MODULE_AUTHOR("Joe Thornber <dm-devel@redhat.com>");
	631	MODULE_DESCRIPTION("Immutable metadata library for dm");
	632
	633	/----------------------------------------------------------------/