drbd_actlog.c

   1// SPDX-License-Identifier: GPL-2.0-only
   2/*
   3   drbd_actlog.c
   4
   5   This file is part of DRBD by Philipp Reisner and Lars Ellenberg.
   6
   7   Copyright (C) 2003-2008, LINBIT Information Technologies GmbH.
   8   Copyright (C) 2003-2008, Philipp Reisner <philipp.reisner@linbit.com>.
   9   Copyright (C) 2003-2008, Lars Ellenberg <lars.ellenberg@linbit.com>.
  10
  11
  12 */
  13
  14#include <linux/slab.h>
  15#include <linux/crc32c.h>
  16#include <linux/drbd.h>
  17#include <linux/drbd_limits.h>
  18#include "drbd_int.h"
  19
  20
  21enum al_transaction_types {
  22	AL_TR_UPDATE = 0,
  23	AL_TR_INITIALIZED = 0xffff
  24};
  25/* all fields on disc in big endian */
  26struct __packed al_transaction_on_disk {
  27	/* don't we all like magic */
  28	__be32	magic;
  29
  30	/* to identify the most recent transaction block
  31	 * in the on disk ring buffer */
  32	__be32	tr_number;
  33
  34	/* checksum on the full 4k block, with this field set to 0. */
  35	__be32	crc32c;
  36
  37	/* type of transaction, special transaction types like:
  38	 * purge-all, set-all-idle, set-all-active, ... to-be-defined
  39	 * see also enum al_transaction_types */
  40	__be16	transaction_type;
  41
  42	/* we currently allow only a few thousand extents,
  43	 * so 16bit will be enough for the slot number. */
  44
  45	/* how many updates in this transaction */
  46	__be16	n_updates;
  47
  48	/* maximum slot number, "al-extents" in drbd.conf speak.
  49	 * Having this in each transaction should make reconfiguration
  50	 * of that parameter easier. */
  51	__be16	context_size;
  52
  53	/* slot number the context starts with */
  54	__be16	context_start_slot_nr;
  55
  56	/* Some reserved bytes.  Expected usage is a 64bit counter of
  57	 * sectors-written since device creation, and other data generation tag
  58	 * supporting usage */
  59	__be32	__reserved[4];
  60
  61	/* --- 36 byte used --- */
  62
  63	/* Reserve space for up to AL_UPDATES_PER_TRANSACTION changes
  64	 * in one transaction, then use the remaining byte in the 4k block for
  65	 * context information.  "Flexible" number of updates per transaction
  66	 * does not help, as we have to account for the case when all update
  67	 * slots are used anyways, so it would only complicate code without
  68	 * additional benefit.
  69	 */
  70	__be16	update_slot_nr[AL_UPDATES_PER_TRANSACTION];
  71
  72	/* but the extent number is 32bit, which at an extent size of 4 MiB
  73	 * allows to cover device sizes of up to 2**54 Byte (16 PiB) */
  74	__be32	update_extent_nr[AL_UPDATES_PER_TRANSACTION];
  75
  76	/* --- 420 bytes used (36 + 64*6) --- */
  77
  78	/* 4096 - 420 = 3676 = 919 * 4 */
  79	__be32	context[AL_CONTEXT_PER_TRANSACTION];
  80};
  81
  82void *drbd_md_get_buffer(struct drbd_device *device, const char *intent)
  83{
  84	int r;
  85
  86	wait_event(device->misc_wait,
  87		   (r = atomic_cmpxchg(&device->md_io.in_use, 0, 1)) == 0 ||
  88		   device->state.disk <= D_FAILED);
  89
  90	if (r)
  91		return NULL;
  92
  93	device->md_io.current_use = intent;
  94	device->md_io.start_jif = jiffies;
  95	device->md_io.submit_jif = device->md_io.start_jif - 1;
  96	return page_address(device->md_io.page);
  97}
  98
  99void drbd_md_put_buffer(struct drbd_device *device)
 100{
 101	if (atomic_dec_and_test(&device->md_io.in_use))
 102		wake_up(&device->misc_wait);
 103}
 104
 105void wait_until_done_or_force_detached(struct drbd_device *device, struct drbd_backing_dev *bdev,
 106				     unsigned int *done)
 107{
 108	long dt;
 109
 110	rcu_read_lock();
 111	dt = rcu_dereference(bdev->disk_conf)->disk_timeout;
 112	rcu_read_unlock();
 113	dt = dt * HZ / 10;
 114	if (dt == 0)
 115		dt = MAX_SCHEDULE_TIMEOUT;
 116
 117	dt = wait_event_timeout(device->misc_wait,
 118			*done || test_bit(FORCE_DETACH, &device->flags), dt);
 119	if (dt == 0) {
 120		drbd_err(device, "meta-data IO operation timed out\n");
 121		drbd_chk_io_error(device, 1, DRBD_FORCE_DETACH);
 122	}
 123}
 124
 125static int _drbd_md_sync_page_io(struct drbd_device *device,
 126				 struct drbd_backing_dev *bdev,
 127				 sector_t sector, enum req_op op)
 128{
 129	struct bio *bio;
 130	/* we do all our meta data IO in aligned 4k blocks. */
 131	const int size = 4096;
 132	int err;
 133	blk_opf_t op_flags = 0;
 134
 135	device->md_io.done = 0;
 136	device->md_io.error = -ENODEV;
 137
 138	if ((op == REQ_OP_WRITE) && !test_bit(MD_NO_FUA, &device->flags))
 139		op_flags |= REQ_FUA | REQ_PREFLUSH;
 140	op_flags |= REQ_SYNC;
 141
 142	bio = bio_alloc_bioset(bdev->md_bdev, 1, op | op_flags, GFP_NOIO,
 143			       &drbd_md_io_bio_set);
 144	bio->bi_iter.bi_sector = sector;
 145	err = -EIO;
 146	if (bio_add_page(bio, device->md_io.page, size, 0) != size)
 147		goto out;
 148	bio->bi_private = device;
 149	bio->bi_end_io = drbd_md_endio;
 150
 151	if (op != REQ_OP_WRITE && device->state.disk == D_DISKLESS && device->ldev == NULL)
 152		/* special case, drbd_md_read() during drbd_adm_attach(): no get_ldev */
 153		;
 154	else if (!get_ldev_if_state(device, D_ATTACHING)) {
 155		/* Corresponding put_ldev in drbd_md_endio() */
 156		drbd_err(device, "ASSERT FAILED: get_ldev_if_state() == 1 in _drbd_md_sync_page_io()\n");
 157		err = -ENODEV;
 158		goto out;
 159	}
 160
 161	bio_get(bio); /* one bio_put() is in the completion handler */
 162	atomic_inc(&device->md_io.in_use); /* drbd_md_put_buffer() is in the completion handler */
 163	device->md_io.submit_jif = jiffies;
 164	if (drbd_insert_fault(device, (op == REQ_OP_WRITE) ? DRBD_FAULT_MD_WR : DRBD_FAULT_MD_RD))
 165		bio_io_error(bio);
 166	else
 167		submit_bio(bio);
 168	wait_until_done_or_force_detached(device, bdev, &device->md_io.done);
 169	if (!bio->bi_status)
 170		err = device->md_io.error;
 171
 172 out:
 173	bio_put(bio);
 174	return err;
 175}
 176
 177int drbd_md_sync_page_io(struct drbd_device *device, struct drbd_backing_dev *bdev,
 178			 sector_t sector, enum req_op op)
 179{
 180	int err;
 181	D_ASSERT(device, atomic_read(&device->md_io.in_use) == 1);
 182
 183	BUG_ON(!bdev->md_bdev);
 184
 185	dynamic_drbd_dbg(device, "meta_data io: %s [%d]:%s(,%llus,%s) %pS\n",
 186	     current->comm, current->pid, __func__,
 187	     (unsigned long long)sector, (op == REQ_OP_WRITE) ? "WRITE" : "READ",
 188	     (void*)_RET_IP_ );
 189
 190	if (sector < drbd_md_first_sector(bdev) ||
 191	    sector + 7 > drbd_md_last_sector(bdev))
 192		drbd_alert(device, "%s [%d]:%s(,%llus,%s) out of range md access!\n",
 193		     current->comm, current->pid, __func__,
 194		     (unsigned long long)sector,
 195		     (op == REQ_OP_WRITE) ? "WRITE" : "READ");
 196
 197	err = _drbd_md_sync_page_io(device, bdev, sector, op);
 198	if (err) {
 199		drbd_err(device, "drbd_md_sync_page_io(,%llus,%s) failed with error %d\n",
 200		    (unsigned long long)sector,
 201		    (op == REQ_OP_WRITE) ? "WRITE" : "READ", err);
 202	}
 203	return err;
 204}
 205
 206static struct bm_extent *find_active_resync_extent(struct drbd_device *device, unsigned int enr)
 207{
 208	struct lc_element *tmp;
 209	tmp = lc_find(device->resync, enr/AL_EXT_PER_BM_SECT);
 210	if (unlikely(tmp != NULL)) {
 211		struct bm_extent  *bm_ext = lc_entry(tmp, struct bm_extent, lce);
 212		if (test_bit(BME_NO_WRITES, &bm_ext->flags))
 213			return bm_ext;
 214	}
 215	return NULL;
 216}
 217
 218static struct lc_element *_al_get(struct drbd_device *device, unsigned int enr, bool nonblock)
 219{
 220	struct lc_element *al_ext;
 221	struct bm_extent *bm_ext;
 222	int wake;
 223
 224	spin_lock_irq(&device->al_lock);
 225	bm_ext = find_active_resync_extent(device, enr);
 226	if (bm_ext) {
 227		wake = !test_and_set_bit(BME_PRIORITY, &bm_ext->flags);
 228		spin_unlock_irq(&device->al_lock);
 229		if (wake)
 230			wake_up(&device->al_wait);
 231		return NULL;
 232	}
 233	if (nonblock)
 234		al_ext = lc_try_get(device->act_log, enr);
 235	else
 236		al_ext = lc_get(device->act_log, enr);
 237	spin_unlock_irq(&device->al_lock);
 238	return al_ext;
 239}
 240
 241bool drbd_al_begin_io_fastpath(struct drbd_device *device, struct drbd_interval *i)
 242{
 243	/* for bios crossing activity log extent boundaries,
 244	 * we may need to activate two extents in one go */
 245	unsigned first = i->sector >> (AL_EXTENT_SHIFT-9);
 246	unsigned last = i->size == 0 ? first : (i->sector + (i->size >> 9) - 1) >> (AL_EXTENT_SHIFT-9);
 247
 248	D_ASSERT(device, first <= last);
 249	D_ASSERT(device, atomic_read(&device->local_cnt) > 0);
 250
 251	/* FIXME figure out a fast path for bios crossing AL extent boundaries */
 252	if (first != last)
 253		return false;
 254
 255	return _al_get(device, first, true);
 256}
 257
 258bool drbd_al_begin_io_prepare(struct drbd_device *device, struct drbd_interval *i)
 259{
 260	/* for bios crossing activity log extent boundaries,
 261	 * we may need to activate two extents in one go */
 262	unsigned first = i->sector >> (AL_EXTENT_SHIFT-9);
 263	unsigned last = i->size == 0 ? first : (i->sector + (i->size >> 9) - 1) >> (AL_EXTENT_SHIFT-9);
 264	unsigned enr;
 265	bool need_transaction = false;
 266
 267	D_ASSERT(device, first <= last);
 268	D_ASSERT(device, atomic_read(&device->local_cnt) > 0);
 269
 270	for (enr = first; enr <= last; enr++) {
 271		struct lc_element *al_ext;
 272		wait_event(device->al_wait,
 273				(al_ext = _al_get(device, enr, false)) != NULL);
 274		if (al_ext->lc_number != enr)
 275			need_transaction = true;
 276	}
 277	return need_transaction;
 278}
 279
 280#if (PAGE_SHIFT + 3) < (AL_EXTENT_SHIFT - BM_BLOCK_SHIFT)
 281/* Currently BM_BLOCK_SHIFT, BM_EXT_SHIFT and AL_EXTENT_SHIFT
 282 * are still coupled, or assume too much about their relation.
 283 * Code below will not work if this is violated.
 284 * Will be cleaned up with some followup patch.
 285 */
 286# error FIXME
 287#endif
 288
 289static unsigned int al_extent_to_bm_page(unsigned int al_enr)
 290{
 291	return al_enr >>
 292		/* bit to page */
 293		((PAGE_SHIFT + 3) -
 294		/* al extent number to bit */
 295		 (AL_EXTENT_SHIFT - BM_BLOCK_SHIFT));
 296}
 297
 298static sector_t al_tr_number_to_on_disk_sector(struct drbd_device *device)
 299{
 300	const unsigned int stripes = device->ldev->md.al_stripes;
 301	const unsigned int stripe_size_4kB = device->ldev->md.al_stripe_size_4k;
 302
 303	/* transaction number, modulo on-disk ring buffer wrap around */
 304	unsigned int t = device->al_tr_number % (device->ldev->md.al_size_4k);
 305
 306	/* ... to aligned 4k on disk block */
 307	t = ((t % stripes) * stripe_size_4kB) + t/stripes;
 308
 309	/* ... to 512 byte sector in activity log */
 310	t *= 8;
 311
 312	/* ... plus offset to the on disk position */
 313	return device->ldev->md.md_offset + device->ldev->md.al_offset + t;
 314}
 315
 316static int __al_write_transaction(struct drbd_device *device, struct al_transaction_on_disk *buffer)
 317{
 318	struct lc_element *e;
 319	sector_t sector;
 320	int i, mx;
 321	unsigned extent_nr;
 322	unsigned crc = 0;
 323	int err = 0;
 324
 325	memset(buffer, 0, sizeof(*buffer));
 326	buffer->magic = cpu_to_be32(DRBD_AL_MAGIC);
 327	buffer->tr_number = cpu_to_be32(device->al_tr_number);
 328
 329	i = 0;
 330
 331	drbd_bm_reset_al_hints(device);
 332
 333	/* Even though no one can start to change this list
 334	 * once we set the LC_LOCKED -- from drbd_al_begin_io(),
 335	 * lc_try_lock_for_transaction() --, someone may still
 336	 * be in the process of changing it. */
 337	spin_lock_irq(&device->al_lock);
 338	list_for_each_entry(e, &device->act_log->to_be_changed, list) {
 339		if (i == AL_UPDATES_PER_TRANSACTION) {
 340			i++;
 341			break;
 342		}
 343		buffer->update_slot_nr[i] = cpu_to_be16(e->lc_index);
 344		buffer->update_extent_nr[i] = cpu_to_be32(e->lc_new_number);
 345		if (e->lc_number != LC_FREE)
 346			drbd_bm_mark_for_writeout(device,
 347					al_extent_to_bm_page(e->lc_number));
 348		i++;
 349	}
 350	spin_unlock_irq(&device->al_lock);
 351	BUG_ON(i > AL_UPDATES_PER_TRANSACTION);
 352
 353	buffer->n_updates = cpu_to_be16(i);
 354	for ( ; i < AL_UPDATES_PER_TRANSACTION; i++) {
 355		buffer->update_slot_nr[i] = cpu_to_be16(-1);
 356		buffer->update_extent_nr[i] = cpu_to_be32(LC_FREE);
 357	}
 358
 359	buffer->context_size = cpu_to_be16(device->act_log->nr_elements);
 360	buffer->context_start_slot_nr = cpu_to_be16(device->al_tr_cycle);
 361
 362	mx = min_t(int, AL_CONTEXT_PER_TRANSACTION,
 363		   device->act_log->nr_elements - device->al_tr_cycle);
 364	for (i = 0; i < mx; i++) {
 365		unsigned idx = device->al_tr_cycle + i;
 366		extent_nr = lc_element_by_index(device->act_log, idx)->lc_number;
 367		buffer->context[i] = cpu_to_be32(extent_nr);
 368	}
 369	for (; i < AL_CONTEXT_PER_TRANSACTION; i++)
 370		buffer->context[i] = cpu_to_be32(LC_FREE);
 371
 372	device->al_tr_cycle += AL_CONTEXT_PER_TRANSACTION;
 373	if (device->al_tr_cycle >= device->act_log->nr_elements)
 374		device->al_tr_cycle = 0;
 375
 376	sector = al_tr_number_to_on_disk_sector(device);
 377
 378	crc = crc32c(0, buffer, 4096);
 379	buffer->crc32c = cpu_to_be32(crc);
 380
 381	if (drbd_bm_write_hinted(device))
 382		err = -EIO;
 383	else {
 384		bool write_al_updates;
 385		rcu_read_lock();
 386		write_al_updates = rcu_dereference(device->ldev->disk_conf)->al_updates;
 387		rcu_read_unlock();
 388		if (write_al_updates) {
 389			if (drbd_md_sync_page_io(device, device->ldev, sector, REQ_OP_WRITE)) {
 390				err = -EIO;
 391				drbd_chk_io_error(device, 1, DRBD_META_IO_ERROR);
 392			} else {
 393				device->al_tr_number++;
 394				device->al_writ_cnt++;
 395			}
 396		}
 397	}
 398
 399	return err;
 400}
 401
 402static int al_write_transaction(struct drbd_device *device)
 403{
 404	struct al_transaction_on_disk *buffer;
 405	int err;
 406
 407	if (!get_ldev(device)) {
 408		drbd_err(device, "disk is %s, cannot start al transaction\n",
 409			drbd_disk_str(device->state.disk));
 410		return -EIO;
 411	}
 412
 413	/* The bitmap write may have failed, causing a state change. */
 414	if (device->state.disk < D_INCONSISTENT) {
 415		drbd_err(device,
 416			"disk is %s, cannot write al transaction\n",
 417			drbd_disk_str(device->state.disk));
 418		put_ldev(device);
 419		return -EIO;
 420	}
 421
 422	/* protects md_io_buffer, al_tr_cycle, ... */
 423	buffer = drbd_md_get_buffer(device, __func__);
 424	if (!buffer) {
 425		drbd_err(device, "disk failed while waiting for md_io buffer\n");
 426		put_ldev(device);
 427		return -ENODEV;
 428	}
 429
 430	err = __al_write_transaction(device, buffer);
 431
 432	drbd_md_put_buffer(device);
 433	put_ldev(device);
 434
 435	return err;
 436}
 437
 438
 439void drbd_al_begin_io_commit(struct drbd_device *device)
 440{
 441	bool locked = false;
 442
 443	/* Serialize multiple transactions.
 444	 * This uses test_and_set_bit, memory barrier is implicit.
 445	 */
 446	wait_event(device->al_wait,
 447			device->act_log->pending_changes == 0 ||
 448			(locked = lc_try_lock_for_transaction(device->act_log)));
 449
 450	if (locked) {
 451		/* Double check: it may have been committed by someone else,
 452		 * while we have been waiting for the lock. */
 453		if (device->act_log->pending_changes) {
 454			bool write_al_updates;
 455
 456			rcu_read_lock();
 457			write_al_updates = rcu_dereference(device->ldev->disk_conf)->al_updates;
 458			rcu_read_unlock();
 459
 460			if (write_al_updates)
 461				al_write_transaction(device);
 462			spin_lock_irq(&device->al_lock);
 463			/* FIXME
 464			if (err)
 465				we need an "lc_cancel" here;
 466			*/
 467			lc_committed(device->act_log);
 468			spin_unlock_irq(&device->al_lock);
 469		}
 470		lc_unlock(device->act_log);
 471		wake_up(&device->al_wait);
 472	}
 473}
 474
 475/*
 476 * @delegate:   delegate activity log I/O to the worker thread
 477 */
 478void drbd_al_begin_io(struct drbd_device *device, struct drbd_interval *i)
 479{
 480	if (drbd_al_begin_io_prepare(device, i))
 481		drbd_al_begin_io_commit(device);
 482}
 483
 484int drbd_al_begin_io_nonblock(struct drbd_device *device, struct drbd_interval *i)
 485{
 486	/* for bios crossing activity log extent boundaries,
 487	 * we may need to activate two extents in one go */
 488	unsigned first = i->sector >> (AL_EXTENT_SHIFT-9);
 489	unsigned last = i->size == 0 ? first : (i->sector + (i->size >> 9) - 1) >> (AL_EXTENT_SHIFT-9);
 490	unsigned enr;
 491
 492	if (i->partially_in_al_next_enr) {
 493		D_ASSERT(device, first < i->partially_in_al_next_enr);
 494		D_ASSERT(device, last >= i->partially_in_al_next_enr);
 495		first = i->partially_in_al_next_enr;
 496	}
 497
 498	D_ASSERT(device, first <= last);
 499
 500	/* Is resync active in this area? */
 501	for (enr = first; enr <= last; enr++) {
 502		struct lc_element *tmp;
 503		tmp = lc_find(device->resync, enr/AL_EXT_PER_BM_SECT);
 504		if (unlikely(tmp != NULL)) {
 505			struct bm_extent  *bm_ext = lc_entry(tmp, struct bm_extent, lce);
 506			if (test_bit(BME_NO_WRITES, &bm_ext->flags)) {
 507				if (!test_and_set_bit(BME_PRIORITY, &bm_ext->flags))
 508					return -EBUSY;
 509				return -EWOULDBLOCK;
 510			}
 511		}
 512	}
 513
 514	/* Try to checkout the refcounts. */
 515	for (enr = first; enr <= last; enr++) {
 516		struct lc_element *al_ext;
 517		al_ext = lc_get_cumulative(device->act_log, enr);
 518
 519		if (!al_ext) {
 520			/* Did not work. We may have exhausted the possible
 521			 * changes per transaction. Or raced with someone
 522			 * "locking" it against changes.
 523			 * Remember where to continue from.
 524			 */
 525			if (enr > first)
 526				i->partially_in_al_next_enr = enr;
 527			return -ENOBUFS;
 528		}
 529	}
 530	return 0;
 531}
 532
 533void drbd_al_complete_io(struct drbd_device *device, struct drbd_interval *i)
 534{
 535	/* for bios crossing activity log extent boundaries,
 536	 * we may need to activate two extents in one go */
 537	unsigned first = i->sector >> (AL_EXTENT_SHIFT-9);
 538	unsigned last = i->size == 0 ? first : (i->sector + (i->size >> 9) - 1) >> (AL_EXTENT_SHIFT-9);
 539	unsigned enr;
 540	struct lc_element *extent;
 541	unsigned long flags;
 542
 543	D_ASSERT(device, first <= last);
 544	spin_lock_irqsave(&device->al_lock, flags);
 545
 546	for (enr = first; enr <= last; enr++) {
 547		extent = lc_find(device->act_log, enr);
 548		/* Yes, this masks a bug elsewhere.  However, during normal
 549		 * operation this is harmless, so no need to crash the kernel
 550		 * by the BUG_ON(refcount == 0) in lc_put().
 551		 */
 552		if (!extent || extent->refcnt == 0) {
 553			drbd_err(device, "al_complete_io() called on inactive extent %u\n", enr);
 554			continue;
 555		}
 556		lc_put(device->act_log, extent);
 557	}
 558	spin_unlock_irqrestore(&device->al_lock, flags);
 559	wake_up(&device->al_wait);
 560}
 561
 562static int _try_lc_del(struct drbd_device *device, struct lc_element *al_ext)
 563{
 564	int rv;
 565
 566	spin_lock_irq(&device->al_lock);
 567	rv = (al_ext->refcnt == 0);
 568	if (likely(rv))
 569		lc_del(device->act_log, al_ext);
 570	spin_unlock_irq(&device->al_lock);
 571
 572	return rv;
 573}
 574
 575/**
 576 * drbd_al_shrink() - Removes all active extents form the activity log
 577 * @device:	DRBD device.
 578 *
 579 * Removes all active extents form the activity log, waiting until
 580 * the reference count of each entry dropped to 0 first, of course.
 581 *
 582 * You need to lock device->act_log with lc_try_lock() / lc_unlock()
 583 */
 584void drbd_al_shrink(struct drbd_device *device)
 585{
 586	struct lc_element *al_ext;
 587	int i;
 588
 589	D_ASSERT(device, test_bit(__LC_LOCKED, &device->act_log->flags));
 590
 591	for (i = 0; i < device->act_log->nr_elements; i++) {
 592		al_ext = lc_element_by_index(device->act_log, i);
 593		if (al_ext->lc_number == LC_FREE)
 594			continue;
 595		wait_event(device->al_wait, _try_lc_del(device, al_ext));
 596	}
 597
 598	wake_up(&device->al_wait);
 599}
 600
 601int drbd_al_initialize(struct drbd_device *device, void *buffer)
 602{
 603	struct al_transaction_on_disk *al = buffer;
 604	struct drbd_md *md = &device->ldev->md;
 605	int al_size_4k = md->al_stripes * md->al_stripe_size_4k;
 606	int i;
 607
 608	__al_write_transaction(device, al);
 609	/* There may or may not have been a pending transaction. */
 610	spin_lock_irq(&device->al_lock);
 611	lc_committed(device->act_log);
 612	spin_unlock_irq(&device->al_lock);
 613
 614	/* The rest of the transactions will have an empty "updates" list, and
 615	 * are written out only to provide the context, and to initialize the
 616	 * on-disk ring buffer. */
 617	for (i = 1; i < al_size_4k; i++) {
 618		int err = __al_write_transaction(device, al);
 619		if (err)
 620			return err;
 621	}
 622	return 0;
 623}
 624
 625static const char *drbd_change_sync_fname[] = {
 626	[RECORD_RS_FAILED] = "drbd_rs_failed_io",
 627	[SET_IN_SYNC] = "drbd_set_in_sync",
 628	[SET_OUT_OF_SYNC] = "drbd_set_out_of_sync"
 629};
 630
 631/* ATTENTION. The AL's extents are 4MB each, while the extents in the
 632 * resync LRU-cache are 16MB each.
 633 * The caller of this function has to hold an get_ldev() reference.
 634 *
 635 * Adjusts the caching members ->rs_left (success) or ->rs_failed (!success),
 636 * potentially pulling in (and recounting the corresponding bits)
 637 * this resync extent into the resync extent lru cache.
 638 *
 639 * Returns whether all bits have been cleared for this resync extent,
 640 * precisely: (rs_left <= rs_failed)
 641 *
 642 * TODO will be obsoleted once we have a caching lru of the on disk bitmap
 643 */
 644static bool update_rs_extent(struct drbd_device *device,
 645		unsigned int enr, int count,
 646		enum update_sync_bits_mode mode)
 647{
 648	struct lc_element *e;
 649
 650	D_ASSERT(device, atomic_read(&device->local_cnt));
 651
 652	/* When setting out-of-sync bits,
 653	 * we don't need it cached (lc_find).
 654	 * But if it is present in the cache,
 655	 * we should update the cached bit count.
 656	 * Otherwise, that extent should be in the resync extent lru cache
 657	 * already -- or we want to pull it in if necessary -- (lc_get),
 658	 * then update and check rs_left and rs_failed. */
 659	if (mode == SET_OUT_OF_SYNC)
 660		e = lc_find(device->resync, enr);
 661	else
 662		e = lc_get(device->resync, enr);
 663	if (e) {
 664		struct bm_extent *ext = lc_entry(e, struct bm_extent, lce);
 665		if (ext->lce.lc_number == enr) {
 666			if (mode == SET_IN_SYNC)
 667				ext->rs_left -= count;
 668			else if (mode == SET_OUT_OF_SYNC)
 669				ext->rs_left += count;
 670			else
 671				ext->rs_failed += count;
 672			if (ext->rs_left < ext->rs_failed) {
 673				drbd_warn(device, "BAD! enr=%u rs_left=%d "
 674				    "rs_failed=%d count=%d cstate=%s\n",
 675				     ext->lce.lc_number, ext->rs_left,
 676				     ext->rs_failed, count,
 677				     drbd_conn_str(device->state.conn));
 678
 679				/* We don't expect to be able to clear more bits
 680				 * than have been set when we originally counted
 681				 * the set bits to cache that value in ext->rs_left.
 682				 * Whatever the reason (disconnect during resync,
 683				 * delayed local completion of an application write),
 684				 * try to fix it up by recounting here. */
 685				ext->rs_left = drbd_bm_e_weight(device, enr);
 686			}
 687		} else {
 688			/* Normally this element should be in the cache,
 689			 * since drbd_rs_begin_io() pulled it already in.
 690			 *
 691			 * But maybe an application write finished, and we set
 692			 * something outside the resync lru_cache in sync.
 693			 */
 694			int rs_left = drbd_bm_e_weight(device, enr);
 695			if (ext->flags != 0) {
 696				drbd_warn(device, "changing resync lce: %d[%u;%02lx]"
 697				     " -> %d[%u;00]\n",
 698				     ext->lce.lc_number, ext->rs_left,
 699				     ext->flags, enr, rs_left);
 700				ext->flags = 0;
 701			}
 702			if (ext->rs_failed) {
 703				drbd_warn(device, "Kicking resync_lru element enr=%u "
 704				     "out with rs_failed=%d\n",
 705				     ext->lce.lc_number, ext->rs_failed);
 706			}
 707			ext->rs_left = rs_left;
 708			ext->rs_failed = (mode == RECORD_RS_FAILED) ? count : 0;
 709			/* we don't keep a persistent log of the resync lru,
 710			 * we can commit any change right away. */
 711			lc_committed(device->resync);
 712		}
 713		if (mode != SET_OUT_OF_SYNC)
 714			lc_put(device->resync, &ext->lce);
 715		/* no race, we are within the al_lock! */
 716
 717		if (ext->rs_left <= ext->rs_failed) {
 718			ext->rs_failed = 0;
 719			return true;
 720		}
 721	} else if (mode != SET_OUT_OF_SYNC) {
 722		/* be quiet if lc_find() did not find it. */
 723		drbd_err(device, "lc_get() failed! locked=%d/%d flags=%lu\n",
 724		    device->resync_locked,
 725		    device->resync->nr_elements,
 726		    device->resync->flags);
 727	}
 728	return false;
 729}
 730
 731void drbd_advance_rs_marks(struct drbd_peer_device *peer_device, unsigned long still_to_go)
 732{
 733	struct drbd_device *device = peer_device->device;
 734	unsigned long now = jiffies;
 735	unsigned long last = device->rs_mark_time[device->rs_last_mark];
 736	int next = (device->rs_last_mark + 1) % DRBD_SYNC_MARKS;
 737	if (time_after_eq(now, last + DRBD_SYNC_MARK_STEP)) {
 738		if (device->rs_mark_left[device->rs_last_mark] != still_to_go &&
 739		    device->state.conn != C_PAUSED_SYNC_T &&
 740		    device->state.conn != C_PAUSED_SYNC_S) {
 741			device->rs_mark_time[next] = now;
 742			device->rs_mark_left[next] = still_to_go;
 743			device->rs_last_mark = next;
 744		}
 745	}
 746}
 747
 748/* It is called lazy update, so don't do write-out too often. */
 749static bool lazy_bitmap_update_due(struct drbd_device *device)
 750{
 751	return time_after(jiffies, device->rs_last_bcast + 2*HZ);
 752}
 753
 754static void maybe_schedule_on_disk_bitmap_update(struct drbd_device *device, bool rs_done)
 755{
 756	if (rs_done) {
 757		struct drbd_connection *connection = first_peer_device(device)->connection;
 758		if (connection->agreed_pro_version <= 95 ||
 759		    is_sync_target_state(device->state.conn))
 760			set_bit(RS_DONE, &device->flags);
 761			/* and also set RS_PROGRESS below */
 762
 763		/* Else: rather wait for explicit notification via receive_state,
 764		 * to avoid uuids-rotated-too-fast causing full resync
 765		 * in next handshake, in case the replication link breaks
 766		 * at the most unfortunate time... */
 767	} else if (!lazy_bitmap_update_due(device))
 768		return;
 769
 770	drbd_device_post_work(device, RS_PROGRESS);
 771}
 772
 773static int update_sync_bits(struct drbd_device *device,
 774		unsigned long sbnr, unsigned long ebnr,
 775		enum update_sync_bits_mode mode)
 776{
 777	/*
 778	 * We keep a count of set bits per resync-extent in the ->rs_left
 779	 * caching member, so we need to loop and work within the resync extent
 780	 * alignment. Typically this loop will execute exactly once.
 781	 */
 782	unsigned long flags;
 783	unsigned long count = 0;
 784	unsigned int cleared = 0;
 785	while (sbnr <= ebnr) {
 786		/* set temporary boundary bit number to last bit number within
 787		 * the resync extent of the current start bit number,
 788		 * but cap at provided end bit number */
 789		unsigned long tbnr = min(ebnr, sbnr | BM_BLOCKS_PER_BM_EXT_MASK);
 790		unsigned long c;
 791
 792		if (mode == RECORD_RS_FAILED)
 793			/* Only called from drbd_rs_failed_io(), bits
 794			 * supposedly still set.  Recount, maybe some
 795			 * of the bits have been successfully cleared
 796			 * by application IO meanwhile.
 797			 */
 798			c = drbd_bm_count_bits(device, sbnr, tbnr);
 799		else if (mode == SET_IN_SYNC)
 800			c = drbd_bm_clear_bits(device, sbnr, tbnr);
 801		else /* if (mode == SET_OUT_OF_SYNC) */
 802			c = drbd_bm_set_bits(device, sbnr, tbnr);
 803
 804		if (c) {
 805			spin_lock_irqsave(&device->al_lock, flags);
 806			cleared += update_rs_extent(device, BM_BIT_TO_EXT(sbnr), c, mode);
 807			spin_unlock_irqrestore(&device->al_lock, flags);
 808			count += c;
 809		}
 810		sbnr = tbnr + 1;
 811	}
 812	if (count) {
 813		if (mode == SET_IN_SYNC) {
 814			unsigned long still_to_go = drbd_bm_total_weight(device);
 815			bool rs_is_done = (still_to_go <= device->rs_failed);
 816			drbd_advance_rs_marks(first_peer_device(device), still_to_go);
 817			if (cleared || rs_is_done)
 818				maybe_schedule_on_disk_bitmap_update(device, rs_is_done);
 819		} else if (mode == RECORD_RS_FAILED)
 820			device->rs_failed += count;
 821		wake_up(&device->al_wait);
 822	}
 823	return count;
 824}
 825
 826static bool plausible_request_size(int size)
 827{
 828	return size > 0
 829		&& size <= DRBD_MAX_BATCH_BIO_SIZE
 830		&& IS_ALIGNED(size, 512);
 831}
 832
 833/* clear the bit corresponding to the piece of storage in question:
 834 * size byte of data starting from sector.  Only clear bits of the affected
 835 * one or more _aligned_ BM_BLOCK_SIZE blocks.
 836 *
 837 * called by worker on C_SYNC_TARGET and receiver on SyncSource.
 838 *
 839 */
 840int __drbd_change_sync(struct drbd_peer_device *peer_device, sector_t sector, int size,
 841		enum update_sync_bits_mode mode)
 842{
 843	/* Is called from worker and receiver context _only_ */
 844	struct drbd_device *device = peer_device->device;
 845	unsigned long sbnr, ebnr, lbnr;
 846	unsigned long count = 0;
 847	sector_t esector, nr_sectors;
 848
 849	/* This would be an empty REQ_PREFLUSH, be silent. */
 850	if ((mode == SET_OUT_OF_SYNC) && size == 0)
 851		return 0;
 852
 853	if (!plausible_request_size(size)) {
 854		drbd_err(device, "%s: sector=%llus size=%d nonsense!\n",
 855				drbd_change_sync_fname[mode],
 856				(unsigned long long)sector, size);
 857		return 0;
 858	}
 859
 860	if (!get_ldev(device))
 861		return 0; /* no disk, no metadata, no bitmap to manipulate bits in */
 862
 863	nr_sectors = get_capacity(device->vdisk);
 864	esector = sector + (size >> 9) - 1;
 865
 866	if (!expect(device, sector < nr_sectors))
 867		goto out;
 868	if (!expect(device, esector < nr_sectors))
 869		esector = nr_sectors - 1;
 870
 871	lbnr = BM_SECT_TO_BIT(nr_sectors-1);
 872
 873	if (mode == SET_IN_SYNC) {
 874		/* Round up start sector, round down end sector.  We make sure
 875		 * we only clear full, aligned, BM_BLOCK_SIZE blocks. */
 876		if (unlikely(esector < BM_SECT_PER_BIT-1))
 877			goto out;
 878		if (unlikely(esector == (nr_sectors-1)))
 879			ebnr = lbnr;
 880		else
 881			ebnr = BM_SECT_TO_BIT(esector - (BM_SECT_PER_BIT-1));
 882		sbnr = BM_SECT_TO_BIT(sector + BM_SECT_PER_BIT-1);
 883	} else {
 884		/* We set it out of sync, or record resync failure.
 885		 * Should not round anything here. */
 886		sbnr = BM_SECT_TO_BIT(sector);
 887		ebnr = BM_SECT_TO_BIT(esector);
 888	}
 889
 890	count = update_sync_bits(device, sbnr, ebnr, mode);
 891out:
 892	put_ldev(device);
 893	return count;
 894}
 895
 896static
 897struct bm_extent *_bme_get(struct drbd_device *device, unsigned int enr)
 898{
 899	struct lc_element *e;
 900	struct bm_extent *bm_ext;
 901	int wakeup = 0;
 902	unsigned long rs_flags;
 903
 904	spin_lock_irq(&device->al_lock);
 905	if (device->resync_locked > device->resync->nr_elements/2) {
 906		spin_unlock_irq(&device->al_lock);
 907		return NULL;
 908	}
 909	e = lc_get(device->resync, enr);
 910	bm_ext = e ? lc_entry(e, struct bm_extent, lce) : NULL;
 911	if (bm_ext) {
 912		if (bm_ext->lce.lc_number != enr) {
 913			bm_ext->rs_left = drbd_bm_e_weight(device, enr);
 914			bm_ext->rs_failed = 0;
 915			lc_committed(device->resync);
 916			wakeup = 1;
 917		}
 918		if (bm_ext->lce.refcnt == 1)
 919			device->resync_locked++;
 920		set_bit(BME_NO_WRITES, &bm_ext->flags);
 921	}
 922	rs_flags = device->resync->flags;
 923	spin_unlock_irq(&device->al_lock);
 924	if (wakeup)
 925		wake_up(&device->al_wait);
 926
 927	if (!bm_ext) {
 928		if (rs_flags & LC_STARVING)
 929			drbd_warn(device, "Have to wait for element"
 930			     " (resync LRU too small?)\n");
 931		BUG_ON(rs_flags & LC_LOCKED);
 932	}
 933
 934	return bm_ext;
 935}
 936
 937static int _is_in_al(struct drbd_device *device, unsigned int enr)
 938{
 939	int rv;
 940
 941	spin_lock_irq(&device->al_lock);
 942	rv = lc_is_used(device->act_log, enr);
 943	spin_unlock_irq(&device->al_lock);
 944
 945	return rv;
 946}
 947
 948/**
 949 * drbd_rs_begin_io() - Gets an extent in the resync LRU cache and sets it to BME_LOCKED
 950 * @device:	DRBD device.
 951 * @sector:	The sector number.
 952 *
 953 * This functions sleeps on al_wait.
 954 *
 955 * Returns: %0 on success, -EINTR if interrupted.
 956 */
 957int drbd_rs_begin_io(struct drbd_device *device, sector_t sector)
 958{
 959	unsigned int enr = BM_SECT_TO_EXT(sector);
 960	struct bm_extent *bm_ext;
 961	int i, sig;
 962	bool sa;
 963
 964retry:
 965	sig = wait_event_interruptible(device->al_wait,
 966			(bm_ext = _bme_get(device, enr)));
 967	if (sig)
 968		return -EINTR;
 969
 970	if (test_bit(BME_LOCKED, &bm_ext->flags))
 971		return 0;
 972
 973	/* step aside only while we are above c-min-rate; unless disabled. */
 974	sa = drbd_rs_c_min_rate_throttle(device);
 975
 976	for (i = 0; i < AL_EXT_PER_BM_SECT; i++) {
 977		sig = wait_event_interruptible(device->al_wait,
 978					       !_is_in_al(device, enr * AL_EXT_PER_BM_SECT + i) ||
 979					       (sa && test_bit(BME_PRIORITY, &bm_ext->flags)));
 980
 981		if (sig || (sa && test_bit(BME_PRIORITY, &bm_ext->flags))) {
 982			spin_lock_irq(&device->al_lock);
 983			if (lc_put(device->resync, &bm_ext->lce) == 0) {
 984				bm_ext->flags = 0; /* clears BME_NO_WRITES and eventually BME_PRIORITY */
 985				device->resync_locked--;
 986				wake_up(&device->al_wait);
 987			}
 988			spin_unlock_irq(&device->al_lock);
 989			if (sig)
 990				return -EINTR;
 991			if (schedule_timeout_interruptible(HZ/10))
 992				return -EINTR;
 993			goto retry;
 994		}
 995	}
 996	set_bit(BME_LOCKED, &bm_ext->flags);
 997	return 0;
 998}
 999
1000/**
1001 * drbd_try_rs_begin_io() - Gets an extent in the resync LRU cache, does not sleep
1002 * @peer_device: DRBD device.
1003 * @sector:	The sector number.
1004 *
1005 * Gets an extent in the resync LRU cache, sets it to BME_NO_WRITES, then
1006 * tries to set it to BME_LOCKED.
1007 *
1008 * Returns: %0 upon success, and -EAGAIN
1009 * if there is still application IO going on in this area.
1010 */
1011int drbd_try_rs_begin_io(struct drbd_peer_device *peer_device, sector_t sector)
1012{
1013	struct drbd_device *device = peer_device->device;
1014	unsigned int enr = BM_SECT_TO_EXT(sector);
1015	const unsigned int al_enr = enr*AL_EXT_PER_BM_SECT;
1016	struct lc_element *e;
1017	struct bm_extent *bm_ext;
1018	int i;
1019	bool throttle = drbd_rs_should_slow_down(peer_device, sector, true);
1020
1021	/* If we need to throttle, a half-locked (only marked BME_NO_WRITES,
1022	 * not yet BME_LOCKED) extent needs to be kicked out explicitly if we
1023	 * need to throttle. There is at most one such half-locked extent,
1024	 * which is remembered in resync_wenr. */
1025
1026	if (throttle && device->resync_wenr != enr)
1027		return -EAGAIN;
1028
1029	spin_lock_irq(&device->al_lock);
1030	if (device->resync_wenr != LC_FREE && device->resync_wenr != enr) {
1031		/* in case you have very heavy scattered io, it may
1032		 * stall the syncer undefined if we give up the ref count
1033		 * when we try again and requeue.
1034		 *
1035		 * if we don't give up the refcount, but the next time
1036		 * we are scheduled this extent has been "synced" by new
1037		 * application writes, we'd miss the lc_put on the
1038		 * extent we keep the refcount on.
1039		 * so we remembered which extent we had to try again, and
1040		 * if the next requested one is something else, we do
1041		 * the lc_put here...
1042		 * we also have to wake_up
1043		 */
1044		e = lc_find(device->resync, device->resync_wenr);
1045		bm_ext = e ? lc_entry(e, struct bm_extent, lce) : NULL;
1046		if (bm_ext) {
1047			D_ASSERT(device, !test_bit(BME_LOCKED, &bm_ext->flags));
1048			D_ASSERT(device, test_bit(BME_NO_WRITES, &bm_ext->flags));
1049			clear_bit(BME_NO_WRITES, &bm_ext->flags);
1050			device->resync_wenr = LC_FREE;
1051			if (lc_put(device->resync, &bm_ext->lce) == 0) {
1052				bm_ext->flags = 0;
1053				device->resync_locked--;
1054			}
1055			wake_up(&device->al_wait);
1056		} else {
1057			drbd_alert(device, "LOGIC BUG\n");
1058		}
1059	}
1060	/* TRY. */
1061	e = lc_try_get(device->resync, enr);
1062	bm_ext = e ? lc_entry(e, struct bm_extent, lce) : NULL;
1063	if (bm_ext) {
1064		if (test_bit(BME_LOCKED, &bm_ext->flags))
1065			goto proceed;
1066		if (!test_and_set_bit(BME_NO_WRITES, &bm_ext->flags)) {
1067			device->resync_locked++;
1068		} else {
1069			/* we did set the BME_NO_WRITES,
1070			 * but then could not set BME_LOCKED,
1071			 * so we tried again.
1072			 * drop the extra reference. */
1073			bm_ext->lce.refcnt--;
1074			D_ASSERT(device, bm_ext->lce.refcnt > 0);
1075		}
1076		goto check_al;
1077	} else {
1078		/* do we rather want to try later? */
1079		if (device->resync_locked > device->resync->nr_elements-3)
1080			goto try_again;
1081		/* Do or do not. There is no try. -- Yoda */
1082		e = lc_get(device->resync, enr);
1083		bm_ext = e ? lc_entry(e, struct bm_extent, lce) : NULL;
1084		if (!bm_ext) {
1085			const unsigned long rs_flags = device->resync->flags;
1086			if (rs_flags & LC_STARVING)
1087				drbd_warn(device, "Have to wait for element"
1088				     " (resync LRU too small?)\n");
1089			BUG_ON(rs_flags & LC_LOCKED);
1090			goto try_again;
1091		}
1092		if (bm_ext->lce.lc_number != enr) {
1093			bm_ext->rs_left = drbd_bm_e_weight(device, enr);
1094			bm_ext->rs_failed = 0;
1095			lc_committed(device->resync);
1096			wake_up(&device->al_wait);
1097			D_ASSERT(device, test_bit(BME_LOCKED, &bm_ext->flags) == 0);
1098		}
1099		set_bit(BME_NO_WRITES, &bm_ext->flags);
1100		D_ASSERT(device, bm_ext->lce.refcnt == 1);
1101		device->resync_locked++;
1102		goto check_al;
1103	}
1104check_al:
1105	for (i = 0; i < AL_EXT_PER_BM_SECT; i++) {
1106		if (lc_is_used(device->act_log, al_enr+i))
1107			goto try_again;
1108	}
1109	set_bit(BME_LOCKED, &bm_ext->flags);
1110proceed:
1111	device->resync_wenr = LC_FREE;
1112	spin_unlock_irq(&device->al_lock);
1113	return 0;
1114
1115try_again:
1116	if (bm_ext) {
1117		if (throttle) {
1118			D_ASSERT(device, !test_bit(BME_LOCKED, &bm_ext->flags));
1119			D_ASSERT(device, test_bit(BME_NO_WRITES, &bm_ext->flags));
1120			clear_bit(BME_NO_WRITES, &bm_ext->flags);
1121			device->resync_wenr = LC_FREE;
1122			if (lc_put(device->resync, &bm_ext->lce) == 0) {
1123				bm_ext->flags = 0;
1124				device->resync_locked--;
1125			}
1126			wake_up(&device->al_wait);
1127		} else
1128			device->resync_wenr = enr;
1129	}
1130	spin_unlock_irq(&device->al_lock);
1131	return -EAGAIN;
1132}
1133
1134void drbd_rs_complete_io(struct drbd_device *device, sector_t sector)
1135{
1136	unsigned int enr = BM_SECT_TO_EXT(sector);
1137	struct lc_element *e;
1138	struct bm_extent *bm_ext;
1139	unsigned long flags;
1140
1141	spin_lock_irqsave(&device->al_lock, flags);
1142	e = lc_find(device->resync, enr);
1143	bm_ext = e ? lc_entry(e, struct bm_extent, lce) : NULL;
1144	if (!bm_ext) {
1145		spin_unlock_irqrestore(&device->al_lock, flags);
1146		if (drbd_ratelimit())
1147			drbd_err(device, "drbd_rs_complete_io() called, but extent not found\n");
1148		return;
1149	}
1150
1151	if (bm_ext->lce.refcnt == 0) {
1152		spin_unlock_irqrestore(&device->al_lock, flags);
1153		drbd_err(device, "drbd_rs_complete_io(,%llu [=%u]) called, "
1154		    "but refcnt is 0!?\n",
1155		    (unsigned long long)sector, enr);
1156		return;
1157	}
1158
1159	if (lc_put(device->resync, &bm_ext->lce) == 0) {
1160		bm_ext->flags = 0; /* clear BME_LOCKED, BME_NO_WRITES and BME_PRIORITY */
1161		device->resync_locked--;
1162		wake_up(&device->al_wait);
1163	}
1164
1165	spin_unlock_irqrestore(&device->al_lock, flags);
1166}
1167
1168/**
1169 * drbd_rs_cancel_all() - Removes all extents from the resync LRU (even BME_LOCKED)
1170 * @device:	DRBD device.
1171 */
1172void drbd_rs_cancel_all(struct drbd_device *device)
1173{
1174	spin_lock_irq(&device->al_lock);
1175
1176	if (get_ldev_if_state(device, D_FAILED)) { /* Makes sure ->resync is there. */
1177		lc_reset(device->resync);
1178		put_ldev(device);
1179	}
1180	device->resync_locked = 0;
1181	device->resync_wenr = LC_FREE;
1182	spin_unlock_irq(&device->al_lock);
1183	wake_up(&device->al_wait);
1184}
1185
1186/**
1187 * drbd_rs_del_all() - Gracefully remove all extents from the resync LRU
1188 * @device:	DRBD device.
1189 *
1190 * Returns: %0 upon success, -EAGAIN if at least one reference count was
1191 * not zero.
1192 */
1193int drbd_rs_del_all(struct drbd_device *device)
1194{
1195	struct lc_element *e;
1196	struct bm_extent *bm_ext;
1197	int i;
1198
1199	spin_lock_irq(&device->al_lock);
1200
1201	if (get_ldev_if_state(device, D_FAILED)) {
1202		/* ok, ->resync is there. */
1203		for (i = 0; i < device->resync->nr_elements; i++) {
1204			e = lc_element_by_index(device->resync, i);
1205			bm_ext = lc_entry(e, struct bm_extent, lce);
1206			if (bm_ext->lce.lc_number == LC_FREE)
1207				continue;
1208			if (bm_ext->lce.lc_number == device->resync_wenr) {
1209				drbd_info(device, "dropping %u in drbd_rs_del_all, apparently"
1210				     " got 'synced' by application io\n",
1211				     device->resync_wenr);
1212				D_ASSERT(device, !test_bit(BME_LOCKED, &bm_ext->flags));
1213				D_ASSERT(device, test_bit(BME_NO_WRITES, &bm_ext->flags));
1214				clear_bit(BME_NO_WRITES, &bm_ext->flags);
1215				device->resync_wenr = LC_FREE;
1216				lc_put(device->resync, &bm_ext->lce);
1217			}
1218			if (bm_ext->lce.refcnt != 0) {
1219				drbd_info(device, "Retrying drbd_rs_del_all() later. "
1220				     "refcnt=%d\n", bm_ext->lce.refcnt);
1221				put_ldev(device);
1222				spin_unlock_irq(&device->al_lock);
1223				return -EAGAIN;
1224			}
1225			D_ASSERT(device, !test_bit(BME_LOCKED, &bm_ext->flags));
1226			D_ASSERT(device, !test_bit(BME_NO_WRITES, &bm_ext->flags));
1227			lc_del(device->resync, &bm_ext->lce);
1228		}
1229		D_ASSERT(device, device->resync->used == 0);
1230		put_ldev(device);
1231	}
1232	spin_unlock_irq(&device->al_lock);
1233	wake_up(&device->al_wait);
1234
1235	return 0;
1236}