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>
39 #include "drbd_protocol.h"
42 static int make_ov_request(struct drbd_device *, int);
43 static int make_resync_request(struct drbd_device *, int);
46 * drbd_md_endio (defined here)
47 * drbd_request_endio (defined here)
48 * drbd_peer_request_endio (defined here)
49 * drbd_bm_endio (defined in drbd_bitmap.c)
51 * For all these callbacks, note the following:
52 * The callbacks will be called in irq context by the IDE drivers,
53 * and in Softirqs/Tasklets/BH context by the SCSI drivers.
54 * Try to get the locking right :)
59 /* About the global_state_lock
60 Each state transition on an device holds a read lock. In case we have
61 to evaluate the resync after dependencies, we grab a write lock, because
62 we need stable states on all devices for that. */
63 rwlock_t global_state_lock;
65 /* used for synchronous meta data and bitmap IO
66 * submitted by drbd_md_sync_page_io()
68 void drbd_md_endio(struct bio *bio, int error)
70 struct drbd_device *device;
72 device = bio->bi_private;
73 device->md_io.error = error;
75 /* We grabbed an extra reference in _drbd_md_sync_page_io() to be able
76 * to timeout on the lower level device, and eventually detach from it.
77 * If this io completion runs after that timeout expired, this
78 * drbd_md_put_buffer() may allow us to finally try and re-attach.
79 * During normal operation, this only puts that extra reference
81 * Make sure we first drop the reference, and only then signal
82 * completion, or we may (in drbd_al_read_log()) cycle so fast into the
83 * next drbd_md_sync_page_io(), that we trigger the
84 * ASSERT(atomic_read(&device->md_io_in_use) == 1) there.
86 drbd_md_put_buffer(device);
87 device->md_io.done = 1;
88 wake_up(&device->misc_wait);
90 if (device->ldev) /* special case: drbd_md_read() during drbd_adm_attach() */
94 /* reads on behalf of the partner,
95 * "submitted" by the receiver
97 static void drbd_endio_read_sec_final(struct drbd_peer_request *peer_req) __releases(local)
99 unsigned long flags = 0;
100 struct drbd_peer_device *peer_device = peer_req->peer_device;
101 struct drbd_device *device = peer_device->device;
103 spin_lock_irqsave(&device->resource->req_lock, flags);
104 device->read_cnt += peer_req->i.size >> 9;
105 list_del(&peer_req->w.list);
106 if (list_empty(&device->read_ee))
107 wake_up(&device->ee_wait);
108 if (test_bit(__EE_WAS_ERROR, &peer_req->flags))
109 __drbd_chk_io_error(device, DRBD_READ_ERROR);
110 spin_unlock_irqrestore(&device->resource->req_lock, flags);
112 drbd_queue_work(&peer_device->connection->sender_work, &peer_req->w);
116 /* writes on behalf of the partner, or resync writes,
117 * "submitted" by the receiver, final stage. */
118 void drbd_endio_write_sec_final(struct drbd_peer_request *peer_req) __releases(local)
120 unsigned long flags = 0;
121 struct drbd_peer_device *peer_device = peer_req->peer_device;
122 struct drbd_device *device = peer_device->device;
123 struct drbd_interval i;
126 int do_al_complete_io;
128 /* after we moved peer_req to done_ee,
129 * we may no longer access it,
130 * it may be freed/reused already!
131 * (as soon as we release the req_lock) */
133 do_al_complete_io = peer_req->flags & EE_CALL_AL_COMPLETE_IO;
134 block_id = peer_req->block_id;
135 peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO;
137 spin_lock_irqsave(&device->resource->req_lock, flags);
138 device->writ_cnt += peer_req->i.size >> 9;
139 list_move_tail(&peer_req->w.list, &device->done_ee);
142 * Do not remove from the write_requests tree here: we did not send the
143 * Ack yet and did not wake possibly waiting conflicting requests.
144 * Removed from the tree from "drbd_process_done_ee" within the
145 * appropriate dw.cb (e_end_block/e_end_resync_block) or from
146 * _drbd_clear_done_ee.
149 do_wake = list_empty(block_id == ID_SYNCER ? &device->sync_ee : &device->active_ee);
151 /* FIXME do we want to detach for failed REQ_DISCARD?
152 * ((peer_req->flags & (EE_WAS_ERROR|EE_IS_TRIM)) == EE_WAS_ERROR) */
153 if (peer_req->flags & EE_WAS_ERROR)
154 __drbd_chk_io_error(device, DRBD_WRITE_ERROR);
155 spin_unlock_irqrestore(&device->resource->req_lock, flags);
157 if (block_id == ID_SYNCER)
158 drbd_rs_complete_io(device, i.sector);
161 wake_up(&device->ee_wait);
163 if (do_al_complete_io)
164 drbd_al_complete_io(device, &i);
166 wake_asender(peer_device->connection);
170 /* writes on behalf of the partner, or resync writes,
171 * "submitted" by the receiver.
173 void drbd_peer_request_endio(struct bio *bio, int error)
175 struct drbd_peer_request *peer_req = bio->bi_private;
176 struct drbd_device *device = peer_req->peer_device->device;
177 int uptodate = bio_flagged(bio, BIO_UPTODATE);
178 int is_write = bio_data_dir(bio) == WRITE;
179 int is_discard = !!(bio->bi_rw & REQ_DISCARD);
181 if (error && __ratelimit(&drbd_ratelimit_state))
182 drbd_warn(device, "%s: error=%d s=%llus\n",
183 is_write ? (is_discard ? "discard" : "write")
185 (unsigned long long)peer_req->i.sector);
186 if (!error && !uptodate) {
187 if (__ratelimit(&drbd_ratelimit_state))
188 drbd_warn(device, "%s: setting error to -EIO s=%llus\n",
189 is_write ? "write" : "read",
190 (unsigned long long)peer_req->i.sector);
191 /* strange behavior of some lower level drivers...
192 * fail the request by clearing the uptodate flag,
193 * but do not return any error?! */
198 set_bit(__EE_WAS_ERROR, &peer_req->flags);
200 bio_put(bio); /* no need for the bio anymore */
201 if (atomic_dec_and_test(&peer_req->pending_bios)) {
203 drbd_endio_write_sec_final(peer_req);
205 drbd_endio_read_sec_final(peer_req);
209 /* read, readA or write requests on R_PRIMARY coming from drbd_make_request
211 void drbd_request_endio(struct bio *bio, int error)
214 struct drbd_request *req = bio->bi_private;
215 struct drbd_device *device = req->device;
216 struct bio_and_error m;
217 enum drbd_req_event what;
218 int uptodate = bio_flagged(bio, BIO_UPTODATE);
220 if (!error && !uptodate) {
221 drbd_warn(device, "p %s: setting error to -EIO\n",
222 bio_data_dir(bio) == WRITE ? "write" : "read");
223 /* strange behavior of some lower level drivers...
224 * fail the request by clearing the uptodate flag,
225 * but do not return any error?! */
230 /* If this request was aborted locally before,
231 * but now was completed "successfully",
232 * chances are that this caused arbitrary data corruption.
234 * "aborting" requests, or force-detaching the disk, is intended for
235 * completely blocked/hung local backing devices which do no longer
236 * complete requests at all, not even do error completions. In this
237 * situation, usually a hard-reset and failover is the only way out.
239 * By "aborting", basically faking a local error-completion,
240 * we allow for a more graceful swichover by cleanly migrating services.
241 * Still the affected node has to be rebooted "soon".
243 * By completing these requests, we allow the upper layers to re-use
244 * the associated data pages.
246 * If later the local backing device "recovers", and now DMAs some data
247 * from disk into the original request pages, in the best case it will
248 * just put random data into unused pages; but typically it will corrupt
249 * meanwhile completely unrelated data, causing all sorts of damage.
251 * Which means delayed successful completion,
252 * especially for READ requests,
253 * is a reason to panic().
255 * We assume that a delayed *error* completion is OK,
256 * though we still will complain noisily about it.
258 if (unlikely(req->rq_state & RQ_LOCAL_ABORTED)) {
259 if (__ratelimit(&drbd_ratelimit_state))
260 drbd_emerg(device, "delayed completion of aborted local request; disk-timeout may be too aggressive\n");
263 panic("possible random memory corruption caused by delayed completion of aborted local request\n");
266 /* to avoid recursion in __req_mod */
267 if (unlikely(error)) {
268 if (bio->bi_rw & REQ_DISCARD)
269 what = (error == -EOPNOTSUPP)
270 ? DISCARD_COMPLETED_NOTSUPP
271 : DISCARD_COMPLETED_WITH_ERROR;
273 what = (bio_data_dir(bio) == WRITE)
274 ? WRITE_COMPLETED_WITH_ERROR
275 : (bio_rw(bio) == READ)
276 ? READ_COMPLETED_WITH_ERROR
277 : READ_AHEAD_COMPLETED_WITH_ERROR;
281 bio_put(req->private_bio);
282 req->private_bio = ERR_PTR(error);
284 /* not req_mod(), we need irqsave here! */
285 spin_lock_irqsave(&device->resource->req_lock, flags);
286 __req_mod(req, what, &m);
287 spin_unlock_irqrestore(&device->resource->req_lock, flags);
291 complete_master_bio(device, &m);
294 void drbd_csum_ee(struct crypto_hash *tfm, struct drbd_peer_request *peer_req, void *digest)
296 struct hash_desc desc;
297 struct scatterlist sg;
298 struct page *page = peer_req->pages;
305 sg_init_table(&sg, 1);
306 crypto_hash_init(&desc);
308 while ((tmp = page_chain_next(page))) {
309 /* all but the last page will be fully used */
310 sg_set_page(&sg, page, PAGE_SIZE, 0);
311 crypto_hash_update(&desc, &sg, sg.length);
314 /* and now the last, possibly only partially used page */
315 len = peer_req->i.size & (PAGE_SIZE - 1);
316 sg_set_page(&sg, page, len ?: PAGE_SIZE, 0);
317 crypto_hash_update(&desc, &sg, sg.length);
318 crypto_hash_final(&desc, digest);
321 void drbd_csum_bio(struct crypto_hash *tfm, struct bio *bio, void *digest)
323 struct hash_desc desc;
324 struct scatterlist sg;
326 struct bvec_iter iter;
331 sg_init_table(&sg, 1);
332 crypto_hash_init(&desc);
334 bio_for_each_segment(bvec, bio, iter) {
335 sg_set_page(&sg, bvec.bv_page, bvec.bv_len, bvec.bv_offset);
336 crypto_hash_update(&desc, &sg, sg.length);
338 crypto_hash_final(&desc, digest);
341 /* MAYBE merge common code with w_e_end_ov_req */
342 static int w_e_send_csum(struct drbd_work *w, int cancel)
344 struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w);
345 struct drbd_peer_device *peer_device = peer_req->peer_device;
346 struct drbd_device *device = peer_device->device;
351 if (unlikely(cancel))
354 if (unlikely((peer_req->flags & EE_WAS_ERROR) != 0))
357 digest_size = crypto_hash_digestsize(peer_device->connection->csums_tfm);
358 digest = kmalloc(digest_size, GFP_NOIO);
360 sector_t sector = peer_req->i.sector;
361 unsigned int size = peer_req->i.size;
362 drbd_csum_ee(peer_device->connection->csums_tfm, peer_req, digest);
363 /* Free peer_req and pages before send.
364 * In case we block on congestion, we could otherwise run into
365 * some distributed deadlock, if the other side blocks on
366 * congestion as well, because our receiver blocks in
367 * drbd_alloc_pages due to pp_in_use > max_buffers. */
368 drbd_free_peer_req(device, peer_req);
370 inc_rs_pending(device);
371 err = drbd_send_drequest_csum(peer_device, sector, size,
376 drbd_err(device, "kmalloc() of digest failed.\n");
382 drbd_free_peer_req(device, peer_req);
385 drbd_err(device, "drbd_send_drequest(..., csum) failed\n");
389 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
391 static int read_for_csum(struct drbd_peer_device *peer_device, sector_t sector, int size)
393 struct drbd_device *device = peer_device->device;
394 struct drbd_peer_request *peer_req;
396 if (!get_ldev(device))
399 /* GFP_TRY, because if there is no memory available right now, this may
400 * be rescheduled for later. It is "only" background resync, after all. */
401 peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER /* unused */, sector,
402 size, true /* has real payload */, GFP_TRY);
406 peer_req->w.cb = w_e_send_csum;
407 spin_lock_irq(&device->resource->req_lock);
408 list_add_tail(&peer_req->w.list, &device->read_ee);
409 spin_unlock_irq(&device->resource->req_lock);
411 atomic_add(size >> 9, &device->rs_sect_ev);
412 if (drbd_submit_peer_request(device, peer_req, READ, DRBD_FAULT_RS_RD) == 0)
415 /* If it failed because of ENOMEM, retry should help. If it failed
416 * because bio_add_page failed (probably broken lower level driver),
417 * retry may or may not help.
418 * If it does not, you may need to force disconnect. */
419 spin_lock_irq(&device->resource->req_lock);
420 list_del(&peer_req->w.list);
421 spin_unlock_irq(&device->resource->req_lock);
423 drbd_free_peer_req(device, peer_req);
429 int w_resync_timer(struct drbd_work *w, int cancel)
431 struct drbd_device *device =
432 container_of(w, struct drbd_device, resync_work);
434 switch (device->state.conn) {
436 make_ov_request(device, cancel);
439 make_resync_request(device, cancel);
446 void resync_timer_fn(unsigned long data)
448 struct drbd_device *device = (struct drbd_device *) data;
450 drbd_queue_work_if_unqueued(
451 &first_peer_device(device)->connection->sender_work,
452 &device->resync_work);
455 static void fifo_set(struct fifo_buffer *fb, int value)
459 for (i = 0; i < fb->size; i++)
460 fb->values[i] = value;
463 static int fifo_push(struct fifo_buffer *fb, int value)
467 ov = fb->values[fb->head_index];
468 fb->values[fb->head_index++] = value;
470 if (fb->head_index >= fb->size)
476 static void fifo_add_val(struct fifo_buffer *fb, int value)
480 for (i = 0; i < fb->size; i++)
481 fb->values[i] += value;
484 struct fifo_buffer *fifo_alloc(int fifo_size)
486 struct fifo_buffer *fb;
488 fb = kzalloc(sizeof(struct fifo_buffer) + sizeof(int) * fifo_size, GFP_NOIO);
493 fb->size = fifo_size;
499 static int drbd_rs_controller(struct drbd_device *device, unsigned int sect_in)
501 struct disk_conf *dc;
502 unsigned int want; /* The number of sectors we want in-flight */
503 int req_sect; /* Number of sectors to request in this turn */
504 int correction; /* Number of sectors more we need in-flight */
505 int cps; /* correction per invocation of drbd_rs_controller() */
506 int steps; /* Number of time steps to plan ahead */
509 struct fifo_buffer *plan;
511 dc = rcu_dereference(device->ldev->disk_conf);
512 plan = rcu_dereference(device->rs_plan_s);
514 steps = plan->size; /* (dc->c_plan_ahead * 10 * SLEEP_TIME) / HZ; */
516 if (device->rs_in_flight + sect_in == 0) { /* At start of resync */
517 want = ((dc->resync_rate * 2 * SLEEP_TIME) / HZ) * steps;
518 } else { /* normal path */
519 want = dc->c_fill_target ? dc->c_fill_target :
520 sect_in * dc->c_delay_target * HZ / (SLEEP_TIME * 10);
523 correction = want - device->rs_in_flight - plan->total;
526 cps = correction / steps;
527 fifo_add_val(plan, cps);
528 plan->total += cps * steps;
530 /* What we do in this step */
531 curr_corr = fifo_push(plan, 0);
532 plan->total -= curr_corr;
534 req_sect = sect_in + curr_corr;
538 max_sect = (dc->c_max_rate * 2 * SLEEP_TIME) / HZ;
539 if (req_sect > max_sect)
543 drbd_warn(device, "si=%u if=%d wa=%u co=%d st=%d cps=%d pl=%d cc=%d rs=%d\n",
544 sect_in, device->rs_in_flight, want, correction,
545 steps, cps, device->rs_planed, curr_corr, req_sect);
551 static int drbd_rs_number_requests(struct drbd_device *device)
553 unsigned int sect_in; /* Number of sectors that came in since the last turn */
556 sect_in = atomic_xchg(&device->rs_sect_in, 0);
557 device->rs_in_flight -= sect_in;
560 mxb = drbd_get_max_buffers(device) / 2;
561 if (rcu_dereference(device->rs_plan_s)->size) {
562 number = drbd_rs_controller(device, sect_in) >> (BM_BLOCK_SHIFT - 9);
563 device->c_sync_rate = number * HZ * (BM_BLOCK_SIZE / 1024) / SLEEP_TIME;
565 device->c_sync_rate = rcu_dereference(device->ldev->disk_conf)->resync_rate;
566 number = SLEEP_TIME * device->c_sync_rate / ((BM_BLOCK_SIZE / 1024) * HZ);
570 /* Don't have more than "max-buffers"/2 in-flight.
571 * Otherwise we may cause the remote site to stall on drbd_alloc_pages(),
572 * potentially causing a distributed deadlock on congestion during
573 * online-verify or (checksum-based) resync, if max-buffers,
574 * socket buffer sizes and resync rate settings are mis-configured. */
576 /* note that "number" is in units of "BM_BLOCK_SIZE" (which is 4k),
577 * mxb (as used here, and in drbd_alloc_pages on the peer) is
578 * "number of pages" (typically also 4k),
579 * but "rs_in_flight" is in "sectors" (512 Byte). */
580 if (mxb - device->rs_in_flight/8 < number)
581 number = mxb - device->rs_in_flight/8;
586 static int make_resync_request(struct drbd_device *const device, int cancel)
588 struct drbd_peer_device *const peer_device = first_peer_device(device);
589 struct drbd_connection *const connection = peer_device ? peer_device->connection : NULL;
592 const sector_t capacity = drbd_get_capacity(device->this_bdev);
594 int number, rollback_i, size;
595 int align, requeue = 0;
598 if (unlikely(cancel))
601 if (device->rs_total == 0) {
603 drbd_resync_finished(device);
607 if (!get_ldev(device)) {
608 /* Since we only need to access device->rsync a
609 get_ldev_if_state(device,D_FAILED) would be sufficient, but
610 to continue resync with a broken disk makes no sense at
612 drbd_err(device, "Disk broke down during resync!\n");
616 max_bio_size = queue_max_hw_sectors(device->rq_queue) << 9;
617 number = drbd_rs_number_requests(device);
621 for (i = 0; i < number; i++) {
622 /* Stop generating RS requests when half of the send buffer is filled,
623 * but notify TCP that we'd like to have more space. */
624 mutex_lock(&connection->data.mutex);
625 if (connection->data.socket) {
626 struct sock *sk = connection->data.socket->sk;
627 int queued = sk->sk_wmem_queued;
628 int sndbuf = sk->sk_sndbuf;
629 if (queued > sndbuf / 2) {
632 set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
636 mutex_unlock(&connection->data.mutex);
641 size = BM_BLOCK_SIZE;
642 bit = drbd_bm_find_next(device, device->bm_resync_fo);
644 if (bit == DRBD_END_OF_BITMAP) {
645 device->bm_resync_fo = drbd_bm_bits(device);
650 sector = BM_BIT_TO_SECT(bit);
652 if (drbd_try_rs_begin_io(device, sector)) {
653 device->bm_resync_fo = bit;
656 device->bm_resync_fo = bit + 1;
658 if (unlikely(drbd_bm_test_bit(device, bit) == 0)) {
659 drbd_rs_complete_io(device, sector);
663 #if DRBD_MAX_BIO_SIZE > BM_BLOCK_SIZE
664 /* try to find some adjacent bits.
665 * we stop if we have already the maximum req size.
667 * Additionally always align bigger requests, in order to
668 * be prepared for all stripe sizes of software RAIDs.
673 if (size + BM_BLOCK_SIZE > max_bio_size)
676 /* Be always aligned */
677 if (sector & ((1<<(align+3))-1))
680 /* do not cross extent boundaries */
681 if (((bit+1) & BM_BLOCKS_PER_BM_EXT_MASK) == 0)
683 /* now, is it actually dirty, after all?
684 * caution, drbd_bm_test_bit is tri-state for some
685 * obscure reason; ( b == 0 ) would get the out-of-band
686 * only accidentally right because of the "oddly sized"
687 * adjustment below */
688 if (drbd_bm_test_bit(device, bit+1) != 1)
691 size += BM_BLOCK_SIZE;
692 if ((BM_BLOCK_SIZE << align) <= size)
696 /* if we merged some,
697 * reset the offset to start the next drbd_bm_find_next from */
698 if (size > BM_BLOCK_SIZE)
699 device->bm_resync_fo = bit + 1;
702 /* adjust very last sectors, in case we are oddly sized */
703 if (sector + (size>>9) > capacity)
704 size = (capacity-sector)<<9;
706 if (device->use_csums) {
707 switch (read_for_csum(peer_device, sector, size)) {
708 case -EIO: /* Disk failure */
711 case -EAGAIN: /* allocation failed, or ldev busy */
712 drbd_rs_complete_io(device, sector);
713 device->bm_resync_fo = BM_SECT_TO_BIT(sector);
725 inc_rs_pending(device);
726 err = drbd_send_drequest(peer_device, P_RS_DATA_REQUEST,
727 sector, size, ID_SYNCER);
729 drbd_err(device, "drbd_send_drequest() failed, aborting...\n");
730 dec_rs_pending(device);
737 if (device->bm_resync_fo >= drbd_bm_bits(device)) {
738 /* last syncer _request_ was sent,
739 * but the P_RS_DATA_REPLY not yet received. sync will end (and
740 * next sync group will resume), as soon as we receive the last
741 * resync data block, and the last bit is cleared.
742 * until then resync "work" is "inactive" ...
749 device->rs_in_flight += (i << (BM_BLOCK_SHIFT - 9));
750 mod_timer(&device->resync_timer, jiffies + SLEEP_TIME);
755 static int make_ov_request(struct drbd_device *device, int cancel)
759 const sector_t capacity = drbd_get_capacity(device->this_bdev);
760 bool stop_sector_reached = false;
762 if (unlikely(cancel))
765 number = drbd_rs_number_requests(device);
767 sector = device->ov_position;
768 for (i = 0; i < number; i++) {
769 if (sector >= capacity)
772 /* We check for "finished" only in the reply path:
773 * w_e_end_ov_reply().
774 * We need to send at least one request out. */
775 stop_sector_reached = i > 0
776 && verify_can_do_stop_sector(device)
777 && sector >= device->ov_stop_sector;
778 if (stop_sector_reached)
781 size = BM_BLOCK_SIZE;
783 if (drbd_try_rs_begin_io(device, sector)) {
784 device->ov_position = sector;
788 if (sector + (size>>9) > capacity)
789 size = (capacity-sector)<<9;
791 inc_rs_pending(device);
792 if (drbd_send_ov_request(first_peer_device(device), sector, size)) {
793 dec_rs_pending(device);
796 sector += BM_SECT_PER_BIT;
798 device->ov_position = sector;
801 device->rs_in_flight += (i << (BM_BLOCK_SHIFT - 9));
802 if (i == 0 || !stop_sector_reached)
803 mod_timer(&device->resync_timer, jiffies + SLEEP_TIME);
807 int w_ov_finished(struct drbd_work *w, int cancel)
809 struct drbd_device_work *dw =
810 container_of(w, struct drbd_device_work, w);
811 struct drbd_device *device = dw->device;
813 ov_out_of_sync_print(device);
814 drbd_resync_finished(device);
819 static int w_resync_finished(struct drbd_work *w, int cancel)
821 struct drbd_device_work *dw =
822 container_of(w, struct drbd_device_work, w);
823 struct drbd_device *device = dw->device;
826 drbd_resync_finished(device);
831 static void ping_peer(struct drbd_device *device)
833 struct drbd_connection *connection = first_peer_device(device)->connection;
835 clear_bit(GOT_PING_ACK, &connection->flags);
836 request_ping(connection);
837 wait_event(connection->ping_wait,
838 test_bit(GOT_PING_ACK, &connection->flags) || device->state.conn < C_CONNECTED);
841 int drbd_resync_finished(struct drbd_device *device)
843 unsigned long db, dt, dbdt;
845 union drbd_state os, ns;
846 struct drbd_device_work *dw;
847 char *khelper_cmd = NULL;
850 /* Remove all elements from the resync LRU. Since future actions
851 * might set bits in the (main) bitmap, then the entries in the
852 * resync LRU would be wrong. */
853 if (drbd_rs_del_all(device)) {
854 /* In case this is not possible now, most probably because
855 * there are P_RS_DATA_REPLY Packets lingering on the worker's
856 * queue (or even the read operations for those packets
857 * is not finished by now). Retry in 100ms. */
859 schedule_timeout_interruptible(HZ / 10);
860 dw = kmalloc(sizeof(struct drbd_device_work), GFP_ATOMIC);
862 dw->w.cb = w_resync_finished;
864 drbd_queue_work(&first_peer_device(device)->connection->sender_work,
868 drbd_err(device, "Warn failed to drbd_rs_del_all() and to kmalloc(dw).\n");
871 dt = (jiffies - device->rs_start - device->rs_paused) / HZ;
875 db = device->rs_total;
876 /* adjust for verify start and stop sectors, respective reached position */
877 if (device->state.conn == C_VERIFY_S || device->state.conn == C_VERIFY_T)
878 db -= device->ov_left;
880 dbdt = Bit2KB(db/dt);
881 device->rs_paused /= HZ;
883 if (!get_ldev(device))
888 spin_lock_irq(&device->resource->req_lock);
889 os = drbd_read_state(device);
891 verify_done = (os.conn == C_VERIFY_S || os.conn == C_VERIFY_T);
893 /* This protects us against multiple calls (that can happen in the presence
894 of application IO), and against connectivity loss just before we arrive here. */
895 if (os.conn <= C_CONNECTED)
899 ns.conn = C_CONNECTED;
901 drbd_info(device, "%s done (total %lu sec; paused %lu sec; %lu K/sec)\n",
902 verify_done ? "Online verify" : "Resync",
903 dt + device->rs_paused, device->rs_paused, dbdt);
905 n_oos = drbd_bm_total_weight(device);
907 if (os.conn == C_VERIFY_S || os.conn == C_VERIFY_T) {
909 drbd_alert(device, "Online verify found %lu %dk block out of sync!\n",
911 khelper_cmd = "out-of-sync";
914 D_ASSERT(device, (n_oos - device->rs_failed) == 0);
916 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T)
917 khelper_cmd = "after-resync-target";
919 if (device->use_csums && device->rs_total) {
920 const unsigned long s = device->rs_same_csum;
921 const unsigned long t = device->rs_total;
924 (t < 100000) ? ((s*100)/t) : (s/(t/100));
925 drbd_info(device, "%u %% had equal checksums, eliminated: %luK; "
926 "transferred %luK total %luK\n",
928 Bit2KB(device->rs_same_csum),
929 Bit2KB(device->rs_total - device->rs_same_csum),
930 Bit2KB(device->rs_total));
934 if (device->rs_failed) {
935 drbd_info(device, " %lu failed blocks\n", device->rs_failed);
937 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T) {
938 ns.disk = D_INCONSISTENT;
939 ns.pdsk = D_UP_TO_DATE;
941 ns.disk = D_UP_TO_DATE;
942 ns.pdsk = D_INCONSISTENT;
945 ns.disk = D_UP_TO_DATE;
946 ns.pdsk = D_UP_TO_DATE;
948 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T) {
949 if (device->p_uuid) {
951 for (i = UI_BITMAP ; i <= UI_HISTORY_END ; i++)
952 _drbd_uuid_set(device, i, device->p_uuid[i]);
953 drbd_uuid_set(device, UI_BITMAP, device->ldev->md.uuid[UI_CURRENT]);
954 _drbd_uuid_set(device, UI_CURRENT, device->p_uuid[UI_CURRENT]);
956 drbd_err(device, "device->p_uuid is NULL! BUG\n");
960 if (!(os.conn == C_VERIFY_S || os.conn == C_VERIFY_T)) {
961 /* for verify runs, we don't update uuids here,
962 * so there would be nothing to report. */
963 drbd_uuid_set_bm(device, 0UL);
964 drbd_print_uuids(device, "updated UUIDs");
965 if (device->p_uuid) {
966 /* Now the two UUID sets are equal, update what we
967 * know of the peer. */
969 for (i = UI_CURRENT ; i <= UI_HISTORY_END ; i++)
970 device->p_uuid[i] = device->ldev->md.uuid[i];
975 _drbd_set_state(device, ns, CS_VERBOSE, NULL);
977 spin_unlock_irq(&device->resource->req_lock);
980 device->rs_total = 0;
981 device->rs_failed = 0;
982 device->rs_paused = 0;
984 /* reset start sector, if we reached end of device */
985 if (verify_done && device->ov_left == 0)
986 device->ov_start_sector = 0;
988 drbd_md_sync(device);
991 drbd_khelper(device, khelper_cmd);
997 static void move_to_net_ee_or_free(struct drbd_device *device, struct drbd_peer_request *peer_req)
999 if (drbd_peer_req_has_active_page(peer_req)) {
1000 /* This might happen if sendpage() has not finished */
1001 int i = (peer_req->i.size + PAGE_SIZE -1) >> PAGE_SHIFT;
1002 atomic_add(i, &device->pp_in_use_by_net);
1003 atomic_sub(i, &device->pp_in_use);
1004 spin_lock_irq(&device->resource->req_lock);
1005 list_add_tail(&peer_req->w.list, &device->net_ee);
1006 spin_unlock_irq(&device->resource->req_lock);
1007 wake_up(&drbd_pp_wait);
1009 drbd_free_peer_req(device, peer_req);
1013 * w_e_end_data_req() - Worker callback, to send a P_DATA_REPLY packet in response to a P_DATA_REQUEST
1014 * @device: DRBD device.
1016 * @cancel: The connection will be closed anyways
1018 int w_e_end_data_req(struct drbd_work *w, int cancel)
1020 struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w);
1021 struct drbd_peer_device *peer_device = peer_req->peer_device;
1022 struct drbd_device *device = peer_device->device;
1025 if (unlikely(cancel)) {
1026 drbd_free_peer_req(device, peer_req);
1027 dec_unacked(device);
1031 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1032 err = drbd_send_block(peer_device, P_DATA_REPLY, peer_req);
1034 if (__ratelimit(&drbd_ratelimit_state))
1035 drbd_err(device, "Sending NegDReply. sector=%llus.\n",
1036 (unsigned long long)peer_req->i.sector);
1038 err = drbd_send_ack(peer_device, P_NEG_DREPLY, peer_req);
1041 dec_unacked(device);
1043 move_to_net_ee_or_free(device, peer_req);
1046 drbd_err(device, "drbd_send_block() failed\n");
1051 * w_e_end_rsdata_req() - Worker callback to send a P_RS_DATA_REPLY packet in response to a P_RS_DATA_REQUEST
1053 * @cancel: The connection will be closed anyways
1055 int w_e_end_rsdata_req(struct drbd_work *w, int cancel)
1057 struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w);
1058 struct drbd_peer_device *peer_device = peer_req->peer_device;
1059 struct drbd_device *device = peer_device->device;
1062 if (unlikely(cancel)) {
1063 drbd_free_peer_req(device, peer_req);
1064 dec_unacked(device);
1068 if (get_ldev_if_state(device, D_FAILED)) {
1069 drbd_rs_complete_io(device, peer_req->i.sector);
1073 if (device->state.conn == C_AHEAD) {
1074 err = drbd_send_ack(peer_device, P_RS_CANCEL, peer_req);
1075 } else if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1076 if (likely(device->state.pdsk >= D_INCONSISTENT)) {
1077 inc_rs_pending(device);
1078 err = drbd_send_block(peer_device, P_RS_DATA_REPLY, peer_req);
1080 if (__ratelimit(&drbd_ratelimit_state))
1081 drbd_err(device, "Not sending RSDataReply, "
1082 "partner DISKLESS!\n");
1086 if (__ratelimit(&drbd_ratelimit_state))
1087 drbd_err(device, "Sending NegRSDReply. sector %llus.\n",
1088 (unsigned long long)peer_req->i.sector);
1090 err = drbd_send_ack(peer_device, P_NEG_RS_DREPLY, peer_req);
1092 /* update resync data with failure */
1093 drbd_rs_failed_io(device, peer_req->i.sector, peer_req->i.size);
1096 dec_unacked(device);
1098 move_to_net_ee_or_free(device, peer_req);
1101 drbd_err(device, "drbd_send_block() failed\n");
1105 int w_e_end_csum_rs_req(struct drbd_work *w, int cancel)
1107 struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w);
1108 struct drbd_peer_device *peer_device = peer_req->peer_device;
1109 struct drbd_device *device = peer_device->device;
1110 struct digest_info *di;
1112 void *digest = NULL;
1115 if (unlikely(cancel)) {
1116 drbd_free_peer_req(device, peer_req);
1117 dec_unacked(device);
1121 if (get_ldev(device)) {
1122 drbd_rs_complete_io(device, peer_req->i.sector);
1126 di = peer_req->digest;
1128 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1129 /* quick hack to try to avoid a race against reconfiguration.
1130 * a real fix would be much more involved,
1131 * introducing more locking mechanisms */
1132 if (peer_device->connection->csums_tfm) {
1133 digest_size = crypto_hash_digestsize(peer_device->connection->csums_tfm);
1134 D_ASSERT(device, digest_size == di->digest_size);
1135 digest = kmalloc(digest_size, GFP_NOIO);
1138 drbd_csum_ee(peer_device->connection->csums_tfm, peer_req, digest);
1139 eq = !memcmp(digest, di->digest, digest_size);
1144 drbd_set_in_sync(device, peer_req->i.sector, peer_req->i.size);
1145 /* rs_same_csums unit is BM_BLOCK_SIZE */
1146 device->rs_same_csum += peer_req->i.size >> BM_BLOCK_SHIFT;
1147 err = drbd_send_ack(peer_device, P_RS_IS_IN_SYNC, peer_req);
1149 inc_rs_pending(device);
1150 peer_req->block_id = ID_SYNCER; /* By setting block_id, digest pointer becomes invalid! */
1151 peer_req->flags &= ~EE_HAS_DIGEST; /* This peer request no longer has a digest pointer */
1153 err = drbd_send_block(peer_device, P_RS_DATA_REPLY, peer_req);
1156 err = drbd_send_ack(peer_device, P_NEG_RS_DREPLY, peer_req);
1157 if (__ratelimit(&drbd_ratelimit_state))
1158 drbd_err(device, "Sending NegDReply. I guess it gets messy.\n");
1161 dec_unacked(device);
1162 move_to_net_ee_or_free(device, peer_req);
1165 drbd_err(device, "drbd_send_block/ack() failed\n");
1169 int w_e_end_ov_req(struct drbd_work *w, int cancel)
1171 struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w);
1172 struct drbd_peer_device *peer_device = peer_req->peer_device;
1173 struct drbd_device *device = peer_device->device;
1174 sector_t sector = peer_req->i.sector;
1175 unsigned int size = peer_req->i.size;
1180 if (unlikely(cancel))
1183 digest_size = crypto_hash_digestsize(peer_device->connection->verify_tfm);
1184 digest = kmalloc(digest_size, GFP_NOIO);
1186 err = 1; /* terminate the connection in case the allocation failed */
1190 if (likely(!(peer_req->flags & EE_WAS_ERROR)))
1191 drbd_csum_ee(peer_device->connection->verify_tfm, peer_req, digest);
1193 memset(digest, 0, digest_size);
1195 /* Free e and pages before send.
1196 * In case we block on congestion, we could otherwise run into
1197 * some distributed deadlock, if the other side blocks on
1198 * congestion as well, because our receiver blocks in
1199 * drbd_alloc_pages due to pp_in_use > max_buffers. */
1200 drbd_free_peer_req(device, peer_req);
1202 inc_rs_pending(device);
1203 err = drbd_send_drequest_csum(peer_device, sector, size, digest, digest_size, P_OV_REPLY);
1205 dec_rs_pending(device);
1210 drbd_free_peer_req(device, peer_req);
1211 dec_unacked(device);
1215 void drbd_ov_out_of_sync_found(struct drbd_device *device, sector_t sector, int size)
1217 if (device->ov_last_oos_start + device->ov_last_oos_size == sector) {
1218 device->ov_last_oos_size += size>>9;
1220 device->ov_last_oos_start = sector;
1221 device->ov_last_oos_size = size>>9;
1223 drbd_set_out_of_sync(device, sector, size);
1226 int w_e_end_ov_reply(struct drbd_work *w, int cancel)
1228 struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w);
1229 struct drbd_peer_device *peer_device = peer_req->peer_device;
1230 struct drbd_device *device = peer_device->device;
1231 struct digest_info *di;
1233 sector_t sector = peer_req->i.sector;
1234 unsigned int size = peer_req->i.size;
1237 bool stop_sector_reached = false;
1239 if (unlikely(cancel)) {
1240 drbd_free_peer_req(device, peer_req);
1241 dec_unacked(device);
1245 /* after "cancel", because after drbd_disconnect/drbd_rs_cancel_all
1246 * the resync lru has been cleaned up already */
1247 if (get_ldev(device)) {
1248 drbd_rs_complete_io(device, peer_req->i.sector);
1252 di = peer_req->digest;
1254 if (likely((peer_req->flags & EE_WAS_ERROR) == 0)) {
1255 digest_size = crypto_hash_digestsize(peer_device->connection->verify_tfm);
1256 digest = kmalloc(digest_size, GFP_NOIO);
1258 drbd_csum_ee(peer_device->connection->verify_tfm, peer_req, digest);
1260 D_ASSERT(device, digest_size == di->digest_size);
1261 eq = !memcmp(digest, di->digest, digest_size);
1266 /* Free peer_req and pages before send.
1267 * In case we block on congestion, we could otherwise run into
1268 * some distributed deadlock, if the other side blocks on
1269 * congestion as well, because our receiver blocks in
1270 * drbd_alloc_pages due to pp_in_use > max_buffers. */
1271 drbd_free_peer_req(device, peer_req);
1273 drbd_ov_out_of_sync_found(device, sector, size);
1275 ov_out_of_sync_print(device);
1277 err = drbd_send_ack_ex(peer_device, P_OV_RESULT, sector, size,
1278 eq ? ID_IN_SYNC : ID_OUT_OF_SYNC);
1280 dec_unacked(device);
1284 /* let's advance progress step marks only for every other megabyte */
1285 if ((device->ov_left & 0x200) == 0x200)
1286 drbd_advance_rs_marks(device, device->ov_left);
1288 stop_sector_reached = verify_can_do_stop_sector(device) &&
1289 (sector + (size>>9)) >= device->ov_stop_sector;
1291 if (device->ov_left == 0 || stop_sector_reached) {
1292 ov_out_of_sync_print(device);
1293 drbd_resync_finished(device);
1300 * We need to track the number of pending barrier acks,
1301 * and to be able to wait for them.
1302 * See also comment in drbd_adm_attach before drbd_suspend_io.
1304 static int drbd_send_barrier(struct drbd_connection *connection)
1306 struct p_barrier *p;
1307 struct drbd_socket *sock;
1309 sock = &connection->data;
1310 p = conn_prepare_command(connection, sock);
1313 p->barrier = connection->send.current_epoch_nr;
1315 connection->send.current_epoch_writes = 0;
1317 return conn_send_command(connection, sock, P_BARRIER, sizeof(*p), NULL, 0);
1320 int w_send_write_hint(struct drbd_work *w, int cancel)
1322 struct drbd_device *device =
1323 container_of(w, struct drbd_device, unplug_work);
1324 struct drbd_socket *sock;
1328 sock = &first_peer_device(device)->connection->data;
1329 if (!drbd_prepare_command(first_peer_device(device), sock))
1331 return drbd_send_command(first_peer_device(device), sock, P_UNPLUG_REMOTE, 0, NULL, 0);
1334 static void re_init_if_first_write(struct drbd_connection *connection, unsigned int epoch)
1336 if (!connection->send.seen_any_write_yet) {
1337 connection->send.seen_any_write_yet = true;
1338 connection->send.current_epoch_nr = epoch;
1339 connection->send.current_epoch_writes = 0;
1343 static void maybe_send_barrier(struct drbd_connection *connection, unsigned int epoch)
1345 /* re-init if first write on this connection */
1346 if (!connection->send.seen_any_write_yet)
1348 if (connection->send.current_epoch_nr != epoch) {
1349 if (connection->send.current_epoch_writes)
1350 drbd_send_barrier(connection);
1351 connection->send.current_epoch_nr = epoch;
1355 int w_send_out_of_sync(struct drbd_work *w, int cancel)
1357 struct drbd_request *req = container_of(w, struct drbd_request, w);
1358 struct drbd_device *device = req->device;
1359 struct drbd_peer_device *const peer_device = first_peer_device(device);
1360 struct drbd_connection *const connection = peer_device->connection;
1363 if (unlikely(cancel)) {
1364 req_mod(req, SEND_CANCELED);
1367 req->pre_send_jif = jiffies;
1369 /* this time, no connection->send.current_epoch_writes++;
1370 * If it was sent, it was the closing barrier for the last
1371 * replicated epoch, before we went into AHEAD mode.
1372 * No more barriers will be sent, until we leave AHEAD mode again. */
1373 maybe_send_barrier(connection, req->epoch);
1375 err = drbd_send_out_of_sync(peer_device, req);
1376 req_mod(req, OOS_HANDED_TO_NETWORK);
1382 * w_send_dblock() - Worker callback to send a P_DATA packet in order to mirror a write request
1384 * @cancel: The connection will be closed anyways
1386 int w_send_dblock(struct drbd_work *w, int cancel)
1388 struct drbd_request *req = container_of(w, struct drbd_request, w);
1389 struct drbd_device *device = req->device;
1390 struct drbd_peer_device *const peer_device = first_peer_device(device);
1391 struct drbd_connection *connection = peer_device->connection;
1394 if (unlikely(cancel)) {
1395 req_mod(req, SEND_CANCELED);
1398 req->pre_send_jif = jiffies;
1400 re_init_if_first_write(connection, req->epoch);
1401 maybe_send_barrier(connection, req->epoch);
1402 connection->send.current_epoch_writes++;
1404 err = drbd_send_dblock(peer_device, req);
1405 req_mod(req, err ? SEND_FAILED : HANDED_OVER_TO_NETWORK);
1411 * w_send_read_req() - Worker callback to send a read request (P_DATA_REQUEST) packet
1413 * @cancel: The connection will be closed anyways
1415 int w_send_read_req(struct drbd_work *w, int cancel)
1417 struct drbd_request *req = container_of(w, struct drbd_request, w);
1418 struct drbd_device *device = req->device;
1419 struct drbd_peer_device *const peer_device = first_peer_device(device);
1420 struct drbd_connection *connection = peer_device->connection;
1423 if (unlikely(cancel)) {
1424 req_mod(req, SEND_CANCELED);
1427 req->pre_send_jif = jiffies;
1429 /* Even read requests may close a write epoch,
1430 * if there was any yet. */
1431 maybe_send_barrier(connection, req->epoch);
1433 err = drbd_send_drequest(peer_device, P_DATA_REQUEST, req->i.sector, req->i.size,
1434 (unsigned long)req);
1436 req_mod(req, err ? SEND_FAILED : HANDED_OVER_TO_NETWORK);
1441 int w_restart_disk_io(struct drbd_work *w, int cancel)
1443 struct drbd_request *req = container_of(w, struct drbd_request, w);
1444 struct drbd_device *device = req->device;
1446 if (bio_data_dir(req->master_bio) == WRITE && req->rq_state & RQ_IN_ACT_LOG)
1447 drbd_al_begin_io(device, &req->i);
1449 drbd_req_make_private_bio(req, req->master_bio);
1450 req->private_bio->bi_bdev = device->ldev->backing_bdev;
1451 generic_make_request(req->private_bio);
1456 static int _drbd_may_sync_now(struct drbd_device *device)
1458 struct drbd_device *odev = device;
1462 if (!odev->ldev || odev->state.disk == D_DISKLESS)
1465 resync_after = rcu_dereference(odev->ldev->disk_conf)->resync_after;
1467 if (resync_after == -1)
1469 odev = minor_to_device(resync_after);
1472 if ((odev->state.conn >= C_SYNC_SOURCE &&
1473 odev->state.conn <= C_PAUSED_SYNC_T) ||
1474 odev->state.aftr_isp || odev->state.peer_isp ||
1475 odev->state.user_isp)
1481 * _drbd_pause_after() - Pause resync on all devices that may not resync now
1482 * @device: DRBD device.
1484 * Called from process context only (admin command and after_state_ch).
1486 static int _drbd_pause_after(struct drbd_device *device)
1488 struct drbd_device *odev;
1492 idr_for_each_entry(&drbd_devices, odev, i) {
1493 if (odev->state.conn == C_STANDALONE && odev->state.disk == D_DISKLESS)
1495 if (!_drbd_may_sync_now(odev))
1496 rv |= (__drbd_set_state(_NS(odev, aftr_isp, 1), CS_HARD, NULL)
1497 != SS_NOTHING_TO_DO);
1505 * _drbd_resume_next() - Resume resync on all devices that may resync now
1506 * @device: DRBD device.
1508 * Called from process context only (admin command and worker).
1510 static int _drbd_resume_next(struct drbd_device *device)
1512 struct drbd_device *odev;
1516 idr_for_each_entry(&drbd_devices, odev, i) {
1517 if (odev->state.conn == C_STANDALONE && odev->state.disk == D_DISKLESS)
1519 if (odev->state.aftr_isp) {
1520 if (_drbd_may_sync_now(odev))
1521 rv |= (__drbd_set_state(_NS(odev, aftr_isp, 0),
1523 != SS_NOTHING_TO_DO) ;
1530 void resume_next_sg(struct drbd_device *device)
1532 write_lock_irq(&global_state_lock);
1533 _drbd_resume_next(device);
1534 write_unlock_irq(&global_state_lock);
1537 void suspend_other_sg(struct drbd_device *device)
1539 write_lock_irq(&global_state_lock);
1540 _drbd_pause_after(device);
1541 write_unlock_irq(&global_state_lock);
1544 /* caller must hold global_state_lock */
1545 enum drbd_ret_code drbd_resync_after_valid(struct drbd_device *device, int o_minor)
1547 struct drbd_device *odev;
1552 if (o_minor < -1 || o_minor > MINORMASK)
1553 return ERR_RESYNC_AFTER;
1555 /* check for loops */
1556 odev = minor_to_device(o_minor);
1559 return ERR_RESYNC_AFTER_CYCLE;
1561 /* You are free to depend on diskless, non-existing,
1562 * or not yet/no longer existing minors.
1563 * We only reject dependency loops.
1564 * We cannot follow the dependency chain beyond a detached or
1567 if (!odev || !odev->ldev || odev->state.disk == D_DISKLESS)
1571 resync_after = rcu_dereference(odev->ldev->disk_conf)->resync_after;
1573 /* dependency chain ends here, no cycles. */
1574 if (resync_after == -1)
1577 /* follow the dependency chain */
1578 odev = minor_to_device(resync_after);
1582 /* caller must hold global_state_lock */
1583 void drbd_resync_after_changed(struct drbd_device *device)
1588 changes = _drbd_pause_after(device);
1589 changes |= _drbd_resume_next(device);
1593 void drbd_rs_controller_reset(struct drbd_device *device)
1595 struct fifo_buffer *plan;
1597 atomic_set(&device->rs_sect_in, 0);
1598 atomic_set(&device->rs_sect_ev, 0);
1599 device->rs_in_flight = 0;
1601 /* Updating the RCU protected object in place is necessary since
1602 this function gets called from atomic context.
1603 It is valid since all other updates also lead to an completely
1606 plan = rcu_dereference(device->rs_plan_s);
1612 void start_resync_timer_fn(unsigned long data)
1614 struct drbd_device *device = (struct drbd_device *) data;
1615 drbd_device_post_work(device, RS_START);
1618 static void do_start_resync(struct drbd_device *device)
1620 if (atomic_read(&device->unacked_cnt) || atomic_read(&device->rs_pending_cnt)) {
1621 drbd_warn(device, "postponing start_resync ...\n");
1622 device->start_resync_timer.expires = jiffies + HZ/10;
1623 add_timer(&device->start_resync_timer);
1627 drbd_start_resync(device, C_SYNC_SOURCE);
1628 clear_bit(AHEAD_TO_SYNC_SOURCE, &device->flags);
1631 static bool use_checksum_based_resync(struct drbd_connection *connection, struct drbd_device *device)
1633 bool csums_after_crash_only;
1635 csums_after_crash_only = rcu_dereference(connection->net_conf)->csums_after_crash_only;
1637 return connection->agreed_pro_version >= 89 && /* supported? */
1638 connection->csums_tfm && /* configured? */
1639 (csums_after_crash_only == 0 /* use for each resync? */
1640 || test_bit(CRASHED_PRIMARY, &device->flags)); /* or only after Primary crash? */
1644 * drbd_start_resync() - Start the resync process
1645 * @device: DRBD device.
1646 * @side: Either C_SYNC_SOURCE or C_SYNC_TARGET
1648 * This function might bring you directly into one of the
1649 * C_PAUSED_SYNC_* states.
1651 void drbd_start_resync(struct drbd_device *device, enum drbd_conns side)
1653 struct drbd_peer_device *peer_device = first_peer_device(device);
1654 struct drbd_connection *connection = peer_device ? peer_device->connection : NULL;
1655 union drbd_state ns;
1658 if (device->state.conn >= C_SYNC_SOURCE && device->state.conn < C_AHEAD) {
1659 drbd_err(device, "Resync already running!\n");
1663 if (!test_bit(B_RS_H_DONE, &device->flags)) {
1664 if (side == C_SYNC_TARGET) {
1665 /* Since application IO was locked out during C_WF_BITMAP_T and
1666 C_WF_SYNC_UUID we are still unmodified. Before going to C_SYNC_TARGET
1667 we check that we might make the data inconsistent. */
1668 r = drbd_khelper(device, "before-resync-target");
1669 r = (r >> 8) & 0xff;
1671 drbd_info(device, "before-resync-target handler returned %d, "
1672 "dropping connection.\n", r);
1673 conn_request_state(connection, NS(conn, C_DISCONNECTING), CS_HARD);
1676 } else /* C_SYNC_SOURCE */ {
1677 r = drbd_khelper(device, "before-resync-source");
1678 r = (r >> 8) & 0xff;
1681 drbd_info(device, "before-resync-source handler returned %d, "
1682 "ignoring. Old userland tools?", r);
1684 drbd_info(device, "before-resync-source handler returned %d, "
1685 "dropping connection.\n", r);
1686 conn_request_state(connection,
1687 NS(conn, C_DISCONNECTING), CS_HARD);
1694 if (current == connection->worker.task) {
1695 /* The worker should not sleep waiting for state_mutex,
1696 that can take long */
1697 if (!mutex_trylock(device->state_mutex)) {
1698 set_bit(B_RS_H_DONE, &device->flags);
1699 device->start_resync_timer.expires = jiffies + HZ/5;
1700 add_timer(&device->start_resync_timer);
1704 mutex_lock(device->state_mutex);
1706 clear_bit(B_RS_H_DONE, &device->flags);
1708 /* req_lock: serialize with drbd_send_and_submit() and others
1709 * global_state_lock: for stable sync-after dependencies */
1710 spin_lock_irq(&device->resource->req_lock);
1711 write_lock(&global_state_lock);
1712 /* Did some connection breakage or IO error race with us? */
1713 if (device->state.conn < C_CONNECTED
1714 || !get_ldev_if_state(device, D_NEGOTIATING)) {
1715 write_unlock(&global_state_lock);
1716 spin_unlock_irq(&device->resource->req_lock);
1717 mutex_unlock(device->state_mutex);
1721 ns = drbd_read_state(device);
1723 ns.aftr_isp = !_drbd_may_sync_now(device);
1727 if (side == C_SYNC_TARGET)
1728 ns.disk = D_INCONSISTENT;
1729 else /* side == C_SYNC_SOURCE */
1730 ns.pdsk = D_INCONSISTENT;
1732 r = __drbd_set_state(device, ns, CS_VERBOSE, NULL);
1733 ns = drbd_read_state(device);
1735 if (ns.conn < C_CONNECTED)
1736 r = SS_UNKNOWN_ERROR;
1738 if (r == SS_SUCCESS) {
1739 unsigned long tw = drbd_bm_total_weight(device);
1740 unsigned long now = jiffies;
1743 device->rs_failed = 0;
1744 device->rs_paused = 0;
1745 device->rs_same_csum = 0;
1746 device->rs_last_events = 0;
1747 device->rs_last_sect_ev = 0;
1748 device->rs_total = tw;
1749 device->rs_start = now;
1750 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
1751 device->rs_mark_left[i] = tw;
1752 device->rs_mark_time[i] = now;
1754 _drbd_pause_after(device);
1755 /* Forget potentially stale cached per resync extent bit-counts.
1756 * Open coded drbd_rs_cancel_all(device), we already have IRQs
1757 * disabled, and know the disk state is ok. */
1758 spin_lock(&device->al_lock);
1759 lc_reset(device->resync);
1760 device->resync_locked = 0;
1761 device->resync_wenr = LC_FREE;
1762 spin_unlock(&device->al_lock);
1764 write_unlock(&global_state_lock);
1765 spin_unlock_irq(&device->resource->req_lock);
1767 if (r == SS_SUCCESS) {
1768 wake_up(&device->al_wait); /* for lc_reset() above */
1769 /* reset rs_last_bcast when a resync or verify is started,
1770 * to deal with potential jiffies wrap. */
1771 device->rs_last_bcast = jiffies - HZ;
1773 drbd_info(device, "Began resync as %s (will sync %lu KB [%lu bits set]).\n",
1774 drbd_conn_str(ns.conn),
1775 (unsigned long) device->rs_total << (BM_BLOCK_SHIFT-10),
1776 (unsigned long) device->rs_total);
1777 if (side == C_SYNC_TARGET) {
1778 device->bm_resync_fo = 0;
1779 device->use_csums = use_checksum_based_resync(connection, device);
1781 device->use_csums = 0;
1784 /* Since protocol 96, we must serialize drbd_gen_and_send_sync_uuid
1785 * with w_send_oos, or the sync target will get confused as to
1786 * how much bits to resync. We cannot do that always, because for an
1787 * empty resync and protocol < 95, we need to do it here, as we call
1788 * drbd_resync_finished from here in that case.
1789 * We drbd_gen_and_send_sync_uuid here for protocol < 96,
1790 * and from after_state_ch otherwise. */
1791 if (side == C_SYNC_SOURCE && connection->agreed_pro_version < 96)
1792 drbd_gen_and_send_sync_uuid(peer_device);
1794 if (connection->agreed_pro_version < 95 && device->rs_total == 0) {
1795 /* This still has a race (about when exactly the peers
1796 * detect connection loss) that can lead to a full sync
1797 * on next handshake. In 8.3.9 we fixed this with explicit
1798 * resync-finished notifications, but the fix
1799 * introduces a protocol change. Sleeping for some
1800 * time longer than the ping interval + timeout on the
1801 * SyncSource, to give the SyncTarget the chance to
1802 * detect connection loss, then waiting for a ping
1803 * response (implicit in drbd_resync_finished) reduces
1804 * the race considerably, but does not solve it. */
1805 if (side == C_SYNC_SOURCE) {
1806 struct net_conf *nc;
1810 nc = rcu_dereference(connection->net_conf);
1811 timeo = nc->ping_int * HZ + nc->ping_timeo * HZ / 9;
1813 schedule_timeout_interruptible(timeo);
1815 drbd_resync_finished(device);
1818 drbd_rs_controller_reset(device);
1819 /* ns.conn may already be != device->state.conn,
1820 * we may have been paused in between, or become paused until
1821 * the timer triggers.
1822 * No matter, that is handled in resync_timer_fn() */
1823 if (ns.conn == C_SYNC_TARGET)
1824 mod_timer(&device->resync_timer, jiffies);
1826 drbd_md_sync(device);
1829 mutex_unlock(device->state_mutex);
1832 static void update_on_disk_bitmap(struct drbd_device *device, bool resync_done)
1834 struct sib_info sib = { .sib_reason = SIB_SYNC_PROGRESS, };
1835 device->rs_last_bcast = jiffies;
1837 if (!get_ldev(device))
1840 drbd_bm_write_lazy(device, 0);
1841 if (resync_done && is_sync_state(device->state.conn))
1842 drbd_resync_finished(device);
1844 drbd_bcast_event(device, &sib);
1845 /* update timestamp, in case it took a while to write out stuff */
1846 device->rs_last_bcast = jiffies;
1850 static void drbd_ldev_destroy(struct drbd_device *device)
1852 lc_destroy(device->resync);
1853 device->resync = NULL;
1854 lc_destroy(device->act_log);
1855 device->act_log = NULL;
1858 drbd_free_ldev(device->ldev);
1859 device->ldev = NULL;
1862 clear_bit(GOING_DISKLESS, &device->flags);
1863 wake_up(&device->misc_wait);
1866 static void go_diskless(struct drbd_device *device)
1868 D_ASSERT(device, device->state.disk == D_FAILED);
1869 /* we cannot assert local_cnt == 0 here, as get_ldev_if_state will
1870 * inc/dec it frequently. Once we are D_DISKLESS, no one will touch
1871 * the protected members anymore, though, so once put_ldev reaches zero
1872 * again, it will be safe to free them. */
1874 /* Try to write changed bitmap pages, read errors may have just
1875 * set some bits outside the area covered by the activity log.
1877 * If we have an IO error during the bitmap writeout,
1878 * we will want a full sync next time, just in case.
1879 * (Do we want a specific meta data flag for this?)
1881 * If that does not make it to stable storage either,
1882 * we cannot do anything about that anymore.
1884 * We still need to check if both bitmap and ldev are present, we may
1885 * end up here after a failed attach, before ldev was even assigned.
1887 if (device->bitmap && device->ldev) {
1888 /* An interrupted resync or similar is allowed to recounts bits
1890 * Any modifications would not be expected anymore, though.
1892 if (drbd_bitmap_io_from_worker(device, drbd_bm_write,
1893 "detach", BM_LOCKED_TEST_ALLOWED)) {
1894 if (test_bit(WAS_READ_ERROR, &device->flags)) {
1895 drbd_md_set_flag(device, MDF_FULL_SYNC);
1896 drbd_md_sync(device);
1901 drbd_force_state(device, NS(disk, D_DISKLESS));
1904 static int do_md_sync(struct drbd_device *device)
1906 drbd_warn(device, "md_sync_timer expired! Worker calls drbd_md_sync().\n");
1907 drbd_md_sync(device);
1911 /* only called from drbd_worker thread, no locking */
1912 void __update_timing_details(
1913 struct drbd_thread_timing_details *tdp,
1914 unsigned int *cb_nr,
1916 const char *fn, const unsigned int line)
1918 unsigned int i = *cb_nr % DRBD_THREAD_DETAILS_HIST;
1919 struct drbd_thread_timing_details *td = tdp + i;
1921 td->start_jif = jiffies;
1927 i = (i+1) % DRBD_THREAD_DETAILS_HIST;
1929 memset(td, 0, sizeof(*td));
1934 static void do_device_work(struct drbd_device *device, const unsigned long todo)
1936 if (test_bit(MD_SYNC, &todo))
1938 if (test_bit(RS_DONE, &todo) ||
1939 test_bit(RS_PROGRESS, &todo))
1940 update_on_disk_bitmap(device, test_bit(RS_DONE, &todo));
1941 if (test_bit(GO_DISKLESS, &todo))
1942 go_diskless(device);
1943 if (test_bit(DESTROY_DISK, &todo))
1944 drbd_ldev_destroy(device);
1945 if (test_bit(RS_START, &todo))
1946 do_start_resync(device);
1949 #define DRBD_DEVICE_WORK_MASK \
1950 ((1UL << GO_DISKLESS) \
1951 |(1UL << DESTROY_DISK) \
1953 |(1UL << RS_START) \
1954 |(1UL << RS_PROGRESS) \
1958 static unsigned long get_work_bits(unsigned long *flags)
1960 unsigned long old, new;
1963 new = old & ~DRBD_DEVICE_WORK_MASK;
1964 } while (cmpxchg(flags, old, new) != old);
1965 return old & DRBD_DEVICE_WORK_MASK;
1968 static void do_unqueued_work(struct drbd_connection *connection)
1970 struct drbd_peer_device *peer_device;
1974 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
1975 struct drbd_device *device = peer_device->device;
1976 unsigned long todo = get_work_bits(&device->flags);
1980 kref_get(&device->kref);
1982 do_device_work(device, todo);
1983 kref_put(&device->kref, drbd_destroy_device);
1989 static bool dequeue_work_batch(struct drbd_work_queue *queue, struct list_head *work_list)
1991 spin_lock_irq(&queue->q_lock);
1992 list_splice_tail_init(&queue->q, work_list);
1993 spin_unlock_irq(&queue->q_lock);
1994 return !list_empty(work_list);
1997 static void wait_for_work(struct drbd_connection *connection, struct list_head *work_list)
2000 struct net_conf *nc;
2003 dequeue_work_batch(&connection->sender_work, work_list);
2004 if (!list_empty(work_list))
2007 /* Still nothing to do?
2008 * Maybe we still need to close the current epoch,
2009 * even if no new requests are queued yet.
2011 * Also, poke TCP, just in case.
2012 * Then wait for new work (or signal). */
2014 nc = rcu_dereference(connection->net_conf);
2015 uncork = nc ? nc->tcp_cork : 0;
2018 mutex_lock(&connection->data.mutex);
2019 if (connection->data.socket)
2020 drbd_tcp_uncork(connection->data.socket);
2021 mutex_unlock(&connection->data.mutex);
2026 prepare_to_wait(&connection->sender_work.q_wait, &wait, TASK_INTERRUPTIBLE);
2027 spin_lock_irq(&connection->resource->req_lock);
2028 spin_lock(&connection->sender_work.q_lock); /* FIXME get rid of this one? */
2029 if (!list_empty(&connection->sender_work.q))
2030 list_splice_tail_init(&connection->sender_work.q, work_list);
2031 spin_unlock(&connection->sender_work.q_lock); /* FIXME get rid of this one? */
2032 if (!list_empty(work_list) || signal_pending(current)) {
2033 spin_unlock_irq(&connection->resource->req_lock);
2037 /* We found nothing new to do, no to-be-communicated request,
2038 * no other work item. We may still need to close the last
2039 * epoch. Next incoming request epoch will be connection ->
2040 * current transfer log epoch number. If that is different
2041 * from the epoch of the last request we communicated, it is
2042 * safe to send the epoch separating barrier now.
2045 atomic_read(&connection->current_tle_nr) !=
2046 connection->send.current_epoch_nr;
2047 spin_unlock_irq(&connection->resource->req_lock);
2050 maybe_send_barrier(connection,
2051 connection->send.current_epoch_nr + 1);
2053 if (test_bit(DEVICE_WORK_PENDING, &connection->flags))
2056 /* drbd_send() may have called flush_signals() */
2057 if (get_t_state(&connection->worker) != RUNNING)
2061 /* may be woken up for other things but new work, too,
2062 * e.g. if the current epoch got closed.
2063 * In which case we send the barrier above. */
2065 finish_wait(&connection->sender_work.q_wait, &wait);
2067 /* someone may have changed the config while we have been waiting above. */
2069 nc = rcu_dereference(connection->net_conf);
2070 cork = nc ? nc->tcp_cork : 0;
2072 mutex_lock(&connection->data.mutex);
2073 if (connection->data.socket) {
2075 drbd_tcp_cork(connection->data.socket);
2077 drbd_tcp_uncork(connection->data.socket);
2079 mutex_unlock(&connection->data.mutex);
2082 int drbd_worker(struct drbd_thread *thi)
2084 struct drbd_connection *connection = thi->connection;
2085 struct drbd_work *w = NULL;
2086 struct drbd_peer_device *peer_device;
2087 LIST_HEAD(work_list);
2090 while (get_t_state(thi) == RUNNING) {
2091 drbd_thread_current_set_cpu(thi);
2093 if (list_empty(&work_list)) {
2094 update_worker_timing_details(connection, wait_for_work);
2095 wait_for_work(connection, &work_list);
2098 if (test_and_clear_bit(DEVICE_WORK_PENDING, &connection->flags)) {
2099 update_worker_timing_details(connection, do_unqueued_work);
2100 do_unqueued_work(connection);
2103 if (signal_pending(current)) {
2104 flush_signals(current);
2105 if (get_t_state(thi) == RUNNING) {
2106 drbd_warn(connection, "Worker got an unexpected signal\n");
2112 if (get_t_state(thi) != RUNNING)
2115 if (!list_empty(&work_list)) {
2116 w = list_first_entry(&work_list, struct drbd_work, list);
2117 list_del_init(&w->list);
2118 update_worker_timing_details(connection, w->cb);
2119 if (w->cb(w, connection->cstate < C_WF_REPORT_PARAMS) == 0)
2121 if (connection->cstate >= C_WF_REPORT_PARAMS)
2122 conn_request_state(connection, NS(conn, C_NETWORK_FAILURE), CS_HARD);
2127 if (test_and_clear_bit(DEVICE_WORK_PENDING, &connection->flags)) {
2128 update_worker_timing_details(connection, do_unqueued_work);
2129 do_unqueued_work(connection);
2131 if (!list_empty(&work_list)) {
2132 w = list_first_entry(&work_list, struct drbd_work, list);
2133 list_del_init(&w->list);
2134 update_worker_timing_details(connection, w->cb);
2137 dequeue_work_batch(&connection->sender_work, &work_list);
2138 } while (!list_empty(&work_list) || test_bit(DEVICE_WORK_PENDING, &connection->flags));
2141 idr_for_each_entry(&connection->peer_devices, peer_device, vnr) {
2142 struct drbd_device *device = peer_device->device;
2143 D_ASSERT(device, device->state.disk == D_DISKLESS && device->state.conn == C_STANDALONE);
2144 kref_get(&device->kref);
2146 drbd_device_cleanup(device);
2147 kref_put(&device->kref, drbd_destroy_device);