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 drbd is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 2, or (at your option)
15 drbd is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with drbd; see the file COPYING. If not, write to
22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
26 #include <linux/module.h>
27 #include <linux/drbd.h>
28 #include <linux/sched.h>
29 #include <linux/wait.h>
31 #include <linux/memcontrol.h>
32 #include <linux/mm_inline.h>
33 #include <linux/slab.h>
34 #include <linux/random.h>
35 #include <linux/string.h>
36 #include <linux/scatterlist.h>
41 static int w_make_ov_request(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
);
42 static int w_make_resync_request(struct drbd_conf
*mdev
,
43 struct drbd_work
*w
, int cancel
);
52 * more endio handlers:
53 atodb_endio in drbd_actlog.c
54 drbd_bm_async_io_complete in drbd_bitmap.c
56 * For all these callbacks, note the following:
57 * The callbacks will be called in irq context by the IDE drivers,
58 * and in Softirqs/Tasklets/BH context by the SCSI drivers.
59 * Try to get the locking right :)
64 /* About the global_state_lock
65 Each state transition on an device holds a read lock. In case we have
66 to evaluate the sync after dependencies, we grab a write lock, because
67 we need stable states on all devices for that. */
68 rwlock_t global_state_lock
;
70 /* used for synchronous meta data and bitmap IO
71 * submitted by drbd_md_sync_page_io()
73 void drbd_md_io_complete(struct bio
*bio
, int error
)
75 struct drbd_md_io
*md_io
;
77 md_io
= (struct drbd_md_io
*)bio
->bi_private
;
80 complete(&md_io
->event
);
83 /* reads on behalf of the partner,
84 * "submitted" by the receiver
86 void drbd_endio_read_sec_final(struct drbd_epoch_entry
*e
) __releases(local
)
88 unsigned long flags
= 0;
89 struct drbd_conf
*mdev
= e
->mdev
;
91 D_ASSERT(e
->block_id
!= ID_VACANT
);
93 spin_lock_irqsave(&mdev
->req_lock
, flags
);
94 mdev
->read_cnt
+= e
->size
>> 9;
96 if (list_empty(&mdev
->read_ee
))
97 wake_up(&mdev
->ee_wait
);
98 if (test_bit(__EE_WAS_ERROR
, &e
->flags
))
99 __drbd_chk_io_error(mdev
, false);
100 spin_unlock_irqrestore(&mdev
->req_lock
, flags
);
102 drbd_queue_work(&mdev
->data
.work
, &e
->w
);
106 /* writes on behalf of the partner, or resync writes,
107 * "submitted" by the receiver, final stage. */
108 static void drbd_endio_write_sec_final(struct drbd_epoch_entry
*e
) __releases(local
)
110 unsigned long flags
= 0;
111 struct drbd_conf
*mdev
= e
->mdev
;
115 int do_al_complete_io
;
117 D_ASSERT(e
->block_id
!= ID_VACANT
);
119 /* after we moved e to done_ee,
120 * we may no longer access it,
121 * it may be freed/reused already!
122 * (as soon as we release the req_lock) */
123 e_sector
= e
->sector
;
124 do_al_complete_io
= e
->flags
& EE_CALL_AL_COMPLETE_IO
;
125 is_syncer_req
= is_syncer_block_id(e
->block_id
);
127 spin_lock_irqsave(&mdev
->req_lock
, flags
);
128 mdev
->writ_cnt
+= e
->size
>> 9;
129 list_del(&e
->w
.list
); /* has been on active_ee or sync_ee */
130 list_add_tail(&e
->w
.list
, &mdev
->done_ee
);
132 /* No hlist_del_init(&e->colision) here, we did not send the Ack yet,
133 * neither did we wake possibly waiting conflicting requests.
134 * done from "drbd_process_done_ee" within the appropriate w.cb
135 * (e_end_block/e_end_resync_block) or from _drbd_clear_done_ee */
137 do_wake
= is_syncer_req
138 ? list_empty(&mdev
->sync_ee
)
139 : list_empty(&mdev
->active_ee
);
141 if (test_bit(__EE_WAS_ERROR
, &e
->flags
))
142 __drbd_chk_io_error(mdev
, false);
143 spin_unlock_irqrestore(&mdev
->req_lock
, flags
);
146 drbd_rs_complete_io(mdev
, e_sector
);
149 wake_up(&mdev
->ee_wait
);
151 if (do_al_complete_io
)
152 drbd_al_complete_io(mdev
, e_sector
);
158 /* writes on behalf of the partner, or resync writes,
159 * "submitted" by the receiver.
161 void drbd_endio_sec(struct bio
*bio
, int error
)
163 struct drbd_epoch_entry
*e
= bio
->bi_private
;
164 struct drbd_conf
*mdev
= e
->mdev
;
165 int uptodate
= bio_flagged(bio
, BIO_UPTODATE
);
166 int is_write
= bio_data_dir(bio
) == WRITE
;
168 if (error
&& __ratelimit(&drbd_ratelimit_state
))
169 dev_warn(DEV
, "%s: error=%d s=%llus\n",
170 is_write
? "write" : "read", error
,
171 (unsigned long long)e
->sector
);
172 if (!error
&& !uptodate
) {
173 if (__ratelimit(&drbd_ratelimit_state
))
174 dev_warn(DEV
, "%s: setting error to -EIO s=%llus\n",
175 is_write
? "write" : "read",
176 (unsigned long long)e
->sector
);
177 /* strange behavior of some lower level drivers...
178 * fail the request by clearing the uptodate flag,
179 * but do not return any error?! */
184 set_bit(__EE_WAS_ERROR
, &e
->flags
);
186 bio_put(bio
); /* no need for the bio anymore */
187 if (atomic_dec_and_test(&e
->pending_bios
)) {
189 drbd_endio_write_sec_final(e
);
191 drbd_endio_read_sec_final(e
);
195 /* read, readA or write requests on R_PRIMARY coming from drbd_make_request
197 void drbd_endio_pri(struct bio
*bio
, int error
)
200 struct drbd_request
*req
= bio
->bi_private
;
201 struct drbd_conf
*mdev
= req
->mdev
;
202 struct bio_and_error m
;
203 enum drbd_req_event what
;
204 int uptodate
= bio_flagged(bio
, BIO_UPTODATE
);
206 if (!error
&& !uptodate
) {
207 dev_warn(DEV
, "p %s: setting error to -EIO\n",
208 bio_data_dir(bio
) == WRITE
? "write" : "read");
209 /* strange behavior of some lower level drivers...
210 * fail the request by clearing the uptodate flag,
211 * but do not return any error?! */
215 /* to avoid recursion in __req_mod */
216 if (unlikely(error
)) {
217 what
= (bio_data_dir(bio
) == WRITE
)
218 ? write_completed_with_error
219 : (bio_rw(bio
) == READ
)
220 ? read_completed_with_error
221 : read_ahead_completed_with_error
;
225 bio_put(req
->private_bio
);
226 req
->private_bio
= ERR_PTR(error
);
228 /* not req_mod(), we need irqsave here! */
229 spin_lock_irqsave(&mdev
->req_lock
, flags
);
230 __req_mod(req
, what
, &m
);
231 spin_unlock_irqrestore(&mdev
->req_lock
, flags
);
234 complete_master_bio(mdev
, &m
);
237 int w_read_retry_remote(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
239 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
241 /* We should not detach for read io-error,
242 * but try to WRITE the P_DATA_REPLY to the failed location,
243 * to give the disk the chance to relocate that block */
245 spin_lock_irq(&mdev
->req_lock
);
246 if (cancel
|| mdev
->state
.pdsk
!= D_UP_TO_DATE
) {
247 _req_mod(req
, read_retry_remote_canceled
);
248 spin_unlock_irq(&mdev
->req_lock
);
251 spin_unlock_irq(&mdev
->req_lock
);
253 return w_send_read_req(mdev
, w
, 0);
256 void drbd_csum_ee(struct drbd_conf
*mdev
, struct crypto_hash
*tfm
, struct drbd_epoch_entry
*e
, void *digest
)
258 struct hash_desc desc
;
259 struct scatterlist sg
;
260 struct page
*page
= e
->pages
;
267 sg_init_table(&sg
, 1);
268 crypto_hash_init(&desc
);
270 while ((tmp
= page_chain_next(page
))) {
271 /* all but the last page will be fully used */
272 sg_set_page(&sg
, page
, PAGE_SIZE
, 0);
273 crypto_hash_update(&desc
, &sg
, sg
.length
);
276 /* and now the last, possibly only partially used page */
277 len
= e
->size
& (PAGE_SIZE
- 1);
278 sg_set_page(&sg
, page
, len
?: PAGE_SIZE
, 0);
279 crypto_hash_update(&desc
, &sg
, sg
.length
);
280 crypto_hash_final(&desc
, digest
);
283 void drbd_csum_bio(struct drbd_conf
*mdev
, struct crypto_hash
*tfm
, struct bio
*bio
, void *digest
)
285 struct hash_desc desc
;
286 struct scatterlist sg
;
287 struct bio_vec
*bvec
;
293 sg_init_table(&sg
, 1);
294 crypto_hash_init(&desc
);
296 __bio_for_each_segment(bvec
, bio
, i
, 0) {
297 sg_set_page(&sg
, bvec
->bv_page
, bvec
->bv_len
, bvec
->bv_offset
);
298 crypto_hash_update(&desc
, &sg
, sg
.length
);
300 crypto_hash_final(&desc
, digest
);
303 static int w_e_send_csum(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
305 struct drbd_epoch_entry
*e
= container_of(w
, struct drbd_epoch_entry
, w
);
310 D_ASSERT(e
->block_id
== DRBD_MAGIC
+ 0xbeef);
312 if (unlikely(cancel
)) {
313 drbd_free_ee(mdev
, e
);
317 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
318 digest_size
= crypto_hash_digestsize(mdev
->csums_tfm
);
319 digest
= kmalloc(digest_size
, GFP_NOIO
);
321 drbd_csum_ee(mdev
, mdev
->csums_tfm
, e
, digest
);
323 inc_rs_pending(mdev
);
324 ok
= drbd_send_drequest_csum(mdev
,
332 dev_err(DEV
, "kmalloc() of digest failed.\n");
338 drbd_free_ee(mdev
, e
);
341 dev_err(DEV
, "drbd_send_drequest(..., csum) failed\n");
345 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
347 static int read_for_csum(struct drbd_conf
*mdev
, sector_t sector
, int size
)
349 struct drbd_epoch_entry
*e
;
354 if (drbd_rs_should_slow_down(mdev
, sector
))
357 /* GFP_TRY, because if there is no memory available right now, this may
358 * be rescheduled for later. It is "only" background resync, after all. */
359 e
= drbd_alloc_ee(mdev
, DRBD_MAGIC
+0xbeef, sector
, size
, GFP_TRY
);
363 e
->w
.cb
= w_e_send_csum
;
364 spin_lock_irq(&mdev
->req_lock
);
365 list_add(&e
->w
.list
, &mdev
->read_ee
);
366 spin_unlock_irq(&mdev
->req_lock
);
368 atomic_add(size
>> 9, &mdev
->rs_sect_ev
);
369 if (drbd_submit_ee(mdev
, e
, READ
, DRBD_FAULT_RS_RD
) == 0)
372 /* drbd_submit_ee currently fails for one reason only:
373 * not being able to allocate enough bios.
374 * Is dropping the connection going to help? */
375 spin_lock_irq(&mdev
->req_lock
);
376 list_del(&e
->w
.list
);
377 spin_unlock_irq(&mdev
->req_lock
);
379 drbd_free_ee(mdev
, e
);
385 int w_resync_timer(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
387 switch (mdev
->state
.conn
) {
389 w_make_ov_request(mdev
, w
, cancel
);
392 w_make_resync_request(mdev
, w
, cancel
);
399 void resync_timer_fn(unsigned long data
)
401 struct drbd_conf
*mdev
= (struct drbd_conf
*) data
;
403 if (list_empty(&mdev
->resync_work
.list
))
404 drbd_queue_work(&mdev
->data
.work
, &mdev
->resync_work
);
407 static void fifo_set(struct fifo_buffer
*fb
, int value
)
411 for (i
= 0; i
< fb
->size
; i
++)
412 fb
->values
[i
] = value
;
415 static int fifo_push(struct fifo_buffer
*fb
, int value
)
419 ov
= fb
->values
[fb
->head_index
];
420 fb
->values
[fb
->head_index
++] = value
;
422 if (fb
->head_index
>= fb
->size
)
428 static void fifo_add_val(struct fifo_buffer
*fb
, int value
)
432 for (i
= 0; i
< fb
->size
; i
++)
433 fb
->values
[i
] += value
;
436 static int drbd_rs_controller(struct drbd_conf
*mdev
)
438 unsigned int sect_in
; /* Number of sectors that came in since the last turn */
439 unsigned int want
; /* The number of sectors we want in the proxy */
440 int req_sect
; /* Number of sectors to request in this turn */
441 int correction
; /* Number of sectors more we need in the proxy*/
442 int cps
; /* correction per invocation of drbd_rs_controller() */
443 int steps
; /* Number of time steps to plan ahead */
447 sect_in
= atomic_xchg(&mdev
->rs_sect_in
, 0); /* Number of sectors that came in */
448 mdev
->rs_in_flight
-= sect_in
;
450 spin_lock(&mdev
->peer_seq_lock
); /* get an atomic view on mdev->rs_plan_s */
452 steps
= mdev
->rs_plan_s
.size
; /* (mdev->sync_conf.c_plan_ahead * 10 * SLEEP_TIME) / HZ; */
454 if (mdev
->rs_in_flight
+ sect_in
== 0) { /* At start of resync */
455 want
= ((mdev
->sync_conf
.rate
* 2 * SLEEP_TIME
) / HZ
) * steps
;
456 } else { /* normal path */
457 want
= mdev
->sync_conf
.c_fill_target
? mdev
->sync_conf
.c_fill_target
:
458 sect_in
* mdev
->sync_conf
.c_delay_target
* HZ
/ (SLEEP_TIME
* 10);
461 correction
= want
- mdev
->rs_in_flight
- mdev
->rs_planed
;
464 cps
= correction
/ steps
;
465 fifo_add_val(&mdev
->rs_plan_s
, cps
);
466 mdev
->rs_planed
+= cps
* steps
;
468 /* What we do in this step */
469 curr_corr
= fifo_push(&mdev
->rs_plan_s
, 0);
470 spin_unlock(&mdev
->peer_seq_lock
);
471 mdev
->rs_planed
-= curr_corr
;
473 req_sect
= sect_in
+ curr_corr
;
477 max_sect
= (mdev
->sync_conf
.c_max_rate
* 2 * SLEEP_TIME
) / HZ
;
478 if (req_sect
> max_sect
)
482 dev_warn(DEV, "si=%u if=%d wa=%u co=%d st=%d cps=%d pl=%d cc=%d rs=%d\n",
483 sect_in, mdev->rs_in_flight, want, correction,
484 steps, cps, mdev->rs_planed, curr_corr, req_sect);
490 static int drbd_rs_number_requests(struct drbd_conf
*mdev
)
493 if (mdev
->rs_plan_s
.size
) { /* mdev->sync_conf.c_plan_ahead */
494 number
= drbd_rs_controller(mdev
) >> (BM_BLOCK_SHIFT
- 9);
495 mdev
->c_sync_rate
= number
* HZ
* (BM_BLOCK_SIZE
/ 1024) / SLEEP_TIME
;
497 mdev
->c_sync_rate
= mdev
->sync_conf
.rate
;
498 number
= SLEEP_TIME
* mdev
->c_sync_rate
/ ((BM_BLOCK_SIZE
/ 1024) * HZ
);
501 /* ignore the amount of pending requests, the resync controller should
502 * throttle down to incoming reply rate soon enough anyways. */
506 static int w_make_resync_request(struct drbd_conf
*mdev
,
507 struct drbd_work
*w
, int cancel
)
511 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
513 int number
, rollback_i
, size
;
514 int align
, queued
, sndbuf
;
517 if (unlikely(cancel
))
520 if (mdev
->rs_total
== 0) {
522 drbd_resync_finished(mdev
);
526 if (!get_ldev(mdev
)) {
527 /* Since we only need to access mdev->rsync a
528 get_ldev_if_state(mdev,D_FAILED) would be sufficient, but
529 to continue resync with a broken disk makes no sense at
531 dev_err(DEV
, "Disk broke down during resync!\n");
535 /* starting with drbd 8.3.8, we can handle multi-bio EEs,
536 * if it should be necessary */
538 mdev
->agreed_pro_version
< 94 ? queue_max_hw_sectors(mdev
->rq_queue
) << 9 :
539 mdev
->agreed_pro_version
< 95 ? DRBD_MAX_SIZE_H80_PACKET
: DRBD_MAX_BIO_SIZE
;
541 number
= drbd_rs_number_requests(mdev
);
545 for (i
= 0; i
< number
; i
++) {
546 /* Stop generating RS requests, when half of the send buffer is filled */
547 mutex_lock(&mdev
->data
.mutex
);
548 if (mdev
->data
.socket
) {
549 queued
= mdev
->data
.socket
->sk
->sk_wmem_queued
;
550 sndbuf
= mdev
->data
.socket
->sk
->sk_sndbuf
;
555 mutex_unlock(&mdev
->data
.mutex
);
556 if (queued
> sndbuf
/ 2)
560 size
= BM_BLOCK_SIZE
;
561 bit
= drbd_bm_find_next(mdev
, mdev
->bm_resync_fo
);
563 if (bit
== DRBD_END_OF_BITMAP
) {
564 mdev
->bm_resync_fo
= drbd_bm_bits(mdev
);
569 sector
= BM_BIT_TO_SECT(bit
);
571 if (drbd_rs_should_slow_down(mdev
, sector
) ||
572 drbd_try_rs_begin_io(mdev
, sector
)) {
573 mdev
->bm_resync_fo
= bit
;
576 mdev
->bm_resync_fo
= bit
+ 1;
578 if (unlikely(drbd_bm_test_bit(mdev
, bit
) == 0)) {
579 drbd_rs_complete_io(mdev
, sector
);
583 #if DRBD_MAX_BIO_SIZE > BM_BLOCK_SIZE
584 /* try to find some adjacent bits.
585 * we stop if we have already the maximum req size.
587 * Additionally always align bigger requests, in order to
588 * be prepared for all stripe sizes of software RAIDs.
593 if (size
+ BM_BLOCK_SIZE
> max_bio_size
)
596 /* Be always aligned */
597 if (sector
& ((1<<(align
+3))-1))
600 /* do not cross extent boundaries */
601 if (((bit
+1) & BM_BLOCKS_PER_BM_EXT_MASK
) == 0)
603 /* now, is it actually dirty, after all?
604 * caution, drbd_bm_test_bit is tri-state for some
605 * obscure reason; ( b == 0 ) would get the out-of-band
606 * only accidentally right because of the "oddly sized"
607 * adjustment below */
608 if (drbd_bm_test_bit(mdev
, bit
+1) != 1)
611 size
+= BM_BLOCK_SIZE
;
612 if ((BM_BLOCK_SIZE
<< align
) <= size
)
616 /* if we merged some,
617 * reset the offset to start the next drbd_bm_find_next from */
618 if (size
> BM_BLOCK_SIZE
)
619 mdev
->bm_resync_fo
= bit
+ 1;
622 /* adjust very last sectors, in case we are oddly sized */
623 if (sector
+ (size
>>9) > capacity
)
624 size
= (capacity
-sector
)<<9;
625 if (mdev
->agreed_pro_version
>= 89 && mdev
->csums_tfm
) {
626 switch (read_for_csum(mdev
, sector
, size
)) {
627 case -EIO
: /* Disk failure */
630 case -EAGAIN
: /* allocation failed, or ldev busy */
631 drbd_rs_complete_io(mdev
, sector
);
632 mdev
->bm_resync_fo
= BM_SECT_TO_BIT(sector
);
642 inc_rs_pending(mdev
);
643 if (!drbd_send_drequest(mdev
, P_RS_DATA_REQUEST
,
644 sector
, size
, ID_SYNCER
)) {
645 dev_err(DEV
, "drbd_send_drequest() failed, aborting...\n");
646 dec_rs_pending(mdev
);
653 if (mdev
->bm_resync_fo
>= drbd_bm_bits(mdev
)) {
654 /* last syncer _request_ was sent,
655 * but the P_RS_DATA_REPLY not yet received. sync will end (and
656 * next sync group will resume), as soon as we receive the last
657 * resync data block, and the last bit is cleared.
658 * until then resync "work" is "inactive" ...
665 mdev
->rs_in_flight
+= (i
<< (BM_BLOCK_SHIFT
- 9));
666 mod_timer(&mdev
->resync_timer
, jiffies
+ SLEEP_TIME
);
671 static int w_make_ov_request(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
675 const sector_t capacity
= drbd_get_capacity(mdev
->this_bdev
);
677 if (unlikely(cancel
))
680 number
= drbd_rs_number_requests(mdev
);
682 sector
= mdev
->ov_position
;
683 for (i
= 0; i
< number
; i
++) {
684 if (sector
>= capacity
) {
688 size
= BM_BLOCK_SIZE
;
690 if (drbd_rs_should_slow_down(mdev
, sector
) ||
691 drbd_try_rs_begin_io(mdev
, sector
)) {
692 mdev
->ov_position
= sector
;
696 if (sector
+ (size
>>9) > capacity
)
697 size
= (capacity
-sector
)<<9;
699 inc_rs_pending(mdev
);
700 if (!drbd_send_ov_request(mdev
, sector
, size
)) {
701 dec_rs_pending(mdev
);
704 sector
+= BM_SECT_PER_BIT
;
706 mdev
->ov_position
= sector
;
709 mdev
->rs_in_flight
+= (i
<< (BM_BLOCK_SHIFT
- 9));
710 mod_timer(&mdev
->resync_timer
, jiffies
+ SLEEP_TIME
);
715 void start_resync_timer_fn(unsigned long data
)
717 struct drbd_conf
*mdev
= (struct drbd_conf
*) data
;
719 drbd_queue_work(&mdev
->data
.work
, &mdev
->start_resync_work
);
722 int w_start_resync(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
724 if (atomic_read(&mdev
->unacked_cnt
) || atomic_read(&mdev
->rs_pending_cnt
)) {
725 dev_warn(DEV
, "w_start_resync later...\n");
726 mdev
->start_resync_timer
.expires
= jiffies
+ HZ
/10;
727 add_timer(&mdev
->start_resync_timer
);
731 drbd_start_resync(mdev
, C_SYNC_SOURCE
);
732 clear_bit(AHEAD_TO_SYNC_SOURCE
, &mdev
->current_epoch
->flags
);
736 int w_ov_finished(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
740 drbd_resync_finished(mdev
);
745 static int w_resync_finished(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
749 drbd_resync_finished(mdev
);
754 static void ping_peer(struct drbd_conf
*mdev
)
756 clear_bit(GOT_PING_ACK
, &mdev
->flags
);
758 wait_event(mdev
->misc_wait
,
759 test_bit(GOT_PING_ACK
, &mdev
->flags
) || mdev
->state
.conn
< C_CONNECTED
);
762 int drbd_resync_finished(struct drbd_conf
*mdev
)
764 unsigned long db
, dt
, dbdt
;
766 union drbd_state os
, ns
;
768 char *khelper_cmd
= NULL
;
771 /* Remove all elements from the resync LRU. Since future actions
772 * might set bits in the (main) bitmap, then the entries in the
773 * resync LRU would be wrong. */
774 if (drbd_rs_del_all(mdev
)) {
775 /* In case this is not possible now, most probably because
776 * there are P_RS_DATA_REPLY Packets lingering on the worker's
777 * queue (or even the read operations for those packets
778 * is not finished by now). Retry in 100ms. */
780 schedule_timeout_interruptible(HZ
/ 10);
781 w
= kmalloc(sizeof(struct drbd_work
), GFP_ATOMIC
);
783 w
->cb
= w_resync_finished
;
784 drbd_queue_work(&mdev
->data
.work
, w
);
787 dev_err(DEV
, "Warn failed to drbd_rs_del_all() and to kmalloc(w).\n");
790 dt
= (jiffies
- mdev
->rs_start
- mdev
->rs_paused
) / HZ
;
794 dbdt
= Bit2KB(db
/dt
);
795 mdev
->rs_paused
/= HZ
;
802 spin_lock_irq(&mdev
->req_lock
);
805 verify_done
= (os
.conn
== C_VERIFY_S
|| os
.conn
== C_VERIFY_T
);
807 /* This protects us against multiple calls (that can happen in the presence
808 of application IO), and against connectivity loss just before we arrive here. */
809 if (os
.conn
<= C_CONNECTED
)
813 ns
.conn
= C_CONNECTED
;
815 dev_info(DEV
, "%s done (total %lu sec; paused %lu sec; %lu K/sec)\n",
816 verify_done
? "Online verify " : "Resync",
817 dt
+ mdev
->rs_paused
, mdev
->rs_paused
, dbdt
);
819 n_oos
= drbd_bm_total_weight(mdev
);
821 if (os
.conn
== C_VERIFY_S
|| os
.conn
== C_VERIFY_T
) {
823 dev_alert(DEV
, "Online verify found %lu %dk block out of sync!\n",
825 khelper_cmd
= "out-of-sync";
828 D_ASSERT((n_oos
- mdev
->rs_failed
) == 0);
830 if (os
.conn
== C_SYNC_TARGET
|| os
.conn
== C_PAUSED_SYNC_T
)
831 khelper_cmd
= "after-resync-target";
833 if (mdev
->csums_tfm
&& mdev
->rs_total
) {
834 const unsigned long s
= mdev
->rs_same_csum
;
835 const unsigned long t
= mdev
->rs_total
;
838 (t
< 100000) ? ((s
*100)/t
) : (s
/(t
/100));
839 dev_info(DEV
, "%u %% had equal check sums, eliminated: %luK; "
840 "transferred %luK total %luK\n",
842 Bit2KB(mdev
->rs_same_csum
),
843 Bit2KB(mdev
->rs_total
- mdev
->rs_same_csum
),
844 Bit2KB(mdev
->rs_total
));
848 if (mdev
->rs_failed
) {
849 dev_info(DEV
, " %lu failed blocks\n", mdev
->rs_failed
);
851 if (os
.conn
== C_SYNC_TARGET
|| os
.conn
== C_PAUSED_SYNC_T
) {
852 ns
.disk
= D_INCONSISTENT
;
853 ns
.pdsk
= D_UP_TO_DATE
;
855 ns
.disk
= D_UP_TO_DATE
;
856 ns
.pdsk
= D_INCONSISTENT
;
859 ns
.disk
= D_UP_TO_DATE
;
860 ns
.pdsk
= D_UP_TO_DATE
;
862 if (os
.conn
== C_SYNC_TARGET
|| os
.conn
== C_PAUSED_SYNC_T
) {
865 for (i
= UI_BITMAP
; i
<= UI_HISTORY_END
; i
++)
866 _drbd_uuid_set(mdev
, i
, mdev
->p_uuid
[i
]);
867 drbd_uuid_set(mdev
, UI_BITMAP
, mdev
->ldev
->md
.uuid
[UI_CURRENT
]);
868 _drbd_uuid_set(mdev
, UI_CURRENT
, mdev
->p_uuid
[UI_CURRENT
]);
870 dev_err(DEV
, "mdev->p_uuid is NULL! BUG\n");
874 drbd_uuid_set_bm(mdev
, 0UL);
877 /* Now the two UUID sets are equal, update what we
878 * know of the peer. */
880 for (i
= UI_CURRENT
; i
<= UI_HISTORY_END
; i
++)
881 mdev
->p_uuid
[i
] = mdev
->ldev
->md
.uuid
[i
];
885 _drbd_set_state(mdev
, ns
, CS_VERBOSE
, NULL
);
887 spin_unlock_irq(&mdev
->req_lock
);
894 mdev
->ov_start_sector
= 0;
899 drbd_khelper(mdev
, khelper_cmd
);
905 static void move_to_net_ee_or_free(struct drbd_conf
*mdev
, struct drbd_epoch_entry
*e
)
907 if (drbd_ee_has_active_page(e
)) {
908 /* This might happen if sendpage() has not finished */
909 int i
= (e
->size
+ PAGE_SIZE
-1) >> PAGE_SHIFT
;
910 atomic_add(i
, &mdev
->pp_in_use_by_net
);
911 atomic_sub(i
, &mdev
->pp_in_use
);
912 spin_lock_irq(&mdev
->req_lock
);
913 list_add_tail(&e
->w
.list
, &mdev
->net_ee
);
914 spin_unlock_irq(&mdev
->req_lock
);
915 wake_up(&drbd_pp_wait
);
917 drbd_free_ee(mdev
, e
);
921 * w_e_end_data_req() - Worker callback, to send a P_DATA_REPLY packet in response to a P_DATA_REQUEST
922 * @mdev: DRBD device.
924 * @cancel: The connection will be closed anyways
926 int w_e_end_data_req(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
928 struct drbd_epoch_entry
*e
= container_of(w
, struct drbd_epoch_entry
, w
);
931 if (unlikely(cancel
)) {
932 drbd_free_ee(mdev
, e
);
937 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
938 ok
= drbd_send_block(mdev
, P_DATA_REPLY
, e
);
940 if (__ratelimit(&drbd_ratelimit_state
))
941 dev_err(DEV
, "Sending NegDReply. sector=%llus.\n",
942 (unsigned long long)e
->sector
);
944 ok
= drbd_send_ack(mdev
, P_NEG_DREPLY
, e
);
949 move_to_net_ee_or_free(mdev
, e
);
952 dev_err(DEV
, "drbd_send_block() failed\n");
957 * w_e_end_rsdata_req() - Worker callback to send a P_RS_DATA_REPLY packet in response to a P_RS_DATA_REQUESTRS
958 * @mdev: DRBD device.
960 * @cancel: The connection will be closed anyways
962 int w_e_end_rsdata_req(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
964 struct drbd_epoch_entry
*e
= container_of(w
, struct drbd_epoch_entry
, w
);
967 if (unlikely(cancel
)) {
968 drbd_free_ee(mdev
, e
);
973 if (get_ldev_if_state(mdev
, D_FAILED
)) {
974 drbd_rs_complete_io(mdev
, e
->sector
);
978 if (mdev
->state
.conn
== C_AHEAD
) {
979 ok
= drbd_send_ack(mdev
, P_RS_CANCEL
, e
);
980 } else if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
981 if (likely(mdev
->state
.pdsk
>= D_INCONSISTENT
)) {
982 inc_rs_pending(mdev
);
983 ok
= drbd_send_block(mdev
, P_RS_DATA_REPLY
, e
);
985 if (__ratelimit(&drbd_ratelimit_state
))
986 dev_err(DEV
, "Not sending RSDataReply, "
987 "partner DISKLESS!\n");
991 if (__ratelimit(&drbd_ratelimit_state
))
992 dev_err(DEV
, "Sending NegRSDReply. sector %llus.\n",
993 (unsigned long long)e
->sector
);
995 ok
= drbd_send_ack(mdev
, P_NEG_RS_DREPLY
, e
);
997 /* update resync data with failure */
998 drbd_rs_failed_io(mdev
, e
->sector
, e
->size
);
1003 move_to_net_ee_or_free(mdev
, e
);
1006 dev_err(DEV
, "drbd_send_block() failed\n");
1010 int w_e_end_csum_rs_req(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1012 struct drbd_epoch_entry
*e
= container_of(w
, struct drbd_epoch_entry
, w
);
1013 struct digest_info
*di
;
1015 void *digest
= NULL
;
1018 if (unlikely(cancel
)) {
1019 drbd_free_ee(mdev
, e
);
1024 if (get_ldev(mdev
)) {
1025 drbd_rs_complete_io(mdev
, e
->sector
);
1031 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
1032 /* quick hack to try to avoid a race against reconfiguration.
1033 * a real fix would be much more involved,
1034 * introducing more locking mechanisms */
1035 if (mdev
->csums_tfm
) {
1036 digest_size
= crypto_hash_digestsize(mdev
->csums_tfm
);
1037 D_ASSERT(digest_size
== di
->digest_size
);
1038 digest
= kmalloc(digest_size
, GFP_NOIO
);
1041 drbd_csum_ee(mdev
, mdev
->csums_tfm
, e
, digest
);
1042 eq
= !memcmp(digest
, di
->digest
, digest_size
);
1047 drbd_set_in_sync(mdev
, e
->sector
, e
->size
);
1048 /* rs_same_csums unit is BM_BLOCK_SIZE */
1049 mdev
->rs_same_csum
+= e
->size
>> BM_BLOCK_SHIFT
;
1050 ok
= drbd_send_ack(mdev
, P_RS_IS_IN_SYNC
, e
);
1052 inc_rs_pending(mdev
);
1053 e
->block_id
= ID_SYNCER
; /* By setting block_id, digest pointer becomes invalid! */
1054 e
->flags
&= ~EE_HAS_DIGEST
; /* This e no longer has a digest pointer */
1056 ok
= drbd_send_block(mdev
, P_RS_DATA_REPLY
, e
);
1059 ok
= drbd_send_ack(mdev
, P_NEG_RS_DREPLY
, e
);
1060 if (__ratelimit(&drbd_ratelimit_state
))
1061 dev_err(DEV
, "Sending NegDReply. I guess it gets messy.\n");
1065 move_to_net_ee_or_free(mdev
, e
);
1068 dev_err(DEV
, "drbd_send_block/ack() failed\n");
1072 int w_e_end_ov_req(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1074 struct drbd_epoch_entry
*e
= container_of(w
, struct drbd_epoch_entry
, w
);
1079 if (unlikely(cancel
))
1082 if (unlikely((e
->flags
& EE_WAS_ERROR
) != 0))
1085 digest_size
= crypto_hash_digestsize(mdev
->verify_tfm
);
1086 /* FIXME if this allocation fails, online verify will not terminate! */
1087 digest
= kmalloc(digest_size
, GFP_NOIO
);
1089 drbd_csum_ee(mdev
, mdev
->verify_tfm
, e
, digest
);
1090 inc_rs_pending(mdev
);
1091 ok
= drbd_send_drequest_csum(mdev
, e
->sector
, e
->size
,
1092 digest
, digest_size
, P_OV_REPLY
);
1094 dec_rs_pending(mdev
);
1099 drbd_free_ee(mdev
, e
);
1106 void drbd_ov_oos_found(struct drbd_conf
*mdev
, sector_t sector
, int size
)
1108 if (mdev
->ov_last_oos_start
+ mdev
->ov_last_oos_size
== sector
) {
1109 mdev
->ov_last_oos_size
+= size
>>9;
1111 mdev
->ov_last_oos_start
= sector
;
1112 mdev
->ov_last_oos_size
= size
>>9;
1114 drbd_set_out_of_sync(mdev
, sector
, size
);
1117 int w_e_end_ov_reply(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1119 struct drbd_epoch_entry
*e
= container_of(w
, struct drbd_epoch_entry
, w
);
1120 struct digest_info
*di
;
1125 if (unlikely(cancel
)) {
1126 drbd_free_ee(mdev
, e
);
1131 /* after "cancel", because after drbd_disconnect/drbd_rs_cancel_all
1132 * the resync lru has been cleaned up already */
1133 if (get_ldev(mdev
)) {
1134 drbd_rs_complete_io(mdev
, e
->sector
);
1140 if (likely((e
->flags
& EE_WAS_ERROR
) == 0)) {
1141 digest_size
= crypto_hash_digestsize(mdev
->verify_tfm
);
1142 digest
= kmalloc(digest_size
, GFP_NOIO
);
1144 drbd_csum_ee(mdev
, mdev
->verify_tfm
, e
, digest
);
1146 D_ASSERT(digest_size
== di
->digest_size
);
1147 eq
= !memcmp(digest
, di
->digest
, digest_size
);
1151 ok
= drbd_send_ack(mdev
, P_NEG_RS_DREPLY
, e
);
1152 if (__ratelimit(&drbd_ratelimit_state
))
1153 dev_err(DEV
, "Sending NegDReply. I guess it gets messy.\n");
1158 drbd_ov_oos_found(mdev
, e
->sector
, e
->size
);
1162 ok
= drbd_send_ack_ex(mdev
, P_OV_RESULT
, e
->sector
, e
->size
,
1163 eq
? ID_IN_SYNC
: ID_OUT_OF_SYNC
);
1165 drbd_free_ee(mdev
, e
);
1169 /* let's advance progress step marks only for every other megabyte */
1170 if ((mdev
->ov_left
& 0x200) == 0x200)
1171 drbd_advance_rs_marks(mdev
, mdev
->ov_left
);
1173 if (mdev
->ov_left
== 0) {
1175 drbd_resync_finished(mdev
);
1181 int w_prev_work_done(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1183 struct drbd_wq_barrier
*b
= container_of(w
, struct drbd_wq_barrier
, w
);
1188 int w_send_barrier(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1190 struct drbd_tl_epoch
*b
= container_of(w
, struct drbd_tl_epoch
, w
);
1191 struct p_barrier
*p
= &mdev
->data
.sbuf
.barrier
;
1194 /* really avoid racing with tl_clear. w.cb may have been referenced
1195 * just before it was reassigned and re-queued, so double check that.
1196 * actually, this race was harmless, since we only try to send the
1197 * barrier packet here, and otherwise do nothing with the object.
1198 * but compare with the head of w_clear_epoch */
1199 spin_lock_irq(&mdev
->req_lock
);
1200 if (w
->cb
!= w_send_barrier
|| mdev
->state
.conn
< C_CONNECTED
)
1202 spin_unlock_irq(&mdev
->req_lock
);
1206 if (!drbd_get_data_sock(mdev
))
1208 p
->barrier
= b
->br_number
;
1209 /* inc_ap_pending was done where this was queued.
1210 * dec_ap_pending will be done in got_BarrierAck
1211 * or (on connection loss) in w_clear_epoch. */
1212 ok
= _drbd_send_cmd(mdev
, mdev
->data
.socket
, P_BARRIER
,
1213 (struct p_header80
*)p
, sizeof(*p
), 0);
1214 drbd_put_data_sock(mdev
);
1219 int w_send_write_hint(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1223 return drbd_send_short_cmd(mdev
, P_UNPLUG_REMOTE
);
1226 int w_send_oos(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1228 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
1231 if (unlikely(cancel
)) {
1232 req_mod(req
, send_canceled
);
1236 ok
= drbd_send_oos(mdev
, req
);
1237 req_mod(req
, oos_handed_to_network
);
1243 * w_send_dblock() - Worker callback to send a P_DATA packet in order to mirror a write request
1244 * @mdev: DRBD device.
1246 * @cancel: The connection will be closed anyways
1248 int w_send_dblock(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1250 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
1253 if (unlikely(cancel
)) {
1254 req_mod(req
, send_canceled
);
1258 ok
= drbd_send_dblock(mdev
, req
);
1259 req_mod(req
, ok
? handed_over_to_network
: send_failed
);
1265 * w_send_read_req() - Worker callback to send a read request (P_DATA_REQUEST) packet
1266 * @mdev: DRBD device.
1268 * @cancel: The connection will be closed anyways
1270 int w_send_read_req(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1272 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
1275 if (unlikely(cancel
)) {
1276 req_mod(req
, send_canceled
);
1280 ok
= drbd_send_drequest(mdev
, P_DATA_REQUEST
, req
->sector
, req
->size
,
1281 (unsigned long)req
);
1284 /* ?? we set C_TIMEOUT or C_BROKEN_PIPE in drbd_send();
1285 * so this is probably redundant */
1286 if (mdev
->state
.conn
>= C_CONNECTED
)
1287 drbd_force_state(mdev
, NS(conn
, C_NETWORK_FAILURE
));
1289 req_mod(req
, ok
? handed_over_to_network
: send_failed
);
1294 int w_restart_disk_io(struct drbd_conf
*mdev
, struct drbd_work
*w
, int cancel
)
1296 struct drbd_request
*req
= container_of(w
, struct drbd_request
, w
);
1298 if (bio_data_dir(req
->master_bio
) == WRITE
&& req
->rq_state
& RQ_IN_ACT_LOG
)
1299 drbd_al_begin_io(mdev
, req
->sector
);
1300 /* Calling drbd_al_begin_io() out of the worker might deadlocks
1301 theoretically. Practically it can not deadlock, since this is
1302 only used when unfreezing IOs. All the extents of the requests
1303 that made it into the TL are already active */
1305 drbd_req_make_private_bio(req
, req
->master_bio
);
1306 req
->private_bio
->bi_bdev
= mdev
->ldev
->backing_bdev
;
1307 generic_make_request(req
->private_bio
);
1312 static int _drbd_may_sync_now(struct drbd_conf
*mdev
)
1314 struct drbd_conf
*odev
= mdev
;
1317 if (odev
->sync_conf
.after
== -1)
1319 odev
= minor_to_mdev(odev
->sync_conf
.after
);
1320 ERR_IF(!odev
) return 1;
1321 if ((odev
->state
.conn
>= C_SYNC_SOURCE
&&
1322 odev
->state
.conn
<= C_PAUSED_SYNC_T
) ||
1323 odev
->state
.aftr_isp
|| odev
->state
.peer_isp
||
1324 odev
->state
.user_isp
)
1330 * _drbd_pause_after() - Pause resync on all devices that may not resync now
1331 * @mdev: DRBD device.
1333 * Called from process context only (admin command and after_state_ch).
1335 static int _drbd_pause_after(struct drbd_conf
*mdev
)
1337 struct drbd_conf
*odev
;
1340 for (i
= 0; i
< minor_count
; i
++) {
1341 odev
= minor_to_mdev(i
);
1344 if (odev
->state
.conn
== C_STANDALONE
&& odev
->state
.disk
== D_DISKLESS
)
1346 if (!_drbd_may_sync_now(odev
))
1347 rv
|= (__drbd_set_state(_NS(odev
, aftr_isp
, 1), CS_HARD
, NULL
)
1348 != SS_NOTHING_TO_DO
);
1355 * _drbd_resume_next() - Resume resync on all devices that may resync now
1356 * @mdev: DRBD device.
1358 * Called from process context only (admin command and worker).
1360 static int _drbd_resume_next(struct drbd_conf
*mdev
)
1362 struct drbd_conf
*odev
;
1365 for (i
= 0; i
< minor_count
; i
++) {
1366 odev
= minor_to_mdev(i
);
1369 if (odev
->state
.conn
== C_STANDALONE
&& odev
->state
.disk
== D_DISKLESS
)
1371 if (odev
->state
.aftr_isp
) {
1372 if (_drbd_may_sync_now(odev
))
1373 rv
|= (__drbd_set_state(_NS(odev
, aftr_isp
, 0),
1375 != SS_NOTHING_TO_DO
) ;
1381 void resume_next_sg(struct drbd_conf
*mdev
)
1383 write_lock_irq(&global_state_lock
);
1384 _drbd_resume_next(mdev
);
1385 write_unlock_irq(&global_state_lock
);
1388 void suspend_other_sg(struct drbd_conf
*mdev
)
1390 write_lock_irq(&global_state_lock
);
1391 _drbd_pause_after(mdev
);
1392 write_unlock_irq(&global_state_lock
);
1395 static int sync_after_error(struct drbd_conf
*mdev
, int o_minor
)
1397 struct drbd_conf
*odev
;
1401 if (o_minor
< -1 || minor_to_mdev(o_minor
) == NULL
)
1402 return ERR_SYNC_AFTER
;
1404 /* check for loops */
1405 odev
= minor_to_mdev(o_minor
);
1408 return ERR_SYNC_AFTER_CYCLE
;
1410 /* dependency chain ends here, no cycles. */
1411 if (odev
->sync_conf
.after
== -1)
1414 /* follow the dependency chain */
1415 odev
= minor_to_mdev(odev
->sync_conf
.after
);
1419 int drbd_alter_sa(struct drbd_conf
*mdev
, int na
)
1424 write_lock_irq(&global_state_lock
);
1425 retcode
= sync_after_error(mdev
, na
);
1426 if (retcode
== NO_ERROR
) {
1427 mdev
->sync_conf
.after
= na
;
1429 changes
= _drbd_pause_after(mdev
);
1430 changes
|= _drbd_resume_next(mdev
);
1433 write_unlock_irq(&global_state_lock
);
1437 void drbd_rs_controller_reset(struct drbd_conf
*mdev
)
1439 atomic_set(&mdev
->rs_sect_in
, 0);
1440 atomic_set(&mdev
->rs_sect_ev
, 0);
1441 mdev
->rs_in_flight
= 0;
1442 mdev
->rs_planed
= 0;
1443 spin_lock(&mdev
->peer_seq_lock
);
1444 fifo_set(&mdev
->rs_plan_s
, 0);
1445 spin_unlock(&mdev
->peer_seq_lock
);
1449 * drbd_start_resync() - Start the resync process
1450 * @mdev: DRBD device.
1451 * @side: Either C_SYNC_SOURCE or C_SYNC_TARGET
1453 * This function might bring you directly into one of the
1454 * C_PAUSED_SYNC_* states.
1456 void drbd_start_resync(struct drbd_conf
*mdev
, enum drbd_conns side
)
1458 union drbd_state ns
;
1461 if (mdev
->state
.conn
>= C_SYNC_SOURCE
&& mdev
->state
.conn
< C_AHEAD
) {
1462 dev_err(DEV
, "Resync already running!\n");
1466 if (mdev
->state
.conn
< C_AHEAD
) {
1467 /* In case a previous resync run was aborted by an IO error/detach on the peer. */
1468 drbd_rs_cancel_all(mdev
);
1469 /* This should be done when we abort the resync. We definitely do not
1470 want to have this for connections going back and forth between
1471 Ahead/Behind and SyncSource/SyncTarget */
1474 if (side
== C_SYNC_TARGET
) {
1475 /* Since application IO was locked out during C_WF_BITMAP_T and
1476 C_WF_SYNC_UUID we are still unmodified. Before going to C_SYNC_TARGET
1477 we check that we might make the data inconsistent. */
1478 r
= drbd_khelper(mdev
, "before-resync-target");
1479 r
= (r
>> 8) & 0xff;
1481 dev_info(DEV
, "before-resync-target handler returned %d, "
1482 "dropping connection.\n", r
);
1483 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
1486 } else /* C_SYNC_SOURCE */ {
1487 r
= drbd_khelper(mdev
, "before-resync-source");
1488 r
= (r
>> 8) & 0xff;
1491 dev_info(DEV
, "before-resync-source handler returned %d, "
1492 "ignoring. Old userland tools?", r
);
1494 dev_info(DEV
, "before-resync-source handler returned %d, "
1495 "dropping connection.\n", r
);
1496 drbd_force_state(mdev
, NS(conn
, C_DISCONNECTING
));
1502 drbd_state_lock(mdev
);
1504 if (!get_ldev_if_state(mdev
, D_NEGOTIATING
)) {
1505 drbd_state_unlock(mdev
);
1509 write_lock_irq(&global_state_lock
);
1512 ns
.aftr_isp
= !_drbd_may_sync_now(mdev
);
1516 if (side
== C_SYNC_TARGET
)
1517 ns
.disk
= D_INCONSISTENT
;
1518 else /* side == C_SYNC_SOURCE */
1519 ns
.pdsk
= D_INCONSISTENT
;
1521 r
= __drbd_set_state(mdev
, ns
, CS_VERBOSE
, NULL
);
1524 if (ns
.conn
< C_CONNECTED
)
1525 r
= SS_UNKNOWN_ERROR
;
1527 if (r
== SS_SUCCESS
) {
1528 unsigned long tw
= drbd_bm_total_weight(mdev
);
1529 unsigned long now
= jiffies
;
1532 mdev
->rs_failed
= 0;
1533 mdev
->rs_paused
= 0;
1534 mdev
->rs_same_csum
= 0;
1535 mdev
->rs_last_events
= 0;
1536 mdev
->rs_last_sect_ev
= 0;
1537 mdev
->rs_total
= tw
;
1538 mdev
->rs_start
= now
;
1539 for (i
= 0; i
< DRBD_SYNC_MARKS
; i
++) {
1540 mdev
->rs_mark_left
[i
] = tw
;
1541 mdev
->rs_mark_time
[i
] = now
;
1543 _drbd_pause_after(mdev
);
1545 write_unlock_irq(&global_state_lock
);
1547 if (side
== C_SYNC_TARGET
)
1548 mdev
->bm_resync_fo
= 0;
1550 /* Since protocol 96, we must serialize drbd_gen_and_send_sync_uuid
1551 * with w_send_oos, or the sync target will get confused as to
1552 * how much bits to resync. We cannot do that always, because for an
1553 * empty resync and protocol < 95, we need to do it here, as we call
1554 * drbd_resync_finished from here in that case.
1555 * We drbd_gen_and_send_sync_uuid here for protocol < 96,
1556 * and from after_state_ch otherwise. */
1557 if (side
== C_SYNC_SOURCE
&& mdev
->agreed_pro_version
< 96)
1558 drbd_gen_and_send_sync_uuid(mdev
);
1560 if (r
== SS_SUCCESS
) {
1561 dev_info(DEV
, "Began resync as %s (will sync %lu KB [%lu bits set]).\n",
1562 drbd_conn_str(ns
.conn
),
1563 (unsigned long) mdev
->rs_total
<< (BM_BLOCK_SHIFT
-10),
1564 (unsigned long) mdev
->rs_total
);
1566 if (mdev
->agreed_pro_version
< 95 && mdev
->rs_total
== 0) {
1567 /* This still has a race (about when exactly the peers
1568 * detect connection loss) that can lead to a full sync
1569 * on next handshake. In 8.3.9 we fixed this with explicit
1570 * resync-finished notifications, but the fix
1571 * introduces a protocol change. Sleeping for some
1572 * time longer than the ping interval + timeout on the
1573 * SyncSource, to give the SyncTarget the chance to
1574 * detect connection loss, then waiting for a ping
1575 * response (implicit in drbd_resync_finished) reduces
1576 * the race considerably, but does not solve it. */
1577 if (side
== C_SYNC_SOURCE
)
1578 schedule_timeout_interruptible(
1579 mdev
->net_conf
->ping_int
* HZ
+
1580 mdev
->net_conf
->ping_timeo
*HZ
/9);
1581 drbd_resync_finished(mdev
);
1584 drbd_rs_controller_reset(mdev
);
1585 /* ns.conn may already be != mdev->state.conn,
1586 * we may have been paused in between, or become paused until
1587 * the timer triggers.
1588 * No matter, that is handled in resync_timer_fn() */
1589 if (ns
.conn
== C_SYNC_TARGET
)
1590 mod_timer(&mdev
->resync_timer
, jiffies
);
1595 drbd_state_unlock(mdev
);
1598 int drbd_worker(struct drbd_thread
*thi
)
1600 struct drbd_conf
*mdev
= thi
->mdev
;
1601 struct drbd_work
*w
= NULL
;
1602 LIST_HEAD(work_list
);
1605 sprintf(current
->comm
, "drbd%d_worker", mdev_to_minor(mdev
));
1607 while (get_t_state(thi
) == Running
) {
1608 drbd_thread_current_set_cpu(mdev
);
1610 if (down_trylock(&mdev
->data
.work
.s
)) {
1611 mutex_lock(&mdev
->data
.mutex
);
1612 if (mdev
->data
.socket
&& !mdev
->net_conf
->no_cork
)
1613 drbd_tcp_uncork(mdev
->data
.socket
);
1614 mutex_unlock(&mdev
->data
.mutex
);
1616 intr
= down_interruptible(&mdev
->data
.work
.s
);
1618 mutex_lock(&mdev
->data
.mutex
);
1619 if (mdev
->data
.socket
&& !mdev
->net_conf
->no_cork
)
1620 drbd_tcp_cork(mdev
->data
.socket
);
1621 mutex_unlock(&mdev
->data
.mutex
);
1625 D_ASSERT(intr
== -EINTR
);
1626 flush_signals(current
);
1627 ERR_IF (get_t_state(thi
) == Running
)
1632 if (get_t_state(thi
) != Running
)
1634 /* With this break, we have done a down() but not consumed
1635 the entry from the list. The cleanup code takes care of
1639 spin_lock_irq(&mdev
->data
.work
.q_lock
);
1640 ERR_IF(list_empty(&mdev
->data
.work
.q
)) {
1641 /* something terribly wrong in our logic.
1642 * we were able to down() the semaphore,
1643 * but the list is empty... doh.
1645 * what is the best thing to do now?
1646 * try again from scratch, restarting the receiver,
1647 * asender, whatnot? could break even more ugly,
1648 * e.g. when we are primary, but no good local data.
1650 * I'll try to get away just starting over this loop.
1652 spin_unlock_irq(&mdev
->data
.work
.q_lock
);
1655 w
= list_entry(mdev
->data
.work
.q
.next
, struct drbd_work
, list
);
1656 list_del_init(&w
->list
);
1657 spin_unlock_irq(&mdev
->data
.work
.q_lock
);
1659 if (!w
->cb(mdev
, w
, mdev
->state
.conn
< C_CONNECTED
)) {
1660 /* dev_warn(DEV, "worker: a callback failed! \n"); */
1661 if (mdev
->state
.conn
>= C_CONNECTED
)
1662 drbd_force_state(mdev
,
1663 NS(conn
, C_NETWORK_FAILURE
));
1666 D_ASSERT(test_bit(DEVICE_DYING
, &mdev
->flags
));
1667 D_ASSERT(test_bit(CONFIG_PENDING
, &mdev
->flags
));
1669 spin_lock_irq(&mdev
->data
.work
.q_lock
);
1671 while (!list_empty(&mdev
->data
.work
.q
)) {
1672 list_splice_init(&mdev
->data
.work
.q
, &work_list
);
1673 spin_unlock_irq(&mdev
->data
.work
.q_lock
);
1675 while (!list_empty(&work_list
)) {
1676 w
= list_entry(work_list
.next
, struct drbd_work
, list
);
1677 list_del_init(&w
->list
);
1679 i
++; /* dead debugging code */
1682 spin_lock_irq(&mdev
->data
.work
.q_lock
);
1684 sema_init(&mdev
->data
.work
.s
, 0);
1685 /* DANGEROUS race: if someone did queue his work within the spinlock,
1686 * but up() ed outside the spinlock, we could get an up() on the
1687 * semaphore without corresponding list entry.
1690 spin_unlock_irq(&mdev
->data
.work
.q_lock
);
1692 D_ASSERT(mdev
->state
.disk
== D_DISKLESS
&& mdev
->state
.conn
== C_STANDALONE
);
1693 /* _drbd_set_state only uses stop_nowait.
1694 * wait here for the Exiting receiver. */
1695 drbd_thread_stop(&mdev
->receiver
);
1696 drbd_mdev_cleanup(mdev
);
1698 dev_info(DEV
, "worker terminated\n");
1700 clear_bit(DEVICE_DYING
, &mdev
->flags
);
1701 clear_bit(CONFIG_PENDING
, &mdev
->flags
);
1702 wake_up(&mdev
->state_wait
);