2 * Copyright (C) 2011 Red Hat, Inc.
4 * This file is released under the GPL.
6 #include "dm-block-manager.h"
7 #include "dm-persistent-data-internal.h"
8 #include "../dm-bufio.h"
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>
17 #define DM_MSG_PREFIX "block manager"
19 /*----------------------------------------------------------------*/
22 * This is a read/write semaphore with a couple of differences.
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.
27 * ii) Recursive locking attempts are detected and return EINVAL. A stack
28 * trace is also emitted for the previous lock acquisition.
30 * iii) Priority is given to write locks.
35 typedef unsigned long stack_entries[MAX_STACK];
40 struct list_head waiters;
41 struct task_struct *holders[MAX_HOLDERS];
43 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
44 struct stack_trace traces[MAX_HOLDERS];
45 stack_entries entries[MAX_HOLDERS];
50 struct list_head list;
51 struct task_struct *task;
55 static unsigned __find_holder(struct block_lock *lock,
56 struct task_struct *task)
60 for (i = 0; i < MAX_HOLDERS; i++)
61 if (lock->holders[i] == task)
64 BUG_ON(i == MAX_HOLDERS);
68 /* call this *after* you increment lock->count */
69 static void __add_holder(struct block_lock *lock, struct task_struct *task)
71 unsigned h = __find_holder(lock, NULL);
72 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
73 struct stack_trace *t;
76 get_task_struct(task);
77 lock->holders[h] = task;
79 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
82 t->max_entries = MAX_STACK;
83 t->entries = lock->entries[h];
89 /* call this *before* you decrement lock->count */
90 static void __del_holder(struct block_lock *lock, struct task_struct *task)
92 unsigned h = __find_holder(lock, task);
93 lock->holders[h] = NULL;
94 put_task_struct(task);
97 static int __check_holder(struct block_lock *lock)
101 for (i = 0; i < MAX_HOLDERS; i++) {
102 if (lock->holders[i] == current) {
103 DMERR("recursive lock detected in metadata");
104 #ifdef CONFIG_DM_DEBUG_BLOCK_STACK_TRACING
105 DMERR("previously held here:");
106 print_stack_trace(lock->traces + i, 4);
108 DMERR("subsequent acquisition attempted here:");
118 static void __wait(struct waiter *w)
121 set_task_state(current, TASK_UNINTERRUPTIBLE);
129 set_task_state(current, TASK_RUNNING);
132 static void __wake_waiter(struct waiter *w)
134 struct task_struct *task;
140 wake_up_process(task);
144 * We either wake a few readers or a single writer.
146 static void __wake_many(struct block_lock *lock)
148 struct waiter *w, *tmp;
150 BUG_ON(lock->count < 0);
151 list_for_each_entry_safe(w, tmp, &lock->waiters, list) {
152 if (lock->count >= MAX_HOLDERS)
155 if (w->wants_write) {
157 return; /* still read locked */
160 __add_holder(lock, w->task);
166 __add_holder(lock, w->task);
171 static void bl_init(struct block_lock *lock)
175 spin_lock_init(&lock->lock);
177 INIT_LIST_HEAD(&lock->waiters);
178 for (i = 0; i < MAX_HOLDERS; i++)
179 lock->holders[i] = NULL;
182 static int __available_for_read(struct block_lock *lock)
184 return lock->count >= 0 &&
185 lock->count < MAX_HOLDERS &&
186 list_empty(&lock->waiters);
189 static int bl_down_read(struct block_lock *lock)
194 spin_lock(&lock->lock);
195 r = __check_holder(lock);
197 spin_unlock(&lock->lock);
201 if (__available_for_read(lock)) {
203 __add_holder(lock, current);
204 spin_unlock(&lock->lock);
208 get_task_struct(current);
212 list_add_tail(&w.list, &lock->waiters);
213 spin_unlock(&lock->lock);
216 put_task_struct(current);
220 static int bl_down_read_nonblock(struct block_lock *lock)
224 spin_lock(&lock->lock);
225 r = __check_holder(lock);
229 if (__available_for_read(lock)) {
231 __add_holder(lock, current);
237 spin_unlock(&lock->lock);
241 static void bl_up_read(struct block_lock *lock)
243 spin_lock(&lock->lock);
244 BUG_ON(lock->count <= 0);
245 __del_holder(lock, current);
247 if (!list_empty(&lock->waiters))
249 spin_unlock(&lock->lock);
252 static int bl_down_write(struct block_lock *lock)
257 spin_lock(&lock->lock);
258 r = __check_holder(lock);
260 spin_unlock(&lock->lock);
264 if (lock->count == 0 && list_empty(&lock->waiters)) {
266 __add_holder(lock, current);
267 spin_unlock(&lock->lock);
271 get_task_struct(current);
276 * Writers given priority. We know there's only one mutator in the
277 * system, so ignoring the ordering reversal.
279 list_add(&w.list, &lock->waiters);
280 spin_unlock(&lock->lock);
283 put_task_struct(current);
288 static void bl_up_write(struct block_lock *lock)
290 spin_lock(&lock->lock);
291 __del_holder(lock, current);
293 if (!list_empty(&lock->waiters))
295 spin_unlock(&lock->lock);
298 static void report_recursive_bug(dm_block_t b, int r)
301 DMERR("recursive acquisition of block %llu requested.",
302 (unsigned long long) b);
305 /*----------------------------------------------------------------*/
308 * Block manager is currently implemented using dm-bufio. struct
309 * dm_block_manager and struct dm_block map directly onto a couple of
310 * structs in the bufio interface. I want to retain the freedom to move
311 * away from bufio in the future. So these structs are just cast within
312 * this .c file, rather than making it through to the public interface.
314 static struct dm_buffer *to_buffer(struct dm_block *b)
316 return (struct dm_buffer *) b;
319 dm_block_t dm_block_location(struct dm_block *b)
321 return dm_bufio_get_block_number(to_buffer(b));
323 EXPORT_SYMBOL_GPL(dm_block_location);
325 void *dm_block_data(struct dm_block *b)
327 return dm_bufio_get_block_data(to_buffer(b));
329 EXPORT_SYMBOL_GPL(dm_block_data);
332 struct dm_block_validator *validator;
333 struct block_lock lock;
337 static void dm_block_manager_alloc_callback(struct dm_buffer *buf)
339 struct buffer_aux *aux = dm_bufio_get_aux_data(buf);
340 aux->validator = NULL;
344 static void dm_block_manager_write_callback(struct dm_buffer *buf)
346 struct buffer_aux *aux = dm_bufio_get_aux_data(buf);
347 if (aux->validator) {
348 aux->validator->prepare_for_write(aux->validator, (struct dm_block *) buf,
349 dm_bufio_get_block_size(dm_bufio_get_client(buf)));
353 /*----------------------------------------------------------------
355 *--------------------------------------------------------------*/
356 struct dm_block_manager {
357 struct dm_bufio_client *bufio;
361 struct dm_block_manager *dm_block_manager_create(struct block_device *bdev,
364 unsigned max_held_per_thread)
367 struct dm_block_manager *bm;
369 bm = kmalloc(sizeof(*bm), GFP_KERNEL);
375 bm->bufio = dm_bufio_client_create(bdev, block_size, max_held_per_thread,
376 sizeof(struct buffer_aux),
377 dm_block_manager_alloc_callback,
378 dm_block_manager_write_callback);
379 if (IS_ERR(bm->bufio)) {
380 r = PTR_ERR(bm->bufio);
385 bm->read_only = false;
392 EXPORT_SYMBOL_GPL(dm_block_manager_create);
394 void dm_block_manager_destroy(struct dm_block_manager *bm)
396 dm_bufio_client_destroy(bm->bufio);
399 EXPORT_SYMBOL_GPL(dm_block_manager_destroy);
401 unsigned dm_bm_block_size(struct dm_block_manager *bm)
403 return dm_bufio_get_block_size(bm->bufio);
405 EXPORT_SYMBOL_GPL(dm_bm_block_size);
407 dm_block_t dm_bm_nr_blocks(struct dm_block_manager *bm)
409 return dm_bufio_get_device_size(bm->bufio);
412 static int dm_bm_validate_buffer(struct dm_block_manager *bm,
413 struct dm_buffer *buf,
414 struct buffer_aux *aux,
415 struct dm_block_validator *v)
417 if (unlikely(!aux->validator)) {
421 r = v->check(v, (struct dm_block *) buf, dm_bufio_get_block_size(bm->bufio));
423 DMERR_LIMIT("%s validator check failed for block %llu", v->name,
424 (unsigned long long) dm_bufio_get_block_number(buf));
429 if (unlikely(aux->validator != v)) {
430 DMERR_LIMIT("validator mismatch (old=%s vs new=%s) for block %llu",
431 aux->validator->name, v ? v->name : "NULL",
432 (unsigned long long) dm_bufio_get_block_number(buf));
439 int dm_bm_read_lock(struct dm_block_manager *bm, dm_block_t b,
440 struct dm_block_validator *v,
441 struct dm_block **result)
443 struct buffer_aux *aux;
447 p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
451 aux = dm_bufio_get_aux_data(to_buffer(*result));
452 r = bl_down_read(&aux->lock);
454 dm_bufio_release(to_buffer(*result));
455 report_recursive_bug(b, r);
459 aux->write_locked = 0;
461 r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
463 bl_up_read(&aux->lock);
464 dm_bufio_release(to_buffer(*result));
470 EXPORT_SYMBOL_GPL(dm_bm_read_lock);
472 int dm_bm_write_lock(struct dm_block_manager *bm,
473 dm_block_t b, struct dm_block_validator *v,
474 struct dm_block **result)
476 struct buffer_aux *aux;
483 p = dm_bufio_read(bm->bufio, b, (struct dm_buffer **) result);
487 aux = dm_bufio_get_aux_data(to_buffer(*result));
488 r = bl_down_write(&aux->lock);
490 dm_bufio_release(to_buffer(*result));
491 report_recursive_bug(b, r);
495 aux->write_locked = 1;
497 r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
499 bl_up_write(&aux->lock);
500 dm_bufio_release(to_buffer(*result));
506 EXPORT_SYMBOL_GPL(dm_bm_write_lock);
508 int dm_bm_read_try_lock(struct dm_block_manager *bm,
509 dm_block_t b, struct dm_block_validator *v,
510 struct dm_block **result)
512 struct buffer_aux *aux;
516 p = dm_bufio_get(bm->bufio, b, (struct dm_buffer **) result);
522 aux = dm_bufio_get_aux_data(to_buffer(*result));
523 r = bl_down_read_nonblock(&aux->lock);
525 dm_bufio_release(to_buffer(*result));
526 report_recursive_bug(b, r);
529 aux->write_locked = 0;
531 r = dm_bm_validate_buffer(bm, to_buffer(*result), aux, v);
533 bl_up_read(&aux->lock);
534 dm_bufio_release(to_buffer(*result));
541 int dm_bm_write_lock_zero(struct dm_block_manager *bm,
542 dm_block_t b, struct dm_block_validator *v,
543 struct dm_block **result)
546 struct buffer_aux *aux;
552 p = dm_bufio_new(bm->bufio, b, (struct dm_buffer **) result);
556 memset(p, 0, dm_bm_block_size(bm));
558 aux = dm_bufio_get_aux_data(to_buffer(*result));
559 r = bl_down_write(&aux->lock);
561 dm_bufio_release(to_buffer(*result));
565 aux->write_locked = 1;
570 EXPORT_SYMBOL_GPL(dm_bm_write_lock_zero);
572 void dm_bm_unlock(struct dm_block *b)
574 struct buffer_aux *aux;
575 aux = dm_bufio_get_aux_data(to_buffer(b));
577 if (aux->write_locked) {
578 dm_bufio_mark_buffer_dirty(to_buffer(b));
579 bl_up_write(&aux->lock);
581 bl_up_read(&aux->lock);
583 dm_bufio_release(to_buffer(b));
585 EXPORT_SYMBOL_GPL(dm_bm_unlock);
587 int dm_bm_flush(struct dm_block_manager *bm)
592 return dm_bufio_write_dirty_buffers(bm->bufio);
594 EXPORT_SYMBOL_GPL(dm_bm_flush);
596 void dm_bm_prefetch(struct dm_block_manager *bm, dm_block_t b)
598 dm_bufio_prefetch(bm->bufio, b, 1);
601 bool dm_bm_is_read_only(struct dm_block_manager *bm)
603 return bm->read_only;
605 EXPORT_SYMBOL_GPL(dm_bm_is_read_only);
607 void dm_bm_set_read_only(struct dm_block_manager *bm)
609 bm->read_only = true;
611 EXPORT_SYMBOL_GPL(dm_bm_set_read_only);
613 void dm_bm_set_read_write(struct dm_block_manager *bm)
615 bm->read_only = false;
617 EXPORT_SYMBOL_GPL(dm_bm_set_read_write);
619 u32 dm_bm_checksum(const void *data, size_t len, u32 init_xor)
621 return crc32c(~(u32) 0, data, len) ^ init_xor;
623 EXPORT_SYMBOL_GPL(dm_bm_checksum);
625 /*----------------------------------------------------------------*/
627 MODULE_LICENSE("GPL");
628 MODULE_AUTHOR("Joe Thornber <dm-devel@redhat.com>");
629 MODULE_DESCRIPTION("Immutable metadata library for dm");
631 /*----------------------------------------------------------------*/