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 struct after_state_chg_work {
63 enum chg_state_flags flags;
64 struct completion *done;
67 static DEFINE_MUTEX(drbd_main_mutex);
68 int drbdd_init(struct drbd_thread *);
69 int drbd_worker(struct drbd_thread *);
70 int drbd_asender(struct drbd_thread *);
73 static int drbd_open(struct block_device *bdev, fmode_t mode);
74 static int drbd_release(struct gendisk *gd, fmode_t mode);
75 static int w_after_state_ch(struct drbd_conf *mdev, struct drbd_work *w, int unused);
76 static void after_state_ch(struct drbd_conf *mdev, union drbd_state os,
77 union drbd_state ns, enum chg_state_flags flags);
78 static int w_md_sync(struct drbd_conf *mdev, struct drbd_work *w, int unused);
79 static void md_sync_timer_fn(unsigned long data);
80 static int w_bitmap_io(struct drbd_conf *mdev, struct drbd_work *w, int unused);
81 static int w_go_diskless(struct drbd_conf *mdev, struct drbd_work *w, int unused);
83 MODULE_AUTHOR("Philipp Reisner <phil@linbit.com>, "
84 "Lars Ellenberg <lars@linbit.com>");
85 MODULE_DESCRIPTION("drbd - Distributed Replicated Block Device v" REL_VERSION);
86 MODULE_VERSION(REL_VERSION);
87 MODULE_LICENSE("GPL");
88 MODULE_PARM_DESC(minor_count, "Maximum number of drbd devices (1-255)");
89 MODULE_ALIAS_BLOCKDEV_MAJOR(DRBD_MAJOR);
91 #include <linux/moduleparam.h>
92 /* allow_open_on_secondary */
93 MODULE_PARM_DESC(allow_oos, "DONT USE!");
94 /* thanks to these macros, if compiled into the kernel (not-module),
95 * this becomes the boot parameter drbd.minor_count */
96 module_param(minor_count, uint, 0444);
97 module_param(disable_sendpage, bool, 0644);
98 module_param(allow_oos, bool, 0);
99 module_param(cn_idx, uint, 0444);
100 module_param(proc_details, int, 0644);
102 #ifdef CONFIG_DRBD_FAULT_INJECTION
105 static int fault_count;
107 /* bitmap of enabled faults */
108 module_param(enable_faults, int, 0664);
109 /* fault rate % value - applies to all enabled faults */
110 module_param(fault_rate, int, 0664);
111 /* count of faults inserted */
112 module_param(fault_count, int, 0664);
113 /* bitmap of devices to insert faults on */
114 module_param(fault_devs, int, 0644);
117 /* module parameter, defined */
118 unsigned int minor_count = 32;
119 int disable_sendpage;
121 unsigned int cn_idx = CN_IDX_DRBD;
122 int proc_details; /* Detail level in proc drbd*/
124 /* Module parameter for setting the user mode helper program
125 * to run. Default is /sbin/drbdadm */
126 char usermode_helper[80] = "/sbin/drbdadm";
128 module_param_string(usermode_helper, usermode_helper, sizeof(usermode_helper), 0644);
130 /* in 2.6.x, our device mapping and config info contains our virtual gendisks
131 * as member "struct gendisk *vdisk;"
133 struct drbd_conf **minor_table;
135 struct kmem_cache *drbd_request_cache;
136 struct kmem_cache *drbd_ee_cache; /* epoch entries */
137 struct kmem_cache *drbd_bm_ext_cache; /* bitmap extents */
138 struct kmem_cache *drbd_al_ext_cache; /* activity log extents */
139 mempool_t *drbd_request_mempool;
140 mempool_t *drbd_ee_mempool;
142 /* I do not use a standard mempool, because:
143 1) I want to hand out the pre-allocated objects first.
144 2) I want to be able to interrupt sleeping allocation with a signal.
145 Note: This is a single linked list, the next pointer is the private
146 member of struct page.
148 struct page *drbd_pp_pool;
149 spinlock_t drbd_pp_lock;
151 wait_queue_head_t drbd_pp_wait;
153 DEFINE_RATELIMIT_STATE(drbd_ratelimit_state, 5 * HZ, 5);
155 static const struct block_device_operations drbd_ops = {
156 .owner = THIS_MODULE,
158 .release = drbd_release,
161 #define ARRY_SIZE(A) (sizeof(A)/sizeof(A[0]))
164 /* When checking with sparse, and this is an inline function, sparse will
165 give tons of false positives. When this is a real functions sparse works.
167 int _get_ldev_if_state(struct drbd_conf *mdev, enum drbd_disk_state mins)
171 atomic_inc(&mdev->local_cnt);
172 io_allowed = (mdev->state.disk >= mins);
174 if (atomic_dec_and_test(&mdev->local_cnt))
175 wake_up(&mdev->misc_wait);
183 * DOC: The transfer log
185 * The transfer log is a single linked list of &struct drbd_tl_epoch objects.
186 * mdev->newest_tle points to the head, mdev->oldest_tle points to the tail
187 * of the list. There is always at least one &struct drbd_tl_epoch object.
189 * Each &struct drbd_tl_epoch has a circular double linked list of requests
192 static int tl_init(struct drbd_conf *mdev)
194 struct drbd_tl_epoch *b;
196 /* during device minor initialization, we may well use GFP_KERNEL */
197 b = kmalloc(sizeof(struct drbd_tl_epoch), GFP_KERNEL);
200 INIT_LIST_HEAD(&b->requests);
201 INIT_LIST_HEAD(&b->w.list);
205 b->w.cb = NULL; /* if this is != NULL, we need to dec_ap_pending in tl_clear */
207 mdev->oldest_tle = b;
208 mdev->newest_tle = b;
209 INIT_LIST_HEAD(&mdev->out_of_sequence_requests);
211 mdev->tl_hash = NULL;
217 static void tl_cleanup(struct drbd_conf *mdev)
219 D_ASSERT(mdev->oldest_tle == mdev->newest_tle);
220 D_ASSERT(list_empty(&mdev->out_of_sequence_requests));
221 kfree(mdev->oldest_tle);
222 mdev->oldest_tle = NULL;
223 kfree(mdev->unused_spare_tle);
224 mdev->unused_spare_tle = NULL;
225 kfree(mdev->tl_hash);
226 mdev->tl_hash = NULL;
231 * _tl_add_barrier() - Adds a barrier to the transfer log
232 * @mdev: DRBD device.
233 * @new: Barrier to be added before the current head of the TL.
235 * The caller must hold the req_lock.
237 void _tl_add_barrier(struct drbd_conf *mdev, struct drbd_tl_epoch *new)
239 struct drbd_tl_epoch *newest_before;
241 INIT_LIST_HEAD(&new->requests);
242 INIT_LIST_HEAD(&new->w.list);
243 new->w.cb = NULL; /* if this is != NULL, we need to dec_ap_pending in tl_clear */
247 newest_before = mdev->newest_tle;
248 /* never send a barrier number == 0, because that is special-cased
249 * when using TCQ for our write ordering code */
250 new->br_number = (newest_before->br_number+1) ?: 1;
251 if (mdev->newest_tle != new) {
252 mdev->newest_tle->next = new;
253 mdev->newest_tle = new;
258 * tl_release() - Free or recycle the oldest &struct drbd_tl_epoch object of the TL
259 * @mdev: DRBD device.
260 * @barrier_nr: Expected identifier of the DRBD write barrier packet.
261 * @set_size: Expected number of requests before that barrier.
263 * In case the passed barrier_nr or set_size does not match the oldest
264 * &struct drbd_tl_epoch objects this function will cause a termination
267 void tl_release(struct drbd_conf *mdev, unsigned int barrier_nr,
268 unsigned int set_size)
270 struct drbd_tl_epoch *b, *nob; /* next old barrier */
271 struct list_head *le, *tle;
272 struct drbd_request *r;
274 spin_lock_irq(&mdev->req_lock);
276 b = mdev->oldest_tle;
278 /* first some paranoia code */
280 dev_err(DEV, "BAD! BarrierAck #%u received, but no epoch in tl!?\n",
284 if (b->br_number != barrier_nr) {
285 dev_err(DEV, "BAD! BarrierAck #%u received, expected #%u!\n",
286 barrier_nr, b->br_number);
289 if (b->n_writes != set_size) {
290 dev_err(DEV, "BAD! BarrierAck #%u received with n_writes=%u, expected n_writes=%u!\n",
291 barrier_nr, set_size, b->n_writes);
295 /* Clean up list of requests processed during current epoch */
296 list_for_each_safe(le, tle, &b->requests) {
297 r = list_entry(le, struct drbd_request, tl_requests);
298 _req_mod(r, barrier_acked);
300 /* There could be requests on the list waiting for completion
301 of the write to the local disk. To avoid corruptions of
302 slab's data structures we have to remove the lists head.
304 Also there could have been a barrier ack out of sequence, overtaking
305 the write acks - which would be a bug and violating write ordering.
306 To not deadlock in case we lose connection while such requests are
307 still pending, we need some way to find them for the
308 _req_mode(connection_lost_while_pending).
310 These have been list_move'd to the out_of_sequence_requests list in
311 _req_mod(, barrier_acked) above.
313 list_del_init(&b->requests);
316 if (test_and_clear_bit(CREATE_BARRIER, &mdev->flags)) {
317 _tl_add_barrier(mdev, b);
319 mdev->oldest_tle = nob;
320 /* if nob == NULL b was the only barrier, and becomes the new
321 barrier. Therefore mdev->oldest_tle points already to b */
323 D_ASSERT(nob != NULL);
324 mdev->oldest_tle = nob;
328 spin_unlock_irq(&mdev->req_lock);
329 dec_ap_pending(mdev);
334 spin_unlock_irq(&mdev->req_lock);
335 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR));
339 * _tl_restart() - Walks the transfer log, and applies an action to all requests
340 * @mdev: DRBD device.
341 * @what: The action/event to perform with all request objects
343 * @what might be one of connection_lost_while_pending, resend, fail_frozen_disk_io,
344 * restart_frozen_disk_io.
346 static void _tl_restart(struct drbd_conf *mdev, enum drbd_req_event what)
348 struct drbd_tl_epoch *b, *tmp, **pn;
349 struct list_head *le, *tle, carry_reads;
350 struct drbd_request *req;
351 int rv, n_writes, n_reads;
353 b = mdev->oldest_tle;
354 pn = &mdev->oldest_tle;
358 INIT_LIST_HEAD(&carry_reads);
359 list_for_each_safe(le, tle, &b->requests) {
360 req = list_entry(le, struct drbd_request, tl_requests);
361 rv = _req_mod(req, what);
363 n_writes += (rv & MR_WRITE) >> MR_WRITE_SHIFT;
364 n_reads += (rv & MR_READ) >> MR_READ_SHIFT;
369 if (what == resend) {
370 b->n_writes = n_writes;
371 if (b->w.cb == NULL) {
372 b->w.cb = w_send_barrier;
373 inc_ap_pending(mdev);
374 set_bit(CREATE_BARRIER, &mdev->flags);
377 drbd_queue_work(&mdev->data.work, &b->w);
382 list_add(&carry_reads, &b->requests);
383 /* there could still be requests on that ring list,
384 * in case local io is still pending */
385 list_del(&b->requests);
387 /* dec_ap_pending corresponding to queue_barrier.
388 * the newest barrier may not have been queued yet,
389 * in which case w.cb is still NULL. */
391 dec_ap_pending(mdev);
393 if (b == mdev->newest_tle) {
394 /* recycle, but reinit! */
395 D_ASSERT(tmp == NULL);
396 INIT_LIST_HEAD(&b->requests);
397 list_splice(&carry_reads, &b->requests);
398 INIT_LIST_HEAD(&b->w.list);
400 b->br_number = net_random();
410 list_splice(&carry_reads, &b->requests);
416 * tl_clear() - Clears all requests and &struct drbd_tl_epoch objects out of the TL
417 * @mdev: DRBD device.
419 * This is called after the connection to the peer was lost. The storage covered
420 * by the requests on the transfer gets marked as our of sync. Called from the
421 * receiver thread and the worker thread.
423 void tl_clear(struct drbd_conf *mdev)
425 struct list_head *le, *tle;
426 struct drbd_request *r;
428 spin_lock_irq(&mdev->req_lock);
430 _tl_restart(mdev, connection_lost_while_pending);
432 /* we expect this list to be empty. */
433 D_ASSERT(list_empty(&mdev->out_of_sequence_requests));
435 /* but just in case, clean it up anyways! */
436 list_for_each_safe(le, tle, &mdev->out_of_sequence_requests) {
437 r = list_entry(le, struct drbd_request, tl_requests);
438 /* It would be nice to complete outside of spinlock.
439 * But this is easier for now. */
440 _req_mod(r, connection_lost_while_pending);
443 /* ensure bit indicating barrier is required is clear */
444 clear_bit(CREATE_BARRIER, &mdev->flags);
446 memset(mdev->app_reads_hash, 0, APP_R_HSIZE*sizeof(void *));
448 spin_unlock_irq(&mdev->req_lock);
451 void tl_restart(struct drbd_conf *mdev, enum drbd_req_event what)
453 spin_lock_irq(&mdev->req_lock);
454 _tl_restart(mdev, what);
455 spin_unlock_irq(&mdev->req_lock);
459 * cl_wide_st_chg() - TRUE if the state change is a cluster wide one
460 * @mdev: DRBD device.
461 * @os: old (current) state.
462 * @ns: new (wanted) state.
464 static int cl_wide_st_chg(struct drbd_conf *mdev,
465 union drbd_state os, union drbd_state ns)
467 return (os.conn >= C_CONNECTED && ns.conn >= C_CONNECTED &&
468 ((os.role != R_PRIMARY && ns.role == R_PRIMARY) ||
469 (os.conn != C_STARTING_SYNC_T && ns.conn == C_STARTING_SYNC_T) ||
470 (os.conn != C_STARTING_SYNC_S && ns.conn == C_STARTING_SYNC_S) ||
471 (os.disk != D_DISKLESS && ns.disk == D_DISKLESS))) ||
472 (os.conn >= C_CONNECTED && ns.conn == C_DISCONNECTING) ||
473 (os.conn == C_CONNECTED && ns.conn == C_VERIFY_S);
476 int drbd_change_state(struct drbd_conf *mdev, enum chg_state_flags f,
477 union drbd_state mask, union drbd_state val)
480 union drbd_state os, ns;
483 spin_lock_irqsave(&mdev->req_lock, flags);
485 ns.i = (os.i & ~mask.i) | val.i;
486 rv = _drbd_set_state(mdev, ns, f, NULL);
488 spin_unlock_irqrestore(&mdev->req_lock, flags);
494 * drbd_force_state() - Impose a change which happens outside our control on our state
495 * @mdev: DRBD device.
496 * @mask: mask of state bits to change.
497 * @val: value of new state bits.
499 void drbd_force_state(struct drbd_conf *mdev,
500 union drbd_state mask, union drbd_state val)
502 drbd_change_state(mdev, CS_HARD, mask, val);
505 static int is_valid_state(struct drbd_conf *mdev, union drbd_state ns);
506 static int is_valid_state_transition(struct drbd_conf *,
507 union drbd_state, union drbd_state);
508 static union drbd_state sanitize_state(struct drbd_conf *mdev, union drbd_state os,
509 union drbd_state ns, const char **warn_sync_abort);
510 int drbd_send_state_req(struct drbd_conf *,
511 union drbd_state, union drbd_state);
513 static enum drbd_state_ret_codes _req_st_cond(struct drbd_conf *mdev,
514 union drbd_state mask, union drbd_state val)
516 union drbd_state os, ns;
520 if (test_and_clear_bit(CL_ST_CHG_SUCCESS, &mdev->flags))
521 return SS_CW_SUCCESS;
523 if (test_and_clear_bit(CL_ST_CHG_FAIL, &mdev->flags))
524 return SS_CW_FAILED_BY_PEER;
527 spin_lock_irqsave(&mdev->req_lock, flags);
529 ns.i = (os.i & ~mask.i) | val.i;
530 ns = sanitize_state(mdev, os, ns, NULL);
532 if (!cl_wide_st_chg(mdev, os, ns))
535 rv = is_valid_state(mdev, ns);
536 if (rv == SS_SUCCESS) {
537 rv = is_valid_state_transition(mdev, ns, os);
538 if (rv == SS_SUCCESS)
539 rv = 0; /* cont waiting, otherwise fail. */
542 spin_unlock_irqrestore(&mdev->req_lock, flags);
548 * drbd_req_state() - Perform an eventually cluster wide state change
549 * @mdev: DRBD device.
550 * @mask: mask of state bits to change.
551 * @val: value of new state bits.
554 * Should not be called directly, use drbd_request_state() or
555 * _drbd_request_state().
557 static int drbd_req_state(struct drbd_conf *mdev,
558 union drbd_state mask, union drbd_state val,
559 enum chg_state_flags f)
561 struct completion done;
563 union drbd_state os, ns;
566 init_completion(&done);
568 if (f & CS_SERIALIZE)
569 mutex_lock(&mdev->state_mutex);
571 spin_lock_irqsave(&mdev->req_lock, flags);
573 ns.i = (os.i & ~mask.i) | val.i;
574 ns = sanitize_state(mdev, os, ns, NULL);
576 if (cl_wide_st_chg(mdev, os, ns)) {
577 rv = is_valid_state(mdev, ns);
578 if (rv == SS_SUCCESS)
579 rv = is_valid_state_transition(mdev, ns, os);
580 spin_unlock_irqrestore(&mdev->req_lock, flags);
582 if (rv < SS_SUCCESS) {
584 print_st_err(mdev, os, ns, rv);
588 drbd_state_lock(mdev);
589 if (!drbd_send_state_req(mdev, mask, val)) {
590 drbd_state_unlock(mdev);
591 rv = SS_CW_FAILED_BY_PEER;
593 print_st_err(mdev, os, ns, rv);
597 wait_event(mdev->state_wait,
598 (rv = _req_st_cond(mdev, mask, val)));
600 if (rv < SS_SUCCESS) {
601 drbd_state_unlock(mdev);
603 print_st_err(mdev, os, ns, rv);
606 spin_lock_irqsave(&mdev->req_lock, flags);
608 ns.i = (os.i & ~mask.i) | val.i;
609 rv = _drbd_set_state(mdev, ns, f, &done);
610 drbd_state_unlock(mdev);
612 rv = _drbd_set_state(mdev, ns, f, &done);
615 spin_unlock_irqrestore(&mdev->req_lock, flags);
617 if (f & CS_WAIT_COMPLETE && rv == SS_SUCCESS) {
618 D_ASSERT(current != mdev->worker.task);
619 wait_for_completion(&done);
623 if (f & CS_SERIALIZE)
624 mutex_unlock(&mdev->state_mutex);
630 * _drbd_request_state() - Request a state change (with flags)
631 * @mdev: DRBD device.
632 * @mask: mask of state bits to change.
633 * @val: value of new state bits.
636 * Cousin of drbd_request_state(), useful with the CS_WAIT_COMPLETE
637 * flag, or when logging of failed state change requests is not desired.
639 int _drbd_request_state(struct drbd_conf *mdev, union drbd_state mask,
640 union drbd_state val, enum chg_state_flags f)
644 wait_event(mdev->state_wait,
645 (rv = drbd_req_state(mdev, mask, val, f)) != SS_IN_TRANSIENT_STATE);
650 static void print_st(struct drbd_conf *mdev, char *name, union drbd_state ns)
652 dev_err(DEV, " %s = { cs:%s ro:%s/%s ds:%s/%s %c%c%c%c }\n",
654 drbd_conn_str(ns.conn),
655 drbd_role_str(ns.role),
656 drbd_role_str(ns.peer),
657 drbd_disk_str(ns.disk),
658 drbd_disk_str(ns.pdsk),
659 is_susp(ns) ? 's' : 'r',
660 ns.aftr_isp ? 'a' : '-',
661 ns.peer_isp ? 'p' : '-',
662 ns.user_isp ? 'u' : '-'
666 void print_st_err(struct drbd_conf *mdev,
667 union drbd_state os, union drbd_state ns, int err)
669 if (err == SS_IN_TRANSIENT_STATE)
671 dev_err(DEV, "State change failed: %s\n", drbd_set_st_err_str(err));
672 print_st(mdev, " state", os);
673 print_st(mdev, "wanted", ns);
677 #define drbd_peer_str drbd_role_str
678 #define drbd_pdsk_str drbd_disk_str
680 #define drbd_susp_str(A) ((A) ? "1" : "0")
681 #define drbd_aftr_isp_str(A) ((A) ? "1" : "0")
682 #define drbd_peer_isp_str(A) ((A) ? "1" : "0")
683 #define drbd_user_isp_str(A) ((A) ? "1" : "0")
686 ({ if (ns.A != os.A) { \
687 pbp += sprintf(pbp, #A "( %s -> %s ) ", \
688 drbd_##A##_str(os.A), \
689 drbd_##A##_str(ns.A)); \
693 * is_valid_state() - Returns an SS_ error code if ns is not valid
694 * @mdev: DRBD device.
695 * @ns: State to consider.
697 static int is_valid_state(struct drbd_conf *mdev, union drbd_state ns)
699 /* See drbd_state_sw_errors in drbd_strings.c */
701 enum drbd_fencing_p fp;
705 if (get_ldev(mdev)) {
706 fp = mdev->ldev->dc.fencing;
710 if (get_net_conf(mdev)) {
711 if (!mdev->net_conf->two_primaries &&
712 ns.role == R_PRIMARY && ns.peer == R_PRIMARY)
713 rv = SS_TWO_PRIMARIES;
718 /* already found a reason to abort */;
719 else if (ns.role == R_SECONDARY && mdev->open_cnt)
720 rv = SS_DEVICE_IN_USE;
722 else if (ns.role == R_PRIMARY && ns.conn < C_CONNECTED && ns.disk < D_UP_TO_DATE)
723 rv = SS_NO_UP_TO_DATE_DISK;
725 else if (fp >= FP_RESOURCE &&
726 ns.role == R_PRIMARY && ns.conn < C_CONNECTED && ns.pdsk >= D_UNKNOWN)
729 else if (ns.role == R_PRIMARY && ns.disk <= D_INCONSISTENT && ns.pdsk <= D_INCONSISTENT)
730 rv = SS_NO_UP_TO_DATE_DISK;
732 else if (ns.conn > C_CONNECTED && ns.disk < D_INCONSISTENT)
733 rv = SS_NO_LOCAL_DISK;
735 else if (ns.conn > C_CONNECTED && ns.pdsk < D_INCONSISTENT)
736 rv = SS_NO_REMOTE_DISK;
738 else if (ns.conn > C_CONNECTED && ns.disk < D_UP_TO_DATE && ns.pdsk < D_UP_TO_DATE)
739 rv = SS_NO_UP_TO_DATE_DISK;
741 else if ((ns.conn == C_CONNECTED ||
742 ns.conn == C_WF_BITMAP_S ||
743 ns.conn == C_SYNC_SOURCE ||
744 ns.conn == C_PAUSED_SYNC_S) &&
745 ns.disk == D_OUTDATED)
746 rv = SS_CONNECTED_OUTDATES;
748 else if ((ns.conn == C_VERIFY_S || ns.conn == C_VERIFY_T) &&
749 (mdev->sync_conf.verify_alg[0] == 0))
750 rv = SS_NO_VERIFY_ALG;
752 else if ((ns.conn == C_VERIFY_S || ns.conn == C_VERIFY_T) &&
753 mdev->agreed_pro_version < 88)
754 rv = SS_NOT_SUPPORTED;
760 * is_valid_state_transition() - Returns an SS_ error code if the state transition is not possible
761 * @mdev: DRBD device.
765 static int is_valid_state_transition(struct drbd_conf *mdev,
766 union drbd_state ns, union drbd_state os)
770 if ((ns.conn == C_STARTING_SYNC_T || ns.conn == C_STARTING_SYNC_S) &&
771 os.conn > C_CONNECTED)
772 rv = SS_RESYNC_RUNNING;
774 if (ns.conn == C_DISCONNECTING && os.conn == C_STANDALONE)
775 rv = SS_ALREADY_STANDALONE;
777 if (ns.disk > D_ATTACHING && os.disk == D_DISKLESS)
780 if (ns.conn == C_WF_CONNECTION && os.conn < C_UNCONNECTED)
781 rv = SS_NO_NET_CONFIG;
783 if (ns.disk == D_OUTDATED && os.disk < D_OUTDATED && os.disk != D_ATTACHING)
784 rv = SS_LOWER_THAN_OUTDATED;
786 if (ns.conn == C_DISCONNECTING && os.conn == C_UNCONNECTED)
787 rv = SS_IN_TRANSIENT_STATE;
789 if (ns.conn == os.conn && ns.conn == C_WF_REPORT_PARAMS)
790 rv = SS_IN_TRANSIENT_STATE;
792 if ((ns.conn == C_VERIFY_S || ns.conn == C_VERIFY_T) && os.conn < C_CONNECTED)
793 rv = SS_NEED_CONNECTION;
795 if ((ns.conn == C_VERIFY_S || ns.conn == C_VERIFY_T) &&
796 ns.conn != os.conn && os.conn > C_CONNECTED)
797 rv = SS_RESYNC_RUNNING;
799 if ((ns.conn == C_STARTING_SYNC_S || ns.conn == C_STARTING_SYNC_T) &&
800 os.conn < C_CONNECTED)
801 rv = SS_NEED_CONNECTION;
807 * sanitize_state() - Resolves implicitly necessary additional changes to a state transition
808 * @mdev: DRBD device.
813 * When we loose connection, we have to set the state of the peers disk (pdsk)
814 * to D_UNKNOWN. This rule and many more along those lines are in this function.
816 static union drbd_state sanitize_state(struct drbd_conf *mdev, union drbd_state os,
817 union drbd_state ns, const char **warn_sync_abort)
819 enum drbd_fencing_p fp;
822 if (get_ldev(mdev)) {
823 fp = mdev->ldev->dc.fencing;
827 /* Disallow Network errors to configure a device's network part */
828 if ((ns.conn >= C_TIMEOUT && ns.conn <= C_TEAR_DOWN) &&
829 os.conn <= C_DISCONNECTING)
832 /* After a network error (+C_TEAR_DOWN) only C_UNCONNECTED or C_DISCONNECTING can follow.
833 * If you try to go into some Sync* state, that shall fail (elsewhere). */
834 if (os.conn >= C_TIMEOUT && os.conn <= C_TEAR_DOWN &&
835 ns.conn != C_UNCONNECTED && ns.conn != C_DISCONNECTING && ns.conn <= C_TEAR_DOWN)
838 /* we cannot fail (again) if we already detached */
839 if (ns.disk == D_FAILED && os.disk == D_DISKLESS)
840 ns.disk = D_DISKLESS;
842 /* if we are only D_ATTACHING yet,
843 * we can (and should) go directly to D_DISKLESS. */
844 if (ns.disk == D_FAILED && os.disk == D_ATTACHING)
845 ns.disk = D_DISKLESS;
847 /* After C_DISCONNECTING only C_STANDALONE may follow */
848 if (os.conn == C_DISCONNECTING && ns.conn != C_STANDALONE)
851 if (ns.conn < C_CONNECTED) {
854 if (ns.pdsk > D_UNKNOWN || ns.pdsk < D_INCONSISTENT)
858 /* Clear the aftr_isp when becoming unconfigured */
859 if (ns.conn == C_STANDALONE && ns.disk == D_DISKLESS && ns.role == R_SECONDARY)
862 /* Abort resync if a disk fails/detaches */
863 if (os.conn > C_CONNECTED && ns.conn > C_CONNECTED &&
864 (ns.disk <= D_FAILED || ns.pdsk <= D_FAILED)) {
867 os.conn == C_VERIFY_S || os.conn == C_VERIFY_T ?
868 "Online-verify" : "Resync";
869 ns.conn = C_CONNECTED;
872 if (ns.conn >= C_CONNECTED &&
873 ((ns.disk == D_CONSISTENT || ns.disk == D_OUTDATED) ||
874 (ns.disk == D_NEGOTIATING && ns.conn == C_WF_BITMAP_T) ||
875 ns.conn >= C_AHEAD)) {
878 case C_PAUSED_SYNC_T:
880 ns.disk = D_OUTDATED;
885 case C_PAUSED_SYNC_S:
887 ns.disk = D_UP_TO_DATE;
890 ns.disk = D_INCONSISTENT;
891 dev_warn(DEV, "Implicitly set disk state Inconsistent!\n");
894 if (os.disk == D_OUTDATED && ns.disk == D_UP_TO_DATE)
895 dev_warn(DEV, "Implicitly set disk from Outdated to UpToDate\n");
898 if (ns.conn >= C_CONNECTED &&
899 (ns.pdsk == D_CONSISTENT || ns.pdsk == D_OUTDATED || ns.conn >= C_AHEAD)) {
903 case C_PAUSED_SYNC_T:
906 ns.pdsk = D_UP_TO_DATE;
909 case C_PAUSED_SYNC_S:
911 /* remap any consistent state to D_OUTDATED,
912 * but disallow "upgrade" of not even consistent states.
915 (D_DISKLESS < os.pdsk && os.pdsk < D_OUTDATED)
916 ? os.pdsk : D_OUTDATED;
919 ns.pdsk = D_INCONSISTENT;
920 dev_warn(DEV, "Implicitly set pdsk Inconsistent!\n");
923 if (os.pdsk == D_OUTDATED && ns.pdsk == D_UP_TO_DATE)
924 dev_warn(DEV, "Implicitly set pdsk from Outdated to UpToDate\n");
927 /* Connection breaks down before we finished "Negotiating" */
928 if (ns.conn < C_CONNECTED && ns.disk == D_NEGOTIATING &&
929 get_ldev_if_state(mdev, D_NEGOTIATING)) {
930 if (mdev->ed_uuid == mdev->ldev->md.uuid[UI_CURRENT]) {
931 ns.disk = mdev->new_state_tmp.disk;
932 ns.pdsk = mdev->new_state_tmp.pdsk;
934 dev_alert(DEV, "Connection lost while negotiating, no data!\n");
935 ns.disk = D_DISKLESS;
941 if (fp == FP_STONITH &&
942 (ns.role == R_PRIMARY && ns.conn < C_CONNECTED && ns.pdsk > D_OUTDATED) &&
943 !(os.role == R_PRIMARY && os.conn < C_CONNECTED && os.pdsk > D_OUTDATED))
944 ns.susp_fen = 1; /* Suspend IO while fence-peer handler runs (peer lost) */
946 if (mdev->sync_conf.on_no_data == OND_SUSPEND_IO &&
947 (ns.role == R_PRIMARY && ns.disk < D_UP_TO_DATE && ns.pdsk < D_UP_TO_DATE) &&
948 !(os.role == R_PRIMARY && os.disk < D_UP_TO_DATE && os.pdsk < D_UP_TO_DATE))
949 ns.susp_nod = 1; /* Suspend IO while no data available (no accessible data available) */
951 if (ns.aftr_isp || ns.peer_isp || ns.user_isp) {
952 if (ns.conn == C_SYNC_SOURCE)
953 ns.conn = C_PAUSED_SYNC_S;
954 if (ns.conn == C_SYNC_TARGET)
955 ns.conn = C_PAUSED_SYNC_T;
957 if (ns.conn == C_PAUSED_SYNC_S)
958 ns.conn = C_SYNC_SOURCE;
959 if (ns.conn == C_PAUSED_SYNC_T)
960 ns.conn = C_SYNC_TARGET;
966 /* helper for __drbd_set_state */
967 static void set_ov_position(struct drbd_conf *mdev, enum drbd_conns cs)
969 if (mdev->agreed_pro_version < 90)
970 mdev->ov_start_sector = 0;
971 mdev->rs_total = drbd_bm_bits(mdev);
972 mdev->ov_position = 0;
973 if (cs == C_VERIFY_T) {
974 /* starting online verify from an arbitrary position
975 * does not fit well into the existing protocol.
976 * on C_VERIFY_T, we initialize ov_left and friends
977 * implicitly in receive_DataRequest once the
978 * first P_OV_REQUEST is received */
979 mdev->ov_start_sector = ~(sector_t)0;
981 unsigned long bit = BM_SECT_TO_BIT(mdev->ov_start_sector);
982 if (bit >= mdev->rs_total) {
983 mdev->ov_start_sector =
984 BM_BIT_TO_SECT(mdev->rs_total - 1);
987 mdev->rs_total -= bit;
988 mdev->ov_position = mdev->ov_start_sector;
990 mdev->ov_left = mdev->rs_total;
993 static void drbd_resume_al(struct drbd_conf *mdev)
995 if (test_and_clear_bit(AL_SUSPENDED, &mdev->flags))
996 dev_info(DEV, "Resumed AL updates\n");
1000 * __drbd_set_state() - Set a new DRBD state
1001 * @mdev: DRBD device.
1004 * @done: Optional completion, that will get completed after the after_state_ch() finished
1006 * Caller needs to hold req_lock, and global_state_lock. Do not call directly.
1008 int __drbd_set_state(struct drbd_conf *mdev,
1009 union drbd_state ns, enum chg_state_flags flags,
1010 struct completion *done)
1012 union drbd_state os;
1013 int rv = SS_SUCCESS;
1014 const char *warn_sync_abort = NULL;
1015 struct after_state_chg_work *ascw;
1019 ns = sanitize_state(mdev, os, ns, &warn_sync_abort);
1022 return SS_NOTHING_TO_DO;
1024 if (!(flags & CS_HARD)) {
1025 /* pre-state-change checks ; only look at ns */
1026 /* See drbd_state_sw_errors in drbd_strings.c */
1028 rv = is_valid_state(mdev, ns);
1029 if (rv < SS_SUCCESS) {
1030 /* If the old state was illegal as well, then let
1033 if (is_valid_state(mdev, os) == rv)
1034 rv = is_valid_state_transition(mdev, ns, os);
1036 rv = is_valid_state_transition(mdev, ns, os);
1039 if (rv < SS_SUCCESS) {
1040 if (flags & CS_VERBOSE)
1041 print_st_err(mdev, os, ns, rv);
1045 if (warn_sync_abort)
1046 dev_warn(DEV, "%s aborted.\n", warn_sync_abort);
1057 if (is_susp(ns) != is_susp(os))
1058 pbp += sprintf(pbp, "susp( %s -> %s ) ",
1059 drbd_susp_str(is_susp(os)),
1060 drbd_susp_str(is_susp(ns)));
1064 dev_info(DEV, "%s\n", pb);
1067 /* solve the race between becoming unconfigured,
1068 * worker doing the cleanup, and
1069 * admin reconfiguring us:
1070 * on (re)configure, first set CONFIG_PENDING,
1071 * then wait for a potentially exiting worker,
1072 * start the worker, and schedule one no_op.
1073 * then proceed with configuration.
1075 if (ns.disk == D_DISKLESS &&
1076 ns.conn == C_STANDALONE &&
1077 ns.role == R_SECONDARY &&
1078 !test_and_set_bit(CONFIG_PENDING, &mdev->flags))
1079 set_bit(DEVICE_DYING, &mdev->flags);
1081 /* if we are going -> D_FAILED or D_DISKLESS, grab one extra reference
1082 * on the ldev here, to be sure the transition -> D_DISKLESS resp.
1083 * drbd_ldev_destroy() won't happen before our corresponding
1084 * after_state_ch works run, where we put_ldev again. */
1085 if ((os.disk != D_FAILED && ns.disk == D_FAILED) ||
1086 (os.disk != D_DISKLESS && ns.disk == D_DISKLESS))
1087 atomic_inc(&mdev->local_cnt);
1090 wake_up(&mdev->misc_wait);
1091 wake_up(&mdev->state_wait);
1093 /* aborted verify run. log the last position */
1094 if ((os.conn == C_VERIFY_S || os.conn == C_VERIFY_T) &&
1095 ns.conn < C_CONNECTED) {
1096 mdev->ov_start_sector =
1097 BM_BIT_TO_SECT(drbd_bm_bits(mdev) - mdev->ov_left);
1098 dev_info(DEV, "Online Verify reached sector %llu\n",
1099 (unsigned long long)mdev->ov_start_sector);
1102 if ((os.conn == C_PAUSED_SYNC_T || os.conn == C_PAUSED_SYNC_S) &&
1103 (ns.conn == C_SYNC_TARGET || ns.conn == C_SYNC_SOURCE)) {
1104 dev_info(DEV, "Syncer continues.\n");
1105 mdev->rs_paused += (long)jiffies
1106 -(long)mdev->rs_mark_time[mdev->rs_last_mark];
1107 if (ns.conn == C_SYNC_TARGET)
1108 mod_timer(&mdev->resync_timer, jiffies);
1111 if ((os.conn == C_SYNC_TARGET || os.conn == C_SYNC_SOURCE) &&
1112 (ns.conn == C_PAUSED_SYNC_T || ns.conn == C_PAUSED_SYNC_S)) {
1113 dev_info(DEV, "Resync suspended\n");
1114 mdev->rs_mark_time[mdev->rs_last_mark] = jiffies;
1117 if (os.conn == C_CONNECTED &&
1118 (ns.conn == C_VERIFY_S || ns.conn == C_VERIFY_T)) {
1119 unsigned long now = jiffies;
1122 set_ov_position(mdev, ns.conn);
1123 mdev->rs_start = now;
1124 mdev->rs_last_events = 0;
1125 mdev->rs_last_sect_ev = 0;
1126 mdev->ov_last_oos_size = 0;
1127 mdev->ov_last_oos_start = 0;
1129 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
1130 mdev->rs_mark_left[i] = mdev->ov_left;
1131 mdev->rs_mark_time[i] = now;
1134 drbd_rs_controller_reset(mdev);
1136 if (ns.conn == C_VERIFY_S) {
1137 dev_info(DEV, "Starting Online Verify from sector %llu\n",
1138 (unsigned long long)mdev->ov_position);
1139 mod_timer(&mdev->resync_timer, jiffies);
1143 if (get_ldev(mdev)) {
1144 u32 mdf = mdev->ldev->md.flags & ~(MDF_CONSISTENT|MDF_PRIMARY_IND|
1145 MDF_CONNECTED_IND|MDF_WAS_UP_TO_DATE|
1146 MDF_PEER_OUT_DATED|MDF_CRASHED_PRIMARY);
1148 if (test_bit(CRASHED_PRIMARY, &mdev->flags))
1149 mdf |= MDF_CRASHED_PRIMARY;
1150 if (mdev->state.role == R_PRIMARY ||
1151 (mdev->state.pdsk < D_INCONSISTENT && mdev->state.peer == R_PRIMARY))
1152 mdf |= MDF_PRIMARY_IND;
1153 if (mdev->state.conn > C_WF_REPORT_PARAMS)
1154 mdf |= MDF_CONNECTED_IND;
1155 if (mdev->state.disk > D_INCONSISTENT)
1156 mdf |= MDF_CONSISTENT;
1157 if (mdev->state.disk > D_OUTDATED)
1158 mdf |= MDF_WAS_UP_TO_DATE;
1159 if (mdev->state.pdsk <= D_OUTDATED && mdev->state.pdsk >= D_INCONSISTENT)
1160 mdf |= MDF_PEER_OUT_DATED;
1161 if (mdf != mdev->ldev->md.flags) {
1162 mdev->ldev->md.flags = mdf;
1163 drbd_md_mark_dirty(mdev);
1165 if (os.disk < D_CONSISTENT && ns.disk >= D_CONSISTENT)
1166 drbd_set_ed_uuid(mdev, mdev->ldev->md.uuid[UI_CURRENT]);
1170 /* Peer was forced D_UP_TO_DATE & R_PRIMARY, consider to resync */
1171 if (os.disk == D_INCONSISTENT && os.pdsk == D_INCONSISTENT &&
1172 os.peer == R_SECONDARY && ns.peer == R_PRIMARY)
1173 set_bit(CONSIDER_RESYNC, &mdev->flags);
1175 /* Receiver should clean up itself */
1176 if (os.conn != C_DISCONNECTING && ns.conn == C_DISCONNECTING)
1177 drbd_thread_stop_nowait(&mdev->receiver);
1179 /* Now the receiver finished cleaning up itself, it should die */
1180 if (os.conn != C_STANDALONE && ns.conn == C_STANDALONE)
1181 drbd_thread_stop_nowait(&mdev->receiver);
1183 /* Upon network failure, we need to restart the receiver. */
1184 if (os.conn > C_TEAR_DOWN &&
1185 ns.conn <= C_TEAR_DOWN && ns.conn >= C_TIMEOUT)
1186 drbd_thread_restart_nowait(&mdev->receiver);
1188 /* Resume AL writing if we get a connection */
1189 if (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED)
1190 drbd_resume_al(mdev);
1192 ascw = kmalloc(sizeof(*ascw), GFP_ATOMIC);
1196 ascw->flags = flags;
1197 ascw->w.cb = w_after_state_ch;
1199 drbd_queue_work(&mdev->data.work, &ascw->w);
1201 dev_warn(DEV, "Could not kmalloc an ascw\n");
1207 static int w_after_state_ch(struct drbd_conf *mdev, struct drbd_work *w, int unused)
1209 struct after_state_chg_work *ascw =
1210 container_of(w, struct after_state_chg_work, w);
1211 after_state_ch(mdev, ascw->os, ascw->ns, ascw->flags);
1212 if (ascw->flags & CS_WAIT_COMPLETE) {
1213 D_ASSERT(ascw->done != NULL);
1214 complete(ascw->done);
1221 static void abw_start_sync(struct drbd_conf *mdev, int rv)
1224 dev_err(DEV, "Writing the bitmap failed not starting resync.\n");
1225 _drbd_request_state(mdev, NS(conn, C_CONNECTED), CS_VERBOSE);
1229 switch (mdev->state.conn) {
1230 case C_STARTING_SYNC_T:
1231 _drbd_request_state(mdev, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE);
1233 case C_STARTING_SYNC_S:
1234 drbd_start_resync(mdev, C_SYNC_SOURCE);
1240 * after_state_ch() - Perform after state change actions that may sleep
1241 * @mdev: DRBD device.
1246 static void after_state_ch(struct drbd_conf *mdev, union drbd_state os,
1247 union drbd_state ns, enum chg_state_flags flags)
1249 enum drbd_fencing_p fp;
1250 enum drbd_req_event what = nothing;
1251 union drbd_state nsm = (union drbd_state){ .i = -1 };
1253 if (os.conn != C_CONNECTED && ns.conn == C_CONNECTED) {
1254 clear_bit(CRASHED_PRIMARY, &mdev->flags);
1256 mdev->p_uuid[UI_FLAGS] &= ~((u64)2);
1260 if (get_ldev(mdev)) {
1261 fp = mdev->ldev->dc.fencing;
1265 /* Inform userspace about the change... */
1266 drbd_bcast_state(mdev, ns);
1268 if (!(os.role == R_PRIMARY && os.disk < D_UP_TO_DATE && os.pdsk < D_UP_TO_DATE) &&
1269 (ns.role == R_PRIMARY && ns.disk < D_UP_TO_DATE && ns.pdsk < D_UP_TO_DATE))
1270 drbd_khelper(mdev, "pri-on-incon-degr");
1272 /* Here we have the actions that are performed after a
1273 state change. This function might sleep */
1277 if (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED) {
1278 if (ns.conn == C_CONNECTED)
1279 what = resend, nsm.susp_nod = 0;
1280 else /* ns.conn > C_CONNECTED */
1281 dev_err(DEV, "Unexpected Resynd going on!\n");
1284 if (os.disk == D_ATTACHING && ns.disk > D_ATTACHING)
1285 what = restart_frozen_disk_io, nsm.susp_nod = 0;
1290 /* case1: The outdate peer handler is successful: */
1291 if (os.pdsk > D_OUTDATED && ns.pdsk <= D_OUTDATED) {
1293 if (test_bit(NEW_CUR_UUID, &mdev->flags)) {
1294 drbd_uuid_new_current(mdev);
1295 clear_bit(NEW_CUR_UUID, &mdev->flags);
1297 spin_lock_irq(&mdev->req_lock);
1298 _drbd_set_state(_NS(mdev, susp_fen, 0), CS_VERBOSE, NULL);
1299 spin_unlock_irq(&mdev->req_lock);
1301 /* case2: The connection was established again: */
1302 if (os.conn < C_CONNECTED && ns.conn >= C_CONNECTED) {
1303 clear_bit(NEW_CUR_UUID, &mdev->flags);
1309 if (what != nothing) {
1310 spin_lock_irq(&mdev->req_lock);
1311 _tl_restart(mdev, what);
1312 nsm.i &= mdev->state.i;
1313 _drbd_set_state(mdev, nsm, CS_VERBOSE, NULL);
1314 spin_unlock_irq(&mdev->req_lock);
1317 /* Do not change the order of the if above and the two below... */
1318 if (os.pdsk == D_DISKLESS && ns.pdsk > D_DISKLESS) { /* attach on the peer */
1319 drbd_send_uuids(mdev);
1320 drbd_send_state(mdev);
1322 if (os.conn != C_WF_BITMAP_S && ns.conn == C_WF_BITMAP_S)
1323 drbd_queue_bitmap_io(mdev, &drbd_send_bitmap, NULL, "send_bitmap (WFBitMapS)");
1325 /* Lost contact to peer's copy of the data */
1326 if ((os.pdsk >= D_INCONSISTENT &&
1327 os.pdsk != D_UNKNOWN &&
1328 os.pdsk != D_OUTDATED)
1329 && (ns.pdsk < D_INCONSISTENT ||
1330 ns.pdsk == D_UNKNOWN ||
1331 ns.pdsk == D_OUTDATED)) {
1332 if (get_ldev(mdev)) {
1333 if ((ns.role == R_PRIMARY || ns.peer == R_PRIMARY) &&
1334 mdev->ldev->md.uuid[UI_BITMAP] == 0 && ns.disk >= D_UP_TO_DATE) {
1335 if (is_susp(mdev->state)) {
1336 set_bit(NEW_CUR_UUID, &mdev->flags);
1338 drbd_uuid_new_current(mdev);
1339 drbd_send_uuids(mdev);
1346 if (ns.pdsk < D_INCONSISTENT && get_ldev(mdev)) {
1347 if (ns.peer == R_PRIMARY && mdev->ldev->md.uuid[UI_BITMAP] == 0) {
1348 drbd_uuid_new_current(mdev);
1349 drbd_send_uuids(mdev);
1352 /* D_DISKLESS Peer becomes secondary */
1353 if (os.peer == R_PRIMARY && ns.peer == R_SECONDARY)
1354 drbd_al_to_on_disk_bm(mdev);
1358 /* Last part of the attaching process ... */
1359 if (ns.conn >= C_CONNECTED &&
1360 os.disk == D_ATTACHING && ns.disk == D_NEGOTIATING) {
1361 drbd_send_sizes(mdev, 0, 0); /* to start sync... */
1362 drbd_send_uuids(mdev);
1363 drbd_send_state(mdev);
1366 /* We want to pause/continue resync, tell peer. */
1367 if (ns.conn >= C_CONNECTED &&
1368 ((os.aftr_isp != ns.aftr_isp) ||
1369 (os.user_isp != ns.user_isp)))
1370 drbd_send_state(mdev);
1372 /* In case one of the isp bits got set, suspend other devices. */
1373 if ((!os.aftr_isp && !os.peer_isp && !os.user_isp) &&
1374 (ns.aftr_isp || ns.peer_isp || ns.user_isp))
1375 suspend_other_sg(mdev);
1377 /* Make sure the peer gets informed about eventual state
1378 changes (ISP bits) while we were in WFReportParams. */
1379 if (os.conn == C_WF_REPORT_PARAMS && ns.conn >= C_CONNECTED)
1380 drbd_send_state(mdev);
1382 if (os.conn != C_AHEAD && ns.conn == C_AHEAD)
1383 drbd_send_state(mdev);
1385 /* We are in the progress to start a full sync... */
1386 if ((os.conn != C_STARTING_SYNC_T && ns.conn == C_STARTING_SYNC_T) ||
1387 (os.conn != C_STARTING_SYNC_S && ns.conn == C_STARTING_SYNC_S))
1388 drbd_queue_bitmap_io(mdev, &drbd_bmio_set_n_write, &abw_start_sync, "set_n_write from StartingSync");
1390 /* We are invalidating our self... */
1391 if (os.conn < C_CONNECTED && ns.conn < C_CONNECTED &&
1392 os.disk > D_INCONSISTENT && ns.disk == D_INCONSISTENT)
1393 drbd_queue_bitmap_io(mdev, &drbd_bmio_set_n_write, NULL, "set_n_write from invalidate");
1395 /* first half of local IO error, failure to attach,
1396 * or administrative detach */
1397 if (os.disk != D_FAILED && ns.disk == D_FAILED) {
1398 enum drbd_io_error_p eh;
1400 /* corresponding get_ldev was in __drbd_set_state, to serialize
1401 * our cleanup here with the transition to D_DISKLESS,
1402 * so it is safe to dreference ldev here. */
1403 eh = mdev->ldev->dc.on_io_error;
1404 was_io_error = test_and_clear_bit(WAS_IO_ERROR, &mdev->flags);
1406 /* current state still has to be D_FAILED,
1407 * there is only one way out: to D_DISKLESS,
1408 * and that may only happen after our put_ldev below. */
1409 if (mdev->state.disk != D_FAILED)
1411 "ASSERT FAILED: disk is %s during detach\n",
1412 drbd_disk_str(mdev->state.disk));
1414 if (drbd_send_state(mdev))
1415 dev_warn(DEV, "Notified peer that I am detaching my disk\n");
1417 dev_err(DEV, "Sending state for detaching disk failed\n");
1419 drbd_rs_cancel_all(mdev);
1421 /* In case we want to get something to stable storage still,
1422 * this may be the last chance.
1423 * Following put_ldev may transition to D_DISKLESS. */
1427 if (was_io_error && eh == EP_CALL_HELPER)
1428 drbd_khelper(mdev, "local-io-error");
1431 /* second half of local IO error, failure to attach,
1432 * or administrative detach,
1433 * after local_cnt references have reached zero again */
1434 if (os.disk != D_DISKLESS && ns.disk == D_DISKLESS) {
1435 /* We must still be diskless,
1436 * re-attach has to be serialized with this! */
1437 if (mdev->state.disk != D_DISKLESS)
1439 "ASSERT FAILED: disk is %s while going diskless\n",
1440 drbd_disk_str(mdev->state.disk));
1443 mdev->rs_failed = 0;
1444 atomic_set(&mdev->rs_pending_cnt, 0);
1446 if (drbd_send_state(mdev))
1447 dev_warn(DEV, "Notified peer that I'm now diskless.\n");
1449 dev_err(DEV, "Sending state for being diskless failed\n");
1450 /* corresponding get_ldev in __drbd_set_state
1451 * this may finaly trigger drbd_ldev_destroy. */
1455 /* Disks got bigger while they were detached */
1456 if (ns.disk > D_NEGOTIATING && ns.pdsk > D_NEGOTIATING &&
1457 test_and_clear_bit(RESYNC_AFTER_NEG, &mdev->flags)) {
1458 if (ns.conn == C_CONNECTED)
1459 resync_after_online_grow(mdev);
1462 /* A resync finished or aborted, wake paused devices... */
1463 if ((os.conn > C_CONNECTED && ns.conn <= C_CONNECTED) ||
1464 (os.peer_isp && !ns.peer_isp) ||
1465 (os.user_isp && !ns.user_isp))
1466 resume_next_sg(mdev);
1468 /* sync target done with resync. Explicitly notify peer, even though
1469 * it should (at least for non-empty resyncs) already know itself. */
1470 if (os.disk < D_UP_TO_DATE && os.conn >= C_SYNC_SOURCE && ns.conn == C_CONNECTED)
1471 drbd_send_state(mdev);
1473 /* free tl_hash if we Got thawed and are C_STANDALONE */
1474 if (ns.conn == C_STANDALONE && !is_susp(ns) && mdev->tl_hash)
1475 drbd_free_tl_hash(mdev);
1477 /* Upon network connection, we need to start the receiver */
1478 if (os.conn == C_STANDALONE && ns.conn == C_UNCONNECTED)
1479 drbd_thread_start(&mdev->receiver);
1481 /* Terminate worker thread if we are unconfigured - it will be
1482 restarted as needed... */
1483 if (ns.disk == D_DISKLESS &&
1484 ns.conn == C_STANDALONE &&
1485 ns.role == R_SECONDARY) {
1486 if (os.aftr_isp != ns.aftr_isp)
1487 resume_next_sg(mdev);
1488 /* set in __drbd_set_state, unless CONFIG_PENDING was set */
1489 if (test_bit(DEVICE_DYING, &mdev->flags))
1490 drbd_thread_stop_nowait(&mdev->worker);
1497 static int drbd_thread_setup(void *arg)
1499 struct drbd_thread *thi = (struct drbd_thread *) arg;
1500 struct drbd_conf *mdev = thi->mdev;
1501 unsigned long flags;
1505 retval = thi->function(thi);
1507 spin_lock_irqsave(&thi->t_lock, flags);
1509 /* if the receiver has been "Exiting", the last thing it did
1510 * was set the conn state to "StandAlone",
1511 * if now a re-connect request comes in, conn state goes C_UNCONNECTED,
1512 * and receiver thread will be "started".
1513 * drbd_thread_start needs to set "Restarting" in that case.
1514 * t_state check and assignment needs to be within the same spinlock,
1515 * so either thread_start sees Exiting, and can remap to Restarting,
1516 * or thread_start see None, and can proceed as normal.
1519 if (thi->t_state == Restarting) {
1520 dev_info(DEV, "Restarting %s\n", current->comm);
1521 thi->t_state = Running;
1522 spin_unlock_irqrestore(&thi->t_lock, flags);
1527 thi->t_state = None;
1529 complete(&thi->stop);
1530 spin_unlock_irqrestore(&thi->t_lock, flags);
1532 dev_info(DEV, "Terminating %s\n", current->comm);
1534 /* Release mod reference taken when thread was started */
1535 module_put(THIS_MODULE);
1539 static void drbd_thread_init(struct drbd_conf *mdev, struct drbd_thread *thi,
1540 int (*func) (struct drbd_thread *))
1542 spin_lock_init(&thi->t_lock);
1544 thi->t_state = None;
1545 thi->function = func;
1549 int drbd_thread_start(struct drbd_thread *thi)
1551 struct drbd_conf *mdev = thi->mdev;
1552 struct task_struct *nt;
1553 unsigned long flags;
1556 thi == &mdev->receiver ? "receiver" :
1557 thi == &mdev->asender ? "asender" :
1558 thi == &mdev->worker ? "worker" : "NONSENSE";
1560 /* is used from state engine doing drbd_thread_stop_nowait,
1561 * while holding the req lock irqsave */
1562 spin_lock_irqsave(&thi->t_lock, flags);
1564 switch (thi->t_state) {
1566 dev_info(DEV, "Starting %s thread (from %s [%d])\n",
1567 me, current->comm, current->pid);
1569 /* Get ref on module for thread - this is released when thread exits */
1570 if (!try_module_get(THIS_MODULE)) {
1571 dev_err(DEV, "Failed to get module reference in drbd_thread_start\n");
1572 spin_unlock_irqrestore(&thi->t_lock, flags);
1576 init_completion(&thi->stop);
1577 D_ASSERT(thi->task == NULL);
1578 thi->reset_cpu_mask = 1;
1579 thi->t_state = Running;
1580 spin_unlock_irqrestore(&thi->t_lock, flags);
1581 flush_signals(current); /* otherw. may get -ERESTARTNOINTR */
1583 nt = kthread_create(drbd_thread_setup, (void *) thi,
1584 "drbd%d_%s", mdev_to_minor(mdev), me);
1587 dev_err(DEV, "Couldn't start thread\n");
1589 module_put(THIS_MODULE);
1592 spin_lock_irqsave(&thi->t_lock, flags);
1594 thi->t_state = Running;
1595 spin_unlock_irqrestore(&thi->t_lock, flags);
1596 wake_up_process(nt);
1599 thi->t_state = Restarting;
1600 dev_info(DEV, "Restarting %s thread (from %s [%d])\n",
1601 me, current->comm, current->pid);
1606 spin_unlock_irqrestore(&thi->t_lock, flags);
1614 void _drbd_thread_stop(struct drbd_thread *thi, int restart, int wait)
1616 unsigned long flags;
1618 enum drbd_thread_state ns = restart ? Restarting : Exiting;
1620 /* may be called from state engine, holding the req lock irqsave */
1621 spin_lock_irqsave(&thi->t_lock, flags);
1623 if (thi->t_state == None) {
1624 spin_unlock_irqrestore(&thi->t_lock, flags);
1626 drbd_thread_start(thi);
1630 if (thi->t_state != ns) {
1631 if (thi->task == NULL) {
1632 spin_unlock_irqrestore(&thi->t_lock, flags);
1638 init_completion(&thi->stop);
1639 if (thi->task != current)
1640 force_sig(DRBD_SIGKILL, thi->task);
1644 spin_unlock_irqrestore(&thi->t_lock, flags);
1647 wait_for_completion(&thi->stop);
1652 * drbd_calc_cpu_mask() - Generate CPU masks, spread over all CPUs
1653 * @mdev: DRBD device.
1655 * Forces all threads of a device onto the same CPU. This is beneficial for
1656 * DRBD's performance. May be overwritten by user's configuration.
1658 void drbd_calc_cpu_mask(struct drbd_conf *mdev)
1662 /* user override. */
1663 if (cpumask_weight(mdev->cpu_mask))
1666 ord = mdev_to_minor(mdev) % cpumask_weight(cpu_online_mask);
1667 for_each_online_cpu(cpu) {
1669 cpumask_set_cpu(cpu, mdev->cpu_mask);
1673 /* should not be reached */
1674 cpumask_setall(mdev->cpu_mask);
1678 * drbd_thread_current_set_cpu() - modifies the cpu mask of the _current_ thread
1679 * @mdev: DRBD device.
1681 * call in the "main loop" of _all_ threads, no need for any mutex, current won't die
1684 void drbd_thread_current_set_cpu(struct drbd_conf *mdev)
1686 struct task_struct *p = current;
1687 struct drbd_thread *thi =
1688 p == mdev->asender.task ? &mdev->asender :
1689 p == mdev->receiver.task ? &mdev->receiver :
1690 p == mdev->worker.task ? &mdev->worker :
1694 if (!thi->reset_cpu_mask)
1696 thi->reset_cpu_mask = 0;
1697 set_cpus_allowed_ptr(p, mdev->cpu_mask);
1701 /* the appropriate socket mutex must be held already */
1702 int _drbd_send_cmd(struct drbd_conf *mdev, struct socket *sock,
1703 enum drbd_packets cmd, struct p_header80 *h,
1704 size_t size, unsigned msg_flags)
1708 ERR_IF(!h) return FALSE;
1709 ERR_IF(!size) return FALSE;
1711 h->magic = BE_DRBD_MAGIC;
1712 h->command = cpu_to_be16(cmd);
1713 h->length = cpu_to_be16(size-sizeof(struct p_header80));
1715 sent = drbd_send(mdev, sock, h, size, msg_flags);
1717 ok = (sent == size);
1719 dev_err(DEV, "short sent %s size=%d sent=%d\n",
1720 cmdname(cmd), (int)size, sent);
1724 /* don't pass the socket. we may only look at it
1725 * when we hold the appropriate socket mutex.
1727 int drbd_send_cmd(struct drbd_conf *mdev, int use_data_socket,
1728 enum drbd_packets cmd, struct p_header80 *h, size_t size)
1731 struct socket *sock;
1733 if (use_data_socket) {
1734 mutex_lock(&mdev->data.mutex);
1735 sock = mdev->data.socket;
1737 mutex_lock(&mdev->meta.mutex);
1738 sock = mdev->meta.socket;
1741 /* drbd_disconnect() could have called drbd_free_sock()
1742 * while we were waiting in down()... */
1743 if (likely(sock != NULL))
1744 ok = _drbd_send_cmd(mdev, sock, cmd, h, size, 0);
1746 if (use_data_socket)
1747 mutex_unlock(&mdev->data.mutex);
1749 mutex_unlock(&mdev->meta.mutex);
1753 int drbd_send_cmd2(struct drbd_conf *mdev, enum drbd_packets cmd, char *data,
1756 struct p_header80 h;
1759 h.magic = BE_DRBD_MAGIC;
1760 h.command = cpu_to_be16(cmd);
1761 h.length = cpu_to_be16(size);
1763 if (!drbd_get_data_sock(mdev))
1767 drbd_send(mdev, mdev->data.socket, &h, sizeof(h), 0));
1769 drbd_send(mdev, mdev->data.socket, data, size, 0));
1771 drbd_put_data_sock(mdev);
1776 int drbd_send_sync_param(struct drbd_conf *mdev, struct syncer_conf *sc)
1778 struct p_rs_param_95 *p;
1779 struct socket *sock;
1781 const int apv = mdev->agreed_pro_version;
1783 size = apv <= 87 ? sizeof(struct p_rs_param)
1784 : apv == 88 ? sizeof(struct p_rs_param)
1785 + strlen(mdev->sync_conf.verify_alg) + 1
1786 : apv <= 94 ? sizeof(struct p_rs_param_89)
1787 : /* apv >= 95 */ sizeof(struct p_rs_param_95);
1789 /* used from admin command context and receiver/worker context.
1790 * to avoid kmalloc, grab the socket right here,
1791 * then use the pre-allocated sbuf there */
1792 mutex_lock(&mdev->data.mutex);
1793 sock = mdev->data.socket;
1795 if (likely(sock != NULL)) {
1796 enum drbd_packets cmd = apv >= 89 ? P_SYNC_PARAM89 : P_SYNC_PARAM;
1798 p = &mdev->data.sbuf.rs_param_95;
1800 /* initialize verify_alg and csums_alg */
1801 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX);
1803 p->rate = cpu_to_be32(sc->rate);
1804 p->c_plan_ahead = cpu_to_be32(sc->c_plan_ahead);
1805 p->c_delay_target = cpu_to_be32(sc->c_delay_target);
1806 p->c_fill_target = cpu_to_be32(sc->c_fill_target);
1807 p->c_max_rate = cpu_to_be32(sc->c_max_rate);
1810 strcpy(p->verify_alg, mdev->sync_conf.verify_alg);
1812 strcpy(p->csums_alg, mdev->sync_conf.csums_alg);
1814 rv = _drbd_send_cmd(mdev, sock, cmd, &p->head, size, 0);
1816 rv = 0; /* not ok */
1818 mutex_unlock(&mdev->data.mutex);
1823 int drbd_send_protocol(struct drbd_conf *mdev)
1825 struct p_protocol *p;
1828 size = sizeof(struct p_protocol);
1830 if (mdev->agreed_pro_version >= 87)
1831 size += strlen(mdev->net_conf->integrity_alg) + 1;
1833 /* we must not recurse into our own queue,
1834 * as that is blocked during handshake */
1835 p = kmalloc(size, GFP_NOIO);
1839 p->protocol = cpu_to_be32(mdev->net_conf->wire_protocol);
1840 p->after_sb_0p = cpu_to_be32(mdev->net_conf->after_sb_0p);
1841 p->after_sb_1p = cpu_to_be32(mdev->net_conf->after_sb_1p);
1842 p->after_sb_2p = cpu_to_be32(mdev->net_conf->after_sb_2p);
1843 p->two_primaries = cpu_to_be32(mdev->net_conf->two_primaries);
1846 if (mdev->net_conf->want_lose)
1848 if (mdev->net_conf->dry_run) {
1849 if (mdev->agreed_pro_version >= 92)
1852 dev_err(DEV, "--dry-run is not supported by peer");
1857 p->conn_flags = cpu_to_be32(cf);
1859 if (mdev->agreed_pro_version >= 87)
1860 strcpy(p->integrity_alg, mdev->net_conf->integrity_alg);
1862 rv = drbd_send_cmd(mdev, USE_DATA_SOCKET, P_PROTOCOL,
1863 (struct p_header80 *)p, size);
1868 int _drbd_send_uuids(struct drbd_conf *mdev, u64 uuid_flags)
1873 if (!get_ldev_if_state(mdev, D_NEGOTIATING))
1876 for (i = UI_CURRENT; i < UI_SIZE; i++)
1877 p.uuid[i] = mdev->ldev ? cpu_to_be64(mdev->ldev->md.uuid[i]) : 0;
1879 mdev->comm_bm_set = drbd_bm_total_weight(mdev);
1880 p.uuid[UI_SIZE] = cpu_to_be64(mdev->comm_bm_set);
1881 uuid_flags |= mdev->net_conf->want_lose ? 1 : 0;
1882 uuid_flags |= test_bit(CRASHED_PRIMARY, &mdev->flags) ? 2 : 0;
1883 uuid_flags |= mdev->new_state_tmp.disk == D_INCONSISTENT ? 4 : 0;
1884 p.uuid[UI_FLAGS] = cpu_to_be64(uuid_flags);
1888 return drbd_send_cmd(mdev, USE_DATA_SOCKET, P_UUIDS,
1889 (struct p_header80 *)&p, sizeof(p));
1892 int drbd_send_uuids(struct drbd_conf *mdev)
1894 return _drbd_send_uuids(mdev, 0);
1897 int drbd_send_uuids_skip_initial_sync(struct drbd_conf *mdev)
1899 return _drbd_send_uuids(mdev, 8);
1903 int drbd_send_sync_uuid(struct drbd_conf *mdev, u64 val)
1907 p.uuid = cpu_to_be64(val);
1909 return drbd_send_cmd(mdev, USE_DATA_SOCKET, P_SYNC_UUID,
1910 (struct p_header80 *)&p, sizeof(p));
1913 int drbd_send_sizes(struct drbd_conf *mdev, int trigger_reply, enum dds_flags flags)
1916 sector_t d_size, u_size;
1920 if (get_ldev_if_state(mdev, D_NEGOTIATING)) {
1921 D_ASSERT(mdev->ldev->backing_bdev);
1922 d_size = drbd_get_max_capacity(mdev->ldev);
1923 u_size = mdev->ldev->dc.disk_size;
1924 q_order_type = drbd_queue_order_type(mdev);
1929 q_order_type = QUEUE_ORDERED_NONE;
1932 p.d_size = cpu_to_be64(d_size);
1933 p.u_size = cpu_to_be64(u_size);
1934 p.c_size = cpu_to_be64(trigger_reply ? 0 : drbd_get_capacity(mdev->this_bdev));
1935 p.max_bio_size = cpu_to_be32(queue_max_hw_sectors(mdev->rq_queue) << 9);
1936 p.queue_order_type = cpu_to_be16(q_order_type);
1937 p.dds_flags = cpu_to_be16(flags);
1939 ok = drbd_send_cmd(mdev, USE_DATA_SOCKET, P_SIZES,
1940 (struct p_header80 *)&p, sizeof(p));
1945 * drbd_send_state() - Sends the drbd state to the peer
1946 * @mdev: DRBD device.
1948 int drbd_send_state(struct drbd_conf *mdev)
1950 struct socket *sock;
1954 /* Grab state lock so we wont send state if we're in the middle
1955 * of a cluster wide state change on another thread */
1956 drbd_state_lock(mdev);
1958 mutex_lock(&mdev->data.mutex);
1960 p.state = cpu_to_be32(mdev->state.i); /* Within the send mutex */
1961 sock = mdev->data.socket;
1963 if (likely(sock != NULL)) {
1964 ok = _drbd_send_cmd(mdev, sock, P_STATE,
1965 (struct p_header80 *)&p, sizeof(p), 0);
1968 mutex_unlock(&mdev->data.mutex);
1970 drbd_state_unlock(mdev);
1974 int drbd_send_state_req(struct drbd_conf *mdev,
1975 union drbd_state mask, union drbd_state val)
1977 struct p_req_state p;
1979 p.mask = cpu_to_be32(mask.i);
1980 p.val = cpu_to_be32(val.i);
1982 return drbd_send_cmd(mdev, USE_DATA_SOCKET, P_STATE_CHG_REQ,
1983 (struct p_header80 *)&p, sizeof(p));
1986 int drbd_send_sr_reply(struct drbd_conf *mdev, int retcode)
1988 struct p_req_state_reply p;
1990 p.retcode = cpu_to_be32(retcode);
1992 return drbd_send_cmd(mdev, USE_META_SOCKET, P_STATE_CHG_REPLY,
1993 (struct p_header80 *)&p, sizeof(p));
1996 int fill_bitmap_rle_bits(struct drbd_conf *mdev,
1997 struct p_compressed_bm *p,
1998 struct bm_xfer_ctx *c)
2000 struct bitstream bs;
2001 unsigned long plain_bits;
2008 /* may we use this feature? */
2009 if ((mdev->sync_conf.use_rle == 0) ||
2010 (mdev->agreed_pro_version < 90))
2013 if (c->bit_offset >= c->bm_bits)
2014 return 0; /* nothing to do. */
2016 /* use at most thus many bytes */
2017 bitstream_init(&bs, p->code, BM_PACKET_VLI_BYTES_MAX, 0);
2018 memset(p->code, 0, BM_PACKET_VLI_BYTES_MAX);
2019 /* plain bits covered in this code string */
2022 /* p->encoding & 0x80 stores whether the first run length is set.
2023 * bit offset is implicit.
2024 * start with toggle == 2 to be able to tell the first iteration */
2027 /* see how much plain bits we can stuff into one packet
2028 * using RLE and VLI. */
2030 tmp = (toggle == 0) ? _drbd_bm_find_next_zero(mdev, c->bit_offset)
2031 : _drbd_bm_find_next(mdev, c->bit_offset);
2034 rl = tmp - c->bit_offset;
2036 if (toggle == 2) { /* first iteration */
2038 /* the first checked bit was set,
2039 * store start value, */
2040 DCBP_set_start(p, 1);
2041 /* but skip encoding of zero run length */
2045 DCBP_set_start(p, 0);
2048 /* paranoia: catch zero runlength.
2049 * can only happen if bitmap is modified while we scan it. */
2051 dev_err(DEV, "unexpected zero runlength while encoding bitmap "
2052 "t:%u bo:%lu\n", toggle, c->bit_offset);
2056 bits = vli_encode_bits(&bs, rl);
2057 if (bits == -ENOBUFS) /* buffer full */
2060 dev_err(DEV, "error while encoding bitmap: %d\n", bits);
2066 c->bit_offset = tmp;
2067 } while (c->bit_offset < c->bm_bits);
2069 len = bs.cur.b - p->code + !!bs.cur.bit;
2071 if (plain_bits < (len << 3)) {
2072 /* incompressible with this method.
2073 * we need to rewind both word and bit position. */
2074 c->bit_offset -= plain_bits;
2075 bm_xfer_ctx_bit_to_word_offset(c);
2076 c->bit_offset = c->word_offset * BITS_PER_LONG;
2080 /* RLE + VLI was able to compress it just fine.
2081 * update c->word_offset. */
2082 bm_xfer_ctx_bit_to_word_offset(c);
2084 /* store pad_bits */
2085 DCBP_set_pad_bits(p, (8 - bs.cur.bit) & 0x7);
2090 enum { OK, FAILED, DONE }
2091 send_bitmap_rle_or_plain(struct drbd_conf *mdev,
2092 struct p_header80 *h, struct bm_xfer_ctx *c)
2094 struct p_compressed_bm *p = (void*)h;
2095 unsigned long num_words;
2099 len = fill_bitmap_rle_bits(mdev, p, c);
2105 DCBP_set_code(p, RLE_VLI_Bits);
2106 ok = _drbd_send_cmd(mdev, mdev->data.socket, P_COMPRESSED_BITMAP, h,
2107 sizeof(*p) + len, 0);
2110 c->bytes[0] += sizeof(*p) + len;
2112 if (c->bit_offset >= c->bm_bits)
2115 /* was not compressible.
2116 * send a buffer full of plain text bits instead. */
2117 num_words = min_t(size_t, BM_PACKET_WORDS, c->bm_words - c->word_offset);
2118 len = num_words * sizeof(long);
2120 drbd_bm_get_lel(mdev, c->word_offset, num_words, (unsigned long*)h->payload);
2121 ok = _drbd_send_cmd(mdev, mdev->data.socket, P_BITMAP,
2122 h, sizeof(struct p_header80) + len, 0);
2123 c->word_offset += num_words;
2124 c->bit_offset = c->word_offset * BITS_PER_LONG;
2127 c->bytes[1] += sizeof(struct p_header80) + len;
2129 if (c->bit_offset > c->bm_bits)
2130 c->bit_offset = c->bm_bits;
2132 ok = ok ? ((len == 0) ? DONE : OK) : FAILED;
2135 INFO_bm_xfer_stats(mdev, "send", c);
2139 /* See the comment at receive_bitmap() */
2140 int _drbd_send_bitmap(struct drbd_conf *mdev)
2142 struct bm_xfer_ctx c;
2143 struct p_header80 *p;
2146 ERR_IF(!mdev->bitmap) return FALSE;
2148 /* maybe we should use some per thread scratch page,
2149 * and allocate that during initial device creation? */
2150 p = (struct p_header80 *) __get_free_page(GFP_NOIO);
2152 dev_err(DEV, "failed to allocate one page buffer in %s\n", __func__);
2156 if (get_ldev(mdev)) {
2157 if (drbd_md_test_flag(mdev->ldev, MDF_FULL_SYNC)) {
2158 dev_info(DEV, "Writing the whole bitmap, MDF_FullSync was set.\n");
2159 drbd_bm_set_all(mdev);
2160 if (drbd_bm_write(mdev)) {
2161 /* write_bm did fail! Leave full sync flag set in Meta P_DATA
2162 * but otherwise process as per normal - need to tell other
2163 * side that a full resync is required! */
2164 dev_err(DEV, "Failed to write bitmap to disk!\n");
2166 drbd_md_clear_flag(mdev, MDF_FULL_SYNC);
2173 c = (struct bm_xfer_ctx) {
2174 .bm_bits = drbd_bm_bits(mdev),
2175 .bm_words = drbd_bm_words(mdev),
2179 ret = send_bitmap_rle_or_plain(mdev, p, &c);
2180 } while (ret == OK);
2182 free_page((unsigned long) p);
2183 return (ret == DONE);
2186 int drbd_send_bitmap(struct drbd_conf *mdev)
2190 if (!drbd_get_data_sock(mdev))
2192 err = !_drbd_send_bitmap(mdev);
2193 drbd_put_data_sock(mdev);
2197 int drbd_send_b_ack(struct drbd_conf *mdev, u32 barrier_nr, u32 set_size)
2200 struct p_barrier_ack p;
2202 p.barrier = barrier_nr;
2203 p.set_size = cpu_to_be32(set_size);
2205 if (mdev->state.conn < C_CONNECTED)
2207 ok = drbd_send_cmd(mdev, USE_META_SOCKET, P_BARRIER_ACK,
2208 (struct p_header80 *)&p, sizeof(p));
2213 * _drbd_send_ack() - Sends an ack packet
2214 * @mdev: DRBD device.
2215 * @cmd: Packet command code.
2216 * @sector: sector, needs to be in big endian byte order
2217 * @blksize: size in byte, needs to be in big endian byte order
2218 * @block_id: Id, big endian byte order
2220 static int _drbd_send_ack(struct drbd_conf *mdev, enum drbd_packets cmd,
2226 struct p_block_ack p;
2229 p.block_id = block_id;
2230 p.blksize = blksize;
2231 p.seq_num = cpu_to_be32(atomic_add_return(1, &mdev->packet_seq));
2233 if (!mdev->meta.socket || mdev->state.conn < C_CONNECTED)
2235 ok = drbd_send_cmd(mdev, USE_META_SOCKET, cmd,
2236 (struct p_header80 *)&p, sizeof(p));
2240 /* dp->sector and dp->block_id already/still in network byte order,
2241 * data_size is payload size according to dp->head,
2242 * and may need to be corrected for digest size. */
2243 int drbd_send_ack_dp(struct drbd_conf *mdev, enum drbd_packets cmd,
2244 struct p_data *dp, int data_size)
2246 data_size -= (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ?
2247 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0;
2248 return _drbd_send_ack(mdev, cmd, dp->sector, cpu_to_be32(data_size),
2252 int drbd_send_ack_rp(struct drbd_conf *mdev, enum drbd_packets cmd,
2253 struct p_block_req *rp)
2255 return _drbd_send_ack(mdev, cmd, rp->sector, rp->blksize, rp->block_id);
2259 * drbd_send_ack() - Sends an ack packet
2260 * @mdev: DRBD device.
2261 * @cmd: Packet command code.
2264 int drbd_send_ack(struct drbd_conf *mdev,
2265 enum drbd_packets cmd, struct drbd_epoch_entry *e)
2267 return _drbd_send_ack(mdev, cmd,
2268 cpu_to_be64(e->sector),
2269 cpu_to_be32(e->size),
2273 /* This function misuses the block_id field to signal if the blocks
2274 * are is sync or not. */
2275 int drbd_send_ack_ex(struct drbd_conf *mdev, enum drbd_packets cmd,
2276 sector_t sector, int blksize, u64 block_id)
2278 return _drbd_send_ack(mdev, cmd,
2279 cpu_to_be64(sector),
2280 cpu_to_be32(blksize),
2281 cpu_to_be64(block_id));
2284 int drbd_send_drequest(struct drbd_conf *mdev, int cmd,
2285 sector_t sector, int size, u64 block_id)
2288 struct p_block_req p;
2290 p.sector = cpu_to_be64(sector);
2291 p.block_id = block_id;
2292 p.blksize = cpu_to_be32(size);
2294 ok = drbd_send_cmd(mdev, USE_DATA_SOCKET, cmd,
2295 (struct p_header80 *)&p, sizeof(p));
2299 int drbd_send_drequest_csum(struct drbd_conf *mdev,
2300 sector_t sector, int size,
2301 void *digest, int digest_size,
2302 enum drbd_packets cmd)
2305 struct p_block_req p;
2307 p.sector = cpu_to_be64(sector);
2308 p.block_id = BE_DRBD_MAGIC + 0xbeef;
2309 p.blksize = cpu_to_be32(size);
2311 p.head.magic = BE_DRBD_MAGIC;
2312 p.head.command = cpu_to_be16(cmd);
2313 p.head.length = cpu_to_be16(sizeof(p) - sizeof(struct p_header80) + digest_size);
2315 mutex_lock(&mdev->data.mutex);
2317 ok = (sizeof(p) == drbd_send(mdev, mdev->data.socket, &p, sizeof(p), 0));
2318 ok = ok && (digest_size == drbd_send(mdev, mdev->data.socket, digest, digest_size, 0));
2320 mutex_unlock(&mdev->data.mutex);
2325 int drbd_send_ov_request(struct drbd_conf *mdev, sector_t sector, int size)
2328 struct p_block_req p;
2330 p.sector = cpu_to_be64(sector);
2331 p.block_id = BE_DRBD_MAGIC + 0xbabe;
2332 p.blksize = cpu_to_be32(size);
2334 ok = drbd_send_cmd(mdev, USE_DATA_SOCKET, P_OV_REQUEST,
2335 (struct p_header80 *)&p, sizeof(p));
2339 /* called on sndtimeo
2340 * returns FALSE if we should retry,
2341 * TRUE if we think connection is dead
2343 static int we_should_drop_the_connection(struct drbd_conf *mdev, struct socket *sock)
2346 /* long elapsed = (long)(jiffies - mdev->last_received); */
2348 drop_it = mdev->meta.socket == sock
2349 || !mdev->asender.task
2350 || get_t_state(&mdev->asender) != Running
2351 || mdev->state.conn < C_CONNECTED;
2356 drop_it = !--mdev->ko_count;
2358 dev_err(DEV, "[%s/%d] sock_sendmsg time expired, ko = %u\n",
2359 current->comm, current->pid, mdev->ko_count);
2363 return drop_it; /* && (mdev->state == R_PRIMARY) */;
2366 /* The idea of sendpage seems to be to put some kind of reference
2367 * to the page into the skb, and to hand it over to the NIC. In
2368 * this process get_page() gets called.
2370 * As soon as the page was really sent over the network put_page()
2371 * gets called by some part of the network layer. [ NIC driver? ]
2373 * [ get_page() / put_page() increment/decrement the count. If count
2374 * reaches 0 the page will be freed. ]
2376 * This works nicely with pages from FSs.
2377 * But this means that in protocol A we might signal IO completion too early!
2379 * In order not to corrupt data during a resync we must make sure
2380 * that we do not reuse our own buffer pages (EEs) to early, therefore
2381 * we have the net_ee list.
2383 * XFS seems to have problems, still, it submits pages with page_count == 0!
2384 * As a workaround, we disable sendpage on pages
2385 * with page_count == 0 or PageSlab.
2387 static int _drbd_no_send_page(struct drbd_conf *mdev, struct page *page,
2388 int offset, size_t size, unsigned msg_flags)
2390 int sent = drbd_send(mdev, mdev->data.socket, kmap(page) + offset, size, msg_flags);
2393 mdev->send_cnt += size>>9;
2394 return sent == size;
2397 static int _drbd_send_page(struct drbd_conf *mdev, struct page *page,
2398 int offset, size_t size, unsigned msg_flags)
2400 mm_segment_t oldfs = get_fs();
2404 /* e.g. XFS meta- & log-data is in slab pages, which have a
2405 * page_count of 0 and/or have PageSlab() set.
2406 * we cannot use send_page for those, as that does get_page();
2407 * put_page(); and would cause either a VM_BUG directly, or
2408 * __page_cache_release a page that would actually still be referenced
2409 * by someone, leading to some obscure delayed Oops somewhere else. */
2410 if (disable_sendpage || (page_count(page) < 1) || PageSlab(page))
2411 return _drbd_no_send_page(mdev, page, offset, size, msg_flags);
2413 msg_flags |= MSG_NOSIGNAL;
2414 drbd_update_congested(mdev);
2417 sent = mdev->data.socket->ops->sendpage(mdev->data.socket, page,
2420 if (sent == -EAGAIN) {
2421 if (we_should_drop_the_connection(mdev,
2428 dev_warn(DEV, "%s: size=%d len=%d sent=%d\n",
2429 __func__, (int)size, len, sent);
2434 } while (len > 0 /* THINK && mdev->cstate >= C_CONNECTED*/);
2436 clear_bit(NET_CONGESTED, &mdev->flags);
2440 mdev->send_cnt += size>>9;
2444 static int _drbd_send_bio(struct drbd_conf *mdev, struct bio *bio)
2446 struct bio_vec *bvec;
2448 /* hint all but last page with MSG_MORE */
2449 __bio_for_each_segment(bvec, bio, i, 0) {
2450 if (!_drbd_no_send_page(mdev, bvec->bv_page,
2451 bvec->bv_offset, bvec->bv_len,
2452 i == bio->bi_vcnt -1 ? 0 : MSG_MORE))
2458 static int _drbd_send_zc_bio(struct drbd_conf *mdev, struct bio *bio)
2460 struct bio_vec *bvec;
2462 /* hint all but last page with MSG_MORE */
2463 __bio_for_each_segment(bvec, bio, i, 0) {
2464 if (!_drbd_send_page(mdev, bvec->bv_page,
2465 bvec->bv_offset, bvec->bv_len,
2466 i == bio->bi_vcnt -1 ? 0 : MSG_MORE))
2472 static int _drbd_send_zc_ee(struct drbd_conf *mdev, struct drbd_epoch_entry *e)
2474 struct page *page = e->pages;
2475 unsigned len = e->size;
2476 /* hint all but last page with MSG_MORE */
2477 page_chain_for_each(page) {
2478 unsigned l = min_t(unsigned, len, PAGE_SIZE);
2479 if (!_drbd_send_page(mdev, page, 0, l,
2480 page_chain_next(page) ? MSG_MORE : 0))
2487 static u32 bio_flags_to_wire(struct drbd_conf *mdev, unsigned long bi_rw)
2489 if (mdev->agreed_pro_version >= 95)
2490 return (bi_rw & REQ_SYNC ? DP_RW_SYNC : 0) |
2491 (bi_rw & REQ_FUA ? DP_FUA : 0) |
2492 (bi_rw & REQ_FLUSH ? DP_FLUSH : 0) |
2493 (bi_rw & REQ_DISCARD ? DP_DISCARD : 0);
2495 return bi_rw & REQ_SYNC ? DP_RW_SYNC : 0;
2498 /* Used to send write requests
2499 * R_PRIMARY -> Peer (P_DATA)
2501 int drbd_send_dblock(struct drbd_conf *mdev, struct drbd_request *req)
2505 unsigned int dp_flags = 0;
2509 if (!drbd_get_data_sock(mdev))
2512 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_w_tfm) ?
2513 crypto_hash_digestsize(mdev->integrity_w_tfm) : 0;
2515 if (req->size <= DRBD_MAX_SIZE_H80_PACKET) {
2516 p.head.h80.magic = BE_DRBD_MAGIC;
2517 p.head.h80.command = cpu_to_be16(P_DATA);
2519 cpu_to_be16(sizeof(p) - sizeof(union p_header) + dgs + req->size);
2521 p.head.h95.magic = BE_DRBD_MAGIC_BIG;
2522 p.head.h95.command = cpu_to_be16(P_DATA);
2524 cpu_to_be32(sizeof(p) - sizeof(union p_header) + dgs + req->size);
2527 p.sector = cpu_to_be64(req->sector);
2528 p.block_id = (unsigned long)req;
2529 p.seq_num = cpu_to_be32(req->seq_num =
2530 atomic_add_return(1, &mdev->packet_seq));
2532 dp_flags = bio_flags_to_wire(mdev, req->master_bio->bi_rw);
2534 if (mdev->state.conn >= C_SYNC_SOURCE &&
2535 mdev->state.conn <= C_PAUSED_SYNC_T)
2536 dp_flags |= DP_MAY_SET_IN_SYNC;
2538 p.dp_flags = cpu_to_be32(dp_flags);
2539 set_bit(UNPLUG_REMOTE, &mdev->flags);
2541 drbd_send(mdev, mdev->data.socket, &p, sizeof(p), dgs ? MSG_MORE : 0));
2543 dgb = mdev->int_dig_out;
2544 drbd_csum_bio(mdev, mdev->integrity_w_tfm, req->master_bio, dgb);
2545 ok = drbd_send(mdev, mdev->data.socket, dgb, dgs, 0);
2548 /* For protocol A, we have to memcpy the payload into
2549 * socket buffers, as we may complete right away
2550 * as soon as we handed it over to tcp, at which point the data
2551 * pages may become invalid.
2553 * For data-integrity enabled, we copy it as well, so we can be
2554 * sure that even if the bio pages may still be modified, it
2555 * won't change the data on the wire, thus if the digest checks
2556 * out ok after sending on this side, but does not fit on the
2557 * receiving side, we sure have detected corruption elsewhere.
2559 if (mdev->net_conf->wire_protocol == DRBD_PROT_A || dgs)
2560 ok = _drbd_send_bio(mdev, req->master_bio);
2562 ok = _drbd_send_zc_bio(mdev, req->master_bio);
2564 /* double check digest, sometimes buffers have been modified in flight. */
2565 if (dgs > 0 && dgs <= 64) {
2566 /* 64 byte, 512 bit, is the larges digest size
2567 * currently supported in kernel crypto. */
2568 unsigned char digest[64];
2569 drbd_csum_bio(mdev, mdev->integrity_w_tfm, req->master_bio, digest);
2570 if (memcmp(mdev->int_dig_out, digest, dgs)) {
2572 "Digest mismatch, buffer modified by upper layers during write: %llus +%u\n",
2573 (unsigned long long)req->sector, req->size);
2575 } /* else if (dgs > 64) {
2576 ... Be noisy about digest too large ...
2580 drbd_put_data_sock(mdev);
2585 /* answer packet, used to send data back for read requests:
2586 * Peer -> (diskless) R_PRIMARY (P_DATA_REPLY)
2587 * C_SYNC_SOURCE -> C_SYNC_TARGET (P_RS_DATA_REPLY)
2589 int drbd_send_block(struct drbd_conf *mdev, enum drbd_packets cmd,
2590 struct drbd_epoch_entry *e)
2597 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_w_tfm) ?
2598 crypto_hash_digestsize(mdev->integrity_w_tfm) : 0;
2600 if (e->size <= DRBD_MAX_SIZE_H80_PACKET) {
2601 p.head.h80.magic = BE_DRBD_MAGIC;
2602 p.head.h80.command = cpu_to_be16(cmd);
2604 cpu_to_be16(sizeof(p) - sizeof(struct p_header80) + dgs + e->size);
2606 p.head.h95.magic = BE_DRBD_MAGIC_BIG;
2607 p.head.h95.command = cpu_to_be16(cmd);
2609 cpu_to_be32(sizeof(p) - sizeof(struct p_header80) + dgs + e->size);
2612 p.sector = cpu_to_be64(e->sector);
2613 p.block_id = e->block_id;
2614 /* p.seq_num = 0; No sequence numbers here.. */
2616 /* Only called by our kernel thread.
2617 * This one may be interrupted by DRBD_SIG and/or DRBD_SIGKILL
2618 * in response to admin command or module unload.
2620 if (!drbd_get_data_sock(mdev))
2623 ok = sizeof(p) == drbd_send(mdev, mdev->data.socket, &p, sizeof(p), dgs ? MSG_MORE : 0);
2625 dgb = mdev->int_dig_out;
2626 drbd_csum_ee(mdev, mdev->integrity_w_tfm, e, dgb);
2627 ok = drbd_send(mdev, mdev->data.socket, dgb, dgs, 0);
2630 ok = _drbd_send_zc_ee(mdev, e);
2632 drbd_put_data_sock(mdev);
2637 int drbd_send_oos(struct drbd_conf *mdev, struct drbd_request *req)
2639 struct p_block_desc p;
2641 p.sector = cpu_to_be64(req->sector);
2642 p.blksize = cpu_to_be32(req->size);
2644 return drbd_send_cmd(mdev, USE_DATA_SOCKET, P_OUT_OF_SYNC, &p.head, sizeof(p));
2648 drbd_send distinguishes two cases:
2650 Packets sent via the data socket "sock"
2651 and packets sent via the meta data socket "msock"
2654 -----------------+-------------------------+------------------------------
2655 timeout conf.timeout / 2 conf.timeout / 2
2656 timeout action send a ping via msock Abort communication
2657 and close all sockets
2661 * you must have down()ed the appropriate [m]sock_mutex elsewhere!
2663 int drbd_send(struct drbd_conf *mdev, struct socket *sock,
2664 void *buf, size_t size, unsigned msg_flags)
2673 /* THINK if (signal_pending) return ... ? */
2678 msg.msg_name = NULL;
2679 msg.msg_namelen = 0;
2680 msg.msg_control = NULL;
2681 msg.msg_controllen = 0;
2682 msg.msg_flags = msg_flags | MSG_NOSIGNAL;
2684 if (sock == mdev->data.socket) {
2685 mdev->ko_count = mdev->net_conf->ko_count;
2686 drbd_update_congested(mdev);
2690 * tcp_sendmsg does _not_ use its size parameter at all ?
2692 * -EAGAIN on timeout, -EINTR on signal.
2695 * do we need to block DRBD_SIG if sock == &meta.socket ??
2696 * otherwise wake_asender() might interrupt some send_*Ack !
2698 rv = kernel_sendmsg(sock, &msg, &iov, 1, size);
2699 if (rv == -EAGAIN) {
2700 if (we_should_drop_the_connection(mdev, sock))
2707 flush_signals(current);
2715 } while (sent < size);
2717 if (sock == mdev->data.socket)
2718 clear_bit(NET_CONGESTED, &mdev->flags);
2721 if (rv != -EAGAIN) {
2722 dev_err(DEV, "%s_sendmsg returned %d\n",
2723 sock == mdev->meta.socket ? "msock" : "sock",
2725 drbd_force_state(mdev, NS(conn, C_BROKEN_PIPE));
2727 drbd_force_state(mdev, NS(conn, C_TIMEOUT));
2733 static int drbd_open(struct block_device *bdev, fmode_t mode)
2735 struct drbd_conf *mdev = bdev->bd_disk->private_data;
2736 unsigned long flags;
2739 mutex_lock(&drbd_main_mutex);
2740 spin_lock_irqsave(&mdev->req_lock, flags);
2741 /* to have a stable mdev->state.role
2742 * and no race with updating open_cnt */
2744 if (mdev->state.role != R_PRIMARY) {
2745 if (mode & FMODE_WRITE)
2747 else if (!allow_oos)
2753 spin_unlock_irqrestore(&mdev->req_lock, flags);
2754 mutex_unlock(&drbd_main_mutex);
2759 static int drbd_release(struct gendisk *gd, fmode_t mode)
2761 struct drbd_conf *mdev = gd->private_data;
2762 mutex_lock(&drbd_main_mutex);
2764 mutex_unlock(&drbd_main_mutex);
2768 static void drbd_set_defaults(struct drbd_conf *mdev)
2770 /* This way we get a compile error when sync_conf grows,
2771 and we forgot to initialize it here */
2772 mdev->sync_conf = (struct syncer_conf) {
2773 /* .rate = */ DRBD_RATE_DEF,
2774 /* .after = */ DRBD_AFTER_DEF,
2775 /* .al_extents = */ DRBD_AL_EXTENTS_DEF,
2776 /* .verify_alg = */ {}, 0,
2777 /* .cpu_mask = */ {}, 0,
2778 /* .csums_alg = */ {}, 0,
2780 /* .on_no_data = */ DRBD_ON_NO_DATA_DEF,
2781 /* .c_plan_ahead = */ DRBD_C_PLAN_AHEAD_DEF,
2782 /* .c_delay_target = */ DRBD_C_DELAY_TARGET_DEF,
2783 /* .c_fill_target = */ DRBD_C_FILL_TARGET_DEF,
2784 /* .c_max_rate = */ DRBD_C_MAX_RATE_DEF,
2785 /* .c_min_rate = */ DRBD_C_MIN_RATE_DEF
2788 /* Have to use that way, because the layout differs between
2789 big endian and little endian */
2790 mdev->state = (union drbd_state) {
2791 { .role = R_SECONDARY,
2793 .conn = C_STANDALONE,
2802 void drbd_init_set_defaults(struct drbd_conf *mdev)
2804 /* the memset(,0,) did most of this.
2805 * note: only assignments, no allocation in here */
2807 drbd_set_defaults(mdev);
2809 atomic_set(&mdev->ap_bio_cnt, 0);
2810 atomic_set(&mdev->ap_pending_cnt, 0);
2811 atomic_set(&mdev->rs_pending_cnt, 0);
2812 atomic_set(&mdev->unacked_cnt, 0);
2813 atomic_set(&mdev->local_cnt, 0);
2814 atomic_set(&mdev->net_cnt, 0);
2815 atomic_set(&mdev->packet_seq, 0);
2816 atomic_set(&mdev->pp_in_use, 0);
2817 atomic_set(&mdev->pp_in_use_by_net, 0);
2818 atomic_set(&mdev->rs_sect_in, 0);
2819 atomic_set(&mdev->rs_sect_ev, 0);
2820 atomic_set(&mdev->ap_in_flight, 0);
2822 mutex_init(&mdev->md_io_mutex);
2823 mutex_init(&mdev->data.mutex);
2824 mutex_init(&mdev->meta.mutex);
2825 sema_init(&mdev->data.work.s, 0);
2826 sema_init(&mdev->meta.work.s, 0);
2827 mutex_init(&mdev->state_mutex);
2829 spin_lock_init(&mdev->data.work.q_lock);
2830 spin_lock_init(&mdev->meta.work.q_lock);
2832 spin_lock_init(&mdev->al_lock);
2833 spin_lock_init(&mdev->req_lock);
2834 spin_lock_init(&mdev->peer_seq_lock);
2835 spin_lock_init(&mdev->epoch_lock);
2837 INIT_LIST_HEAD(&mdev->active_ee);
2838 INIT_LIST_HEAD(&mdev->sync_ee);
2839 INIT_LIST_HEAD(&mdev->done_ee);
2840 INIT_LIST_HEAD(&mdev->read_ee);
2841 INIT_LIST_HEAD(&mdev->net_ee);
2842 INIT_LIST_HEAD(&mdev->resync_reads);
2843 INIT_LIST_HEAD(&mdev->data.work.q);
2844 INIT_LIST_HEAD(&mdev->meta.work.q);
2845 INIT_LIST_HEAD(&mdev->resync_work.list);
2846 INIT_LIST_HEAD(&mdev->unplug_work.list);
2847 INIT_LIST_HEAD(&mdev->go_diskless.list);
2848 INIT_LIST_HEAD(&mdev->md_sync_work.list);
2849 INIT_LIST_HEAD(&mdev->start_resync_work.list);
2850 INIT_LIST_HEAD(&mdev->bm_io_work.w.list);
2852 mdev->resync_work.cb = w_resync_inactive;
2853 mdev->unplug_work.cb = w_send_write_hint;
2854 mdev->go_diskless.cb = w_go_diskless;
2855 mdev->md_sync_work.cb = w_md_sync;
2856 mdev->bm_io_work.w.cb = w_bitmap_io;
2857 init_timer(&mdev->resync_timer);
2858 init_timer(&mdev->md_sync_timer);
2859 mdev->resync_timer.function = resync_timer_fn;
2860 mdev->resync_timer.data = (unsigned long) mdev;
2861 mdev->md_sync_timer.function = md_sync_timer_fn;
2862 mdev->md_sync_timer.data = (unsigned long) mdev;
2864 init_waitqueue_head(&mdev->misc_wait);
2865 init_waitqueue_head(&mdev->state_wait);
2866 init_waitqueue_head(&mdev->net_cnt_wait);
2867 init_waitqueue_head(&mdev->ee_wait);
2868 init_waitqueue_head(&mdev->al_wait);
2869 init_waitqueue_head(&mdev->seq_wait);
2871 drbd_thread_init(mdev, &mdev->receiver, drbdd_init);
2872 drbd_thread_init(mdev, &mdev->worker, drbd_worker);
2873 drbd_thread_init(mdev, &mdev->asender, drbd_asender);
2875 mdev->agreed_pro_version = PRO_VERSION_MAX;
2876 mdev->write_ordering = WO_bdev_flush;
2877 mdev->resync_wenr = LC_FREE;
2880 void drbd_mdev_cleanup(struct drbd_conf *mdev)
2883 if (mdev->receiver.t_state != None)
2884 dev_err(DEV, "ASSERT FAILED: receiver t_state == %d expected 0.\n",
2885 mdev->receiver.t_state);
2887 /* no need to lock it, I'm the only thread alive */
2888 if (atomic_read(&mdev->current_epoch->epoch_size) != 0)
2889 dev_err(DEV, "epoch_size:%d\n", atomic_read(&mdev->current_epoch->epoch_size));
2899 mdev->rs_failed = 0;
2900 mdev->rs_last_events = 0;
2901 mdev->rs_last_sect_ev = 0;
2902 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
2903 mdev->rs_mark_left[i] = 0;
2904 mdev->rs_mark_time[i] = 0;
2906 D_ASSERT(mdev->net_conf == NULL);
2908 drbd_set_my_capacity(mdev, 0);
2910 /* maybe never allocated. */
2911 drbd_bm_resize(mdev, 0, 1);
2912 drbd_bm_cleanup(mdev);
2915 drbd_free_resources(mdev);
2916 clear_bit(AL_SUSPENDED, &mdev->flags);
2919 * currently we drbd_init_ee only on module load, so
2920 * we may do drbd_release_ee only on module unload!
2922 D_ASSERT(list_empty(&mdev->active_ee));
2923 D_ASSERT(list_empty(&mdev->sync_ee));
2924 D_ASSERT(list_empty(&mdev->done_ee));
2925 D_ASSERT(list_empty(&mdev->read_ee));
2926 D_ASSERT(list_empty(&mdev->net_ee));
2927 D_ASSERT(list_empty(&mdev->resync_reads));
2928 D_ASSERT(list_empty(&mdev->data.work.q));
2929 D_ASSERT(list_empty(&mdev->meta.work.q));
2930 D_ASSERT(list_empty(&mdev->resync_work.list));
2931 D_ASSERT(list_empty(&mdev->unplug_work.list));
2932 D_ASSERT(list_empty(&mdev->go_diskless.list));
2936 static void drbd_destroy_mempools(void)
2940 while (drbd_pp_pool) {
2941 page = drbd_pp_pool;
2942 drbd_pp_pool = (struct page *)page_private(page);
2947 /* D_ASSERT(atomic_read(&drbd_pp_vacant)==0); */
2949 if (drbd_ee_mempool)
2950 mempool_destroy(drbd_ee_mempool);
2951 if (drbd_request_mempool)
2952 mempool_destroy(drbd_request_mempool);
2954 kmem_cache_destroy(drbd_ee_cache);
2955 if (drbd_request_cache)
2956 kmem_cache_destroy(drbd_request_cache);
2957 if (drbd_bm_ext_cache)
2958 kmem_cache_destroy(drbd_bm_ext_cache);
2959 if (drbd_al_ext_cache)
2960 kmem_cache_destroy(drbd_al_ext_cache);
2962 drbd_ee_mempool = NULL;
2963 drbd_request_mempool = NULL;
2964 drbd_ee_cache = NULL;
2965 drbd_request_cache = NULL;
2966 drbd_bm_ext_cache = NULL;
2967 drbd_al_ext_cache = NULL;
2972 static int drbd_create_mempools(void)
2975 const int number = (DRBD_MAX_BIO_SIZE/PAGE_SIZE) * minor_count;
2978 /* prepare our caches and mempools */
2979 drbd_request_mempool = NULL;
2980 drbd_ee_cache = NULL;
2981 drbd_request_cache = NULL;
2982 drbd_bm_ext_cache = NULL;
2983 drbd_al_ext_cache = NULL;
2984 drbd_pp_pool = NULL;
2987 drbd_request_cache = kmem_cache_create(
2988 "drbd_req", sizeof(struct drbd_request), 0, 0, NULL);
2989 if (drbd_request_cache == NULL)
2992 drbd_ee_cache = kmem_cache_create(
2993 "drbd_ee", sizeof(struct drbd_epoch_entry), 0, 0, NULL);
2994 if (drbd_ee_cache == NULL)
2997 drbd_bm_ext_cache = kmem_cache_create(
2998 "drbd_bm", sizeof(struct bm_extent), 0, 0, NULL);
2999 if (drbd_bm_ext_cache == NULL)
3002 drbd_al_ext_cache = kmem_cache_create(
3003 "drbd_al", sizeof(struct lc_element), 0, 0, NULL);
3004 if (drbd_al_ext_cache == NULL)
3008 drbd_request_mempool = mempool_create(number,
3009 mempool_alloc_slab, mempool_free_slab, drbd_request_cache);
3010 if (drbd_request_mempool == NULL)
3013 drbd_ee_mempool = mempool_create(number,
3014 mempool_alloc_slab, mempool_free_slab, drbd_ee_cache);
3015 if (drbd_ee_mempool == NULL)
3018 /* drbd's page pool */
3019 spin_lock_init(&drbd_pp_lock);
3021 for (i = 0; i < number; i++) {
3022 page = alloc_page(GFP_HIGHUSER);
3025 set_page_private(page, (unsigned long)drbd_pp_pool);
3026 drbd_pp_pool = page;
3028 drbd_pp_vacant = number;
3033 drbd_destroy_mempools(); /* in case we allocated some */
3037 static int drbd_notify_sys(struct notifier_block *this, unsigned long code,
3040 /* just so we have it. you never know what interesting things we
3041 * might want to do here some day...
3047 static struct notifier_block drbd_notifier = {
3048 .notifier_call = drbd_notify_sys,
3051 static void drbd_release_ee_lists(struct drbd_conf *mdev)
3055 rr = drbd_release_ee(mdev, &mdev->active_ee);
3057 dev_err(DEV, "%d EEs in active list found!\n", rr);
3059 rr = drbd_release_ee(mdev, &mdev->sync_ee);
3061 dev_err(DEV, "%d EEs in sync list found!\n", rr);
3063 rr = drbd_release_ee(mdev, &mdev->read_ee);
3065 dev_err(DEV, "%d EEs in read list found!\n", rr);
3067 rr = drbd_release_ee(mdev, &mdev->done_ee);
3069 dev_err(DEV, "%d EEs in done list found!\n", rr);
3071 rr = drbd_release_ee(mdev, &mdev->net_ee);
3073 dev_err(DEV, "%d EEs in net list found!\n", rr);
3076 /* caution. no locking.
3077 * currently only used from module cleanup code. */
3078 static void drbd_delete_device(unsigned int minor)
3080 struct drbd_conf *mdev = minor_to_mdev(minor);
3085 /* paranoia asserts */
3086 if (mdev->open_cnt != 0)
3087 dev_err(DEV, "open_cnt = %d in %s:%u", mdev->open_cnt,
3088 __FILE__ , __LINE__);
3090 ERR_IF (!list_empty(&mdev->data.work.q)) {
3091 struct list_head *lp;
3092 list_for_each(lp, &mdev->data.work.q) {
3093 dev_err(DEV, "lp = %p\n", lp);
3096 /* end paranoia asserts */
3098 del_gendisk(mdev->vdisk);
3100 /* cleanup stuff that may have been allocated during
3101 * device (re-)configuration or state changes */
3103 if (mdev->this_bdev)
3104 bdput(mdev->this_bdev);
3106 drbd_free_resources(mdev);
3108 drbd_release_ee_lists(mdev);
3110 /* should be free'd on disconnect? */
3111 kfree(mdev->ee_hash);
3113 mdev->ee_hash_s = 0;
3114 mdev->ee_hash = NULL;
3117 lc_destroy(mdev->act_log);
3118 lc_destroy(mdev->resync);
3120 kfree(mdev->p_uuid);
3121 /* mdev->p_uuid = NULL; */
3123 kfree(mdev->int_dig_out);
3124 kfree(mdev->int_dig_in);
3125 kfree(mdev->int_dig_vv);
3127 /* cleanup the rest that has been
3128 * allocated from drbd_new_device
3129 * and actually free the mdev itself */
3130 drbd_free_mdev(mdev);
3133 static void drbd_cleanup(void)
3137 unregister_reboot_notifier(&drbd_notifier);
3143 remove_proc_entry("drbd", NULL);
3146 drbd_delete_device(i);
3147 drbd_destroy_mempools();
3152 unregister_blkdev(DRBD_MAJOR, "drbd");
3154 printk(KERN_INFO "drbd: module cleanup done.\n");
3158 * drbd_congested() - Callback for pdflush
3159 * @congested_data: User data
3160 * @bdi_bits: Bits pdflush is currently interested in
3162 * Returns 1<<BDI_async_congested and/or 1<<BDI_sync_congested if we are congested.
3164 static int drbd_congested(void *congested_data, int bdi_bits)
3166 struct drbd_conf *mdev = congested_data;
3167 struct request_queue *q;
3171 if (!__inc_ap_bio_cond(mdev)) {
3172 /* DRBD has frozen IO */
3178 if (get_ldev(mdev)) {
3179 q = bdev_get_queue(mdev->ldev->backing_bdev);
3180 r = bdi_congested(&q->backing_dev_info, bdi_bits);
3186 if (bdi_bits & (1 << BDI_async_congested) && test_bit(NET_CONGESTED, &mdev->flags)) {
3187 r |= (1 << BDI_async_congested);
3188 reason = reason == 'b' ? 'a' : 'n';
3192 mdev->congestion_reason = reason;
3196 struct drbd_conf *drbd_new_device(unsigned int minor)
3198 struct drbd_conf *mdev;
3199 struct gendisk *disk;
3200 struct request_queue *q;
3202 /* GFP_KERNEL, we are outside of all write-out paths */
3203 mdev = kzalloc(sizeof(struct drbd_conf), GFP_KERNEL);
3206 if (!zalloc_cpumask_var(&mdev->cpu_mask, GFP_KERNEL))
3207 goto out_no_cpumask;
3209 mdev->minor = minor;
3211 drbd_init_set_defaults(mdev);
3213 q = blk_alloc_queue(GFP_KERNEL);
3217 q->queuedata = mdev;
3219 disk = alloc_disk(1);
3224 set_disk_ro(disk, TRUE);
3227 disk->major = DRBD_MAJOR;
3228 disk->first_minor = minor;
3229 disk->fops = &drbd_ops;
3230 sprintf(disk->disk_name, "drbd%d", minor);
3231 disk->private_data = mdev;
3233 mdev->this_bdev = bdget(MKDEV(DRBD_MAJOR, minor));
3234 /* we have no partitions. we contain only ourselves. */
3235 mdev->this_bdev->bd_contains = mdev->this_bdev;
3237 q->backing_dev_info.congested_fn = drbd_congested;
3238 q->backing_dev_info.congested_data = mdev;
3240 blk_queue_make_request(q, drbd_make_request_26);
3241 blk_queue_max_hw_sectors(q, DRBD_MAX_BIO_SIZE >> 9);
3242 blk_queue_bounce_limit(q, BLK_BOUNCE_ANY);
3243 blk_queue_merge_bvec(q, drbd_merge_bvec);
3244 q->queue_lock = &mdev->req_lock;
3246 mdev->md_io_page = alloc_page(GFP_KERNEL);
3247 if (!mdev->md_io_page)
3248 goto out_no_io_page;
3250 if (drbd_bm_init(mdev))
3252 /* no need to lock access, we are still initializing this minor device. */
3256 mdev->app_reads_hash = kzalloc(APP_R_HSIZE*sizeof(void *), GFP_KERNEL);
3257 if (!mdev->app_reads_hash)
3258 goto out_no_app_reads;
3260 mdev->current_epoch = kzalloc(sizeof(struct drbd_epoch), GFP_KERNEL);
3261 if (!mdev->current_epoch)
3264 INIT_LIST_HEAD(&mdev->current_epoch->list);
3269 /* out_whatever_else:
3270 kfree(mdev->current_epoch); */
3272 kfree(mdev->app_reads_hash);
3276 drbd_bm_cleanup(mdev);
3278 __free_page(mdev->md_io_page);
3282 blk_cleanup_queue(q);
3284 free_cpumask_var(mdev->cpu_mask);
3290 /* counterpart of drbd_new_device.
3291 * last part of drbd_delete_device. */
3292 void drbd_free_mdev(struct drbd_conf *mdev)
3294 kfree(mdev->current_epoch);
3295 kfree(mdev->app_reads_hash);
3297 if (mdev->bitmap) /* should no longer be there. */
3298 drbd_bm_cleanup(mdev);
3299 __free_page(mdev->md_io_page);
3300 put_disk(mdev->vdisk);
3301 blk_cleanup_queue(mdev->rq_queue);
3302 free_cpumask_var(mdev->cpu_mask);
3307 int __init drbd_init(void)
3311 if (sizeof(struct p_handshake) != 80) {
3313 "drbd: never change the size or layout "
3314 "of the HandShake packet.\n");
3318 if (1 > minor_count || minor_count > 255) {
3320 "drbd: invalid minor_count (%d)\n", minor_count);
3328 err = drbd_nl_init();
3332 err = register_blkdev(DRBD_MAJOR, "drbd");
3335 "drbd: unable to register block device major %d\n",
3340 register_reboot_notifier(&drbd_notifier);
3343 * allocate all necessary structs
3347 init_waitqueue_head(&drbd_pp_wait);
3349 drbd_proc = NULL; /* play safe for drbd_cleanup */
3350 minor_table = kzalloc(sizeof(struct drbd_conf *)*minor_count,
3355 err = drbd_create_mempools();
3359 drbd_proc = proc_create_data("drbd", S_IFREG | S_IRUGO , NULL, &drbd_proc_fops, NULL);
3361 printk(KERN_ERR "drbd: unable to register proc file\n");
3365 rwlock_init(&global_state_lock);
3367 printk(KERN_INFO "drbd: initialized. "
3368 "Version: " REL_VERSION " (api:%d/proto:%d-%d)\n",
3369 API_VERSION, PRO_VERSION_MIN, PRO_VERSION_MAX);
3370 printk(KERN_INFO "drbd: %s\n", drbd_buildtag());
3371 printk(KERN_INFO "drbd: registered as block device major %d\n",
3373 printk(KERN_INFO "drbd: minor_table @ 0x%p\n", minor_table);
3375 return 0; /* Success! */
3380 /* currently always the case */
3381 printk(KERN_ERR "drbd: ran out of memory\n");
3383 printk(KERN_ERR "drbd: initialization failure\n");
3387 void drbd_free_bc(struct drbd_backing_dev *ldev)
3392 blkdev_put(ldev->backing_bdev, FMODE_READ | FMODE_WRITE | FMODE_EXCL);
3393 blkdev_put(ldev->md_bdev, FMODE_READ | FMODE_WRITE | FMODE_EXCL);
3398 void drbd_free_sock(struct drbd_conf *mdev)
3400 if (mdev->data.socket) {
3401 mutex_lock(&mdev->data.mutex);
3402 kernel_sock_shutdown(mdev->data.socket, SHUT_RDWR);
3403 sock_release(mdev->data.socket);
3404 mdev->data.socket = NULL;
3405 mutex_unlock(&mdev->data.mutex);
3407 if (mdev->meta.socket) {
3408 mutex_lock(&mdev->meta.mutex);
3409 kernel_sock_shutdown(mdev->meta.socket, SHUT_RDWR);
3410 sock_release(mdev->meta.socket);
3411 mdev->meta.socket = NULL;
3412 mutex_unlock(&mdev->meta.mutex);
3417 void drbd_free_resources(struct drbd_conf *mdev)
3419 crypto_free_hash(mdev->csums_tfm);
3420 mdev->csums_tfm = NULL;
3421 crypto_free_hash(mdev->verify_tfm);
3422 mdev->verify_tfm = NULL;
3423 crypto_free_hash(mdev->cram_hmac_tfm);
3424 mdev->cram_hmac_tfm = NULL;
3425 crypto_free_hash(mdev->integrity_w_tfm);
3426 mdev->integrity_w_tfm = NULL;
3427 crypto_free_hash(mdev->integrity_r_tfm);
3428 mdev->integrity_r_tfm = NULL;
3430 drbd_free_sock(mdev);
3433 drbd_free_bc(mdev->ldev);
3434 mdev->ldev = NULL;);
3437 /* meta data management */
3439 struct meta_data_on_disk {
3440 u64 la_size; /* last agreed size. */
3441 u64 uuid[UI_SIZE]; /* UUIDs. */
3444 u32 flags; /* MDF */
3447 u32 al_offset; /* offset to this block */
3448 u32 al_nr_extents; /* important for restoring the AL */
3449 /* `-- act_log->nr_elements <-- sync_conf.al_extents */
3450 u32 bm_offset; /* offset to the bitmap, from here */
3451 u32 bm_bytes_per_bit; /* BM_BLOCK_SIZE */
3452 u32 reserved_u32[4];
3457 * drbd_md_sync() - Writes the meta data super block if the MD_DIRTY flag bit is set
3458 * @mdev: DRBD device.
3460 void drbd_md_sync(struct drbd_conf *mdev)
3462 struct meta_data_on_disk *buffer;
3466 del_timer(&mdev->md_sync_timer);
3467 /* timer may be rearmed by drbd_md_mark_dirty() now. */
3468 if (!test_and_clear_bit(MD_DIRTY, &mdev->flags))
3471 /* We use here D_FAILED and not D_ATTACHING because we try to write
3472 * metadata even if we detach due to a disk failure! */
3473 if (!get_ldev_if_state(mdev, D_FAILED))
3476 mutex_lock(&mdev->md_io_mutex);
3477 buffer = (struct meta_data_on_disk *)page_address(mdev->md_io_page);
3478 memset(buffer, 0, 512);
3480 buffer->la_size = cpu_to_be64(drbd_get_capacity(mdev->this_bdev));
3481 for (i = UI_CURRENT; i < UI_SIZE; i++)
3482 buffer->uuid[i] = cpu_to_be64(mdev->ldev->md.uuid[i]);
3483 buffer->flags = cpu_to_be32(mdev->ldev->md.flags);
3484 buffer->magic = cpu_to_be32(DRBD_MD_MAGIC);
3486 buffer->md_size_sect = cpu_to_be32(mdev->ldev->md.md_size_sect);
3487 buffer->al_offset = cpu_to_be32(mdev->ldev->md.al_offset);
3488 buffer->al_nr_extents = cpu_to_be32(mdev->act_log->nr_elements);
3489 buffer->bm_bytes_per_bit = cpu_to_be32(BM_BLOCK_SIZE);
3490 buffer->device_uuid = cpu_to_be64(mdev->ldev->md.device_uuid);
3492 buffer->bm_offset = cpu_to_be32(mdev->ldev->md.bm_offset);
3494 D_ASSERT(drbd_md_ss__(mdev, mdev->ldev) == mdev->ldev->md.md_offset);
3495 sector = mdev->ldev->md.md_offset;
3497 if (!drbd_md_sync_page_io(mdev, mdev->ldev, sector, WRITE)) {
3498 /* this was a try anyways ... */
3499 dev_err(DEV, "meta data update failed!\n");
3500 drbd_chk_io_error(mdev, 1, TRUE);
3503 /* Update mdev->ldev->md.la_size_sect,
3504 * since we updated it on metadata. */
3505 mdev->ldev->md.la_size_sect = drbd_get_capacity(mdev->this_bdev);
3507 mutex_unlock(&mdev->md_io_mutex);
3512 * drbd_md_read() - Reads in the meta data super block
3513 * @mdev: DRBD device.
3514 * @bdev: Device from which the meta data should be read in.
3516 * Return 0 (NO_ERROR) on success, and an enum drbd_ret_codes in case
3517 * something goes wrong. Currently only: ERR_IO_MD_DISK, ERR_MD_INVALID.
3519 int drbd_md_read(struct drbd_conf *mdev, struct drbd_backing_dev *bdev)
3521 struct meta_data_on_disk *buffer;
3522 int i, rv = NO_ERROR;
3524 if (!get_ldev_if_state(mdev, D_ATTACHING))
3525 return ERR_IO_MD_DISK;
3527 mutex_lock(&mdev->md_io_mutex);
3528 buffer = (struct meta_data_on_disk *)page_address(mdev->md_io_page);
3530 if (!drbd_md_sync_page_io(mdev, bdev, bdev->md.md_offset, READ)) {
3531 /* NOTE: cant do normal error processing here as this is
3532 called BEFORE disk is attached */
3533 dev_err(DEV, "Error while reading metadata.\n");
3534 rv = ERR_IO_MD_DISK;
3538 if (be32_to_cpu(buffer->magic) != DRBD_MD_MAGIC) {
3539 dev_err(DEV, "Error while reading metadata, magic not found.\n");
3540 rv = ERR_MD_INVALID;
3543 if (be32_to_cpu(buffer->al_offset) != bdev->md.al_offset) {
3544 dev_err(DEV, "unexpected al_offset: %d (expected %d)\n",
3545 be32_to_cpu(buffer->al_offset), bdev->md.al_offset);
3546 rv = ERR_MD_INVALID;
3549 if (be32_to_cpu(buffer->bm_offset) != bdev->md.bm_offset) {
3550 dev_err(DEV, "unexpected bm_offset: %d (expected %d)\n",
3551 be32_to_cpu(buffer->bm_offset), bdev->md.bm_offset);
3552 rv = ERR_MD_INVALID;
3555 if (be32_to_cpu(buffer->md_size_sect) != bdev->md.md_size_sect) {
3556 dev_err(DEV, "unexpected md_size: %u (expected %u)\n",
3557 be32_to_cpu(buffer->md_size_sect), bdev->md.md_size_sect);
3558 rv = ERR_MD_INVALID;
3562 if (be32_to_cpu(buffer->bm_bytes_per_bit) != BM_BLOCK_SIZE) {
3563 dev_err(DEV, "unexpected bm_bytes_per_bit: %u (expected %u)\n",
3564 be32_to_cpu(buffer->bm_bytes_per_bit), BM_BLOCK_SIZE);
3565 rv = ERR_MD_INVALID;
3569 bdev->md.la_size_sect = be64_to_cpu(buffer->la_size);
3570 for (i = UI_CURRENT; i < UI_SIZE; i++)
3571 bdev->md.uuid[i] = be64_to_cpu(buffer->uuid[i]);
3572 bdev->md.flags = be32_to_cpu(buffer->flags);
3573 mdev->sync_conf.al_extents = be32_to_cpu(buffer->al_nr_extents);
3574 bdev->md.device_uuid = be64_to_cpu(buffer->device_uuid);
3576 if (mdev->sync_conf.al_extents < 7)
3577 mdev->sync_conf.al_extents = 127;
3580 mutex_unlock(&mdev->md_io_mutex);
3586 static void debug_drbd_uuid(struct drbd_conf *mdev, enum drbd_uuid_index index)
3588 static char *uuid_str[UI_EXTENDED_SIZE] = {
3589 [UI_CURRENT] = "CURRENT",
3590 [UI_BITMAP] = "BITMAP",
3591 [UI_HISTORY_START] = "HISTORY_START",
3592 [UI_HISTORY_END] = "HISTORY_END",
3594 [UI_FLAGS] = "FLAGS",
3597 if (index >= UI_EXTENDED_SIZE) {
3598 dev_warn(DEV, " uuid_index >= EXTENDED_SIZE\n");
3602 dynamic_dev_dbg(DEV, " uuid[%s] now %016llX\n",
3604 (unsigned long long)mdev->ldev->md.uuid[index]);
3609 * drbd_md_mark_dirty() - Mark meta data super block as dirty
3610 * @mdev: DRBD device.
3612 * Call this function if you change anything that should be written to
3613 * the meta-data super block. This function sets MD_DIRTY, and starts a
3614 * timer that ensures that within five seconds you have to call drbd_md_sync().
3617 void drbd_md_mark_dirty_(struct drbd_conf *mdev, unsigned int line, const char *func)
3619 if (!test_and_set_bit(MD_DIRTY, &mdev->flags)) {
3620 mod_timer(&mdev->md_sync_timer, jiffies + HZ);
3621 mdev->last_md_mark_dirty.line = line;
3622 mdev->last_md_mark_dirty.func = func;
3626 void drbd_md_mark_dirty(struct drbd_conf *mdev)
3628 if (!test_and_set_bit(MD_DIRTY, &mdev->flags))
3629 mod_timer(&mdev->md_sync_timer, jiffies + 5*HZ);
3633 static void drbd_uuid_move_history(struct drbd_conf *mdev) __must_hold(local)
3637 for (i = UI_HISTORY_START; i < UI_HISTORY_END; i++) {
3638 mdev->ldev->md.uuid[i+1] = mdev->ldev->md.uuid[i];
3639 debug_drbd_uuid(mdev, i+1);
3643 void _drbd_uuid_set(struct drbd_conf *mdev, int idx, u64 val) __must_hold(local)
3645 if (idx == UI_CURRENT) {
3646 if (mdev->state.role == R_PRIMARY)
3651 drbd_set_ed_uuid(mdev, val);
3654 mdev->ldev->md.uuid[idx] = val;
3655 debug_drbd_uuid(mdev, idx);
3656 drbd_md_mark_dirty(mdev);
3660 void drbd_uuid_set(struct drbd_conf *mdev, int idx, u64 val) __must_hold(local)
3662 if (mdev->ldev->md.uuid[idx]) {
3663 drbd_uuid_move_history(mdev);
3664 mdev->ldev->md.uuid[UI_HISTORY_START] = mdev->ldev->md.uuid[idx];
3665 debug_drbd_uuid(mdev, UI_HISTORY_START);
3667 _drbd_uuid_set(mdev, idx, val);
3671 * drbd_uuid_new_current() - Creates a new current UUID
3672 * @mdev: DRBD device.
3674 * Creates a new current UUID, and rotates the old current UUID into
3675 * the bitmap slot. Causes an incremental resync upon next connect.
3677 void drbd_uuid_new_current(struct drbd_conf *mdev) __must_hold(local)
3681 dev_info(DEV, "Creating new current UUID\n");
3682 D_ASSERT(mdev->ldev->md.uuid[UI_BITMAP] == 0);
3683 mdev->ldev->md.uuid[UI_BITMAP] = mdev->ldev->md.uuid[UI_CURRENT];
3684 debug_drbd_uuid(mdev, UI_BITMAP);
3686 get_random_bytes(&val, sizeof(u64));
3687 _drbd_uuid_set(mdev, UI_CURRENT, val);
3688 /* get it to stable storage _now_ */
3692 void drbd_uuid_set_bm(struct drbd_conf *mdev, u64 val) __must_hold(local)
3694 if (mdev->ldev->md.uuid[UI_BITMAP] == 0 && val == 0)
3698 drbd_uuid_move_history(mdev);
3699 mdev->ldev->md.uuid[UI_HISTORY_START] = mdev->ldev->md.uuid[UI_BITMAP];
3700 mdev->ldev->md.uuid[UI_BITMAP] = 0;
3701 debug_drbd_uuid(mdev, UI_HISTORY_START);
3702 debug_drbd_uuid(mdev, UI_BITMAP);
3704 if (mdev->ldev->md.uuid[UI_BITMAP])
3705 dev_warn(DEV, "bm UUID already set");
3707 mdev->ldev->md.uuid[UI_BITMAP] = val;
3708 mdev->ldev->md.uuid[UI_BITMAP] &= ~((u64)1);
3710 debug_drbd_uuid(mdev, UI_BITMAP);
3712 drbd_md_mark_dirty(mdev);
3716 * drbd_bmio_set_n_write() - io_fn for drbd_queue_bitmap_io() or drbd_bitmap_io()
3717 * @mdev: DRBD device.
3719 * Sets all bits in the bitmap and writes the whole bitmap to stable storage.
3721 int drbd_bmio_set_n_write(struct drbd_conf *mdev)
3725 if (get_ldev_if_state(mdev, D_ATTACHING)) {
3726 drbd_md_set_flag(mdev, MDF_FULL_SYNC);
3728 drbd_bm_set_all(mdev);
3730 rv = drbd_bm_write(mdev);
3733 drbd_md_clear_flag(mdev, MDF_FULL_SYNC);
3744 * drbd_bmio_clear_n_write() - io_fn for drbd_queue_bitmap_io() or drbd_bitmap_io()
3745 * @mdev: DRBD device.
3747 * Clears all bits in the bitmap and writes the whole bitmap to stable storage.
3749 int drbd_bmio_clear_n_write(struct drbd_conf *mdev)
3753 drbd_resume_al(mdev);
3754 if (get_ldev_if_state(mdev, D_ATTACHING)) {
3755 drbd_bm_clear_all(mdev);
3756 rv = drbd_bm_write(mdev);
3763 static int w_bitmap_io(struct drbd_conf *mdev, struct drbd_work *w, int unused)
3765 struct bm_io_work *work = container_of(w, struct bm_io_work, w);
3768 D_ASSERT(atomic_read(&mdev->ap_bio_cnt) == 0);
3770 drbd_bm_lock(mdev, work->why);
3771 rv = work->io_fn(mdev);
3772 drbd_bm_unlock(mdev);
3774 clear_bit(BITMAP_IO, &mdev->flags);
3775 smp_mb__after_clear_bit();
3776 wake_up(&mdev->misc_wait);
3779 work->done(mdev, rv);
3781 clear_bit(BITMAP_IO_QUEUED, &mdev->flags);
3787 void drbd_ldev_destroy(struct drbd_conf *mdev)
3789 lc_destroy(mdev->resync);
3790 mdev->resync = NULL;
3791 lc_destroy(mdev->act_log);
3792 mdev->act_log = NULL;
3794 drbd_free_bc(mdev->ldev);
3795 mdev->ldev = NULL;);
3797 if (mdev->md_io_tmpp) {
3798 __free_page(mdev->md_io_tmpp);
3799 mdev->md_io_tmpp = NULL;
3801 clear_bit(GO_DISKLESS, &mdev->flags);
3804 static int w_go_diskless(struct drbd_conf *mdev, struct drbd_work *w, int unused)
3806 D_ASSERT(mdev->state.disk == D_FAILED);
3807 /* we cannot assert local_cnt == 0 here, as get_ldev_if_state will
3808 * inc/dec it frequently. Once we are D_DISKLESS, no one will touch
3809 * the protected members anymore, though, so once put_ldev reaches zero
3810 * again, it will be safe to free them. */
3811 drbd_force_state(mdev, NS(disk, D_DISKLESS));
3815 void drbd_go_diskless(struct drbd_conf *mdev)
3817 D_ASSERT(mdev->state.disk == D_FAILED);
3818 if (!test_and_set_bit(GO_DISKLESS, &mdev->flags))
3819 drbd_queue_work(&mdev->data.work, &mdev->go_diskless);
3823 * drbd_queue_bitmap_io() - Queues an IO operation on the whole bitmap
3824 * @mdev: DRBD device.
3825 * @io_fn: IO callback to be called when bitmap IO is possible
3826 * @done: callback to be called after the bitmap IO was performed
3827 * @why: Descriptive text of the reason for doing the IO
3829 * While IO on the bitmap happens we freeze application IO thus we ensure
3830 * that drbd_set_out_of_sync() can not be called. This function MAY ONLY be
3831 * called from worker context. It MUST NOT be used while a previous such
3832 * work is still pending!
3834 void drbd_queue_bitmap_io(struct drbd_conf *mdev,
3835 int (*io_fn)(struct drbd_conf *),
3836 void (*done)(struct drbd_conf *, int),
3839 D_ASSERT(current == mdev->worker.task);
3841 D_ASSERT(!test_bit(BITMAP_IO_QUEUED, &mdev->flags));
3842 D_ASSERT(!test_bit(BITMAP_IO, &mdev->flags));
3843 D_ASSERT(list_empty(&mdev->bm_io_work.w.list));
3844 if (mdev->bm_io_work.why)
3845 dev_err(DEV, "FIXME going to queue '%s' but '%s' still pending?\n",
3846 why, mdev->bm_io_work.why);
3848 mdev->bm_io_work.io_fn = io_fn;
3849 mdev->bm_io_work.done = done;
3850 mdev->bm_io_work.why = why;
3852 set_bit(BITMAP_IO, &mdev->flags);
3853 if (atomic_read(&mdev->ap_bio_cnt) == 0) {
3854 if (!test_and_set_bit(BITMAP_IO_QUEUED, &mdev->flags))
3855 drbd_queue_work(&mdev->data.work, &mdev->bm_io_work.w);
3860 * drbd_bitmap_io() - Does an IO operation on the whole bitmap
3861 * @mdev: DRBD device.
3862 * @io_fn: IO callback to be called when bitmap IO is possible
3863 * @why: Descriptive text of the reason for doing the IO
3865 * freezes application IO while that the actual IO operations runs. This
3866 * functions MAY NOT be called from worker context.
3868 int drbd_bitmap_io(struct drbd_conf *mdev, int (*io_fn)(struct drbd_conf *), char *why)
3872 D_ASSERT(current != mdev->worker.task);
3874 drbd_suspend_io(mdev);
3876 drbd_bm_lock(mdev, why);
3878 drbd_bm_unlock(mdev);
3880 drbd_resume_io(mdev);
3885 void drbd_md_set_flag(struct drbd_conf *mdev, int flag) __must_hold(local)
3887 if ((mdev->ldev->md.flags & flag) != flag) {
3888 drbd_md_mark_dirty(mdev);
3889 mdev->ldev->md.flags |= flag;
3893 void drbd_md_clear_flag(struct drbd_conf *mdev, int flag) __must_hold(local)
3895 if ((mdev->ldev->md.flags & flag) != 0) {
3896 drbd_md_mark_dirty(mdev);
3897 mdev->ldev->md.flags &= ~flag;
3900 int drbd_md_test_flag(struct drbd_backing_dev *bdev, int flag)
3902 return (bdev->md.flags & flag) != 0;
3905 static void md_sync_timer_fn(unsigned long data)
3907 struct drbd_conf *mdev = (struct drbd_conf *) data;
3909 drbd_queue_work_front(&mdev->data.work, &mdev->md_sync_work);
3912 static int w_md_sync(struct drbd_conf *mdev, struct drbd_work *w, int unused)
3914 dev_warn(DEV, "md_sync_timer expired! Worker calls drbd_md_sync().\n");
3916 dev_warn(DEV, "last md_mark_dirty: %s:%u\n",
3917 mdev->last_md_mark_dirty.func, mdev->last_md_mark_dirty.line);
3923 #ifdef CONFIG_DRBD_FAULT_INJECTION
3924 /* Fault insertion support including random number generator shamelessly
3925 * stolen from kernel/rcutorture.c */
3926 struct fault_random_state {
3927 unsigned long state;
3928 unsigned long count;
3931 #define FAULT_RANDOM_MULT 39916801 /* prime */
3932 #define FAULT_RANDOM_ADD 479001701 /* prime */
3933 #define FAULT_RANDOM_REFRESH 10000
3936 * Crude but fast random-number generator. Uses a linear congruential
3937 * generator, with occasional help from get_random_bytes().
3939 static unsigned long
3940 _drbd_fault_random(struct fault_random_state *rsp)
3944 if (!rsp->count--) {
3945 get_random_bytes(&refresh, sizeof(refresh));
3946 rsp->state += refresh;
3947 rsp->count = FAULT_RANDOM_REFRESH;
3949 rsp->state = rsp->state * FAULT_RANDOM_MULT + FAULT_RANDOM_ADD;
3950 return swahw32(rsp->state);
3954 _drbd_fault_str(unsigned int type) {
3955 static char *_faults[] = {
3956 [DRBD_FAULT_MD_WR] = "Meta-data write",
3957 [DRBD_FAULT_MD_RD] = "Meta-data read",
3958 [DRBD_FAULT_RS_WR] = "Resync write",
3959 [DRBD_FAULT_RS_RD] = "Resync read",
3960 [DRBD_FAULT_DT_WR] = "Data write",
3961 [DRBD_FAULT_DT_RD] = "Data read",
3962 [DRBD_FAULT_DT_RA] = "Data read ahead",
3963 [DRBD_FAULT_BM_ALLOC] = "BM allocation",
3964 [DRBD_FAULT_AL_EE] = "EE allocation",
3965 [DRBD_FAULT_RECEIVE] = "receive data corruption",
3968 return (type < DRBD_FAULT_MAX) ? _faults[type] : "**Unknown**";
3972 _drbd_insert_fault(struct drbd_conf *mdev, unsigned int type)
3974 static struct fault_random_state rrs = {0, 0};
3976 unsigned int ret = (
3978 ((1 << mdev_to_minor(mdev)) & fault_devs) != 0) &&
3979 (((_drbd_fault_random(&rrs) % 100) + 1) <= fault_rate));
3984 if (__ratelimit(&drbd_ratelimit_state))
3985 dev_warn(DEV, "***Simulating %s failure\n",
3986 _drbd_fault_str(type));
3993 const char *drbd_buildtag(void)
3995 /* DRBD built from external sources has here a reference to the
3996 git hash of the source code. */
3998 static char buildtag[38] = "\0uilt-in";
4000 if (buildtag[0] == 0) {
4001 #ifdef CONFIG_MODULES
4002 if (THIS_MODULE != NULL)
4003 sprintf(buildtag, "srcversion: %-24s", THIS_MODULE->srcversion);
4012 module_init(drbd_init)
4013 module_exit(drbd_cleanup)
4015 EXPORT_SYMBOL(drbd_conn_str);
4016 EXPORT_SYMBOL(drbd_role_str);
4017 EXPORT_SYMBOL(drbd_disk_str);
4018 EXPORT_SYMBOL(drbd_set_st_err_str);