4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH.
7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>.
8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
10 Thanks to Carter Burden, Bart Grantham and Gennadiy Nerubayev
11 from Logicworks, Inc. for making SDP replication support possible.
13 drbd is free software; you can redistribute it and/or modify
14 it under the terms of the GNU General Public License as published by
15 the Free Software Foundation; either version 2, or (at your option)
18 drbd is distributed in the hope that it will be useful,
19 but WITHOUT ANY WARRANTY; without even the implied warranty of
20 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
21 GNU General Public License for more details.
23 You should have received a copy of the GNU General Public License
24 along with drbd; see the file COPYING. If not, write to
25 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
29 #include <linux/module.h>
30 #include <linux/drbd.h>
31 #include <asm/uaccess.h>
32 #include <asm/types.h>
34 #include <linux/ctype.h>
35 #include <linux/mutex.h>
37 #include <linux/file.h>
38 #include <linux/proc_fs.h>
39 #include <linux/init.h>
41 #include <linux/memcontrol.h>
42 #include <linux/mm_inline.h>
43 #include <linux/slab.h>
44 #include <linux/random.h>
45 #include <linux/reboot.h>
46 #include <linux/notifier.h>
47 #include <linux/kthread.h>
49 #define __KERNEL_SYSCALLS__
50 #include <linux/unistd.h>
51 #include <linux/vmalloc.h>
53 #include <linux/drbd_limits.h>
55 #include "drbd_req.h" /* only for _req_mod in tl_release and tl_clear */
59 static DEFINE_MUTEX(drbd_main_mutex);
60 int drbdd_init(struct drbd_thread *);
61 int drbd_worker(struct drbd_thread *);
62 int drbd_asender(struct drbd_thread *);
65 static int drbd_open(struct block_device *bdev, fmode_t mode);
66 static int drbd_release(struct gendisk *gd, fmode_t mode);
67 static int w_md_sync(struct drbd_work *w, int unused);
68 static void md_sync_timer_fn(unsigned long data);
69 static int w_bitmap_io(struct drbd_work *w, int unused);
70 static int w_go_diskless(struct drbd_work *w, int unused);
72 MODULE_AUTHOR("Philipp Reisner <phil@linbit.com>, "
73 "Lars Ellenberg <lars@linbit.com>");
74 MODULE_DESCRIPTION("drbd - Distributed Replicated Block Device v" REL_VERSION);
75 MODULE_VERSION(REL_VERSION);
76 MODULE_LICENSE("GPL");
77 MODULE_PARM_DESC(minor_count, "Approximate number of drbd devices ("
78 __stringify(DRBD_MINOR_COUNT_MIN) "-" __stringify(DRBD_MINOR_COUNT_MAX) ")");
79 MODULE_ALIAS_BLOCKDEV_MAJOR(DRBD_MAJOR);
81 #include <linux/moduleparam.h>
82 /* allow_open_on_secondary */
83 MODULE_PARM_DESC(allow_oos, "DONT USE!");
84 /* thanks to these macros, if compiled into the kernel (not-module),
85 * this becomes the boot parameter drbd.minor_count */
86 module_param(minor_count, uint, 0444);
87 module_param(disable_sendpage, bool, 0644);
88 module_param(allow_oos, bool, 0);
89 module_param(proc_details, int, 0644);
91 #ifdef CONFIG_DRBD_FAULT_INJECTION
94 static int fault_count;
96 /* bitmap of enabled faults */
97 module_param(enable_faults, int, 0664);
98 /* fault rate % value - applies to all enabled faults */
99 module_param(fault_rate, int, 0664);
100 /* count of faults inserted */
101 module_param(fault_count, int, 0664);
102 /* bitmap of devices to insert faults on */
103 module_param(fault_devs, int, 0644);
106 /* module parameter, defined */
107 unsigned int minor_count = DRBD_MINOR_COUNT_DEF;
108 int disable_sendpage;
110 int proc_details; /* Detail level in proc drbd*/
112 /* Module parameter for setting the user mode helper program
113 * to run. Default is /sbin/drbdadm */
114 char usermode_helper[80] = "/sbin/drbdadm";
116 module_param_string(usermode_helper, usermode_helper, sizeof(usermode_helper), 0644);
118 /* in 2.6.x, our device mapping and config info contains our virtual gendisks
119 * as member "struct gendisk *vdisk;"
122 struct list_head drbd_tconns; /* list of struct drbd_tconn */
123 DEFINE_MUTEX(drbd_cfg_mutex);
125 struct kmem_cache *drbd_request_cache;
126 struct kmem_cache *drbd_ee_cache; /* peer requests */
127 struct kmem_cache *drbd_bm_ext_cache; /* bitmap extents */
128 struct kmem_cache *drbd_al_ext_cache; /* activity log extents */
129 mempool_t *drbd_request_mempool;
130 mempool_t *drbd_ee_mempool;
131 mempool_t *drbd_md_io_page_pool;
132 struct bio_set *drbd_md_io_bio_set;
134 /* I do not use a standard mempool, because:
135 1) I want to hand out the pre-allocated objects first.
136 2) I want to be able to interrupt sleeping allocation with a signal.
137 Note: This is a single linked list, the next pointer is the private
138 member of struct page.
140 struct page *drbd_pp_pool;
141 spinlock_t drbd_pp_lock;
143 wait_queue_head_t drbd_pp_wait;
145 DEFINE_RATELIMIT_STATE(drbd_ratelimit_state, 5 * HZ, 5);
147 static const struct block_device_operations drbd_ops = {
148 .owner = THIS_MODULE,
150 .release = drbd_release,
153 static void bio_destructor_drbd(struct bio *bio)
155 bio_free(bio, drbd_md_io_bio_set);
158 struct bio *bio_alloc_drbd(gfp_t gfp_mask)
162 if (!drbd_md_io_bio_set)
163 return bio_alloc(gfp_mask, 1);
165 bio = bio_alloc_bioset(gfp_mask, 1, drbd_md_io_bio_set);
168 bio->bi_destructor = bio_destructor_drbd;
173 /* When checking with sparse, and this is an inline function, sparse will
174 give tons of false positives. When this is a real functions sparse works.
176 int _get_ldev_if_state(struct drbd_conf *mdev, enum drbd_disk_state mins)
180 atomic_inc(&mdev->local_cnt);
181 io_allowed = (mdev->state.disk >= mins);
183 if (atomic_dec_and_test(&mdev->local_cnt))
184 wake_up(&mdev->misc_wait);
192 * DOC: The transfer log
194 * The transfer log is a single linked list of &struct drbd_tl_epoch objects.
195 * mdev->tconn->newest_tle points to the head, mdev->tconn->oldest_tle points to the tail
196 * of the list. There is always at least one &struct drbd_tl_epoch object.
198 * Each &struct drbd_tl_epoch has a circular double linked list of requests
201 static int tl_init(struct drbd_tconn *tconn)
203 struct drbd_tl_epoch *b;
205 /* during device minor initialization, we may well use GFP_KERNEL */
206 b = kmalloc(sizeof(struct drbd_tl_epoch), GFP_KERNEL);
209 INIT_LIST_HEAD(&b->requests);
210 INIT_LIST_HEAD(&b->w.list);
214 b->w.cb = NULL; /* if this is != NULL, we need to dec_ap_pending in tl_clear */
216 tconn->oldest_tle = b;
217 tconn->newest_tle = b;
218 INIT_LIST_HEAD(&tconn->out_of_sequence_requests);
223 static void tl_cleanup(struct drbd_tconn *tconn)
225 if (tconn->oldest_tle != tconn->newest_tle)
226 conn_err(tconn, "ASSERT FAILED: oldest_tle == newest_tle\n");
227 if (!list_empty(&tconn->out_of_sequence_requests))
228 conn_err(tconn, "ASSERT FAILED: list_empty(out_of_sequence_requests)\n");
229 kfree(tconn->oldest_tle);
230 tconn->oldest_tle = NULL;
231 kfree(tconn->unused_spare_tle);
232 tconn->unused_spare_tle = NULL;
236 * _tl_add_barrier() - Adds a barrier to the transfer log
237 * @mdev: DRBD device.
238 * @new: Barrier to be added before the current head of the TL.
240 * The caller must hold the req_lock.
242 void _tl_add_barrier(struct drbd_tconn *tconn, struct drbd_tl_epoch *new)
244 struct drbd_tl_epoch *newest_before;
246 INIT_LIST_HEAD(&new->requests);
247 INIT_LIST_HEAD(&new->w.list);
248 new->w.cb = NULL; /* if this is != NULL, we need to dec_ap_pending in tl_clear */
252 newest_before = tconn->newest_tle;
253 /* never send a barrier number == 0, because that is special-cased
254 * when using TCQ for our write ordering code */
255 new->br_number = (newest_before->br_number+1) ?: 1;
256 if (tconn->newest_tle != new) {
257 tconn->newest_tle->next = new;
258 tconn->newest_tle = new;
263 * tl_release() - Free or recycle the oldest &struct drbd_tl_epoch object of the TL
264 * @mdev: DRBD device.
265 * @barrier_nr: Expected identifier of the DRBD write barrier packet.
266 * @set_size: Expected number of requests before that barrier.
268 * In case the passed barrier_nr or set_size does not match the oldest
269 * &struct drbd_tl_epoch objects this function will cause a termination
272 void tl_release(struct drbd_tconn *tconn, unsigned int barrier_nr,
273 unsigned int set_size)
275 struct drbd_conf *mdev;
276 struct drbd_tl_epoch *b, *nob; /* next old barrier */
277 struct list_head *le, *tle;
278 struct drbd_request *r;
280 spin_lock_irq(&tconn->req_lock);
282 b = tconn->oldest_tle;
284 /* first some paranoia code */
286 conn_err(tconn, "BAD! BarrierAck #%u received, but no epoch in tl!?\n",
290 if (b->br_number != barrier_nr) {
291 conn_err(tconn, "BAD! BarrierAck #%u received, expected #%u!\n",
292 barrier_nr, b->br_number);
295 if (b->n_writes != set_size) {
296 conn_err(tconn, "BAD! BarrierAck #%u received with n_writes=%u, expected n_writes=%u!\n",
297 barrier_nr, set_size, b->n_writes);
301 /* Clean up list of requests processed during current epoch */
302 list_for_each_safe(le, tle, &b->requests) {
303 r = list_entry(le, struct drbd_request, tl_requests);
304 _req_mod(r, BARRIER_ACKED);
306 /* There could be requests on the list waiting for completion
307 of the write to the local disk. To avoid corruptions of
308 slab's data structures we have to remove the lists head.
310 Also there could have been a barrier ack out of sequence, overtaking
311 the write acks - which would be a bug and violating write ordering.
312 To not deadlock in case we lose connection while such requests are
313 still pending, we need some way to find them for the
314 _req_mode(CONNECTION_LOST_WHILE_PENDING).
316 These have been list_move'd to the out_of_sequence_requests list in
317 _req_mod(, BARRIER_ACKED) above.
319 list_del_init(&b->requests);
323 if (test_and_clear_bit(CREATE_BARRIER, &mdev->flags)) {
324 _tl_add_barrier(tconn, b);
326 tconn->oldest_tle = nob;
327 /* if nob == NULL b was the only barrier, and becomes the new
328 barrier. Therefore tconn->oldest_tle points already to b */
330 D_ASSERT(nob != NULL);
331 tconn->oldest_tle = nob;
335 spin_unlock_irq(&tconn->req_lock);
336 dec_ap_pending(mdev);
341 spin_unlock_irq(&tconn->req_lock);
342 conn_request_state(tconn, NS(conn, C_PROTOCOL_ERROR), CS_HARD);
347 * _tl_restart() - Walks the transfer log, and applies an action to all requests
348 * @mdev: DRBD device.
349 * @what: The action/event to perform with all request objects
351 * @what might be one of CONNECTION_LOST_WHILE_PENDING, RESEND, FAIL_FROZEN_DISK_IO,
352 * RESTART_FROZEN_DISK_IO.
354 void _tl_restart(struct drbd_tconn *tconn, enum drbd_req_event what)
356 struct drbd_tl_epoch *b, *tmp, **pn;
357 struct list_head *le, *tle, carry_reads;
358 struct drbd_request *req;
359 int rv, n_writes, n_reads;
361 b = tconn->oldest_tle;
362 pn = &tconn->oldest_tle;
366 INIT_LIST_HEAD(&carry_reads);
367 list_for_each_safe(le, tle, &b->requests) {
368 req = list_entry(le, struct drbd_request, tl_requests);
369 rv = _req_mod(req, what);
371 n_writes += (rv & MR_WRITE) >> MR_WRITE_SHIFT;
372 n_reads += (rv & MR_READ) >> MR_READ_SHIFT;
377 if (what == RESEND) {
378 b->n_writes = n_writes;
379 if (b->w.cb == NULL) {
380 b->w.cb = w_send_barrier;
381 inc_ap_pending(b->w.mdev);
382 set_bit(CREATE_BARRIER, &b->w.mdev->flags);
385 drbd_queue_work(&tconn->data.work, &b->w);
390 list_add(&carry_reads, &b->requests);
391 /* there could still be requests on that ring list,
392 * in case local io is still pending */
393 list_del(&b->requests);
395 /* dec_ap_pending corresponding to queue_barrier.
396 * the newest barrier may not have been queued yet,
397 * in which case w.cb is still NULL. */
399 dec_ap_pending(b->w.mdev);
401 if (b == tconn->newest_tle) {
402 /* recycle, but reinit! */
404 conn_err(tconn, "ASSERT FAILED tmp == NULL");
405 INIT_LIST_HEAD(&b->requests);
406 list_splice(&carry_reads, &b->requests);
407 INIT_LIST_HEAD(&b->w.list);
409 b->br_number = net_random();
419 list_splice(&carry_reads, &b->requests);
425 * tl_clear() - Clears all requests and &struct drbd_tl_epoch objects out of the TL
426 * @mdev: DRBD device.
428 * This is called after the connection to the peer was lost. The storage covered
429 * by the requests on the transfer gets marked as our of sync. Called from the
430 * receiver thread and the worker thread.
432 void tl_clear(struct drbd_tconn *tconn)
434 struct drbd_conf *mdev;
435 struct list_head *le, *tle;
436 struct drbd_request *r;
439 spin_lock_irq(&tconn->req_lock);
441 _tl_restart(tconn, CONNECTION_LOST_WHILE_PENDING);
443 /* we expect this list to be empty. */
444 if (!list_empty(&tconn->out_of_sequence_requests))
445 conn_err(tconn, "ASSERT FAILED list_empty(&out_of_sequence_requests)\n");
447 /* but just in case, clean it up anyways! */
448 list_for_each_safe(le, tle, &tconn->out_of_sequence_requests) {
449 r = list_entry(le, struct drbd_request, tl_requests);
450 /* It would be nice to complete outside of spinlock.
451 * But this is easier for now. */
452 _req_mod(r, CONNECTION_LOST_WHILE_PENDING);
455 /* ensure bit indicating barrier is required is clear */
456 idr_for_each_entry(&tconn->volumes, mdev, minor)
457 clear_bit(CREATE_BARRIER, &mdev->flags);
459 spin_unlock_irq(&tconn->req_lock);
462 void tl_restart(struct drbd_tconn *tconn, enum drbd_req_event what)
464 spin_lock_irq(&tconn->req_lock);
465 _tl_restart(tconn, what);
466 spin_unlock_irq(&tconn->req_lock);
469 static int drbd_thread_setup(void *arg)
471 struct drbd_thread *thi = (struct drbd_thread *) arg;
472 struct drbd_tconn *tconn = thi->tconn;
476 snprintf(current->comm, sizeof(current->comm), "drbd_%c_%s",
477 thi->name[0], thi->tconn->name);
480 retval = thi->function(thi);
482 spin_lock_irqsave(&thi->t_lock, flags);
484 /* if the receiver has been "EXITING", the last thing it did
485 * was set the conn state to "StandAlone",
486 * if now a re-connect request comes in, conn state goes C_UNCONNECTED,
487 * and receiver thread will be "started".
488 * drbd_thread_start needs to set "RESTARTING" in that case.
489 * t_state check and assignment needs to be within the same spinlock,
490 * so either thread_start sees EXITING, and can remap to RESTARTING,
491 * or thread_start see NONE, and can proceed as normal.
494 if (thi->t_state == RESTARTING) {
495 conn_info(tconn, "Restarting %s thread\n", thi->name);
496 thi->t_state = RUNNING;
497 spin_unlock_irqrestore(&thi->t_lock, flags);
504 complete(&thi->stop);
505 spin_unlock_irqrestore(&thi->t_lock, flags);
507 conn_info(tconn, "Terminating %s\n", current->comm);
509 /* Release mod reference taken when thread was started */
510 module_put(THIS_MODULE);
514 static void drbd_thread_init(struct drbd_tconn *tconn, struct drbd_thread *thi,
515 int (*func) (struct drbd_thread *), char *name)
517 spin_lock_init(&thi->t_lock);
520 thi->function = func;
522 strncpy(thi->name, name, ARRAY_SIZE(thi->name));
525 int drbd_thread_start(struct drbd_thread *thi)
527 struct drbd_tconn *tconn = thi->tconn;
528 struct task_struct *nt;
531 /* is used from state engine doing drbd_thread_stop_nowait,
532 * while holding the req lock irqsave */
533 spin_lock_irqsave(&thi->t_lock, flags);
535 switch (thi->t_state) {
537 conn_info(tconn, "Starting %s thread (from %s [%d])\n",
538 thi->name, current->comm, current->pid);
540 /* Get ref on module for thread - this is released when thread exits */
541 if (!try_module_get(THIS_MODULE)) {
542 conn_err(tconn, "Failed to get module reference in drbd_thread_start\n");
543 spin_unlock_irqrestore(&thi->t_lock, flags);
547 init_completion(&thi->stop);
548 thi->reset_cpu_mask = 1;
549 thi->t_state = RUNNING;
550 spin_unlock_irqrestore(&thi->t_lock, flags);
551 flush_signals(current); /* otherw. may get -ERESTARTNOINTR */
553 nt = kthread_create(drbd_thread_setup, (void *) thi,
554 "drbd_%c_%s", thi->name[0], thi->tconn->name);
557 conn_err(tconn, "Couldn't start thread\n");
559 module_put(THIS_MODULE);
562 spin_lock_irqsave(&thi->t_lock, flags);
564 thi->t_state = RUNNING;
565 spin_unlock_irqrestore(&thi->t_lock, flags);
569 thi->t_state = RESTARTING;
570 conn_info(tconn, "Restarting %s thread (from %s [%d])\n",
571 thi->name, current->comm, current->pid);
576 spin_unlock_irqrestore(&thi->t_lock, flags);
584 void _drbd_thread_stop(struct drbd_thread *thi, int restart, int wait)
588 enum drbd_thread_state ns = restart ? RESTARTING : EXITING;
590 /* may be called from state engine, holding the req lock irqsave */
591 spin_lock_irqsave(&thi->t_lock, flags);
593 if (thi->t_state == NONE) {
594 spin_unlock_irqrestore(&thi->t_lock, flags);
596 drbd_thread_start(thi);
600 if (thi->t_state != ns) {
601 if (thi->task == NULL) {
602 spin_unlock_irqrestore(&thi->t_lock, flags);
608 init_completion(&thi->stop);
609 if (thi->task != current)
610 force_sig(DRBD_SIGKILL, thi->task);
613 spin_unlock_irqrestore(&thi->t_lock, flags);
616 wait_for_completion(&thi->stop);
619 static struct drbd_thread *drbd_task_to_thread(struct drbd_tconn *tconn, struct task_struct *task)
621 struct drbd_thread *thi =
622 task == tconn->receiver.task ? &tconn->receiver :
623 task == tconn->asender.task ? &tconn->asender :
624 task == tconn->worker.task ? &tconn->worker : NULL;
629 char *drbd_task_to_thread_name(struct drbd_tconn *tconn, struct task_struct *task)
631 struct drbd_thread *thi = drbd_task_to_thread(tconn, task);
632 return thi ? thi->name : task->comm;
635 int conn_lowest_minor(struct drbd_tconn *tconn)
639 if (!idr_get_next(&tconn->volumes, &minor))
646 * drbd_calc_cpu_mask() - Generate CPU masks, spread over all CPUs
647 * @mdev: DRBD device.
649 * Forces all threads of a device onto the same CPU. This is beneficial for
650 * DRBD's performance. May be overwritten by user's configuration.
652 void drbd_calc_cpu_mask(struct drbd_tconn *tconn)
657 if (cpumask_weight(tconn->cpu_mask))
660 ord = conn_lowest_minor(tconn) % cpumask_weight(cpu_online_mask);
661 for_each_online_cpu(cpu) {
663 cpumask_set_cpu(cpu, tconn->cpu_mask);
667 /* should not be reached */
668 cpumask_setall(tconn->cpu_mask);
672 * drbd_thread_current_set_cpu() - modifies the cpu mask of the _current_ thread
673 * @mdev: DRBD device.
674 * @thi: drbd_thread object
676 * call in the "main loop" of _all_ threads, no need for any mutex, current won't die
679 void drbd_thread_current_set_cpu(struct drbd_thread *thi)
681 struct task_struct *p = current;
683 if (!thi->reset_cpu_mask)
685 thi->reset_cpu_mask = 0;
686 set_cpus_allowed_ptr(p, thi->tconn->cpu_mask);
690 static void prepare_header80(struct p_header80 *h, enum drbd_packet cmd, int size)
692 h->magic = cpu_to_be32(DRBD_MAGIC);
693 h->command = cpu_to_be16(cmd);
694 h->length = cpu_to_be16(size);
697 static void prepare_header95(struct p_header95 *h, enum drbd_packet cmd, int size)
699 h->magic = cpu_to_be16(DRBD_MAGIC_BIG);
700 h->command = cpu_to_be16(cmd);
701 h->length = cpu_to_be32(size);
704 static void _prepare_header(struct drbd_tconn *tconn, int vnr, struct p_header *h,
705 enum drbd_packet cmd, int size)
707 if (tconn->agreed_pro_version >= 100 || size > DRBD_MAX_SIZE_H80_PACKET)
708 prepare_header95(&h->h95, cmd, size);
710 prepare_header80(&h->h80, cmd, size);
713 static void prepare_header(struct drbd_conf *mdev, struct p_header *h,
714 enum drbd_packet cmd, int size)
716 _prepare_header(mdev->tconn, mdev->vnr, h, cmd, size);
719 /* the appropriate socket mutex must be held already */
720 int _conn_send_cmd(struct drbd_tconn *tconn, int vnr, struct socket *sock,
721 enum drbd_packet cmd, struct p_header *h, size_t size,
726 _prepare_header(tconn, vnr, h, cmd, size - sizeof(struct p_header));
728 sent = drbd_send(tconn, sock, h, size, msg_flags);
731 if (!ok && !signal_pending(current))
732 conn_warn(tconn, "short sent %s size=%d sent=%d\n",
733 cmdname(cmd), (int)size, sent);
737 /* don't pass the socket. we may only look at it
738 * when we hold the appropriate socket mutex.
740 int conn_send_cmd(struct drbd_tconn *tconn, int vnr, int use_data_socket,
741 enum drbd_packet cmd, struct p_header *h, size_t size)
746 if (use_data_socket) {
747 mutex_lock(&tconn->data.mutex);
748 sock = tconn->data.socket;
750 mutex_lock(&tconn->meta.mutex);
751 sock = tconn->meta.socket;
754 /* drbd_disconnect() could have called drbd_free_sock()
755 * while we were waiting in down()... */
756 if (likely(sock != NULL))
757 ok = _conn_send_cmd(tconn, vnr, sock, cmd, h, size, 0);
760 mutex_unlock(&tconn->data.mutex);
762 mutex_unlock(&tconn->meta.mutex);
766 int conn_send_cmd2(struct drbd_tconn *tconn, enum drbd_packet cmd, char *data,
772 prepare_header80(&h, cmd, size);
774 if (!drbd_get_data_sock(tconn))
778 drbd_send(tconn, tconn->data.socket, &h, sizeof(h), 0));
780 drbd_send(tconn, tconn->data.socket, data, size, 0));
782 drbd_put_data_sock(tconn);
787 int drbd_send_sync_param(struct drbd_conf *mdev, struct syncer_conf *sc)
789 struct p_rs_param_95 *p;
792 const int apv = mdev->tconn->agreed_pro_version;
794 size = apv <= 87 ? sizeof(struct p_rs_param)
795 : apv == 88 ? sizeof(struct p_rs_param)
796 + strlen(mdev->sync_conf.verify_alg) + 1
797 : apv <= 94 ? sizeof(struct p_rs_param_89)
798 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
800 /* used from admin command context and receiver/worker context.
801 * to avoid kmalloc, grab the socket right here,
802 * then use the pre-allocated sbuf there */
803 mutex_lock(&mdev->tconn->data.mutex);
804 sock = mdev->tconn->data.socket;
806 if (likely(sock != NULL)) {
807 enum drbd_packet cmd =
808 apv >= 89 ? P_SYNC_PARAM89 : P_SYNC_PARAM;
810 p = &mdev->tconn->data.sbuf.rs_param_95;
812 /* initialize verify_alg and csums_alg */
813 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
815 p->rate = cpu_to_be32(sc->rate);
816 p->c_plan_ahead = cpu_to_be32(sc->c_plan_ahead);
817 p->c_delay_target = cpu_to_be32(sc->c_delay_target);
818 p->c_fill_target = cpu_to_be32(sc->c_fill_target);
819 p->c_max_rate = cpu_to_be32(sc->c_max_rate);
822 strcpy(p->verify_alg, mdev->sync_conf.verify_alg);
824 strcpy(p->csums_alg, mdev->sync_conf.csums_alg);
826 rv = _drbd_send_cmd(mdev, sock, cmd, &p->head, size, 0);
830 mutex_unlock(&mdev->tconn->data.mutex);
835 int drbd_send_protocol(struct drbd_tconn *tconn)
837 struct p_protocol *p;
840 size = sizeof(struct p_protocol);
842 if (tconn->agreed_pro_version >= 87)
843 size += strlen(tconn->net_conf->integrity_alg) + 1;
845 /* we must not recurse into our own queue,
846 * as that is blocked during handshake */
847 p = kmalloc(size, GFP_NOIO);
851 p->protocol = cpu_to_be32(tconn->net_conf->wire_protocol);
852 p->after_sb_0p = cpu_to_be32(tconn->net_conf->after_sb_0p);
853 p->after_sb_1p = cpu_to_be32(tconn->net_conf->after_sb_1p);
854 p->after_sb_2p = cpu_to_be32(tconn->net_conf->after_sb_2p);
855 p->two_primaries = cpu_to_be32(tconn->net_conf->two_primaries);
858 if (tconn->net_conf->want_lose)
860 if (tconn->net_conf->dry_run) {
861 if (tconn->agreed_pro_version >= 92)
864 conn_err(tconn, "--dry-run is not supported by peer");
869 p->conn_flags = cpu_to_be32(cf);
871 if (tconn->agreed_pro_version >= 87)
872 strcpy(p->integrity_alg, tconn->net_conf->integrity_alg);
874 rv = conn_send_cmd2(tconn, P_PROTOCOL, p->head.payload, size - sizeof(struct p_header));
879 int _drbd_send_uuids(struct drbd_conf *mdev, u64 uuid_flags)
884 if (!get_ldev_if_state(mdev, D_NEGOTIATING))
887 for (i = UI_CURRENT; i < UI_SIZE; i++)
888 p.uuid[i] = mdev->ldev ? cpu_to_be64(mdev->ldev->md.uuid[i]) : 0;
890 mdev->comm_bm_set = drbd_bm_total_weight(mdev);
891 p.uuid[UI_SIZE] = cpu_to_be64(mdev->comm_bm_set);
892 uuid_flags |= mdev->tconn->net_conf->want_lose ? 1 : 0;
893 uuid_flags |= test_bit(CRASHED_PRIMARY, &mdev->flags) ? 2 : 0;
894 uuid_flags |= mdev->new_state_tmp.disk == D_INCONSISTENT ? 4 : 0;
895 p.uuid[UI_FLAGS] = cpu_to_be64(uuid_flags);
899 return drbd_send_cmd(mdev, USE_DATA_SOCKET, P_UUIDS, &p.head, sizeof(p));
902 int drbd_send_uuids(struct drbd_conf *mdev)
904 return _drbd_send_uuids(mdev, 0);
907 int drbd_send_uuids_skip_initial_sync(struct drbd_conf *mdev)
909 return _drbd_send_uuids(mdev, 8);
912 void drbd_print_uuids(struct drbd_conf *mdev, const char *text)
914 if (get_ldev_if_state(mdev, D_NEGOTIATING)) {
915 u64 *uuid = mdev->ldev->md.uuid;
916 dev_info(DEV, "%s %016llX:%016llX:%016llX:%016llX\n",
918 (unsigned long long)uuid[UI_CURRENT],
919 (unsigned long long)uuid[UI_BITMAP],
920 (unsigned long long)uuid[UI_HISTORY_START],
921 (unsigned long long)uuid[UI_HISTORY_END]);
924 dev_info(DEV, "%s effective data uuid: %016llX\n",
926 (unsigned long long)mdev->ed_uuid);
930 int drbd_gen_and_send_sync_uuid(struct drbd_conf *mdev)
935 D_ASSERT(mdev->state.disk == D_UP_TO_DATE);
937 uuid = mdev->ldev->md.uuid[UI_BITMAP] + UUID_NEW_BM_OFFSET;
938 drbd_uuid_set(mdev, UI_BITMAP, uuid);
939 drbd_print_uuids(mdev, "updated sync UUID");
941 p.uuid = cpu_to_be64(uuid);
943 return drbd_send_cmd(mdev, USE_DATA_SOCKET, P_SYNC_UUID, &p.head, sizeof(p));
946 int drbd_send_sizes(struct drbd_conf *mdev, int trigger_reply, enum dds_flags flags)
949 sector_t d_size, u_size;
950 int q_order_type, max_bio_size;
953 if (get_ldev_if_state(mdev, D_NEGOTIATING)) {
954 D_ASSERT(mdev->ldev->backing_bdev);
955 d_size = drbd_get_max_capacity(mdev->ldev);
956 u_size = mdev->ldev->dc.disk_size;
957 q_order_type = drbd_queue_order_type(mdev);
958 max_bio_size = queue_max_hw_sectors(mdev->ldev->backing_bdev->bd_disk->queue) << 9;
959 max_bio_size = min_t(int, max_bio_size, DRBD_MAX_BIO_SIZE);
964 q_order_type = QUEUE_ORDERED_NONE;
965 max_bio_size = DRBD_MAX_BIO_SIZE; /* ... multiple BIOs per peer_request */
968 p.d_size = cpu_to_be64(d_size);
969 p.u_size = cpu_to_be64(u_size);
970 p.c_size = cpu_to_be64(trigger_reply ? 0 : drbd_get_capacity(mdev->this_bdev));
971 p.max_bio_size = cpu_to_be32(max_bio_size);
972 p.queue_order_type = cpu_to_be16(q_order_type);
973 p.dds_flags = cpu_to_be16(flags);
975 ok = drbd_send_cmd(mdev, USE_DATA_SOCKET, P_SIZES, &p.head, sizeof(p));
980 * drbd_send_state() - Sends the drbd state to the peer
981 * @mdev: DRBD device.
983 int drbd_send_state(struct drbd_conf *mdev)
989 mutex_lock(&mdev->tconn->data.mutex);
991 p.state = cpu_to_be32(mdev->state.i); /* Within the send mutex */
992 sock = mdev->tconn->data.socket;
994 if (likely(sock != NULL)) {
995 ok = _drbd_send_cmd(mdev, sock, P_STATE, &p.head, sizeof(p), 0);
998 mutex_unlock(&mdev->tconn->data.mutex);
1003 int _conn_send_state_req(struct drbd_tconn *tconn, int vnr, enum drbd_packet cmd,
1004 union drbd_state mask, union drbd_state val)
1006 struct p_req_state p;
1008 p.mask = cpu_to_be32(mask.i);
1009 p.val = cpu_to_be32(val.i);
1011 return conn_send_cmd(tconn, vnr, USE_DATA_SOCKET, cmd, &p.head, sizeof(p));
1014 int drbd_send_sr_reply(struct drbd_conf *mdev, enum drbd_state_rv retcode)
1016 struct p_req_state_reply p;
1018 p.retcode = cpu_to_be32(retcode);
1020 return drbd_send_cmd(mdev, USE_META_SOCKET, P_STATE_CHG_REPLY, &p.head, sizeof(p));
1023 int conn_send_sr_reply(struct drbd_tconn *tconn, enum drbd_state_rv retcode)
1025 struct p_req_state_reply p;
1026 enum drbd_packet cmd = tconn->agreed_pro_version < 100 ? P_STATE_CHG_REPLY : P_CONN_ST_CHG_REPLY;
1028 p.retcode = cpu_to_be32(retcode);
1030 return conn_send_cmd(tconn, 0, USE_META_SOCKET, cmd, &p.head, sizeof(p));
1033 int fill_bitmap_rle_bits(struct drbd_conf *mdev,
1034 struct p_compressed_bm *p,
1035 struct bm_xfer_ctx *c)
1037 struct bitstream bs;
1038 unsigned long plain_bits;
1045 /* may we use this feature? */
1046 if ((mdev->sync_conf.use_rle == 0) ||
1047 (mdev->tconn->agreed_pro_version < 90))
1050 if (c->bit_offset >= c->bm_bits)
1051 return 0; /* nothing to do. */
1053 /* use at most thus many bytes */
1054 bitstream_init(&bs, p->code, BM_PACKET_VLI_BYTES_MAX, 0);
1055 memset(p->code, 0, BM_PACKET_VLI_BYTES_MAX);
1056 /* plain bits covered in this code string */
1059 /* p->encoding & 0x80 stores whether the first run length is set.
1060 * bit offset is implicit.
1061 * start with toggle == 2 to be able to tell the first iteration */
1064 /* see how much plain bits we can stuff into one packet
1065 * using RLE and VLI. */
1067 tmp = (toggle == 0) ? _drbd_bm_find_next_zero(mdev, c->bit_offset)
1068 : _drbd_bm_find_next(mdev, c->bit_offset);
1071 rl = tmp - c->bit_offset;
1073 if (toggle == 2) { /* first iteration */
1075 /* the first checked bit was set,
1076 * store start value, */
1077 DCBP_set_start(p, 1);
1078 /* but skip encoding of zero run length */
1082 DCBP_set_start(p, 0);
1085 /* paranoia: catch zero runlength.
1086 * can only happen if bitmap is modified while we scan it. */
1088 dev_err(DEV, "unexpected zero runlength while encoding bitmap "
1089 "t:%u bo:%lu\n", toggle, c->bit_offset);
1093 bits = vli_encode_bits(&bs, rl);
1094 if (bits == -ENOBUFS) /* buffer full */
1097 dev_err(DEV, "error while encoding bitmap: %d\n", bits);
1103 c->bit_offset = tmp;
1104 } while (c->bit_offset < c->bm_bits);
1106 len = bs.cur.b - p->code + !!bs.cur.bit;
1108 if (plain_bits < (len << 3)) {
1109 /* incompressible with this method.
1110 * we need to rewind both word and bit position. */
1111 c->bit_offset -= plain_bits;
1112 bm_xfer_ctx_bit_to_word_offset(c);
1113 c->bit_offset = c->word_offset * BITS_PER_LONG;
1117 /* RLE + VLI was able to compress it just fine.
1118 * update c->word_offset. */
1119 bm_xfer_ctx_bit_to_word_offset(c);
1121 /* store pad_bits */
1122 DCBP_set_pad_bits(p, (8 - bs.cur.bit) & 0x7);
1128 * send_bitmap_rle_or_plain
1130 * Return 0 when done, 1 when another iteration is needed, and a negative error
1131 * code upon failure.
1134 send_bitmap_rle_or_plain(struct drbd_conf *mdev,
1135 struct p_header *h, struct bm_xfer_ctx *c)
1137 struct p_compressed_bm *p = (void*)h;
1138 unsigned long num_words;
1142 len = fill_bitmap_rle_bits(mdev, p, c);
1148 DCBP_set_code(p, RLE_VLI_Bits);
1149 ok = _drbd_send_cmd(mdev, mdev->tconn->data.socket, P_COMPRESSED_BITMAP, h,
1150 sizeof(*p) + len, 0);
1153 c->bytes[0] += sizeof(*p) + len;
1155 if (c->bit_offset >= c->bm_bits)
1158 /* was not compressible.
1159 * send a buffer full of plain text bits instead. */
1160 num_words = min_t(size_t, BM_PACKET_WORDS, c->bm_words - c->word_offset);
1161 len = num_words * sizeof(long);
1163 drbd_bm_get_lel(mdev, c->word_offset, num_words, (unsigned long*)h->payload);
1164 ok = _drbd_send_cmd(mdev, mdev->tconn->data.socket, P_BITMAP,
1165 h, sizeof(struct p_header80) + len, 0);
1166 c->word_offset += num_words;
1167 c->bit_offset = c->word_offset * BITS_PER_LONG;
1170 c->bytes[1] += sizeof(struct p_header80) + len;
1172 if (c->bit_offset > c->bm_bits)
1173 c->bit_offset = c->bm_bits;
1177 INFO_bm_xfer_stats(mdev, "send", c);
1185 /* See the comment at receive_bitmap() */
1186 int _drbd_send_bitmap(struct drbd_conf *mdev)
1188 struct bm_xfer_ctx c;
1192 if (!expect(mdev->bitmap))
1195 /* maybe we should use some per thread scratch page,
1196 * and allocate that during initial device creation? */
1197 p = (struct p_header *) __get_free_page(GFP_NOIO);
1199 dev_err(DEV, "failed to allocate one page buffer in %s\n", __func__);
1203 if (get_ldev(mdev)) {
1204 if (drbd_md_test_flag(mdev->ldev, MDF_FULL_SYNC)) {
1205 dev_info(DEV, "Writing the whole bitmap, MDF_FullSync was set.\n");
1206 drbd_bm_set_all(mdev);
1207 if (drbd_bm_write(mdev)) {
1208 /* write_bm did fail! Leave full sync flag set in Meta P_DATA
1209 * but otherwise process as per normal - need to tell other
1210 * side that a full resync is required! */
1211 dev_err(DEV, "Failed to write bitmap to disk!\n");
1213 drbd_md_clear_flag(mdev, MDF_FULL_SYNC);
1220 c = (struct bm_xfer_ctx) {
1221 .bm_bits = drbd_bm_bits(mdev),
1222 .bm_words = drbd_bm_words(mdev),
1226 err = send_bitmap_rle_or_plain(mdev, p, &c);
1229 free_page((unsigned long) p);
1233 int drbd_send_bitmap(struct drbd_conf *mdev)
1237 if (!drbd_get_data_sock(mdev->tconn))
1239 err = !_drbd_send_bitmap(mdev);
1240 drbd_put_data_sock(mdev->tconn);
1244 int drbd_send_b_ack(struct drbd_conf *mdev, u32 barrier_nr, u32 set_size)
1247 struct p_barrier_ack p;
1249 p.barrier = barrier_nr;
1250 p.set_size = cpu_to_be32(set_size);
1252 if (mdev->state.conn < C_CONNECTED)
1254 ok = drbd_send_cmd(mdev, USE_META_SOCKET, P_BARRIER_ACK, &p.head, sizeof(p));
1259 * _drbd_send_ack() - Sends an ack packet
1260 * @mdev: DRBD device.
1261 * @cmd: Packet command code.
1262 * @sector: sector, needs to be in big endian byte order
1263 * @blksize: size in byte, needs to be in big endian byte order
1264 * @block_id: Id, big endian byte order
1266 static int _drbd_send_ack(struct drbd_conf *mdev, enum drbd_packet cmd,
1267 u64 sector, u32 blksize, u64 block_id)
1270 struct p_block_ack p;
1273 p.block_id = block_id;
1274 p.blksize = blksize;
1275 p.seq_num = cpu_to_be32(atomic_inc_return(&mdev->packet_seq));
1277 if (!mdev->tconn->meta.socket || mdev->state.conn < C_CONNECTED)
1279 ok = drbd_send_cmd(mdev, USE_META_SOCKET, cmd, &p.head, sizeof(p));
1283 /* dp->sector and dp->block_id already/still in network byte order,
1284 * data_size is payload size according to dp->head,
1285 * and may need to be corrected for digest size. */
1286 int drbd_send_ack_dp(struct drbd_conf *mdev, enum drbd_packet cmd,
1287 struct p_data *dp, int data_size)
1289 data_size -= (mdev->tconn->agreed_pro_version >= 87 && mdev->tconn->integrity_r_tfm) ?
1290 crypto_hash_digestsize(mdev->tconn->integrity_r_tfm) : 0;
1291 return _drbd_send_ack(mdev, cmd, dp->sector, cpu_to_be32(data_size),
1295 int drbd_send_ack_rp(struct drbd_conf *mdev, enum drbd_packet cmd,
1296 struct p_block_req *rp)
1298 return _drbd_send_ack(mdev, cmd, rp->sector, rp->blksize, rp->block_id);
1302 * drbd_send_ack() - Sends an ack packet
1303 * @mdev: DRBD device
1304 * @cmd: packet command code
1305 * @peer_req: peer request
1307 int drbd_send_ack(struct drbd_conf *mdev, enum drbd_packet cmd,
1308 struct drbd_peer_request *peer_req)
1310 return _drbd_send_ack(mdev, cmd,
1311 cpu_to_be64(peer_req->i.sector),
1312 cpu_to_be32(peer_req->i.size),
1313 peer_req->block_id);
1316 /* This function misuses the block_id field to signal if the blocks
1317 * are is sync or not. */
1318 int drbd_send_ack_ex(struct drbd_conf *mdev, enum drbd_packet cmd,
1319 sector_t sector, int blksize, u64 block_id)
1321 return _drbd_send_ack(mdev, cmd,
1322 cpu_to_be64(sector),
1323 cpu_to_be32(blksize),
1324 cpu_to_be64(block_id));
1327 int drbd_send_drequest(struct drbd_conf *mdev, int cmd,
1328 sector_t sector, int size, u64 block_id)
1331 struct p_block_req p;
1333 p.sector = cpu_to_be64(sector);
1334 p.block_id = block_id;
1335 p.blksize = cpu_to_be32(size);
1337 ok = drbd_send_cmd(mdev, USE_DATA_SOCKET, cmd, &p.head, sizeof(p));
1341 int drbd_send_drequest_csum(struct drbd_conf *mdev, sector_t sector, int size,
1342 void *digest, int digest_size, enum drbd_packet cmd)
1345 struct p_block_req p;
1347 prepare_header(mdev, &p.head, cmd, sizeof(p) - sizeof(struct p_header) + digest_size);
1348 p.sector = cpu_to_be64(sector);
1349 p.block_id = ID_SYNCER /* unused */;
1350 p.blksize = cpu_to_be32(size);
1352 mutex_lock(&mdev->tconn->data.mutex);
1354 ok = (sizeof(p) == drbd_send(mdev->tconn, mdev->tconn->data.socket, &p, sizeof(p), 0));
1355 ok = ok && (digest_size == drbd_send(mdev->tconn, mdev->tconn->data.socket, digest, digest_size, 0));
1357 mutex_unlock(&mdev->tconn->data.mutex);
1362 int drbd_send_ov_request(struct drbd_conf *mdev, sector_t sector, int size)
1365 struct p_block_req p;
1367 p.sector = cpu_to_be64(sector);
1368 p.block_id = ID_SYNCER /* unused */;
1369 p.blksize = cpu_to_be32(size);
1371 ok = drbd_send_cmd(mdev, USE_DATA_SOCKET, P_OV_REQUEST, &p.head, sizeof(p));
1375 /* called on sndtimeo
1376 * returns false if we should retry,
1377 * true if we think connection is dead
1379 static int we_should_drop_the_connection(struct drbd_tconn *tconn, struct socket *sock)
1382 /* long elapsed = (long)(jiffies - mdev->last_received); */
1384 drop_it = tconn->meta.socket == sock
1385 || !tconn->asender.task
1386 || get_t_state(&tconn->asender) != RUNNING
1387 || tconn->cstate < C_WF_REPORT_PARAMS;
1392 drop_it = !--tconn->ko_count;
1394 conn_err(tconn, "[%s/%d] sock_sendmsg time expired, ko = %u\n",
1395 current->comm, current->pid, tconn->ko_count);
1396 request_ping(tconn);
1399 return drop_it; /* && (mdev->state == R_PRIMARY) */;
1402 static void drbd_update_congested(struct drbd_tconn *tconn)
1404 struct sock *sk = tconn->data.socket->sk;
1405 if (sk->sk_wmem_queued > sk->sk_sndbuf * 4 / 5)
1406 set_bit(NET_CONGESTED, &tconn->flags);
1409 /* The idea of sendpage seems to be to put some kind of reference
1410 * to the page into the skb, and to hand it over to the NIC. In
1411 * this process get_page() gets called.
1413 * As soon as the page was really sent over the network put_page()
1414 * gets called by some part of the network layer. [ NIC driver? ]
1416 * [ get_page() / put_page() increment/decrement the count. If count
1417 * reaches 0 the page will be freed. ]
1419 * This works nicely with pages from FSs.
1420 * But this means that in protocol A we might signal IO completion too early!
1422 * In order not to corrupt data during a resync we must make sure
1423 * that we do not reuse our own buffer pages (EEs) to early, therefore
1424 * we have the net_ee list.
1426 * XFS seems to have problems, still, it submits pages with page_count == 0!
1427 * As a workaround, we disable sendpage on pages
1428 * with page_count == 0 or PageSlab.
1430 static int _drbd_no_send_page(struct drbd_conf *mdev, struct page *page,
1431 int offset, size_t size, unsigned msg_flags)
1433 int sent = drbd_send(mdev->tconn, mdev->tconn->data.socket, kmap(page) + offset, size, msg_flags);
1436 mdev->send_cnt += size>>9;
1437 return sent == size;
1440 static int _drbd_send_page(struct drbd_conf *mdev, struct page *page,
1441 int offset, size_t size, unsigned msg_flags)
1443 mm_segment_t oldfs = get_fs();
1447 /* e.g. XFS meta- & log-data is in slab pages, which have a
1448 * page_count of 0 and/or have PageSlab() set.
1449 * we cannot use send_page for those, as that does get_page();
1450 * put_page(); and would cause either a VM_BUG directly, or
1451 * __page_cache_release a page that would actually still be referenced
1452 * by someone, leading to some obscure delayed Oops somewhere else. */
1453 if (disable_sendpage || (page_count(page) < 1) || PageSlab(page))
1454 return _drbd_no_send_page(mdev, page, offset, size, msg_flags);
1456 msg_flags |= MSG_NOSIGNAL;
1457 drbd_update_congested(mdev->tconn);
1460 sent = mdev->tconn->data.socket->ops->sendpage(mdev->tconn->data.socket, page,
1463 if (sent == -EAGAIN) {
1464 if (we_should_drop_the_connection(mdev->tconn,
1465 mdev->tconn->data.socket))
1471 dev_warn(DEV, "%s: size=%d len=%d sent=%d\n",
1472 __func__, (int)size, len, sent);
1477 } while (len > 0 /* THINK && mdev->cstate >= C_CONNECTED*/);
1479 clear_bit(NET_CONGESTED, &mdev->tconn->flags);
1483 mdev->send_cnt += size>>9;
1487 static int _drbd_send_bio(struct drbd_conf *mdev, struct bio *bio)
1489 struct bio_vec *bvec;
1491 /* hint all but last page with MSG_MORE */
1492 __bio_for_each_segment(bvec, bio, i, 0) {
1493 if (!_drbd_no_send_page(mdev, bvec->bv_page,
1494 bvec->bv_offset, bvec->bv_len,
1495 i == bio->bi_vcnt -1 ? 0 : MSG_MORE))
1501 static int _drbd_send_zc_bio(struct drbd_conf *mdev, struct bio *bio)
1503 struct bio_vec *bvec;
1505 /* hint all but last page with MSG_MORE */
1506 __bio_for_each_segment(bvec, bio, i, 0) {
1507 if (!_drbd_send_page(mdev, bvec->bv_page,
1508 bvec->bv_offset, bvec->bv_len,
1509 i == bio->bi_vcnt -1 ? 0 : MSG_MORE))
1515 static int _drbd_send_zc_ee(struct drbd_conf *mdev,
1516 struct drbd_peer_request *peer_req)
1518 struct page *page = peer_req->pages;
1519 unsigned len = peer_req->i.size;
1521 /* hint all but last page with MSG_MORE */
1522 page_chain_for_each(page) {
1523 unsigned l = min_t(unsigned, len, PAGE_SIZE);
1524 if (!_drbd_send_page(mdev, page, 0, l,
1525 page_chain_next(page) ? MSG_MORE : 0))
1532 static u32 bio_flags_to_wire(struct drbd_conf *mdev, unsigned long bi_rw)
1534 if (mdev->tconn->agreed_pro_version >= 95)
1535 return (bi_rw & REQ_SYNC ? DP_RW_SYNC : 0) |
1536 (bi_rw & REQ_FUA ? DP_FUA : 0) |
1537 (bi_rw & REQ_FLUSH ? DP_FLUSH : 0) |
1538 (bi_rw & REQ_DISCARD ? DP_DISCARD : 0);
1540 return bi_rw & REQ_SYNC ? DP_RW_SYNC : 0;
1543 /* Used to send write requests
1544 * R_PRIMARY -> Peer (P_DATA)
1546 int drbd_send_dblock(struct drbd_conf *mdev, struct drbd_request *req)
1550 unsigned int dp_flags = 0;
1554 if (!drbd_get_data_sock(mdev->tconn))
1557 dgs = (mdev->tconn->agreed_pro_version >= 87 && mdev->tconn->integrity_w_tfm) ?
1558 crypto_hash_digestsize(mdev->tconn->integrity_w_tfm) : 0;
1560 prepare_header(mdev, &p.head, P_DATA, sizeof(p) - sizeof(struct p_header) + dgs + req->i.size);
1561 p.sector = cpu_to_be64(req->i.sector);
1562 p.block_id = (unsigned long)req;
1563 p.seq_num = cpu_to_be32(req->seq_num = atomic_inc_return(&mdev->packet_seq));
1565 dp_flags = bio_flags_to_wire(mdev, req->master_bio->bi_rw);
1567 if (mdev->state.conn >= C_SYNC_SOURCE &&
1568 mdev->state.conn <= C_PAUSED_SYNC_T)
1569 dp_flags |= DP_MAY_SET_IN_SYNC;
1571 p.dp_flags = cpu_to_be32(dp_flags);
1572 set_bit(UNPLUG_REMOTE, &mdev->flags);
1574 drbd_send(mdev->tconn, mdev->tconn->data.socket, &p, sizeof(p), dgs ? MSG_MORE : 0));
1576 dgb = mdev->tconn->int_dig_out;
1577 drbd_csum_bio(mdev, mdev->tconn->integrity_w_tfm, req->master_bio, dgb);
1578 ok = dgs == drbd_send(mdev->tconn, mdev->tconn->data.socket, dgb, dgs, 0);
1581 /* For protocol A, we have to memcpy the payload into
1582 * socket buffers, as we may complete right away
1583 * as soon as we handed it over to tcp, at which point the data
1584 * pages may become invalid.
1586 * For data-integrity enabled, we copy it as well, so we can be
1587 * sure that even if the bio pages may still be modified, it
1588 * won't change the data on the wire, thus if the digest checks
1589 * out ok after sending on this side, but does not fit on the
1590 * receiving side, we sure have detected corruption elsewhere.
1592 if (mdev->tconn->net_conf->wire_protocol == DRBD_PROT_A || dgs)
1593 ok = _drbd_send_bio(mdev, req->master_bio);
1595 ok = _drbd_send_zc_bio(mdev, req->master_bio);
1597 /* double check digest, sometimes buffers have been modified in flight. */
1598 if (dgs > 0 && dgs <= 64) {
1599 /* 64 byte, 512 bit, is the largest digest size
1600 * currently supported in kernel crypto. */
1601 unsigned char digest[64];
1602 drbd_csum_bio(mdev, mdev->tconn->integrity_w_tfm, req->master_bio, digest);
1603 if (memcmp(mdev->tconn->int_dig_out, digest, dgs)) {
1605 "Digest mismatch, buffer modified by upper layers during write: %llus +%u\n",
1606 (unsigned long long)req->i.sector, req->i.size);
1608 } /* else if (dgs > 64) {
1609 ... Be noisy about digest too large ...
1613 drbd_put_data_sock(mdev->tconn);
1618 /* answer packet, used to send data back for read requests:
1619 * Peer -> (diskless) R_PRIMARY (P_DATA_REPLY)
1620 * C_SYNC_SOURCE -> C_SYNC_TARGET (P_RS_DATA_REPLY)
1622 int drbd_send_block(struct drbd_conf *mdev, enum drbd_packet cmd,
1623 struct drbd_peer_request *peer_req)
1630 dgs = (mdev->tconn->agreed_pro_version >= 87 && mdev->tconn->integrity_w_tfm) ?
1631 crypto_hash_digestsize(mdev->tconn->integrity_w_tfm) : 0;
1633 prepare_header(mdev, &p.head, cmd, sizeof(p) -
1634 sizeof(struct p_header80) +
1635 dgs + peer_req->i.size);
1636 p.sector = cpu_to_be64(peer_req->i.sector);
1637 p.block_id = peer_req->block_id;
1638 p.seq_num = 0; /* unused */
1640 /* Only called by our kernel thread.
1641 * This one may be interrupted by DRBD_SIG and/or DRBD_SIGKILL
1642 * in response to admin command or module unload.
1644 if (!drbd_get_data_sock(mdev->tconn))
1647 ok = sizeof(p) == drbd_send(mdev->tconn, mdev->tconn->data.socket, &p, sizeof(p), dgs ? MSG_MORE : 0);
1649 dgb = mdev->tconn->int_dig_out;
1650 drbd_csum_ee(mdev, mdev->tconn->integrity_w_tfm, peer_req, dgb);
1651 ok = dgs == drbd_send(mdev->tconn, mdev->tconn->data.socket, dgb, dgs, 0);
1654 ok = _drbd_send_zc_ee(mdev, peer_req);
1656 drbd_put_data_sock(mdev->tconn);
1661 int drbd_send_oos(struct drbd_conf *mdev, struct drbd_request *req)
1663 struct p_block_desc p;
1665 p.sector = cpu_to_be64(req->i.sector);
1666 p.blksize = cpu_to_be32(req->i.size);
1668 return drbd_send_cmd(mdev, USE_DATA_SOCKET, P_OUT_OF_SYNC, &p.head, sizeof(p));
1672 drbd_send distinguishes two cases:
1674 Packets sent via the data socket "sock"
1675 and packets sent via the meta data socket "msock"
1678 -----------------+-------------------------+------------------------------
1679 timeout conf.timeout / 2 conf.timeout / 2
1680 timeout action send a ping via msock Abort communication
1681 and close all sockets
1685 * you must have down()ed the appropriate [m]sock_mutex elsewhere!
1687 int drbd_send(struct drbd_tconn *tconn, struct socket *sock,
1688 void *buf, size_t size, unsigned msg_flags)
1697 /* THINK if (signal_pending) return ... ? */
1702 msg.msg_name = NULL;
1703 msg.msg_namelen = 0;
1704 msg.msg_control = NULL;
1705 msg.msg_controllen = 0;
1706 msg.msg_flags = msg_flags | MSG_NOSIGNAL;
1708 if (sock == tconn->data.socket) {
1709 tconn->ko_count = tconn->net_conf->ko_count;
1710 drbd_update_congested(tconn);
1714 * tcp_sendmsg does _not_ use its size parameter at all ?
1716 * -EAGAIN on timeout, -EINTR on signal.
1719 * do we need to block DRBD_SIG if sock == &meta.socket ??
1720 * otherwise wake_asender() might interrupt some send_*Ack !
1722 rv = kernel_sendmsg(sock, &msg, &iov, 1, size);
1723 if (rv == -EAGAIN) {
1724 if (we_should_drop_the_connection(tconn, sock))
1730 flush_signals(current);
1738 } while (sent < size);
1740 if (sock == tconn->data.socket)
1741 clear_bit(NET_CONGESTED, &tconn->flags);
1744 if (rv != -EAGAIN) {
1745 conn_err(tconn, "%s_sendmsg returned %d\n",
1746 sock == tconn->meta.socket ? "msock" : "sock",
1748 conn_request_state(tconn, NS(conn, C_BROKEN_PIPE), CS_HARD);
1750 conn_request_state(tconn, NS(conn, C_TIMEOUT), CS_HARD);
1756 static int drbd_open(struct block_device *bdev, fmode_t mode)
1758 struct drbd_conf *mdev = bdev->bd_disk->private_data;
1759 unsigned long flags;
1762 mutex_lock(&drbd_main_mutex);
1763 spin_lock_irqsave(&mdev->tconn->req_lock, flags);
1764 /* to have a stable mdev->state.role
1765 * and no race with updating open_cnt */
1767 if (mdev->state.role != R_PRIMARY) {
1768 if (mode & FMODE_WRITE)
1770 else if (!allow_oos)
1776 spin_unlock_irqrestore(&mdev->tconn->req_lock, flags);
1777 mutex_unlock(&drbd_main_mutex);
1782 static int drbd_release(struct gendisk *gd, fmode_t mode)
1784 struct drbd_conf *mdev = gd->private_data;
1785 mutex_lock(&drbd_main_mutex);
1787 mutex_unlock(&drbd_main_mutex);
1791 static void drbd_set_defaults(struct drbd_conf *mdev)
1793 /* This way we get a compile error when sync_conf grows,
1794 and we forgot to initialize it here */
1795 mdev->sync_conf = (struct syncer_conf) {
1796 /* .rate = */ DRBD_RATE_DEF,
1797 /* .after = */ DRBD_AFTER_DEF,
1798 /* .al_extents = */ DRBD_AL_EXTENTS_DEF,
1799 /* .verify_alg = */ {}, 0,
1800 /* .cpu_mask = */ {}, 0,
1801 /* .csums_alg = */ {}, 0,
1803 /* .on_no_data = */ DRBD_ON_NO_DATA_DEF,
1804 /* .c_plan_ahead = */ DRBD_C_PLAN_AHEAD_DEF,
1805 /* .c_delay_target = */ DRBD_C_DELAY_TARGET_DEF,
1806 /* .c_fill_target = */ DRBD_C_FILL_TARGET_DEF,
1807 /* .c_max_rate = */ DRBD_C_MAX_RATE_DEF,
1808 /* .c_min_rate = */ DRBD_C_MIN_RATE_DEF
1811 /* Have to use that way, because the layout differs between
1812 big endian and little endian */
1813 mdev->state = (union drbd_state) {
1814 { .role = R_SECONDARY,
1816 .conn = C_STANDALONE,
1825 void drbd_init_set_defaults(struct drbd_conf *mdev)
1827 /* the memset(,0,) did most of this.
1828 * note: only assignments, no allocation in here */
1830 drbd_set_defaults(mdev);
1832 atomic_set(&mdev->ap_bio_cnt, 0);
1833 atomic_set(&mdev->ap_pending_cnt, 0);
1834 atomic_set(&mdev->rs_pending_cnt, 0);
1835 atomic_set(&mdev->unacked_cnt, 0);
1836 atomic_set(&mdev->local_cnt, 0);
1837 atomic_set(&mdev->pp_in_use, 0);
1838 atomic_set(&mdev->pp_in_use_by_net, 0);
1839 atomic_set(&mdev->rs_sect_in, 0);
1840 atomic_set(&mdev->rs_sect_ev, 0);
1841 atomic_set(&mdev->ap_in_flight, 0);
1843 mutex_init(&mdev->md_io_mutex);
1844 mutex_init(&mdev->own_state_mutex);
1845 mdev->state_mutex = &mdev->own_state_mutex;
1847 spin_lock_init(&mdev->al_lock);
1848 spin_lock_init(&mdev->peer_seq_lock);
1849 spin_lock_init(&mdev->epoch_lock);
1851 INIT_LIST_HEAD(&mdev->active_ee);
1852 INIT_LIST_HEAD(&mdev->sync_ee);
1853 INIT_LIST_HEAD(&mdev->done_ee);
1854 INIT_LIST_HEAD(&mdev->read_ee);
1855 INIT_LIST_HEAD(&mdev->net_ee);
1856 INIT_LIST_HEAD(&mdev->resync_reads);
1857 INIT_LIST_HEAD(&mdev->resync_work.list);
1858 INIT_LIST_HEAD(&mdev->unplug_work.list);
1859 INIT_LIST_HEAD(&mdev->go_diskless.list);
1860 INIT_LIST_HEAD(&mdev->md_sync_work.list);
1861 INIT_LIST_HEAD(&mdev->start_resync_work.list);
1862 INIT_LIST_HEAD(&mdev->bm_io_work.w.list);
1864 mdev->resync_work.cb = w_resync_timer;
1865 mdev->unplug_work.cb = w_send_write_hint;
1866 mdev->go_diskless.cb = w_go_diskless;
1867 mdev->md_sync_work.cb = w_md_sync;
1868 mdev->bm_io_work.w.cb = w_bitmap_io;
1869 mdev->start_resync_work.cb = w_start_resync;
1871 mdev->resync_work.mdev = mdev;
1872 mdev->unplug_work.mdev = mdev;
1873 mdev->go_diskless.mdev = mdev;
1874 mdev->md_sync_work.mdev = mdev;
1875 mdev->bm_io_work.w.mdev = mdev;
1876 mdev->start_resync_work.mdev = mdev;
1878 init_timer(&mdev->resync_timer);
1879 init_timer(&mdev->md_sync_timer);
1880 init_timer(&mdev->start_resync_timer);
1881 init_timer(&mdev->request_timer);
1882 mdev->resync_timer.function = resync_timer_fn;
1883 mdev->resync_timer.data = (unsigned long) mdev;
1884 mdev->md_sync_timer.function = md_sync_timer_fn;
1885 mdev->md_sync_timer.data = (unsigned long) mdev;
1886 mdev->start_resync_timer.function = start_resync_timer_fn;
1887 mdev->start_resync_timer.data = (unsigned long) mdev;
1888 mdev->request_timer.function = request_timer_fn;
1889 mdev->request_timer.data = (unsigned long) mdev;
1891 init_waitqueue_head(&mdev->misc_wait);
1892 init_waitqueue_head(&mdev->state_wait);
1893 init_waitqueue_head(&mdev->ee_wait);
1894 init_waitqueue_head(&mdev->al_wait);
1895 init_waitqueue_head(&mdev->seq_wait);
1897 /* mdev->tconn->agreed_pro_version gets initialized in drbd_connect() */
1898 mdev->write_ordering = WO_bdev_flush;
1899 mdev->resync_wenr = LC_FREE;
1900 mdev->peer_max_bio_size = DRBD_MAX_BIO_SIZE_SAFE;
1901 mdev->local_max_bio_size = DRBD_MAX_BIO_SIZE_SAFE;
1904 void drbd_mdev_cleanup(struct drbd_conf *mdev)
1907 if (mdev->tconn->receiver.t_state != NONE)
1908 dev_err(DEV, "ASSERT FAILED: receiver t_state == %d expected 0.\n",
1909 mdev->tconn->receiver.t_state);
1911 /* no need to lock it, I'm the only thread alive */
1912 if (atomic_read(&mdev->current_epoch->epoch_size) != 0)
1913 dev_err(DEV, "epoch_size:%d\n", atomic_read(&mdev->current_epoch->epoch_size));
1923 mdev->rs_failed = 0;
1924 mdev->rs_last_events = 0;
1925 mdev->rs_last_sect_ev = 0;
1926 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
1927 mdev->rs_mark_left[i] = 0;
1928 mdev->rs_mark_time[i] = 0;
1930 D_ASSERT(mdev->tconn->net_conf == NULL);
1932 drbd_set_my_capacity(mdev, 0);
1934 /* maybe never allocated. */
1935 drbd_bm_resize(mdev, 0, 1);
1936 drbd_bm_cleanup(mdev);
1939 drbd_free_resources(mdev);
1940 clear_bit(AL_SUSPENDED, &mdev->flags);
1943 * currently we drbd_init_ee only on module load, so
1944 * we may do drbd_release_ee only on module unload!
1946 D_ASSERT(list_empty(&mdev->active_ee));
1947 D_ASSERT(list_empty(&mdev->sync_ee));
1948 D_ASSERT(list_empty(&mdev->done_ee));
1949 D_ASSERT(list_empty(&mdev->read_ee));
1950 D_ASSERT(list_empty(&mdev->net_ee));
1951 D_ASSERT(list_empty(&mdev->resync_reads));
1952 D_ASSERT(list_empty(&mdev->tconn->data.work.q));
1953 D_ASSERT(list_empty(&mdev->tconn->meta.work.q));
1954 D_ASSERT(list_empty(&mdev->resync_work.list));
1955 D_ASSERT(list_empty(&mdev->unplug_work.list));
1956 D_ASSERT(list_empty(&mdev->go_diskless.list));
1958 drbd_set_defaults(mdev);
1962 static void drbd_destroy_mempools(void)
1966 while (drbd_pp_pool) {
1967 page = drbd_pp_pool;
1968 drbd_pp_pool = (struct page *)page_private(page);
1973 /* D_ASSERT(atomic_read(&drbd_pp_vacant)==0); */
1975 if (drbd_md_io_bio_set)
1976 bioset_free(drbd_md_io_bio_set);
1977 if (drbd_md_io_page_pool)
1978 mempool_destroy(drbd_md_io_page_pool);
1979 if (drbd_ee_mempool)
1980 mempool_destroy(drbd_ee_mempool);
1981 if (drbd_request_mempool)
1982 mempool_destroy(drbd_request_mempool);
1984 kmem_cache_destroy(drbd_ee_cache);
1985 if (drbd_request_cache)
1986 kmem_cache_destroy(drbd_request_cache);
1987 if (drbd_bm_ext_cache)
1988 kmem_cache_destroy(drbd_bm_ext_cache);
1989 if (drbd_al_ext_cache)
1990 kmem_cache_destroy(drbd_al_ext_cache);
1992 drbd_md_io_bio_set = NULL;
1993 drbd_md_io_page_pool = NULL;
1994 drbd_ee_mempool = NULL;
1995 drbd_request_mempool = NULL;
1996 drbd_ee_cache = NULL;
1997 drbd_request_cache = NULL;
1998 drbd_bm_ext_cache = NULL;
1999 drbd_al_ext_cache = NULL;
2004 static int drbd_create_mempools(void)
2007 const int number = (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * minor_count;
2010 /* prepare our caches and mempools */
2011 drbd_request_mempool = NULL;
2012 drbd_ee_cache = NULL;
2013 drbd_request_cache = NULL;
2014 drbd_bm_ext_cache = NULL;
2015 drbd_al_ext_cache = NULL;
2016 drbd_pp_pool = NULL;
2017 drbd_md_io_page_pool = NULL;
2018 drbd_md_io_bio_set = NULL;
2021 drbd_request_cache = kmem_cache_create(
2022 "drbd_req", sizeof(struct drbd_request), 0, 0, NULL);
2023 if (drbd_request_cache == NULL)
2026 drbd_ee_cache = kmem_cache_create(
2027 "drbd_ee", sizeof(struct drbd_peer_request), 0, 0, NULL);
2028 if (drbd_ee_cache == NULL)
2031 drbd_bm_ext_cache = kmem_cache_create(
2032 "drbd_bm", sizeof(struct bm_extent), 0, 0, NULL);
2033 if (drbd_bm_ext_cache == NULL)
2036 drbd_al_ext_cache = kmem_cache_create(
2037 "drbd_al", sizeof(struct lc_element), 0, 0, NULL);
2038 if (drbd_al_ext_cache == NULL)
2042 drbd_md_io_bio_set = bioset_create(DRBD_MIN_POOL_PAGES, 0);
2043 if (drbd_md_io_bio_set == NULL)
2046 drbd_md_io_page_pool = mempool_create_page_pool(DRBD_MIN_POOL_PAGES, 0);
2047 if (drbd_md_io_page_pool == NULL)
2050 drbd_request_mempool = mempool_create(number,
2051 mempool_alloc_slab, mempool_free_slab, drbd_request_cache);
2052 if (drbd_request_mempool == NULL)
2055 drbd_ee_mempool = mempool_create(number,
2056 mempool_alloc_slab, mempool_free_slab, drbd_ee_cache);
2057 if (drbd_ee_mempool == NULL)
2060 /* drbd's page pool */
2061 spin_lock_init(&drbd_pp_lock);
2063 for (i = 0; i < number; i++) {
2064 page = alloc_page(GFP_HIGHUSER);
2067 set_page_private(page, (unsigned long)drbd_pp_pool);
2068 drbd_pp_pool = page;
2070 drbd_pp_vacant = number;
2075 drbd_destroy_mempools(); /* in case we allocated some */
2079 static int drbd_notify_sys(struct notifier_block *this, unsigned long code,
2082 /* just so we have it. you never know what interesting things we
2083 * might want to do here some day...
2089 static struct notifier_block drbd_notifier = {
2090 .notifier_call = drbd_notify_sys,
2093 static void drbd_release_ee_lists(struct drbd_conf *mdev)
2097 rr = drbd_release_ee(mdev, &mdev->active_ee);
2099 dev_err(DEV, "%d EEs in active list found!\n", rr);
2101 rr = drbd_release_ee(mdev, &mdev->sync_ee);
2103 dev_err(DEV, "%d EEs in sync list found!\n", rr);
2105 rr = drbd_release_ee(mdev, &mdev->read_ee);
2107 dev_err(DEV, "%d EEs in read list found!\n", rr);
2109 rr = drbd_release_ee(mdev, &mdev->done_ee);
2111 dev_err(DEV, "%d EEs in done list found!\n", rr);
2113 rr = drbd_release_ee(mdev, &mdev->net_ee);
2115 dev_err(DEV, "%d EEs in net list found!\n", rr);
2118 /* caution. no locking. */
2119 void drbd_delete_device(unsigned int minor)
2121 struct drbd_conf *mdev = minor_to_mdev(minor);
2126 idr_remove(&mdev->tconn->volumes, mdev->vnr);
2127 idr_remove(&minors, minor);
2130 /* paranoia asserts */
2131 D_ASSERT(mdev->open_cnt == 0);
2132 D_ASSERT(list_empty(&mdev->tconn->data.work.q));
2133 /* end paranoia asserts */
2135 del_gendisk(mdev->vdisk);
2137 /* cleanup stuff that may have been allocated during
2138 * device (re-)configuration or state changes */
2140 if (mdev->this_bdev)
2141 bdput(mdev->this_bdev);
2143 drbd_free_resources(mdev);
2145 drbd_release_ee_lists(mdev);
2147 lc_destroy(mdev->act_log);
2148 lc_destroy(mdev->resync);
2150 kfree(mdev->p_uuid);
2151 /* mdev->p_uuid = NULL; */
2153 /* cleanup the rest that has been
2154 * allocated from drbd_new_device
2155 * and actually free the mdev itself */
2156 drbd_free_mdev(mdev);
2159 static void drbd_cleanup(void)
2162 struct drbd_conf *mdev;
2164 unregister_reboot_notifier(&drbd_notifier);
2166 /* first remove proc,
2167 * drbdsetup uses it's presence to detect
2168 * whether DRBD is loaded.
2169 * If we would get stuck in proc removal,
2170 * but have netlink already deregistered,
2171 * some drbdsetup commands may wait forever
2175 remove_proc_entry("drbd", NULL);
2177 drbd_genl_unregister();
2179 idr_for_each_entry(&minors, mdev, i)
2180 drbd_delete_device(i);
2181 drbd_destroy_mempools();
2182 unregister_blkdev(DRBD_MAJOR, "drbd");
2184 idr_destroy(&minors);
2186 printk(KERN_INFO "drbd: module cleanup done.\n");
2190 * drbd_congested() - Callback for pdflush
2191 * @congested_data: User data
2192 * @bdi_bits: Bits pdflush is currently interested in
2194 * Returns 1<<BDI_async_congested and/or 1<<BDI_sync_congested if we are congested.
2196 static int drbd_congested(void *congested_data, int bdi_bits)
2198 struct drbd_conf *mdev = congested_data;
2199 struct request_queue *q;
2203 if (!may_inc_ap_bio(mdev)) {
2204 /* DRBD has frozen IO */
2210 if (get_ldev(mdev)) {
2211 q = bdev_get_queue(mdev->ldev->backing_bdev);
2212 r = bdi_congested(&q->backing_dev_info, bdi_bits);
2218 if (bdi_bits & (1 << BDI_async_congested) && test_bit(NET_CONGESTED, &mdev->tconn->flags)) {
2219 r |= (1 << BDI_async_congested);
2220 reason = reason == 'b' ? 'a' : 'n';
2224 mdev->congestion_reason = reason;
2228 static void drbd_init_workqueue(struct drbd_work_queue* wq)
2230 sema_init(&wq->s, 0);
2231 spin_lock_init(&wq->q_lock);
2232 INIT_LIST_HEAD(&wq->q);
2235 struct drbd_tconn *conn_by_name(const char *name)
2237 struct drbd_tconn *tconn;
2239 if (!name || !name[0])
2242 mutex_lock(&drbd_cfg_mutex);
2243 list_for_each_entry(tconn, &drbd_tconns, all_tconn) {
2244 if (!strcmp(tconn->name, name))
2249 mutex_unlock(&drbd_cfg_mutex);
2253 struct drbd_tconn *drbd_new_tconn(const char *name)
2255 struct drbd_tconn *tconn;
2257 tconn = kzalloc(sizeof(struct drbd_tconn), GFP_KERNEL);
2261 tconn->name = kstrdup(name, GFP_KERNEL);
2265 if (!zalloc_cpumask_var(&tconn->cpu_mask, GFP_KERNEL))
2268 if (!tl_init(tconn))
2271 tconn->cstate = C_STANDALONE;
2272 mutex_init(&tconn->cstate_mutex);
2273 spin_lock_init(&tconn->req_lock);
2274 atomic_set(&tconn->net_cnt, 0);
2275 init_waitqueue_head(&tconn->net_cnt_wait);
2276 init_waitqueue_head(&tconn->ping_wait);
2277 idr_init(&tconn->volumes);
2279 drbd_init_workqueue(&tconn->data.work);
2280 mutex_init(&tconn->data.mutex);
2282 drbd_init_workqueue(&tconn->meta.work);
2283 mutex_init(&tconn->meta.mutex);
2285 drbd_thread_init(tconn, &tconn->receiver, drbdd_init, "receiver");
2286 drbd_thread_init(tconn, &tconn->worker, drbd_worker, "worker");
2287 drbd_thread_init(tconn, &tconn->asender, drbd_asender, "asender");
2289 mutex_lock(&drbd_cfg_mutex);
2290 list_add_tail(&tconn->all_tconn, &drbd_tconns);
2291 mutex_unlock(&drbd_cfg_mutex);
2297 free_cpumask_var(tconn->cpu_mask);
2304 void drbd_free_tconn(struct drbd_tconn *tconn)
2306 mutex_lock(&drbd_cfg_mutex);
2307 list_del(&tconn->all_tconn);
2308 mutex_unlock(&drbd_cfg_mutex);
2309 idr_destroy(&tconn->volumes);
2311 free_cpumask_var(tconn->cpu_mask);
2313 kfree(tconn->int_dig_out);
2314 kfree(tconn->int_dig_in);
2315 kfree(tconn->int_dig_vv);
2319 enum drbd_ret_code conn_new_minor(struct drbd_tconn *tconn, unsigned int minor, int vnr)
2321 struct drbd_conf *mdev;
2322 struct gendisk *disk;
2323 struct request_queue *q;
2325 int minor_got = minor;
2326 enum drbd_ret_code err = ERR_NOMEM;
2328 mdev = minor_to_mdev(minor);
2330 return ERR_MINOR_EXISTS;
2332 /* GFP_KERNEL, we are outside of all write-out paths */
2333 mdev = kzalloc(sizeof(struct drbd_conf), GFP_KERNEL);
2337 mdev->tconn = tconn;
2338 mdev->minor = minor;
2341 drbd_init_set_defaults(mdev);
2343 q = blk_alloc_queue(GFP_KERNEL);
2347 q->queuedata = mdev;
2349 disk = alloc_disk(1);
2354 set_disk_ro(disk, true);
2357 disk->major = DRBD_MAJOR;
2358 disk->first_minor = minor;
2359 disk->fops = &drbd_ops;
2360 sprintf(disk->disk_name, "drbd%d", minor);
2361 disk->private_data = mdev;
2363 mdev->this_bdev = bdget(MKDEV(DRBD_MAJOR, minor));
2364 /* we have no partitions. we contain only ourselves. */
2365 mdev->this_bdev->bd_contains = mdev->this_bdev;
2367 q->backing_dev_info.congested_fn = drbd_congested;
2368 q->backing_dev_info.congested_data = mdev;
2370 blk_queue_make_request(q, drbd_make_request);
2371 /* Setting the max_hw_sectors to an odd value of 8kibyte here
2372 This triggers a max_bio_size message upon first attach or connect */
2373 blk_queue_max_hw_sectors(q, DRBD_MAX_BIO_SIZE_SAFE >> 8);
2374 blk_queue_bounce_limit(q, BLK_BOUNCE_ANY);
2375 blk_queue_merge_bvec(q, drbd_merge_bvec);
2376 q->queue_lock = &mdev->tconn->req_lock; /* needed since we use */
2378 mdev->md_io_page = alloc_page(GFP_KERNEL);
2379 if (!mdev->md_io_page)
2380 goto out_no_io_page;
2382 if (drbd_bm_init(mdev))
2384 mdev->read_requests = RB_ROOT;
2385 mdev->write_requests = RB_ROOT;
2387 mdev->current_epoch = kzalloc(sizeof(struct drbd_epoch), GFP_KERNEL);
2388 if (!mdev->current_epoch)
2391 INIT_LIST_HEAD(&mdev->current_epoch->list);
2394 if (!idr_pre_get(&minors, GFP_KERNEL))
2395 goto out_no_minor_idr;
2396 if (idr_get_new_above(&minors, mdev, minor, &minor_got))
2397 goto out_no_minor_idr;
2398 if (minor_got != minor) {
2399 err = ERR_MINOR_EXISTS;
2400 drbd_msg_put_info("requested minor exists already");
2401 goto out_idr_remove_minor;
2404 if (!idr_pre_get(&tconn->volumes, GFP_KERNEL))
2405 goto out_idr_remove_minor;
2406 if (idr_get_new_above(&tconn->volumes, mdev, vnr, &vnr_got))
2407 goto out_idr_remove_minor;
2408 if (vnr_got != vnr) {
2409 err = ERR_INVALID_REQUEST;
2410 drbd_msg_put_info("requested volume exists already");
2411 goto out_idr_remove_vol;
2418 idr_remove(&tconn->volumes, vnr_got);
2419 out_idr_remove_minor:
2420 idr_remove(&minors, minor_got);
2423 kfree(mdev->current_epoch);
2425 drbd_bm_cleanup(mdev);
2427 __free_page(mdev->md_io_page);
2431 blk_cleanup_queue(q);
2437 /* counterpart of drbd_new_device.
2438 * last part of drbd_delete_device. */
2439 void drbd_free_mdev(struct drbd_conf *mdev)
2441 kfree(mdev->current_epoch);
2442 if (mdev->bitmap) /* should no longer be there. */
2443 drbd_bm_cleanup(mdev);
2444 __free_page(mdev->md_io_page);
2445 put_disk(mdev->vdisk);
2446 blk_cleanup_queue(mdev->rq_queue);
2451 int __init drbd_init(void)
2455 BUILD_BUG_ON(sizeof(struct p_header80) != sizeof(struct p_header95));
2456 BUILD_BUG_ON(sizeof(struct p_handshake) != 80);
2458 if (minor_count < DRBD_MINOR_COUNT_MIN || minor_count > DRBD_MINOR_COUNT_MAX) {
2460 "drbd: invalid minor_count (%d)\n", minor_count);
2468 err = register_blkdev(DRBD_MAJOR, "drbd");
2471 "drbd: unable to register block device major %d\n",
2476 err = drbd_genl_register();
2478 printk(KERN_ERR "drbd: unable to register generic netlink family\n");
2483 register_reboot_notifier(&drbd_notifier);
2486 * allocate all necessary structs
2490 init_waitqueue_head(&drbd_pp_wait);
2492 drbd_proc = NULL; /* play safe for drbd_cleanup */
2495 err = drbd_create_mempools();
2499 drbd_proc = proc_create_data("drbd", S_IFREG | S_IRUGO , NULL, &drbd_proc_fops, NULL);
2501 printk(KERN_ERR "drbd: unable to register proc file\n");
2505 rwlock_init(&global_state_lock);
2506 INIT_LIST_HEAD(&drbd_tconns);
2508 printk(KERN_INFO "drbd: initialized. "
2509 "Version: " REL_VERSION " (api:%d/proto:%d-%d)\n",
2510 API_VERSION, PRO_VERSION_MIN, PRO_VERSION_MAX);
2511 printk(KERN_INFO "drbd: %s\n", drbd_buildtag());
2512 printk(KERN_INFO "drbd: registered as block device major %d\n",
2515 return 0; /* Success! */
2520 /* currently always the case */
2521 printk(KERN_ERR "drbd: ran out of memory\n");
2523 printk(KERN_ERR "drbd: initialization failure\n");
2527 void drbd_free_bc(struct drbd_backing_dev *ldev)
2532 blkdev_put(ldev->backing_bdev, FMODE_READ | FMODE_WRITE | FMODE_EXCL);
2533 blkdev_put(ldev->md_bdev, FMODE_READ | FMODE_WRITE | FMODE_EXCL);
2538 void drbd_free_sock(struct drbd_tconn *tconn)
2540 if (tconn->data.socket) {
2541 mutex_lock(&tconn->data.mutex);
2542 kernel_sock_shutdown(tconn->data.socket, SHUT_RDWR);
2543 sock_release(tconn->data.socket);
2544 tconn->data.socket = NULL;
2545 mutex_unlock(&tconn->data.mutex);
2547 if (tconn->meta.socket) {
2548 mutex_lock(&tconn->meta.mutex);
2549 kernel_sock_shutdown(tconn->meta.socket, SHUT_RDWR);
2550 sock_release(tconn->meta.socket);
2551 tconn->meta.socket = NULL;
2552 mutex_unlock(&tconn->meta.mutex);
2557 void drbd_free_resources(struct drbd_conf *mdev)
2559 crypto_free_hash(mdev->csums_tfm);
2560 mdev->csums_tfm = NULL;
2561 crypto_free_hash(mdev->verify_tfm);
2562 mdev->verify_tfm = NULL;
2563 crypto_free_hash(mdev->tconn->cram_hmac_tfm);
2564 mdev->tconn->cram_hmac_tfm = NULL;
2565 crypto_free_hash(mdev->tconn->integrity_w_tfm);
2566 mdev->tconn->integrity_w_tfm = NULL;
2567 crypto_free_hash(mdev->tconn->integrity_r_tfm);
2568 mdev->tconn->integrity_r_tfm = NULL;
2570 drbd_free_sock(mdev->tconn);
2573 drbd_free_bc(mdev->ldev);
2574 mdev->ldev = NULL;);
2577 /* meta data management */
2579 struct meta_data_on_disk {
2580 u64 la_size; /* last agreed size. */
2581 u64 uuid[UI_SIZE]; /* UUIDs. */
2584 u32 flags; /* MDF */
2587 u32 al_offset; /* offset to this block */
2588 u32 al_nr_extents; /* important for restoring the AL */
2589 /* `-- act_log->nr_elements <-- sync_conf.al_extents */
2590 u32 bm_offset; /* offset to the bitmap, from here */
2591 u32 bm_bytes_per_bit; /* BM_BLOCK_SIZE */
2592 u32 la_peer_max_bio_size; /* last peer max_bio_size */
2593 u32 reserved_u32[3];
2598 * drbd_md_sync() - Writes the meta data super block if the MD_DIRTY flag bit is set
2599 * @mdev: DRBD device.
2601 void drbd_md_sync(struct drbd_conf *mdev)
2603 struct meta_data_on_disk *buffer;
2607 del_timer(&mdev->md_sync_timer);
2608 /* timer may be rearmed by drbd_md_mark_dirty() now. */
2609 if (!test_and_clear_bit(MD_DIRTY, &mdev->flags))
2612 /* We use here D_FAILED and not D_ATTACHING because we try to write
2613 * metadata even if we detach due to a disk failure! */
2614 if (!get_ldev_if_state(mdev, D_FAILED))
2617 mutex_lock(&mdev->md_io_mutex);
2618 buffer = (struct meta_data_on_disk *)page_address(mdev->md_io_page);
2619 memset(buffer, 0, 512);
2621 buffer->la_size = cpu_to_be64(drbd_get_capacity(mdev->this_bdev));
2622 for (i = UI_CURRENT; i < UI_SIZE; i++)
2623 buffer->uuid[i] = cpu_to_be64(mdev->ldev->md.uuid[i]);
2624 buffer->flags = cpu_to_be32(mdev->ldev->md.flags);
2625 buffer->magic = cpu_to_be32(DRBD_MD_MAGIC);
2627 buffer->md_size_sect = cpu_to_be32(mdev->ldev->md.md_size_sect);
2628 buffer->al_offset = cpu_to_be32(mdev->ldev->md.al_offset);
2629 buffer->al_nr_extents = cpu_to_be32(mdev->act_log->nr_elements);
2630 buffer->bm_bytes_per_bit = cpu_to_be32(BM_BLOCK_SIZE);
2631 buffer->device_uuid = cpu_to_be64(mdev->ldev->md.device_uuid);
2633 buffer->bm_offset = cpu_to_be32(mdev->ldev->md.bm_offset);
2634 buffer->la_peer_max_bio_size = cpu_to_be32(mdev->peer_max_bio_size);
2636 D_ASSERT(drbd_md_ss__(mdev, mdev->ldev) == mdev->ldev->md.md_offset);
2637 sector = mdev->ldev->md.md_offset;
2639 if (!drbd_md_sync_page_io(mdev, mdev->ldev, sector, WRITE)) {
2640 /* this was a try anyways ... */
2641 dev_err(DEV, "meta data update failed!\n");
2642 drbd_chk_io_error(mdev, 1, true);
2645 /* Update mdev->ldev->md.la_size_sect,
2646 * since we updated it on metadata. */
2647 mdev->ldev->md.la_size_sect = drbd_get_capacity(mdev->this_bdev);
2649 mutex_unlock(&mdev->md_io_mutex);
2654 * drbd_md_read() - Reads in the meta data super block
2655 * @mdev: DRBD device.
2656 * @bdev: Device from which the meta data should be read in.
2658 * Return 0 (NO_ERROR) on success, and an enum drbd_ret_code in case
2659 * something goes wrong. Currently only: ERR_IO_MD_DISK, ERR_MD_INVALID.
2661 int drbd_md_read(struct drbd_conf *mdev, struct drbd_backing_dev *bdev)
2663 struct meta_data_on_disk *buffer;
2664 int i, rv = NO_ERROR;
2666 if (!get_ldev_if_state(mdev, D_ATTACHING))
2667 return ERR_IO_MD_DISK;
2669 mutex_lock(&mdev->md_io_mutex);
2670 buffer = (struct meta_data_on_disk *)page_address(mdev->md_io_page);
2672 if (!drbd_md_sync_page_io(mdev, bdev, bdev->md.md_offset, READ)) {
2673 /* NOTE: can't do normal error processing here as this is
2674 called BEFORE disk is attached */
2675 dev_err(DEV, "Error while reading metadata.\n");
2676 rv = ERR_IO_MD_DISK;
2680 if (buffer->magic != cpu_to_be32(DRBD_MD_MAGIC)) {
2681 dev_err(DEV, "Error while reading metadata, magic not found.\n");
2682 rv = ERR_MD_INVALID;
2685 if (be32_to_cpu(buffer->al_offset) != bdev->md.al_offset) {
2686 dev_err(DEV, "unexpected al_offset: %d (expected %d)\n",
2687 be32_to_cpu(buffer->al_offset), bdev->md.al_offset);
2688 rv = ERR_MD_INVALID;
2691 if (be32_to_cpu(buffer->bm_offset) != bdev->md.bm_offset) {
2692 dev_err(DEV, "unexpected bm_offset: %d (expected %d)\n",
2693 be32_to_cpu(buffer->bm_offset), bdev->md.bm_offset);
2694 rv = ERR_MD_INVALID;
2697 if (be32_to_cpu(buffer->md_size_sect) != bdev->md.md_size_sect) {
2698 dev_err(DEV, "unexpected md_size: %u (expected %u)\n",
2699 be32_to_cpu(buffer->md_size_sect), bdev->md.md_size_sect);
2700 rv = ERR_MD_INVALID;
2704 if (be32_to_cpu(buffer->bm_bytes_per_bit) != BM_BLOCK_SIZE) {
2705 dev_err(DEV, "unexpected bm_bytes_per_bit: %u (expected %u)\n",
2706 be32_to_cpu(buffer->bm_bytes_per_bit), BM_BLOCK_SIZE);
2707 rv = ERR_MD_INVALID;
2711 bdev->md.la_size_sect = be64_to_cpu(buffer->la_size);
2712 for (i = UI_CURRENT; i < UI_SIZE; i++)
2713 bdev->md.uuid[i] = be64_to_cpu(buffer->uuid[i]);
2714 bdev->md.flags = be32_to_cpu(buffer->flags);
2715 mdev->sync_conf.al_extents = be32_to_cpu(buffer->al_nr_extents);
2716 bdev->md.device_uuid = be64_to_cpu(buffer->device_uuid);
2718 spin_lock_irq(&mdev->tconn->req_lock);
2719 if (mdev->state.conn < C_CONNECTED) {
2721 peer = be32_to_cpu(buffer->la_peer_max_bio_size);
2722 peer = max_t(int, peer, DRBD_MAX_BIO_SIZE_SAFE);
2723 mdev->peer_max_bio_size = peer;
2725 spin_unlock_irq(&mdev->tconn->req_lock);
2727 if (mdev->sync_conf.al_extents < 7)
2728 mdev->sync_conf.al_extents = 127;
2731 mutex_unlock(&mdev->md_io_mutex);
2738 * drbd_md_mark_dirty() - Mark meta data super block as dirty
2739 * @mdev: DRBD device.
2741 * Call this function if you change anything that should be written to
2742 * the meta-data super block. This function sets MD_DIRTY, and starts a
2743 * timer that ensures that within five seconds you have to call drbd_md_sync().
2746 void drbd_md_mark_dirty_(struct drbd_conf *mdev, unsigned int line, const char *func)
2748 if (!test_and_set_bit(MD_DIRTY, &mdev->flags)) {
2749 mod_timer(&mdev->md_sync_timer, jiffies + HZ);
2750 mdev->last_md_mark_dirty.line = line;
2751 mdev->last_md_mark_dirty.func = func;
2755 void drbd_md_mark_dirty(struct drbd_conf *mdev)
2757 if (!test_and_set_bit(MD_DIRTY, &mdev->flags))
2758 mod_timer(&mdev->md_sync_timer, jiffies + 5*HZ);
2762 static void drbd_uuid_move_history(struct drbd_conf *mdev) __must_hold(local)
2766 for (i = UI_HISTORY_START; i < UI_HISTORY_END; i++)
2767 mdev->ldev->md.uuid[i+1] = mdev->ldev->md.uuid[i];
2770 void _drbd_uuid_set(struct drbd_conf *mdev, int idx, u64 val) __must_hold(local)
2772 if (idx == UI_CURRENT) {
2773 if (mdev->state.role == R_PRIMARY)
2778 drbd_set_ed_uuid(mdev, val);
2781 mdev->ldev->md.uuid[idx] = val;
2782 drbd_md_mark_dirty(mdev);
2786 void drbd_uuid_set(struct drbd_conf *mdev, int idx, u64 val) __must_hold(local)
2788 if (mdev->ldev->md.uuid[idx]) {
2789 drbd_uuid_move_history(mdev);
2790 mdev->ldev->md.uuid[UI_HISTORY_START] = mdev->ldev->md.uuid[idx];
2792 _drbd_uuid_set(mdev, idx, val);
2796 * drbd_uuid_new_current() - Creates a new current UUID
2797 * @mdev: DRBD device.
2799 * Creates a new current UUID, and rotates the old current UUID into
2800 * the bitmap slot. Causes an incremental resync upon next connect.
2802 void drbd_uuid_new_current(struct drbd_conf *mdev) __must_hold(local)
2805 unsigned long long bm_uuid = mdev->ldev->md.uuid[UI_BITMAP];
2808 dev_warn(DEV, "bm UUID was already set: %llX\n", bm_uuid);
2810 mdev->ldev->md.uuid[UI_BITMAP] = mdev->ldev->md.uuid[UI_CURRENT];
2812 get_random_bytes(&val, sizeof(u64));
2813 _drbd_uuid_set(mdev, UI_CURRENT, val);
2814 drbd_print_uuids(mdev, "new current UUID");
2815 /* get it to stable storage _now_ */
2819 void drbd_uuid_set_bm(struct drbd_conf *mdev, u64 val) __must_hold(local)
2821 if (mdev->ldev->md.uuid[UI_BITMAP] == 0 && val == 0)
2825 drbd_uuid_move_history(mdev);
2826 mdev->ldev->md.uuid[UI_HISTORY_START] = mdev->ldev->md.uuid[UI_BITMAP];
2827 mdev->ldev->md.uuid[UI_BITMAP] = 0;
2829 unsigned long long bm_uuid = mdev->ldev->md.uuid[UI_BITMAP];
2831 dev_warn(DEV, "bm UUID was already set: %llX\n", bm_uuid);
2833 mdev->ldev->md.uuid[UI_BITMAP] = val & ~((u64)1);
2835 drbd_md_mark_dirty(mdev);
2839 * drbd_bmio_set_n_write() - io_fn for drbd_queue_bitmap_io() or drbd_bitmap_io()
2840 * @mdev: DRBD device.
2842 * Sets all bits in the bitmap and writes the whole bitmap to stable storage.
2844 int drbd_bmio_set_n_write(struct drbd_conf *mdev)
2848 if (get_ldev_if_state(mdev, D_ATTACHING)) {
2849 drbd_md_set_flag(mdev, MDF_FULL_SYNC);
2851 drbd_bm_set_all(mdev);
2853 rv = drbd_bm_write(mdev);
2856 drbd_md_clear_flag(mdev, MDF_FULL_SYNC);
2867 * drbd_bmio_clear_n_write() - io_fn for drbd_queue_bitmap_io() or drbd_bitmap_io()
2868 * @mdev: DRBD device.
2870 * Clears all bits in the bitmap and writes the whole bitmap to stable storage.
2872 int drbd_bmio_clear_n_write(struct drbd_conf *mdev)
2876 drbd_resume_al(mdev);
2877 if (get_ldev_if_state(mdev, D_ATTACHING)) {
2878 drbd_bm_clear_all(mdev);
2879 rv = drbd_bm_write(mdev);
2886 static int w_bitmap_io(struct drbd_work *w, int unused)
2888 struct bm_io_work *work = container_of(w, struct bm_io_work, w);
2889 struct drbd_conf *mdev = w->mdev;
2892 D_ASSERT(atomic_read(&mdev->ap_bio_cnt) == 0);
2894 if (get_ldev(mdev)) {
2895 drbd_bm_lock(mdev, work->why, work->flags);
2896 rv = work->io_fn(mdev);
2897 drbd_bm_unlock(mdev);
2901 clear_bit_unlock(BITMAP_IO, &mdev->flags);
2902 wake_up(&mdev->misc_wait);
2905 work->done(mdev, rv);
2907 clear_bit(BITMAP_IO_QUEUED, &mdev->flags);
2914 void drbd_ldev_destroy(struct drbd_conf *mdev)
2916 lc_destroy(mdev->resync);
2917 mdev->resync = NULL;
2918 lc_destroy(mdev->act_log);
2919 mdev->act_log = NULL;
2921 drbd_free_bc(mdev->ldev);
2922 mdev->ldev = NULL;);
2924 clear_bit(GO_DISKLESS, &mdev->flags);
2927 static int w_go_diskless(struct drbd_work *w, int unused)
2929 struct drbd_conf *mdev = w->mdev;
2931 D_ASSERT(mdev->state.disk == D_FAILED);
2932 /* we cannot assert local_cnt == 0 here, as get_ldev_if_state will
2933 * inc/dec it frequently. Once we are D_DISKLESS, no one will touch
2934 * the protected members anymore, though, so once put_ldev reaches zero
2935 * again, it will be safe to free them. */
2936 drbd_force_state(mdev, NS(disk, D_DISKLESS));
2940 void drbd_go_diskless(struct drbd_conf *mdev)
2942 D_ASSERT(mdev->state.disk == D_FAILED);
2943 if (!test_and_set_bit(GO_DISKLESS, &mdev->flags))
2944 drbd_queue_work(&mdev->tconn->data.work, &mdev->go_diskless);
2948 * drbd_queue_bitmap_io() - Queues an IO operation on the whole bitmap
2949 * @mdev: DRBD device.
2950 * @io_fn: IO callback to be called when bitmap IO is possible
2951 * @done: callback to be called after the bitmap IO was performed
2952 * @why: Descriptive text of the reason for doing the IO
2954 * While IO on the bitmap happens we freeze application IO thus we ensure
2955 * that drbd_set_out_of_sync() can not be called. This function MAY ONLY be
2956 * called from worker context. It MUST NOT be used while a previous such
2957 * work is still pending!
2959 void drbd_queue_bitmap_io(struct drbd_conf *mdev,
2960 int (*io_fn)(struct drbd_conf *),
2961 void (*done)(struct drbd_conf *, int),
2962 char *why, enum bm_flag flags)
2964 D_ASSERT(current == mdev->tconn->worker.task);
2966 D_ASSERT(!test_bit(BITMAP_IO_QUEUED, &mdev->flags));
2967 D_ASSERT(!test_bit(BITMAP_IO, &mdev->flags));
2968 D_ASSERT(list_empty(&mdev->bm_io_work.w.list));
2969 if (mdev->bm_io_work.why)
2970 dev_err(DEV, "FIXME going to queue '%s' but '%s' still pending?\n",
2971 why, mdev->bm_io_work.why);
2973 mdev->bm_io_work.io_fn = io_fn;
2974 mdev->bm_io_work.done = done;
2975 mdev->bm_io_work.why = why;
2976 mdev->bm_io_work.flags = flags;
2978 spin_lock_irq(&mdev->tconn->req_lock);
2979 set_bit(BITMAP_IO, &mdev->flags);
2980 if (atomic_read(&mdev->ap_bio_cnt) == 0) {
2981 if (!test_and_set_bit(BITMAP_IO_QUEUED, &mdev->flags))
2982 drbd_queue_work(&mdev->tconn->data.work, &mdev->bm_io_work.w);
2984 spin_unlock_irq(&mdev->tconn->req_lock);
2988 * drbd_bitmap_io() - Does an IO operation on the whole bitmap
2989 * @mdev: DRBD device.
2990 * @io_fn: IO callback to be called when bitmap IO is possible
2991 * @why: Descriptive text of the reason for doing the IO
2993 * freezes application IO while that the actual IO operations runs. This
2994 * functions MAY NOT be called from worker context.
2996 int drbd_bitmap_io(struct drbd_conf *mdev, int (*io_fn)(struct drbd_conf *),
2997 char *why, enum bm_flag flags)
3001 D_ASSERT(current != mdev->tconn->worker.task);
3003 if ((flags & BM_LOCKED_SET_ALLOWED) == 0)
3004 drbd_suspend_io(mdev);
3006 drbd_bm_lock(mdev, why, flags);
3008 drbd_bm_unlock(mdev);
3010 if ((flags & BM_LOCKED_SET_ALLOWED) == 0)
3011 drbd_resume_io(mdev);
3016 void drbd_md_set_flag(struct drbd_conf *mdev, int flag) __must_hold(local)
3018 if ((mdev->ldev->md.flags & flag) != flag) {
3019 drbd_md_mark_dirty(mdev);
3020 mdev->ldev->md.flags |= flag;
3024 void drbd_md_clear_flag(struct drbd_conf *mdev, int flag) __must_hold(local)
3026 if ((mdev->ldev->md.flags & flag) != 0) {
3027 drbd_md_mark_dirty(mdev);
3028 mdev->ldev->md.flags &= ~flag;
3031 int drbd_md_test_flag(struct drbd_backing_dev *bdev, int flag)
3033 return (bdev->md.flags & flag) != 0;
3036 static void md_sync_timer_fn(unsigned long data)
3038 struct drbd_conf *mdev = (struct drbd_conf *) data;
3040 drbd_queue_work_front(&mdev->tconn->data.work, &mdev->md_sync_work);
3043 static int w_md_sync(struct drbd_work *w, int unused)
3045 struct drbd_conf *mdev = w->mdev;
3047 dev_warn(DEV, "md_sync_timer expired! Worker calls drbd_md_sync().\n");
3049 dev_warn(DEV, "last md_mark_dirty: %s:%u\n",
3050 mdev->last_md_mark_dirty.func, mdev->last_md_mark_dirty.line);
3056 const char *cmdname(enum drbd_packet cmd)
3058 /* THINK may need to become several global tables
3059 * when we want to support more than
3060 * one PRO_VERSION */
3061 static const char *cmdnames[] = {
3063 [P_DATA_REPLY] = "DataReply",
3064 [P_RS_DATA_REPLY] = "RSDataReply",
3065 [P_BARRIER] = "Barrier",
3066 [P_BITMAP] = "ReportBitMap",
3067 [P_BECOME_SYNC_TARGET] = "BecomeSyncTarget",
3068 [P_BECOME_SYNC_SOURCE] = "BecomeSyncSource",
3069 [P_UNPLUG_REMOTE] = "UnplugRemote",
3070 [P_DATA_REQUEST] = "DataRequest",
3071 [P_RS_DATA_REQUEST] = "RSDataRequest",
3072 [P_SYNC_PARAM] = "SyncParam",
3073 [P_SYNC_PARAM89] = "SyncParam89",
3074 [P_PROTOCOL] = "ReportProtocol",
3075 [P_UUIDS] = "ReportUUIDs",
3076 [P_SIZES] = "ReportSizes",
3077 [P_STATE] = "ReportState",
3078 [P_SYNC_UUID] = "ReportSyncUUID",
3079 [P_AUTH_CHALLENGE] = "AuthChallenge",
3080 [P_AUTH_RESPONSE] = "AuthResponse",
3082 [P_PING_ACK] = "PingAck",
3083 [P_RECV_ACK] = "RecvAck",
3084 [P_WRITE_ACK] = "WriteAck",
3085 [P_RS_WRITE_ACK] = "RSWriteAck",
3086 [P_DISCARD_WRITE] = "DiscardWrite",
3087 [P_NEG_ACK] = "NegAck",
3088 [P_NEG_DREPLY] = "NegDReply",
3089 [P_NEG_RS_DREPLY] = "NegRSDReply",
3090 [P_BARRIER_ACK] = "BarrierAck",
3091 [P_STATE_CHG_REQ] = "StateChgRequest",
3092 [P_STATE_CHG_REPLY] = "StateChgReply",
3093 [P_OV_REQUEST] = "OVRequest",
3094 [P_OV_REPLY] = "OVReply",
3095 [P_OV_RESULT] = "OVResult",
3096 [P_CSUM_RS_REQUEST] = "CsumRSRequest",
3097 [P_RS_IS_IN_SYNC] = "CsumRSIsInSync",
3098 [P_COMPRESSED_BITMAP] = "CBitmap",
3099 [P_DELAY_PROBE] = "DelayProbe",
3100 [P_OUT_OF_SYNC] = "OutOfSync",
3101 [P_RETRY_WRITE] = "RetryWrite",
3104 if (cmd == P_HAND_SHAKE_M)
3105 return "HandShakeM";
3106 if (cmd == P_HAND_SHAKE_S)
3107 return "HandShakeS";
3108 if (cmd == P_HAND_SHAKE)
3110 if (cmd >= ARRAY_SIZE(cmdnames))
3112 return cmdnames[cmd];
3116 * drbd_wait_misc - wait for a request to make progress
3117 * @mdev: device associated with the request
3118 * @i: the struct drbd_interval embedded in struct drbd_request or
3119 * struct drbd_peer_request
3121 int drbd_wait_misc(struct drbd_conf *mdev, struct drbd_interval *i)
3123 struct net_conf *net_conf = mdev->tconn->net_conf;
3129 timeout = MAX_SCHEDULE_TIMEOUT;
3130 if (net_conf->ko_count)
3131 timeout = net_conf->timeout * HZ / 10 * net_conf->ko_count;
3133 /* Indicate to wake up mdev->misc_wait on progress. */
3135 prepare_to_wait(&mdev->misc_wait, &wait, TASK_INTERRUPTIBLE);
3136 spin_unlock_irq(&mdev->tconn->req_lock);
3137 timeout = schedule_timeout(timeout);
3138 finish_wait(&mdev->misc_wait, &wait);
3139 spin_lock_irq(&mdev->tconn->req_lock);
3140 if (!timeout || mdev->state.conn < C_CONNECTED)
3142 if (signal_pending(current))
3143 return -ERESTARTSYS;
3147 #ifdef CONFIG_DRBD_FAULT_INJECTION
3148 /* Fault insertion support including random number generator shamelessly
3149 * stolen from kernel/rcutorture.c */
3150 struct fault_random_state {
3151 unsigned long state;
3152 unsigned long count;
3155 #define FAULT_RANDOM_MULT 39916801 /* prime */
3156 #define FAULT_RANDOM_ADD 479001701 /* prime */
3157 #define FAULT_RANDOM_REFRESH 10000
3160 * Crude but fast random-number generator. Uses a linear congruential
3161 * generator, with occasional help from get_random_bytes().
3163 static unsigned long
3164 _drbd_fault_random(struct fault_random_state *rsp)
3168 if (!rsp->count--) {
3169 get_random_bytes(&refresh, sizeof(refresh));
3170 rsp->state += refresh;
3171 rsp->count = FAULT_RANDOM_REFRESH;
3173 rsp->state = rsp->state * FAULT_RANDOM_MULT + FAULT_RANDOM_ADD;
3174 return swahw32(rsp->state);
3178 _drbd_fault_str(unsigned int type) {
3179 static char *_faults[] = {
3180 [DRBD_FAULT_MD_WR] = "Meta-data write",
3181 [DRBD_FAULT_MD_RD] = "Meta-data read",
3182 [DRBD_FAULT_RS_WR] = "Resync write",
3183 [DRBD_FAULT_RS_RD] = "Resync read",
3184 [DRBD_FAULT_DT_WR] = "Data write",
3185 [DRBD_FAULT_DT_RD] = "Data read",
3186 [DRBD_FAULT_DT_RA] = "Data read ahead",
3187 [DRBD_FAULT_BM_ALLOC] = "BM allocation",
3188 [DRBD_FAULT_AL_EE] = "EE allocation",
3189 [DRBD_FAULT_RECEIVE] = "receive data corruption",
3192 return (type < DRBD_FAULT_MAX) ? _faults[type] : "**Unknown**";
3196 _drbd_insert_fault(struct drbd_conf *mdev, unsigned int type)
3198 static struct fault_random_state rrs = {0, 0};
3200 unsigned int ret = (
3202 ((1 << mdev_to_minor(mdev)) & fault_devs) != 0) &&
3203 (((_drbd_fault_random(&rrs) % 100) + 1) <= fault_rate));
3208 if (__ratelimit(&drbd_ratelimit_state))
3209 dev_warn(DEV, "***Simulating %s failure\n",
3210 _drbd_fault_str(type));
3217 const char *drbd_buildtag(void)
3219 /* DRBD built from external sources has here a reference to the
3220 git hash of the source code. */
3222 static char buildtag[38] = "\0uilt-in";
3224 if (buildtag[0] == 0) {
3225 #ifdef CONFIG_MODULES
3226 if (THIS_MODULE != NULL)
3227 sprintf(buildtag, "srcversion: %-24s", THIS_MODULE->srcversion);
3236 module_init(drbd_init)
3237 module_exit(drbd_cleanup)
3239 EXPORT_SYMBOL(drbd_conn_str);
3240 EXPORT_SYMBOL(drbd_role_str);
3241 EXPORT_SYMBOL(drbd_disk_str);
3242 EXPORT_SYMBOL(drbd_set_st_err_str);