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);
50 * more endio handlers:
51 atodb_endio in drbd_actlog.c
52 drbd_bm_async_io_complete in drbd_bitmap.c
54 * For all these callbacks, note the following:
55 * The callbacks will be called in irq context by the IDE drivers,
56 * and in Softirqs/Tasklets/BH context by the SCSI drivers.
57 * Try to get the locking right :)
62 /* About the global_state_lock
63 Each state transition on an device holds a read lock. In case we have
64 to evaluate the sync after dependencies, we grab a write lock, because
65 we need stable states on all devices for that. */
66 rwlock_t global_state_lock;
68 /* used for synchronous meta data and bitmap IO
69 * submitted by drbd_md_sync_page_io()
71 void drbd_md_io_complete(struct bio *bio, int error)
73 struct drbd_md_io *md_io;
75 md_io = (struct drbd_md_io *)bio->bi_private;
78 complete(&md_io->event);
81 /* reads on behalf of the partner,
82 * "submitted" by the receiver
84 void drbd_endio_read_sec_final(struct drbd_epoch_entry *e) __releases(local)
86 unsigned long flags = 0;
87 struct drbd_conf *mdev = e->mdev;
89 D_ASSERT(e->block_id != ID_VACANT);
91 spin_lock_irqsave(&mdev->req_lock, flags);
92 mdev->read_cnt += e->size >> 9;
94 if (list_empty(&mdev->read_ee))
95 wake_up(&mdev->ee_wait);
96 if (test_bit(__EE_WAS_ERROR, &e->flags))
97 __drbd_chk_io_error(mdev, FALSE);
98 spin_unlock_irqrestore(&mdev->req_lock, flags);
100 drbd_queue_work(&mdev->data.work, &e->w);
104 /* writes on behalf of the partner, or resync writes,
105 * "submitted" by the receiver, final stage. */
106 static void drbd_endio_write_sec_final(struct drbd_epoch_entry *e) __releases(local)
108 unsigned long flags = 0;
109 struct drbd_conf *mdev = e->mdev;
113 int do_al_complete_io;
115 D_ASSERT(e->block_id != ID_VACANT);
117 /* after we moved e to done_ee,
118 * we may no longer access it,
119 * it may be freed/reused already!
120 * (as soon as we release the req_lock) */
121 e_sector = e->sector;
122 do_al_complete_io = e->flags & EE_CALL_AL_COMPLETE_IO;
123 is_syncer_req = is_syncer_block_id(e->block_id);
125 spin_lock_irqsave(&mdev->req_lock, flags);
126 mdev->writ_cnt += e->size >> 9;
127 list_del(&e->w.list); /* has been on active_ee or sync_ee */
128 list_add_tail(&e->w.list, &mdev->done_ee);
130 /* No hlist_del_init(&e->colision) here, we did not send the Ack yet,
131 * neither did we wake possibly waiting conflicting requests.
132 * done from "drbd_process_done_ee" within the appropriate w.cb
133 * (e_end_block/e_end_resync_block) or from _drbd_clear_done_ee */
135 do_wake = is_syncer_req
136 ? list_empty(&mdev->sync_ee)
137 : list_empty(&mdev->active_ee);
139 if (test_bit(__EE_WAS_ERROR, &e->flags))
140 __drbd_chk_io_error(mdev, FALSE);
141 spin_unlock_irqrestore(&mdev->req_lock, flags);
144 drbd_rs_complete_io(mdev, e_sector);
147 wake_up(&mdev->ee_wait);
149 if (do_al_complete_io)
150 drbd_al_complete_io(mdev, e_sector);
156 /* writes on behalf of the partner, or resync writes,
157 * "submitted" by the receiver.
159 void drbd_endio_sec(struct bio *bio, int error)
161 struct drbd_epoch_entry *e = bio->bi_private;
162 struct drbd_conf *mdev = e->mdev;
163 int uptodate = bio_flagged(bio, BIO_UPTODATE);
164 int is_write = bio_data_dir(bio) == WRITE;
167 dev_warn(DEV, "%s: error=%d s=%llus\n",
168 is_write ? "write" : "read", error,
169 (unsigned long long)e->sector);
170 if (!error && !uptodate) {
171 dev_warn(DEV, "%s: setting error to -EIO s=%llus\n",
172 is_write ? "write" : "read",
173 (unsigned long long)e->sector);
174 /* strange behavior of some lower level drivers...
175 * fail the request by clearing the uptodate flag,
176 * but do not return any error?! */
181 set_bit(__EE_WAS_ERROR, &e->flags);
183 bio_put(bio); /* no need for the bio anymore */
184 if (atomic_dec_and_test(&e->pending_bios)) {
186 drbd_endio_write_sec_final(e);
188 drbd_endio_read_sec_final(e);
192 /* read, readA or write requests on R_PRIMARY coming from drbd_make_request
194 void drbd_endio_pri(struct bio *bio, int error)
196 struct drbd_request *req = bio->bi_private;
197 struct drbd_conf *mdev = req->mdev;
198 enum drbd_req_event what;
199 int uptodate = bio_flagged(bio, BIO_UPTODATE);
201 if (!error && !uptodate) {
202 dev_warn(DEV, "p %s: setting error to -EIO\n",
203 bio_data_dir(bio) == WRITE ? "write" : "read");
204 /* strange behavior of some lower level drivers...
205 * fail the request by clearing the uptodate flag,
206 * but do not return any error?! */
210 /* to avoid recursion in __req_mod */
211 if (unlikely(error)) {
212 what = (bio_data_dir(bio) == WRITE)
213 ? write_completed_with_error
214 : (bio_rw(bio) == READ)
215 ? read_completed_with_error
216 : read_ahead_completed_with_error;
220 bio_put(req->private_bio);
221 req->private_bio = ERR_PTR(error);
226 int w_read_retry_remote(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
228 struct drbd_request *req = container_of(w, struct drbd_request, w);
230 /* We should not detach for read io-error,
231 * but try to WRITE the P_DATA_REPLY to the failed location,
232 * to give the disk the chance to relocate that block */
234 spin_lock_irq(&mdev->req_lock);
235 if (cancel || mdev->state.pdsk != D_UP_TO_DATE) {
236 _req_mod(req, read_retry_remote_canceled);
237 spin_unlock_irq(&mdev->req_lock);
240 spin_unlock_irq(&mdev->req_lock);
242 return w_send_read_req(mdev, w, 0);
245 int w_resync_inactive(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
247 ERR_IF(cancel) return 1;
248 dev_err(DEV, "resync inactive, but callback triggered??\n");
249 return 1; /* Simply ignore this! */
252 void drbd_csum_ee(struct drbd_conf *mdev, struct crypto_hash *tfm, struct drbd_epoch_entry *e, void *digest)
254 struct hash_desc desc;
255 struct scatterlist sg;
256 struct page *page = e->pages;
263 sg_init_table(&sg, 1);
264 crypto_hash_init(&desc);
266 while ((tmp = page_chain_next(page))) {
267 /* all but the last page will be fully used */
268 sg_set_page(&sg, page, PAGE_SIZE, 0);
269 crypto_hash_update(&desc, &sg, sg.length);
272 /* and now the last, possibly only partially used page */
273 len = e->size & (PAGE_SIZE - 1);
274 sg_set_page(&sg, page, len ?: PAGE_SIZE, 0);
275 crypto_hash_update(&desc, &sg, sg.length);
276 crypto_hash_final(&desc, digest);
279 void drbd_csum_bio(struct drbd_conf *mdev, struct crypto_hash *tfm, struct bio *bio, void *digest)
281 struct hash_desc desc;
282 struct scatterlist sg;
283 struct bio_vec *bvec;
289 sg_init_table(&sg, 1);
290 crypto_hash_init(&desc);
292 __bio_for_each_segment(bvec, bio, i, 0) {
293 sg_set_page(&sg, bvec->bv_page, bvec->bv_len, bvec->bv_offset);
294 crypto_hash_update(&desc, &sg, sg.length);
296 crypto_hash_final(&desc, digest);
299 static int w_e_send_csum(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
301 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w);
306 D_ASSERT(e->block_id == DRBD_MAGIC + 0xbeef);
308 if (unlikely(cancel)) {
309 drbd_free_ee(mdev, e);
313 if (likely((e->flags & EE_WAS_ERROR) == 0)) {
314 digest_size = crypto_hash_digestsize(mdev->csums_tfm);
315 digest = kmalloc(digest_size, GFP_NOIO);
317 drbd_csum_ee(mdev, mdev->csums_tfm, e, digest);
319 inc_rs_pending(mdev);
320 ok = drbd_send_drequest_csum(mdev,
328 dev_err(DEV, "kmalloc() of digest failed.\n");
334 drbd_free_ee(mdev, e);
337 dev_err(DEV, "drbd_send_drequest(..., csum) failed\n");
341 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN)
343 static int read_for_csum(struct drbd_conf *mdev, sector_t sector, int size)
345 struct drbd_epoch_entry *e;
350 if (drbd_rs_should_slow_down(mdev))
353 /* GFP_TRY, because if there is no memory available right now, this may
354 * be rescheduled for later. It is "only" background resync, after all. */
355 e = drbd_alloc_ee(mdev, DRBD_MAGIC+0xbeef, sector, size, GFP_TRY);
359 e->w.cb = w_e_send_csum;
360 spin_lock_irq(&mdev->req_lock);
361 list_add(&e->w.list, &mdev->read_ee);
362 spin_unlock_irq(&mdev->req_lock);
364 atomic_add(size >> 9, &mdev->rs_sect_ev);
365 if (drbd_submit_ee(mdev, e, READ, DRBD_FAULT_RS_RD) == 0)
368 /* drbd_submit_ee currently fails for one reason only:
369 * not being able to allocate enough bios.
370 * Is dropping the connection going to help? */
371 spin_lock_irq(&mdev->req_lock);
372 list_del(&e->w.list);
373 spin_unlock_irq(&mdev->req_lock);
375 drbd_free_ee(mdev, e);
381 void resync_timer_fn(unsigned long data)
383 struct drbd_conf *mdev = (struct drbd_conf *) data;
387 switch (mdev->state.conn) {
389 mdev->resync_work.cb = w_make_ov_request;
392 mdev->resync_work.cb = w_make_resync_request;
396 mdev->resync_work.cb = w_resync_inactive;
399 /* harmless race: list_empty outside data.work.q_lock */
400 if (list_empty(&mdev->resync_work.list) && queue)
401 drbd_queue_work(&mdev->data.work, &mdev->resync_work);
404 static void fifo_set(struct fifo_buffer *fb, int value)
408 for (i = 0; i < fb->size; i++)
409 fb->values[i] = value;
412 static int fifo_push(struct fifo_buffer *fb, int value)
416 ov = fb->values[fb->head_index];
417 fb->values[fb->head_index++] = value;
419 if (fb->head_index >= fb->size)
425 static void fifo_add_val(struct fifo_buffer *fb, int value)
429 for (i = 0; i < fb->size; i++)
430 fb->values[i] += value;
433 int drbd_rs_controller(struct drbd_conf *mdev)
435 unsigned int sect_in; /* Number of sectors that came in since the last turn */
436 unsigned int want; /* The number of sectors we want in the proxy */
437 int req_sect; /* Number of sectors to request in this turn */
438 int correction; /* Number of sectors more we need in the proxy*/
439 int cps; /* correction per invocation of drbd_rs_controller() */
440 int steps; /* Number of time steps to plan ahead */
444 sect_in = atomic_xchg(&mdev->rs_sect_in, 0); /* Number of sectors that came in */
445 mdev->rs_in_flight -= sect_in;
447 spin_lock(&mdev->peer_seq_lock); /* get an atomic view on mdev->rs_plan_s */
449 steps = mdev->rs_plan_s.size; /* (mdev->sync_conf.c_plan_ahead * 10 * SLEEP_TIME) / HZ; */
451 if (mdev->rs_in_flight + sect_in == 0) { /* At start of resync */
452 want = ((mdev->sync_conf.rate * 2 * SLEEP_TIME) / HZ) * steps;
453 } else { /* normal path */
454 want = mdev->sync_conf.c_fill_target ? mdev->sync_conf.c_fill_target :
455 sect_in * mdev->sync_conf.c_delay_target * HZ / (SLEEP_TIME * 10);
458 correction = want - mdev->rs_in_flight - mdev->rs_planed;
461 cps = correction / steps;
462 fifo_add_val(&mdev->rs_plan_s, cps);
463 mdev->rs_planed += cps * steps;
465 /* What we do in this step */
466 curr_corr = fifo_push(&mdev->rs_plan_s, 0);
467 spin_unlock(&mdev->peer_seq_lock);
468 mdev->rs_planed -= curr_corr;
470 req_sect = sect_in + curr_corr;
474 max_sect = (mdev->sync_conf.c_max_rate * 2 * SLEEP_TIME) / HZ;
475 if (req_sect > max_sect)
479 dev_warn(DEV, "si=%u if=%d wa=%u co=%d st=%d cps=%d pl=%d cc=%d rs=%d\n",
480 sect_in, mdev->rs_in_flight, want, correction,
481 steps, cps, mdev->rs_planed, curr_corr, req_sect);
487 int w_make_resync_request(struct drbd_conf *mdev,
488 struct drbd_work *w, int cancel)
492 const sector_t capacity = drbd_get_capacity(mdev->this_bdev);
493 int max_segment_size;
494 int number, rollback_i, size, pe, mx;
495 int align, queued, sndbuf;
498 if (unlikely(cancel))
501 if (unlikely(mdev->state.conn < C_CONNECTED)) {
502 dev_err(DEV, "Confused in w_make_resync_request()! cstate < Connected");
506 if (mdev->state.conn != C_SYNC_TARGET)
507 dev_err(DEV, "%s in w_make_resync_request\n",
508 drbd_conn_str(mdev->state.conn));
510 if (mdev->rs_total == 0) {
512 drbd_resync_finished(mdev);
516 if (!get_ldev(mdev)) {
517 /* Since we only need to access mdev->rsync a
518 get_ldev_if_state(mdev,D_FAILED) would be sufficient, but
519 to continue resync with a broken disk makes no sense at
521 dev_err(DEV, "Disk broke down during resync!\n");
522 mdev->resync_work.cb = w_resync_inactive;
526 /* starting with drbd 8.3.8, we can handle multi-bio EEs,
527 * if it should be necessary */
529 mdev->agreed_pro_version < 94 ? queue_max_segment_size(mdev->rq_queue) :
530 mdev->agreed_pro_version < 95 ? DRBD_MAX_SIZE_H80_PACKET : DRBD_MAX_SEGMENT_SIZE;
532 if (mdev->rs_plan_s.size) { /* mdev->sync_conf.c_plan_ahead */
533 number = drbd_rs_controller(mdev) >> (BM_BLOCK_SHIFT - 9);
534 mdev->c_sync_rate = number * HZ * (BM_BLOCK_SIZE / 1024) / SLEEP_TIME;
536 mdev->c_sync_rate = mdev->sync_conf.rate;
537 number = SLEEP_TIME * mdev->c_sync_rate / ((BM_BLOCK_SIZE / 1024) * HZ);
540 /* Throttle resync on lower level disk activity, which may also be
541 * caused by application IO on Primary/SyncTarget.
542 * Keep this after the call to drbd_rs_controller, as that assumes
543 * to be called as precisely as possible every SLEEP_TIME,
544 * and would be confused otherwise. */
545 if (drbd_rs_should_slow_down(mdev))
548 mutex_lock(&mdev->data.mutex);
549 if (mdev->data.socket)
550 mx = mdev->data.socket->sk->sk_rcvbuf / sizeof(struct p_block_req);
553 mutex_unlock(&mdev->data.mutex);
555 /* For resync rates >160MB/sec, allow more pending RS requests */
559 /* Limit the number of pending RS requests to no more than the peer's receive buffer */
560 pe = atomic_read(&mdev->rs_pending_cnt);
561 if ((pe + number) > mx) {
565 for (i = 0; i < number; i++) {
566 /* Stop generating RS requests, when half of the send buffer is filled */
567 mutex_lock(&mdev->data.mutex);
568 if (mdev->data.socket) {
569 queued = mdev->data.socket->sk->sk_wmem_queued;
570 sndbuf = mdev->data.socket->sk->sk_sndbuf;
575 mutex_unlock(&mdev->data.mutex);
576 if (queued > sndbuf / 2)
580 size = BM_BLOCK_SIZE;
581 bit = drbd_bm_find_next(mdev, mdev->bm_resync_fo);
584 mdev->bm_resync_fo = drbd_bm_bits(mdev);
585 mdev->resync_work.cb = w_resync_inactive;
590 sector = BM_BIT_TO_SECT(bit);
592 if (drbd_try_rs_begin_io(mdev, sector)) {
593 mdev->bm_resync_fo = bit;
596 mdev->bm_resync_fo = bit + 1;
598 if (unlikely(drbd_bm_test_bit(mdev, bit) == 0)) {
599 drbd_rs_complete_io(mdev, sector);
603 #if DRBD_MAX_SEGMENT_SIZE > BM_BLOCK_SIZE
604 /* try to find some adjacent bits.
605 * we stop if we have already the maximum req size.
607 * Additionally always align bigger requests, in order to
608 * be prepared for all stripe sizes of software RAIDs.
613 if (size + BM_BLOCK_SIZE > max_segment_size)
616 /* Be always aligned */
617 if (sector & ((1<<(align+3))-1))
620 /* do not cross extent boundaries */
621 if (((bit+1) & BM_BLOCKS_PER_BM_EXT_MASK) == 0)
623 /* now, is it actually dirty, after all?
624 * caution, drbd_bm_test_bit is tri-state for some
625 * obscure reason; ( b == 0 ) would get the out-of-band
626 * only accidentally right because of the "oddly sized"
627 * adjustment below */
628 if (drbd_bm_test_bit(mdev, bit+1) != 1)
631 size += BM_BLOCK_SIZE;
632 if ((BM_BLOCK_SIZE << align) <= size)
636 /* if we merged some,
637 * reset the offset to start the next drbd_bm_find_next from */
638 if (size > BM_BLOCK_SIZE)
639 mdev->bm_resync_fo = bit + 1;
642 /* adjust very last sectors, in case we are oddly sized */
643 if (sector + (size>>9) > capacity)
644 size = (capacity-sector)<<9;
645 if (mdev->agreed_pro_version >= 89 && mdev->csums_tfm) {
646 switch (read_for_csum(mdev, sector, size)) {
647 case -EIO: /* Disk failure */
650 case -EAGAIN: /* allocation failed, or ldev busy */
651 drbd_rs_complete_io(mdev, sector);
652 mdev->bm_resync_fo = BM_SECT_TO_BIT(sector);
662 inc_rs_pending(mdev);
663 if (!drbd_send_drequest(mdev, P_RS_DATA_REQUEST,
664 sector, size, ID_SYNCER)) {
665 dev_err(DEV, "drbd_send_drequest() failed, aborting...\n");
666 dec_rs_pending(mdev);
673 if (mdev->bm_resync_fo >= drbd_bm_bits(mdev)) {
674 /* last syncer _request_ was sent,
675 * but the P_RS_DATA_REPLY not yet received. sync will end (and
676 * next sync group will resume), as soon as we receive the last
677 * resync data block, and the last bit is cleared.
678 * until then resync "work" is "inactive" ...
680 mdev->resync_work.cb = w_resync_inactive;
686 mdev->rs_in_flight += (i << (BM_BLOCK_SHIFT - 9));
687 mod_timer(&mdev->resync_timer, jiffies + SLEEP_TIME);
692 static int w_make_ov_request(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
696 const sector_t capacity = drbd_get_capacity(mdev->this_bdev);
698 if (unlikely(cancel))
701 if (unlikely(mdev->state.conn < C_CONNECTED)) {
702 dev_err(DEV, "Confused in w_make_ov_request()! cstate < Connected");
706 number = SLEEP_TIME*mdev->sync_conf.rate / ((BM_BLOCK_SIZE/1024)*HZ);
707 if (atomic_read(&mdev->rs_pending_cnt) > number)
710 number -= atomic_read(&mdev->rs_pending_cnt);
712 sector = mdev->ov_position;
713 for (i = 0; i < number; i++) {
714 if (sector >= capacity) {
715 mdev->resync_work.cb = w_resync_inactive;
719 size = BM_BLOCK_SIZE;
721 if (drbd_try_rs_begin_io(mdev, sector)) {
722 mdev->ov_position = sector;
726 if (sector + (size>>9) > capacity)
727 size = (capacity-sector)<<9;
729 inc_rs_pending(mdev);
730 if (!drbd_send_ov_request(mdev, sector, size)) {
731 dec_rs_pending(mdev);
734 sector += BM_SECT_PER_BIT;
736 mdev->ov_position = sector;
739 mod_timer(&mdev->resync_timer, jiffies + SLEEP_TIME);
744 int w_ov_finished(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
748 drbd_resync_finished(mdev);
753 static int w_resync_finished(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
757 drbd_resync_finished(mdev);
762 static void ping_peer(struct drbd_conf *mdev)
764 clear_bit(GOT_PING_ACK, &mdev->flags);
766 wait_event(mdev->misc_wait,
767 test_bit(GOT_PING_ACK, &mdev->flags) || mdev->state.conn < C_CONNECTED);
770 int drbd_resync_finished(struct drbd_conf *mdev)
772 unsigned long db, dt, dbdt;
774 union drbd_state os, ns;
776 char *khelper_cmd = NULL;
778 /* Remove all elements from the resync LRU. Since future actions
779 * might set bits in the (main) bitmap, then the entries in the
780 * resync LRU would be wrong. */
781 if (drbd_rs_del_all(mdev)) {
782 /* In case this is not possible now, most probably because
783 * there are P_RS_DATA_REPLY Packets lingering on the worker's
784 * queue (or even the read operations for those packets
785 * is not finished by now). Retry in 100ms. */
788 __set_current_state(TASK_INTERRUPTIBLE);
789 schedule_timeout(HZ / 10);
790 w = kmalloc(sizeof(struct drbd_work), GFP_ATOMIC);
792 w->cb = w_resync_finished;
793 drbd_queue_work(&mdev->data.work, w);
796 dev_err(DEV, "Warn failed to drbd_rs_del_all() and to kmalloc(w).\n");
799 dt = (jiffies - mdev->rs_start - mdev->rs_paused) / HZ;
803 dbdt = Bit2KB(db/dt);
804 mdev->rs_paused /= HZ;
811 spin_lock_irq(&mdev->req_lock);
814 /* This protects us against multiple calls (that can happen in the presence
815 of application IO), and against connectivity loss just before we arrive here. */
816 if (os.conn <= C_CONNECTED)
820 ns.conn = C_CONNECTED;
822 dev_info(DEV, "%s done (total %lu sec; paused %lu sec; %lu K/sec)\n",
823 (os.conn == C_VERIFY_S || os.conn == C_VERIFY_T) ?
824 "Online verify " : "Resync",
825 dt + mdev->rs_paused, mdev->rs_paused, dbdt);
827 n_oos = drbd_bm_total_weight(mdev);
829 if (os.conn == C_VERIFY_S || os.conn == C_VERIFY_T) {
831 dev_alert(DEV, "Online verify found %lu %dk block out of sync!\n",
833 khelper_cmd = "out-of-sync";
836 D_ASSERT((n_oos - mdev->rs_failed) == 0);
838 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T)
839 khelper_cmd = "after-resync-target";
841 if (mdev->csums_tfm && mdev->rs_total) {
842 const unsigned long s = mdev->rs_same_csum;
843 const unsigned long t = mdev->rs_total;
846 (t < 100000) ? ((s*100)/t) : (s/(t/100));
847 dev_info(DEV, "%u %% had equal check sums, eliminated: %luK; "
848 "transferred %luK total %luK\n",
850 Bit2KB(mdev->rs_same_csum),
851 Bit2KB(mdev->rs_total - mdev->rs_same_csum),
852 Bit2KB(mdev->rs_total));
856 if (mdev->rs_failed) {
857 dev_info(DEV, " %lu failed blocks\n", mdev->rs_failed);
859 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T) {
860 ns.disk = D_INCONSISTENT;
861 ns.pdsk = D_UP_TO_DATE;
863 ns.disk = D_UP_TO_DATE;
864 ns.pdsk = D_INCONSISTENT;
867 ns.disk = D_UP_TO_DATE;
868 ns.pdsk = D_UP_TO_DATE;
870 if (os.conn == C_SYNC_TARGET || os.conn == C_PAUSED_SYNC_T) {
873 for (i = UI_BITMAP ; i <= UI_HISTORY_END ; i++)
874 _drbd_uuid_set(mdev, i, mdev->p_uuid[i]);
875 drbd_uuid_set(mdev, UI_BITMAP, mdev->ldev->md.uuid[UI_CURRENT]);
876 _drbd_uuid_set(mdev, UI_CURRENT, mdev->p_uuid[UI_CURRENT]);
878 dev_err(DEV, "mdev->p_uuid is NULL! BUG\n");
882 drbd_uuid_set_bm(mdev, 0UL);
885 /* Now the two UUID sets are equal, update what we
886 * know of the peer. */
888 for (i = UI_CURRENT ; i <= UI_HISTORY_END ; i++)
889 mdev->p_uuid[i] = mdev->ldev->md.uuid[i];
893 _drbd_set_state(mdev, ns, CS_VERBOSE, NULL);
895 spin_unlock_irq(&mdev->req_lock);
901 mdev->ov_start_sector = 0;
905 if (test_and_clear_bit(WRITE_BM_AFTER_RESYNC, &mdev->flags)) {
906 dev_info(DEV, "Writing the whole bitmap\n");
907 drbd_queue_bitmap_io(mdev, &drbd_bm_write, NULL, "write from resync_finished");
911 drbd_khelper(mdev, khelper_cmd);
917 static void move_to_net_ee_or_free(struct drbd_conf *mdev, struct drbd_epoch_entry *e)
919 if (drbd_ee_has_active_page(e)) {
920 /* This might happen if sendpage() has not finished */
921 int i = (e->size + PAGE_SIZE -1) >> PAGE_SHIFT;
922 atomic_add(i, &mdev->pp_in_use_by_net);
923 atomic_sub(i, &mdev->pp_in_use);
924 spin_lock_irq(&mdev->req_lock);
925 list_add_tail(&e->w.list, &mdev->net_ee);
926 spin_unlock_irq(&mdev->req_lock);
927 wake_up(&drbd_pp_wait);
929 drbd_free_ee(mdev, e);
933 * w_e_end_data_req() - Worker callback, to send a P_DATA_REPLY packet in response to a P_DATA_REQUEST
934 * @mdev: DRBD device.
936 * @cancel: The connection will be closed anyways
938 int w_e_end_data_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
940 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w);
943 if (unlikely(cancel)) {
944 drbd_free_ee(mdev, e);
949 if (likely((e->flags & EE_WAS_ERROR) == 0)) {
950 ok = drbd_send_block(mdev, P_DATA_REPLY, e);
952 if (__ratelimit(&drbd_ratelimit_state))
953 dev_err(DEV, "Sending NegDReply. sector=%llus.\n",
954 (unsigned long long)e->sector);
956 ok = drbd_send_ack(mdev, P_NEG_DREPLY, e);
961 move_to_net_ee_or_free(mdev, e);
964 dev_err(DEV, "drbd_send_block() failed\n");
969 * w_e_end_rsdata_req() - Worker callback to send a P_RS_DATA_REPLY packet in response to a P_RS_DATA_REQUESTRS
970 * @mdev: DRBD device.
972 * @cancel: The connection will be closed anyways
974 int w_e_end_rsdata_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
976 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w);
979 if (unlikely(cancel)) {
980 drbd_free_ee(mdev, e);
985 if (get_ldev_if_state(mdev, D_FAILED)) {
986 drbd_rs_complete_io(mdev, e->sector);
990 if (likely((e->flags & EE_WAS_ERROR) == 0)) {
991 if (likely(mdev->state.pdsk >= D_INCONSISTENT)) {
992 inc_rs_pending(mdev);
993 ok = drbd_send_block(mdev, P_RS_DATA_REPLY, e);
995 if (__ratelimit(&drbd_ratelimit_state))
996 dev_err(DEV, "Not sending RSDataReply, "
997 "partner DISKLESS!\n");
1001 if (__ratelimit(&drbd_ratelimit_state))
1002 dev_err(DEV, "Sending NegRSDReply. sector %llus.\n",
1003 (unsigned long long)e->sector);
1005 ok = drbd_send_ack(mdev, P_NEG_RS_DREPLY, e);
1007 /* update resync data with failure */
1008 drbd_rs_failed_io(mdev, e->sector, e->size);
1013 move_to_net_ee_or_free(mdev, e);
1016 dev_err(DEV, "drbd_send_block() failed\n");
1020 int w_e_end_csum_rs_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1022 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w);
1023 struct digest_info *di;
1025 void *digest = NULL;
1028 if (unlikely(cancel)) {
1029 drbd_free_ee(mdev, e);
1034 if (get_ldev(mdev)) {
1035 drbd_rs_complete_io(mdev, e->sector);
1041 if (likely((e->flags & EE_WAS_ERROR) == 0)) {
1042 /* quick hack to try to avoid a race against reconfiguration.
1043 * a real fix would be much more involved,
1044 * introducing more locking mechanisms */
1045 if (mdev->csums_tfm) {
1046 digest_size = crypto_hash_digestsize(mdev->csums_tfm);
1047 D_ASSERT(digest_size == di->digest_size);
1048 digest = kmalloc(digest_size, GFP_NOIO);
1051 drbd_csum_ee(mdev, mdev->csums_tfm, e, digest);
1052 eq = !memcmp(digest, di->digest, digest_size);
1057 drbd_set_in_sync(mdev, e->sector, e->size);
1058 /* rs_same_csums unit is BM_BLOCK_SIZE */
1059 mdev->rs_same_csum += e->size >> BM_BLOCK_SHIFT;
1060 ok = drbd_send_ack(mdev, P_RS_IS_IN_SYNC, e);
1062 inc_rs_pending(mdev);
1063 e->block_id = ID_SYNCER; /* By setting block_id, digest pointer becomes invalid! */
1064 e->flags &= ~EE_HAS_DIGEST; /* This e no longer has a digest pointer */
1066 ok = drbd_send_block(mdev, P_RS_DATA_REPLY, e);
1069 ok = drbd_send_ack(mdev, P_NEG_RS_DREPLY, e);
1070 if (__ratelimit(&drbd_ratelimit_state))
1071 dev_err(DEV, "Sending NegDReply. I guess it gets messy.\n");
1075 move_to_net_ee_or_free(mdev, e);
1078 dev_err(DEV, "drbd_send_block/ack() failed\n");
1082 int w_e_end_ov_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1084 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w);
1089 if (unlikely(cancel))
1092 if (unlikely((e->flags & EE_WAS_ERROR) != 0))
1095 digest_size = crypto_hash_digestsize(mdev->verify_tfm);
1096 /* FIXME if this allocation fails, online verify will not terminate! */
1097 digest = kmalloc(digest_size, GFP_NOIO);
1099 drbd_csum_ee(mdev, mdev->verify_tfm, e, digest);
1100 inc_rs_pending(mdev);
1101 ok = drbd_send_drequest_csum(mdev, e->sector, e->size,
1102 digest, digest_size, P_OV_REPLY);
1104 dec_rs_pending(mdev);
1109 drbd_free_ee(mdev, e);
1116 void drbd_ov_oos_found(struct drbd_conf *mdev, sector_t sector, int size)
1118 if (mdev->ov_last_oos_start + mdev->ov_last_oos_size == sector) {
1119 mdev->ov_last_oos_size += size>>9;
1121 mdev->ov_last_oos_start = sector;
1122 mdev->ov_last_oos_size = size>>9;
1124 drbd_set_out_of_sync(mdev, sector, size);
1125 set_bit(WRITE_BM_AFTER_RESYNC, &mdev->flags);
1128 int w_e_end_ov_reply(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1130 struct drbd_epoch_entry *e = container_of(w, struct drbd_epoch_entry, w);
1131 struct digest_info *di;
1136 if (unlikely(cancel)) {
1137 drbd_free_ee(mdev, e);
1142 /* after "cancel", because after drbd_disconnect/drbd_rs_cancel_all
1143 * the resync lru has been cleaned up already */
1144 if (get_ldev(mdev)) {
1145 drbd_rs_complete_io(mdev, e->sector);
1151 if (likely((e->flags & EE_WAS_ERROR) == 0)) {
1152 digest_size = crypto_hash_digestsize(mdev->verify_tfm);
1153 digest = kmalloc(digest_size, GFP_NOIO);
1155 drbd_csum_ee(mdev, mdev->verify_tfm, e, digest);
1157 D_ASSERT(digest_size == di->digest_size);
1158 eq = !memcmp(digest, di->digest, digest_size);
1162 ok = drbd_send_ack(mdev, P_NEG_RS_DREPLY, e);
1163 if (__ratelimit(&drbd_ratelimit_state))
1164 dev_err(DEV, "Sending NegDReply. I guess it gets messy.\n");
1169 drbd_ov_oos_found(mdev, e->sector, e->size);
1173 ok = drbd_send_ack_ex(mdev, P_OV_RESULT, e->sector, e->size,
1174 eq ? ID_IN_SYNC : ID_OUT_OF_SYNC);
1176 drbd_free_ee(mdev, e);
1178 if (--mdev->ov_left == 0) {
1180 drbd_resync_finished(mdev);
1186 int w_prev_work_done(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1188 struct drbd_wq_barrier *b = container_of(w, struct drbd_wq_barrier, w);
1193 int w_send_barrier(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1195 struct drbd_tl_epoch *b = container_of(w, struct drbd_tl_epoch, w);
1196 struct p_barrier *p = &mdev->data.sbuf.barrier;
1199 /* really avoid racing with tl_clear. w.cb may have been referenced
1200 * just before it was reassigned and re-queued, so double check that.
1201 * actually, this race was harmless, since we only try to send the
1202 * barrier packet here, and otherwise do nothing with the object.
1203 * but compare with the head of w_clear_epoch */
1204 spin_lock_irq(&mdev->req_lock);
1205 if (w->cb != w_send_barrier || mdev->state.conn < C_CONNECTED)
1207 spin_unlock_irq(&mdev->req_lock);
1211 if (!drbd_get_data_sock(mdev))
1213 p->barrier = b->br_number;
1214 /* inc_ap_pending was done where this was queued.
1215 * dec_ap_pending will be done in got_BarrierAck
1216 * or (on connection loss) in w_clear_epoch. */
1217 ok = _drbd_send_cmd(mdev, mdev->data.socket, P_BARRIER,
1218 (struct p_header80 *)p, sizeof(*p), 0);
1219 drbd_put_data_sock(mdev);
1224 int w_send_write_hint(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1228 return drbd_send_short_cmd(mdev, P_UNPLUG_REMOTE);
1232 * w_send_dblock() - Worker callback to send a P_DATA packet in order to mirror a write request
1233 * @mdev: DRBD device.
1235 * @cancel: The connection will be closed anyways
1237 int w_send_dblock(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1239 struct drbd_request *req = container_of(w, struct drbd_request, w);
1242 if (unlikely(cancel)) {
1243 req_mod(req, send_canceled);
1247 ok = drbd_send_dblock(mdev, req);
1248 req_mod(req, ok ? handed_over_to_network : send_failed);
1254 * w_send_read_req() - Worker callback to send a read request (P_DATA_REQUEST) packet
1255 * @mdev: DRBD device.
1257 * @cancel: The connection will be closed anyways
1259 int w_send_read_req(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1261 struct drbd_request *req = container_of(w, struct drbd_request, w);
1264 if (unlikely(cancel)) {
1265 req_mod(req, send_canceled);
1269 ok = drbd_send_drequest(mdev, P_DATA_REQUEST, req->sector, req->size,
1270 (unsigned long)req);
1273 /* ?? we set C_TIMEOUT or C_BROKEN_PIPE in drbd_send();
1274 * so this is probably redundant */
1275 if (mdev->state.conn >= C_CONNECTED)
1276 drbd_force_state(mdev, NS(conn, C_NETWORK_FAILURE));
1278 req_mod(req, ok ? handed_over_to_network : send_failed);
1283 int w_restart_disk_io(struct drbd_conf *mdev, struct drbd_work *w, int cancel)
1285 struct drbd_request *req = container_of(w, struct drbd_request, w);
1287 if (bio_data_dir(req->master_bio) == WRITE && req->rq_state & RQ_IN_ACT_LOG)
1288 drbd_al_begin_io(mdev, req->sector);
1289 /* Calling drbd_al_begin_io() out of the worker might deadlocks
1290 theoretically. Practically it can not deadlock, since this is
1291 only used when unfreezing IOs. All the extents of the requests
1292 that made it into the TL are already active */
1294 drbd_req_make_private_bio(req, req->master_bio);
1295 req->private_bio->bi_bdev = mdev->ldev->backing_bdev;
1296 generic_make_request(req->private_bio);
1301 static int _drbd_may_sync_now(struct drbd_conf *mdev)
1303 struct drbd_conf *odev = mdev;
1306 if (odev->sync_conf.after == -1)
1308 odev = minor_to_mdev(odev->sync_conf.after);
1309 ERR_IF(!odev) return 1;
1310 if ((odev->state.conn >= C_SYNC_SOURCE &&
1311 odev->state.conn <= C_PAUSED_SYNC_T) ||
1312 odev->state.aftr_isp || odev->state.peer_isp ||
1313 odev->state.user_isp)
1319 * _drbd_pause_after() - Pause resync on all devices that may not resync now
1320 * @mdev: DRBD device.
1322 * Called from process context only (admin command and after_state_ch).
1324 static int _drbd_pause_after(struct drbd_conf *mdev)
1326 struct drbd_conf *odev;
1329 for (i = 0; i < minor_count; i++) {
1330 odev = minor_to_mdev(i);
1333 if (odev->state.conn == C_STANDALONE && odev->state.disk == D_DISKLESS)
1335 if (!_drbd_may_sync_now(odev))
1336 rv |= (__drbd_set_state(_NS(odev, aftr_isp, 1), CS_HARD, NULL)
1337 != SS_NOTHING_TO_DO);
1344 * _drbd_resume_next() - Resume resync on all devices that may resync now
1345 * @mdev: DRBD device.
1347 * Called from process context only (admin command and worker).
1349 static int _drbd_resume_next(struct drbd_conf *mdev)
1351 struct drbd_conf *odev;
1354 for (i = 0; i < minor_count; i++) {
1355 odev = minor_to_mdev(i);
1358 if (odev->state.conn == C_STANDALONE && odev->state.disk == D_DISKLESS)
1360 if (odev->state.aftr_isp) {
1361 if (_drbd_may_sync_now(odev))
1362 rv |= (__drbd_set_state(_NS(odev, aftr_isp, 0),
1364 != SS_NOTHING_TO_DO) ;
1370 void resume_next_sg(struct drbd_conf *mdev)
1372 write_lock_irq(&global_state_lock);
1373 _drbd_resume_next(mdev);
1374 write_unlock_irq(&global_state_lock);
1377 void suspend_other_sg(struct drbd_conf *mdev)
1379 write_lock_irq(&global_state_lock);
1380 _drbd_pause_after(mdev);
1381 write_unlock_irq(&global_state_lock);
1384 static int sync_after_error(struct drbd_conf *mdev, int o_minor)
1386 struct drbd_conf *odev;
1390 if (o_minor < -1 || minor_to_mdev(o_minor) == NULL)
1391 return ERR_SYNC_AFTER;
1393 /* check for loops */
1394 odev = minor_to_mdev(o_minor);
1397 return ERR_SYNC_AFTER_CYCLE;
1399 /* dependency chain ends here, no cycles. */
1400 if (odev->sync_conf.after == -1)
1403 /* follow the dependency chain */
1404 odev = minor_to_mdev(odev->sync_conf.after);
1408 int drbd_alter_sa(struct drbd_conf *mdev, int na)
1413 write_lock_irq(&global_state_lock);
1414 retcode = sync_after_error(mdev, na);
1415 if (retcode == NO_ERROR) {
1416 mdev->sync_conf.after = na;
1418 changes = _drbd_pause_after(mdev);
1419 changes |= _drbd_resume_next(mdev);
1422 write_unlock_irq(&global_state_lock);
1427 * drbd_start_resync() - Start the resync process
1428 * @mdev: DRBD device.
1429 * @side: Either C_SYNC_SOURCE or C_SYNC_TARGET
1431 * This function might bring you directly into one of the
1432 * C_PAUSED_SYNC_* states.
1434 void drbd_start_resync(struct drbd_conf *mdev, enum drbd_conns side)
1436 union drbd_state ns;
1439 if (mdev->state.conn >= C_SYNC_SOURCE) {
1440 dev_err(DEV, "Resync already running!\n");
1444 /* In case a previous resync run was aborted by an IO error/detach on the peer. */
1445 drbd_rs_cancel_all(mdev);
1447 if (side == C_SYNC_TARGET) {
1448 /* Since application IO was locked out during C_WF_BITMAP_T and
1449 C_WF_SYNC_UUID we are still unmodified. Before going to C_SYNC_TARGET
1450 we check that we might make the data inconsistent. */
1451 r = drbd_khelper(mdev, "before-resync-target");
1452 r = (r >> 8) & 0xff;
1454 dev_info(DEV, "before-resync-target handler returned %d, "
1455 "dropping connection.\n", r);
1456 drbd_force_state(mdev, NS(conn, C_DISCONNECTING));
1461 drbd_state_lock(mdev);
1463 if (!get_ldev_if_state(mdev, D_NEGOTIATING)) {
1464 drbd_state_unlock(mdev);
1468 if (side == C_SYNC_TARGET) {
1469 mdev->bm_resync_fo = 0;
1470 } else /* side == C_SYNC_SOURCE */ {
1473 get_random_bytes(&uuid, sizeof(u64));
1474 drbd_uuid_set(mdev, UI_BITMAP, uuid);
1475 drbd_send_sync_uuid(mdev, uuid);
1477 D_ASSERT(mdev->state.disk == D_UP_TO_DATE);
1480 write_lock_irq(&global_state_lock);
1483 ns.aftr_isp = !_drbd_may_sync_now(mdev);
1487 if (side == C_SYNC_TARGET)
1488 ns.disk = D_INCONSISTENT;
1489 else /* side == C_SYNC_SOURCE */
1490 ns.pdsk = D_INCONSISTENT;
1492 r = __drbd_set_state(mdev, ns, CS_VERBOSE, NULL);
1495 if (ns.conn < C_CONNECTED)
1496 r = SS_UNKNOWN_ERROR;
1498 if (r == SS_SUCCESS) {
1499 unsigned long tw = drbd_bm_total_weight(mdev);
1500 unsigned long now = jiffies;
1503 mdev->rs_failed = 0;
1504 mdev->rs_paused = 0;
1505 mdev->rs_same_csum = 0;
1506 mdev->rs_last_events = 0;
1507 mdev->rs_last_sect_ev = 0;
1508 mdev->rs_total = tw;
1509 mdev->rs_start = now;
1510 for (i = 0; i < DRBD_SYNC_MARKS; i++) {
1511 mdev->rs_mark_left[i] = tw;
1512 mdev->rs_mark_time[i] = now;
1514 _drbd_pause_after(mdev);
1516 write_unlock_irq(&global_state_lock);
1519 if (r == SS_SUCCESS) {
1520 dev_info(DEV, "Began resync as %s (will sync %lu KB [%lu bits set]).\n",
1521 drbd_conn_str(ns.conn),
1522 (unsigned long) mdev->rs_total << (BM_BLOCK_SHIFT-10),
1523 (unsigned long) mdev->rs_total);
1525 if (mdev->agreed_pro_version < 95 && mdev->rs_total == 0) {
1526 /* This still has a race (about when exactly the peers
1527 * detect connection loss) that can lead to a full sync
1528 * on next handshake. In 8.3.9 we fixed this with explicit
1529 * resync-finished notifications, but the fix
1530 * introduces a protocol change. Sleeping for some
1531 * time longer than the ping interval + timeout on the
1532 * SyncSource, to give the SyncTarget the chance to
1533 * detect connection loss, then waiting for a ping
1534 * response (implicit in drbd_resync_finished) reduces
1535 * the race considerably, but does not solve it. */
1536 if (side == C_SYNC_SOURCE)
1537 schedule_timeout_interruptible(
1538 mdev->net_conf->ping_int * HZ +
1539 mdev->net_conf->ping_timeo*HZ/9);
1540 drbd_resync_finished(mdev);
1543 atomic_set(&mdev->rs_sect_in, 0);
1544 atomic_set(&mdev->rs_sect_ev, 0);
1545 mdev->rs_in_flight = 0;
1546 mdev->rs_planed = 0;
1547 spin_lock(&mdev->peer_seq_lock);
1548 fifo_set(&mdev->rs_plan_s, 0);
1549 spin_unlock(&mdev->peer_seq_lock);
1550 /* ns.conn may already be != mdev->state.conn,
1551 * we may have been paused in between, or become paused until
1552 * the timer triggers.
1553 * No matter, that is handled in resync_timer_fn() */
1554 if (ns.conn == C_SYNC_TARGET)
1555 mod_timer(&mdev->resync_timer, jiffies);
1559 drbd_state_unlock(mdev);
1562 int drbd_worker(struct drbd_thread *thi)
1564 struct drbd_conf *mdev = thi->mdev;
1565 struct drbd_work *w = NULL;
1566 LIST_HEAD(work_list);
1569 sprintf(current->comm, "drbd%d_worker", mdev_to_minor(mdev));
1571 while (get_t_state(thi) == Running) {
1572 drbd_thread_current_set_cpu(mdev);
1574 if (down_trylock(&mdev->data.work.s)) {
1575 mutex_lock(&mdev->data.mutex);
1576 if (mdev->data.socket && !mdev->net_conf->no_cork)
1577 drbd_tcp_uncork(mdev->data.socket);
1578 mutex_unlock(&mdev->data.mutex);
1580 intr = down_interruptible(&mdev->data.work.s);
1582 mutex_lock(&mdev->data.mutex);
1583 if (mdev->data.socket && !mdev->net_conf->no_cork)
1584 drbd_tcp_cork(mdev->data.socket);
1585 mutex_unlock(&mdev->data.mutex);
1589 D_ASSERT(intr == -EINTR);
1590 flush_signals(current);
1591 ERR_IF (get_t_state(thi) == Running)
1596 if (get_t_state(thi) != Running)
1598 /* With this break, we have done a down() but not consumed
1599 the entry from the list. The cleanup code takes care of
1603 spin_lock_irq(&mdev->data.work.q_lock);
1604 ERR_IF(list_empty(&mdev->data.work.q)) {
1605 /* something terribly wrong in our logic.
1606 * we were able to down() the semaphore,
1607 * but the list is empty... doh.
1609 * what is the best thing to do now?
1610 * try again from scratch, restarting the receiver,
1611 * asender, whatnot? could break even more ugly,
1612 * e.g. when we are primary, but no good local data.
1614 * I'll try to get away just starting over this loop.
1616 spin_unlock_irq(&mdev->data.work.q_lock);
1619 w = list_entry(mdev->data.work.q.next, struct drbd_work, list);
1620 list_del_init(&w->list);
1621 spin_unlock_irq(&mdev->data.work.q_lock);
1623 if (!w->cb(mdev, w, mdev->state.conn < C_CONNECTED)) {
1624 /* dev_warn(DEV, "worker: a callback failed! \n"); */
1625 if (mdev->state.conn >= C_CONNECTED)
1626 drbd_force_state(mdev,
1627 NS(conn, C_NETWORK_FAILURE));
1630 D_ASSERT(test_bit(DEVICE_DYING, &mdev->flags));
1631 D_ASSERT(test_bit(CONFIG_PENDING, &mdev->flags));
1633 spin_lock_irq(&mdev->data.work.q_lock);
1635 while (!list_empty(&mdev->data.work.q)) {
1636 list_splice_init(&mdev->data.work.q, &work_list);
1637 spin_unlock_irq(&mdev->data.work.q_lock);
1639 while (!list_empty(&work_list)) {
1640 w = list_entry(work_list.next, struct drbd_work, list);
1641 list_del_init(&w->list);
1643 i++; /* dead debugging code */
1646 spin_lock_irq(&mdev->data.work.q_lock);
1648 sema_init(&mdev->data.work.s, 0);
1649 /* DANGEROUS race: if someone did queue his work within the spinlock,
1650 * but up() ed outside the spinlock, we could get an up() on the
1651 * semaphore without corresponding list entry.
1654 spin_unlock_irq(&mdev->data.work.q_lock);
1656 D_ASSERT(mdev->state.disk == D_DISKLESS && mdev->state.conn == C_STANDALONE);
1657 /* _drbd_set_state only uses stop_nowait.
1658 * wait here for the Exiting receiver. */
1659 drbd_thread_stop(&mdev->receiver);
1660 drbd_mdev_cleanup(mdev);
1662 dev_info(DEV, "worker terminated\n");
1664 clear_bit(DEVICE_DYING, &mdev->flags);
1665 clear_bit(CONFIG_PENDING, &mdev->flags);
1666 wake_up(&mdev->state_wait);